LeadNurture Pro — Testing Suite Temporal

# tests/conftest.py
# LeadNurture Pro — Fixtures compartidos para Temporal testing

import pytest
from temporalio.testing import WorkflowEnvironment, ActivityEnvironment
from unittest.mock import AsyncMock, MagicMock
from datetime import datetime
from app.models import Lead, LeadStatus, EmailResult


@pytest.fixture(scope="module")
async def workflow_env():
    """Entorno de time-skipping reutilizable por módulo.

    Permite que workflow.sleep(days=3) se ejecute al instante.
    Se comparte entre tests del mismo módulo para mayor velocidad.
    """
    env = await WorkflowEnvironment.start_time_skipping()
    yield env
    await env.shutdown()


@pytest.fixture
def activity_env():
    """Entorno de actividades, fresco por cada test."""
    return ActivityEnvironment()


@pytest.fixture
def lead_valido() -> Lead:
    """Lead de ejemplo con datos completos y válidos."""
    return Lead(
        id="lead-001",
        email="carlos.perez@pyme.es",
        nombre="Carlos Pérez",
        empresa="Pyme Solutions SL",
        telefono="+34 600 123 456",
        empleados=25,
        sector="marketing",
        creado_en=datetime.utcnow(),
    )


@pytest.fixture
def lead_invalido() -> Lead:
    """Lead con email malformado para testar validación."""
    return Lead(
        id="lead-002",
        email="no-es-un-email",
        nombre="Sin Email",
        empresa="X",
    )


@pytest.fixture
def mock_clearbit() -> AsyncMock:
    """Mock del cliente Clearbit para enriquecimiento de datos."""
    mock = AsyncMock()
    mock.enrich.return_value = {"icp_score": 0.82, "employees": 25}
    return mock


@pytest.fixture
def mock_mailgun() -> AsyncMock:
    """Mock del cliente Mailgun para envío de emails."""
    mock = AsyncMock()
    mock.send.return_value = EmailResult(
        message_id="mg-abc123", status="queued"
    )
    return mock

# tests/unit/test_onboarding_unit.py

import pytest
from datetime import timedelta
from temporalio import workflow, activity
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
from app.workflows import OnboardingLeadWorkflow
from app.activities import (
    validar_lead, segmentar_lead,
    enviar_email_bienvenida, notificar_crm
)


# ─── TEST 1: Happy path completo ─────────────────────────────────────────────

@pytest.mark.asyncio
async def test_onboarding_completo_lead_icp(workflow_env, lead_valido):
    """Lead ICP válido: todo el pipeline se ejecuta y crea registro en CRM."""

    async with Worker(
        workflow_env.client,
        task_queue="test-onboarding",
        workflows=[OnboardingLeadWorkflow],
        activities=[validar_lead, segmentar_lead,
                    enviar_email_bienvenida, notificar_crm],
    ):
        resultado = await workflow_env.client.execute_workflow(
            OnboardingLeadWorkflow.run,
            lead_valido,
            id=f"onb-{lead_valido.id}",
            task_queue="test-onboarding",
        )

    assert resultado.status == "completado"
    assert resultado.crm_id is not None
    assert resultado.emails_enviados >= 1


# ─── TEST 2: Time-skipping — espera de 3 días ────────────────────────────────

@pytest.mark.asyncio
async def test_espera_interaccion_se_resuelve_al_instante(workflow_env, lead_valido):
    """workflow.sleep(days=3) debe completar en millisegundos con time-skipping."""
    import time

    inicio = time.monotonic()

    async with Worker(
        workflow_env.client,
        task_queue="test-timeskip",
        workflows=[OnboardingLeadWorkflow],
        activities=[validar_lead, segmentar_lead,
                    enviar_email_bienvenida, notificar_crm],
    ):
        await workflow_env.client.execute_workflow(
            OnboardingLeadWorkflow.run,
            lead_valido,
            id="onb-timeskip-001",
            task_queue="test-timeskip",
        )

    duracion = time.monotonic() - inicio
    assert duracion < 2.0, f"Demasiado lento: {duracion:.2f}s (esperado < 2s)"


# ─── TEST 3: Lead no-ICP → no se notifica CRM ────────────────────────────────

@pytest.mark.asyncio
async def test_lead_no_icp_sin_notificacion_crm(workflow_env):
    """Leads con ICP score bajo no deben enviarse al CRM."""
    from app.models import Lead

    lead_no_icp = Lead(
        id="lead-no-icp",
        email="info@freelancer.es",
        nombre="Freelancer Suelto",
        empresa="Autónomo",
        empleados=1,
        sector="diseño",
    )

    async with Worker(
        workflow_env.client,
        task_queue="test-no-icp",
        workflows=[OnboardingLeadWorkflow],
        activities=[validar_lead, segmentar_lead,
                    enviar_email_bienvenida, notificar_crm],
    ):
        resultado = await workflow_env.client.execute_workflow(
            OnboardingLeadWorkflow.run,
            lead_no_icp,
            id="onb-no-icp",
            task_queue="test-no-icp",
        )

    assert resultado.status == "nurturing"
    assert resultado.crm_id is None  # No debe llegar a CRM


# ─── TEST 4: Control manual de tiempo (avance intermedio) ────────────────────

@pytest.mark.asyncio
async def test_estado_intermedio_tras_1_dia(workflow_env, lead_valido):
    """Verifica estado 'esperando_interaccion' tras avanzar 1 día manualmente."""

    async with Worker(
        workflow_env.client,
        task_queue="test-manual-time",
        workflows=[OnboardingLeadWorkflow],
        activities=[validar_lead, segmentar_lead,
                    enviar_email_bienvenida, notificar_crm],
    ):
        handle = await workflow_env.client.start_workflow(
            OnboardingLeadWorkflow.run,
            lead_valido,
            id="onb-manual-time",
            task_queue="test-manual-time",
        )

        # Avanzar sólo 1 día (quedan 2 días de espera)
        await workflow_env.sleep(timedelta(days=1))

        estado = await handle.query(OnboardingLeadWorkflow.get_status)
        assert estado == "esperando_interaccion"

        # Avanzar los 2 días restantes
        await workflow_env.sleep(timedelta(days=2))
        resultado = await handle.result()
        assert resultado.status != "esperando_interaccion"

# tests/integration/test_onboarding_integration.py

import pytest
from datetime import timedelta
from unittest.mock import AsyncMock, patch, call
from temporalio import workflow, activity
from temporalio.common import RetryPolicy
from temporalio.exceptions import ApplicationError
from temporalio.worker import Worker
from app.workflows import OnboardingLeadWorkflow


# ─── TEST 5: ValidarLead — email inválido ────────────────────────────────────

@pytest.mark.asyncio
async def test_validar_lead_email_invalido(activity_env, lead_invalido):
    """Actividad de validación debe lanzar ApplicationError con email malformado."""
    from app.activities import validar_lead

    with pytest.raises(ApplicationError) as exc:
        await activity_env.run(validar_lead, lead_invalido)

    assert "email" in str(exc.value).lower()
    assert exc.value.non_retryable == True  # No reintentar errores de validación


# ─── TEST 6: Mock de Clearbit con error transitorio ──────────────────────────

@pytest.mark.asyncio
async def test_clearbit_reintento_tras_error_transitorio(workflow_env, lead_valido):
    """Clearbit falla 2 veces, éxito al 3er intento — workflow no se interrumpe."""

    llamadas = []

    @activity.defn(name="validar_lead")
    async def validar_lead_mock(lead):
        llamadas.append(len(llamadas) + 1)
        if len(llamadas) < 3:
            raise Exception(f"Clearbit timeout #{len(llamadas)}")
        return lead  # Éxito al 3er intento

    async with Worker(
        workflow_env.client,
        task_queue="test-retry",
        workflows=[OnboardingLeadWorkflow],
        activities=[validar_lead_mock, /* otras actividades... */],
    ):
        resultado = await workflow_env.client.execute_workflow(
            OnboardingLeadWorkflow.run,
            lead_valido,
            id="onb-retry-test",
            task_queue="test-retry",
        )

    assert resultado.status == "completado"
    assert len(llamadas) == 3  # Confirmar 3 intentos


# ─── TEST 7: Señal de interacción del lead ───────────────────────────────────

@pytest.mark.asyncio
async def test_senal_click_email_adelanta_workflow(workflow_env, lead_valido):
    """Al recibir señal 'lead_clicked', el workflow salta la espera de 3 días."""

    async with Worker(
        workflow_env.client,
        task_queue="test-signal",
        workflows=[OnboardingLeadWorkflow],
        activities=[validar_lead, segmentar_lead,
                    enviar_email_bienvenida, notificar_crm],
    ):
        handle = await workflow_env.client.start_workflow(
            OnboardingLeadWorkflow.run,
            lead_valido,
            id="onb-click-signal",
            task_queue="test-signal",
        )

        # Esperar a que empiece la pausa de 3 días
        await workflow_env.sleep(timedelta(hours=1))

        # Simular click del lead en el email
        await handle.signal(
            OnboardingLeadWorkflow.lead_interaccion,
            {"tipo": "click", "url": "https://leadnurturepro.es/demo"}
        )

        resultado = await handle.result()

    assert resultado.status == "convertido"
    assert resultado.interaccion.tipo == "click"


# ─── TEST 8: Envío de follow-up al no haber interacción ─────────────────────

@pytest.mark.asyncio
async def test_followup_si_no_hay_interaccion_en_3_dias(workflow_env, lead_valido):
    """Sin señal de interacción tras 3 días, se envía email follow-up."""

    emails_enviados = []

    @activity.defn(name="enviar_email_bienvenida")
    async def mock_email(lead, tipo):
        emails_enviados.append(tipo)
        return {"message_id": f"mg-{tipo}", "status": "queued"}

    async with Worker(...):
        resultado = await workflow_env.client.execute_workflow(
            OnboardingLeadWorkflow.run,
            lead_valido,
            id="onb-followup",
            task_queue="test-followup",
        )

    # Debe haber enviado: bienvenida + follow-up
    assert "bienvenida" in emails_enviados
    assert "followup_dia3" in emails_enviados
    assert len(emails_enviados) == 2


# ─── TEST 9: Parametrizado — segmentación de leads ───────────────────────────

@pytest.mark.parametrize("empleados,sector,icp_esperado", [
    (50, "marketing", "ICP"),
    (200, "retail", "ICP"),
    (1, "diseño", "No-ICP"),
    (5000, "banca", "No-ICP"),  # Enterprise fuera de objetivo
])
@pytest.mark.asyncio
async def test_segmentacion_parametrizada(
    activity_env, empleados, sector, icp_esperado
):
    """Verifica la lógica de segmentación según tamaño y sector."""
    from app.activities import segmentar_lead
    from app.models import Lead

    lead = Lead(
        id="test",
        email="test@empresa.es",
        nombre="Test",
        empresa="Empresa Test",
        empleados=empleados,
        sector=sector,
    )

    resultado = await activity_env.run(segmentar_lead, lead)
    assert resultado.segmento == icp_esperado

# .github/workflows/tests.yml
name: Tests LeadNurture Pro

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        python-version: ["3.11", "3.12"]

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python ${{ matrix.python-version }}
        uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}
          cache: 'pip'

      - name: Install dependencies
        run: |
          pip install -e ".[test]"

      - name: Type check (mypy)
        run: mypy app/

      - name: Unit tests (sin Temporal server)
        run: |
          pytest tests/unit/ -m unit -v --tb=short

      - name: Integration tests (WorkflowEnvironment time-skipping)
        run: |
          pytest tests/integration/ -m integration -v --tb=short

      - name: Coverage report
        run: |
          pytest --cov=app --cov-report=xml --cov-fail-under=80

      - name: Upload coverage to Codecov
        uses: codecov/codecov-action@v4
        with:
          file: ./coverage.xml
          fail_ci_if_error: true