from __future__ import annotations import os from contextlib import asynccontextmanager from datetime import datetime, timezone from typing import Any from fastapi import FastAPI from fastapi.testclient import TestClient os.environ.setdefault("VELOCITY_JWT_SECRET", "test-secret") from backend.api import routes_crm from backend.auth.dependencies import UserPrincipal, get_current_user def _now() -> datetime: return datetime.now(timezone.utc) class FakeConn: def __init__(self) -> None: self.leads: dict[str, dict[str, Any]] = { "legacy-1": { "id": "legacy-1", "tenant_id": "tenant_alpha", "name": "Legacy Duplicate", "email": "legacy.duplicate@example.com", "phone": "+971500000001", "source": "website", "notes": "Old legacy note", "qualification": "HOT", "score": 82, "kanban_status": "Qualifying", "budget": "AED 5M", "unit_interest": "Legacy Tower", "metadata": {}, "created_at": _now(), "updated_at": _now(), }, "legacy-2": { "id": "legacy-2", "tenant_id": "tenant_alpha", "name": "Legacy Only", "email": "legacy.only@example.com", "phone": "+971500000002", "source": "walkin", "notes": "Pure legacy lead", "qualification": "POTENTIAL", "score": 74, "kanban_status": "Negotiation", "budget": "AED 7M", "unit_interest": "Legacy Residence", "metadata": {}, "created_at": _now(), "updated_at": _now(), }, } self.chat_logs: dict[str, dict[str, Any]] = {} async def execute(self, query: str, *args): normalized = " ".join(query.strip().split()) if "CREATE TABLE IF NOT EXISTS leads" in normalized or "CREATE TABLE IF NOT EXISTS chat_logs" in normalized: return "CREATE" if normalized.startswith("ALTER TABLE leads ADD COLUMN IF NOT EXISTS tenant_id"): return "ALTER TABLE" if normalized.startswith("ALTER TABLE chat_logs ADD COLUMN IF NOT EXISTS tenant_id"): return "ALTER TABLE" if normalized.startswith("UPDATE leads SET tenant_id = $1 WHERE tenant_id IS NULL OR tenant_id = ''"): return "UPDATE" if normalized.startswith("UPDATE chat_logs SET tenant_id = $1 WHERE tenant_id IS NULL OR tenant_id = ''"): return "UPDATE" if normalized.startswith("ALTER TABLE leads ALTER COLUMN tenant_id SET DEFAULT"): return "ALTER TABLE" if normalized.startswith("ALTER TABLE chat_logs ALTER COLUMN tenant_id SET DEFAULT"): return "ALTER TABLE" if "CREATE INDEX IF NOT EXISTS" in normalized: return "CREATE INDEX" raise AssertionError(f"Unexpected execute query: {query}") async def fetch(self, query: str, *args): normalized = " ".join(query.strip().split()) if "FROM leads" in normalized: rows = [row for row in self.leads.values() if row["tenant_id"] == args[0]] return rows if "FROM chat_logs" in normalized: rows = [row for row in self.chat_logs.values() if row["tenant_id"] == args[0]] if len(args) >= 2: rows = [row for row in rows if row["lead_id"] == args[1]] return rows raise AssertionError(f"Unexpected fetch query: {query}") async def fetchrow(self, query: str, *args): normalized = " ".join(query.strip().split()) if "FROM leads" in normalized and "WHERE id = $1 AND tenant_id = $2" in normalized: row = self.leads.get(args[0]) return row if row and row["tenant_id"] == args[1] else None raise AssertionError(f"Unexpected fetchrow query: {query}") class FakePool: def __init__(self) -> None: self.conn = FakeConn() @asynccontextmanager async def acquire(self): yield self.conn def _build_client() -> TestClient: app = FastAPI() app.state.db_pool = FakePool() app.include_router(routes_crm.crm_router, prefix="/api") app.dependency_overrides[get_current_user] = lambda: UserPrincipal( "user-1", "ADMIN", "tenant_alpha", ) return TestClient(app) def test_list_leads_merges_canonical_and_legacy_without_duplicate_shadow(monkeypatch) -> None: client = _build_client() async def fake_fetch_canonical_leads(conn: Any, tenant_id: str, search: str | None = None) -> list[dict[str, Any]]: assert tenant_id == "tenant_alpha" return [ { "id": "legacy-1", "name": "Canonical Preferred", "email": "canonical@example.com", "phone": "+971500009999", "source": "website", "notes": "Canonical note", "qualification": "WHALE", "score": 96, "kanban_status": "Negotiation", "stage": "negotiation", "budget": "AED 18M", "unit_interest": "Sky Deck Penthouse", "metadata": { "legacy_lead_id": "legacy-1", "canonical_lead_id": "canon-1", "canonical_person_id": "person-1", }, "created_at": "2026-04-22T10:00:00+00:00", "updated_at": "2026-04-22T11:00:00+00:00", } ] monkeypatch.setattr(routes_crm, "_fetch_canonical_leads", fake_fetch_canonical_leads) response = client.get("/api/leads") assert response.status_code == 200 payload = response.json()["data"] assert len(payload) == 2 assert payload[0]["name"] == "Canonical Preferred" assert payload[1]["id"] == "legacy-2" def test_get_lead_resolves_canonical_record_by_canonical_id(monkeypatch) -> None: client = _build_client() async def fake_fetch_canonical_leads(conn: Any, tenant_id: str, search: str | None = None) -> list[dict[str, Any]]: return [ { "id": "legacy-1", "name": "Canonical Preferred", "email": "canonical@example.com", "phone": "+971500009999", "source": "website", "notes": "Canonical note", "qualification": "WHALE", "score": 96, "kanban_status": "Negotiation", "stage": "negotiation", "budget": "AED 18M", "unit_interest": "Sky Deck Penthouse", "metadata": { "legacy_lead_id": "legacy-1", "canonical_lead_id": "canon-1", "canonical_person_id": "person-1", }, "created_at": "2026-04-22T10:00:00+00:00", "updated_at": "2026-04-22T11:00:00+00:00", } ] monkeypatch.setattr(routes_crm, "_fetch_canonical_leads", fake_fetch_canonical_leads) response = client.get("/api/leads/canon-1") assert response.status_code == 200 assert response.json()["data"]["name"] == "Canonical Preferred" def test_chat_logs_fall_back_to_canonical_interactions(monkeypatch) -> None: client = _build_client() async def fake_fetch_canonical_chat_logs(conn: Any, tenant_id: str, lead_id: str, channel: str | None = None) -> list[dict[str, Any]]: assert tenant_id == "tenant_alpha" assert lead_id == "canon-1" return [ { "id": "interaction-1", "lead_id": "canon-1", "sender": "oracle", "channel": "whatsapp", "content": "Canonical interaction summary", "metadata": {"source_of_truth": "canonical_crm"}, "created_at": "2026-04-22T12:00:00+00:00", } ] monkeypatch.setattr(routes_crm, "_fetch_canonical_chat_logs", fake_fetch_canonical_chat_logs) response = client.get("/api/chat-logs", params={"lead_id": "canon-1"}) assert response.status_code == 200 payload = response.json()["data"] assert len(payload) == 1 assert payload[0]["content"] == "Canonical interaction summary" def test_list_leads_falls_back_to_legacy_when_canonical_bridge_unavailable(monkeypatch) -> None: client = _build_client() async def failing_fetch_canonical_leads(conn: Any, tenant_id: str, search: str | None = None) -> list[dict[str, Any]]: raise RuntimeError("canonical tables unavailable") monkeypatch.setattr(routes_crm, "_fetch_canonical_leads", failing_fetch_canonical_leads) response = client.get("/api/leads") assert response.status_code == 200 payload = response.json()["data"] assert [lead["id"] for lead in payload] == ["legacy-1", "legacy-2"]