Arquitectura Backend — LeadFlow SaaS

from abc import ABC, abstractmethod
from dataclasses import dataclass
from ..entities.lead import Lead


@dataclass
class CRMContact:
    """DTO de salida — no depende de ningún SDK externo."""
    external_id: str
    synced_at: str
    crm_url: str


class ICRMPort(ABC):
    """Puerto de salida para sincronización con CRM.

    El dominio define el contrato; el adaptador lo implementa.
    Si mañana cambiamos de HubSpot a Salesforce, solo cambia el adaptador.
    """

    @abstractmethod
    async def push_lead(self, lead: Lead) -> CRMContact:
        """Envía un lead al CRM. Lanza CRMConnectionError si falla."""
        ...

    @abstractmethod
    async def update_lead_score(self, external_id: str, score: int) -> bool:
        """Actualiza el score de un contacto ya existente en el CRM."""
        ...

from hubspot import HubSpot  # SDK solo en esta capa
from ...domain.interfaces.i_crm_port import ICRMPort, CRMContact
from ...domain.entities.lead import Lead


class HubSpotCRMAdapter(ICRMPort):
    """Implementación concreta del puerto ICRMPort usando HubSpot SDK."""

    def __init__(self, api_key: str) -> None:
        self._client = HubSpot(access_token=api_key)

    async def push_lead(self, lead: Lead) -> CRMContact:
        props = {
            "email": str(lead.email),
            "firstname": lead.first_name,
            "lastname": lead.last_name,
            "leadscore_cultiva": str(lead.score.value),
            "source_channel": lead.source.value,
        }
        response = await self._client.crm.contacts.basic_api.create(
            simple_public_object_input_for_create={"properties": props}
        )
        return CRMContact(
            external_id=response.id,
            synced_at=response.created_at.isoformat(),
            crm_url=f"https://app.hubspot.com/contacts/{response.id}",
        )

    async def update_lead_score(self, external_id: str, score: int) -> bool:
        await self._client.crm.contacts.basic_api.update(
            contact_id=external_id,
            simple_public_object_input={"properties": {"leadscore_cultiva": str(score)}}
        )
        return True

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from uuid import uuid4
# ⚠ Solo imports de stdlib y otros módulos del dominio
from .value_objects.email import Email
from .value_objects.lead_score import LeadScore


class LeadStatus(Enum):
    NEW = "new"
    ENRICHED = "enriched"
    SCORED = "scored"
    DISTRIBUTED = "distributed"


class LeadSource(Enum):
    WEB_FORM = "web_form"
    LINKEDIN = "linkedin"
    META_ADS = "meta_ads"


@dataclass
class Lead:
    """Aggregate Root del contexto Lead Management."""
    id: str
    email: Email
    first_name: str
    last_name: str
    source: LeadSource
    status: LeadStatus
    score: LeadScore | None
    agency_id: str
    captured_at: datetime
    _events: list = field(default_factory=list, repr=False)

    @classmethod
    def capture(
        cls,
        email: str,
        first_name: str,
        last_name: str,
        source: LeadSource,
        agency_id: str,
    ) -> Lead:
        """Factory method — único punto de creación de un lead."""
        lead = cls(
            id=str(uuid4()),
            email=Email(email),  # valida aquí, no en el controller
            first_name=first_name,
            last_name=last_name,
            source=source,
            status=LeadStatus.NEW,
            score=None,
            agency_id=agency_id,
            captured_at=datetime.now(timezone.utc),
        )
        lead._events.append(LeadCaptured(lead_id=lead.id, source=source))
        return lead

    def apply_score(self, score: int, model_version: str) -> None:
        """Aplica el scoring de IA. Cambia el estado del aggregate."""
        self.score = LeadScore(score)
        self.status = LeadStatus.SCORED
        self._events.append(LeadScored(
            lead_id=self.id, score=score, model_version=model_version
        ))

    def pull_events(self) -> list:
        events, self._events = self._events, []
        return events

import pytest
from lead_management.adapters.repositories.in_memory_lead_repository import InMemoryLeadRepository
from lead_management.use_cases.capture_lead import CaptureLeadUseCase, CaptureLeadRequest
from lead_management.domain.entities.lead import LeadSource


@pytest.fixture
def use_case():
    return CaptureLeadUseCase(lead_repository=InMemoryLeadRepository())


@pytest.mark.asyncio
async def test_capture_lead_from_web_form(use_case):
    request = CaptureLeadRequest(
        email="carlos@agencianexo.com",
        first_name="Carlos",
        last_name="Morales",
        source=LeadSource.WEB_FORM,
        agency_id="agency-nexo-001",
    )
    result = await use_case.execute(request)

    assert result.success is True
    assert result.lead.email == "carlos@agencianexo.com"
    assert result.lead.status.value == "new"
    assert result.lead.id is not None
    assert len(result.lead.pull_events()) == 1  # LeadCaptured emitted


@pytest.mark.asyncio
async def test_duplicate_email_same_agency_rejected(use_case):
    req = CaptureLeadRequest(
        email="carlos@agencianexo.com", first_name="Carlos",
        last_name="M", source=LeadSource.WEB_FORM, agency_id="agency-nexo-001"
    )
    await use_case.execute(req)  # primer capture
    result = await use_case.execute(req)  # segundo — debe rechazarse

    assert result.success is False
    assert "already exists" in result.error


@pytest.mark.asyncio
async def test_invalid_email_rejected_at_boundary(use_case):
    """El VO Email debe lanzar ValueError antes de llegar al repositorio."""
    req = CaptureLeadRequest(
        email="not-an-email", first_name="X", last_name="Y",
        source=LeadSource.META_ADS, agency_id="agency-nexo-001"
    )
    result = await use_case.execute(req)

    assert result.success is False
    assert "Invalid email" in result.error