163 lines
5.1 KiB
Python
163 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import os
|
|
from contextlib import asynccontextmanager
|
|
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_imports
|
|
from backend.auth.dependencies import UserPrincipal, get_current_user
|
|
|
|
|
|
class FakeConn:
|
|
async def execute(self, query: str, *args):
|
|
return "OK"
|
|
|
|
|
|
class FakePool:
|
|
def __init__(self) -> None:
|
|
self.conn = FakeConn()
|
|
|
|
@asynccontextmanager
|
|
async def acquire(self):
|
|
yield self.conn
|
|
|
|
|
|
def _build_app(*, authenticated: bool) -> TestClient:
|
|
app = FastAPI()
|
|
app.state.db_pool = FakePool()
|
|
app.include_router(routes_crm_imports.router, prefix="/api")
|
|
if authenticated:
|
|
app.dependency_overrides[get_current_user] = lambda: UserPrincipal(
|
|
"00000000-0000-0000-0000-000000000001",
|
|
"ADMIN",
|
|
"tenant_alpha",
|
|
)
|
|
return TestClient(app)
|
|
|
|
|
|
def test_canonical_crm_routes_require_authentication() -> None:
|
|
client = _build_app(authenticated=False)
|
|
|
|
response = client.get("/api/crm/contacts")
|
|
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == "Missing or malformed Authorization header."
|
|
|
|
|
|
def test_canonical_crm_task_routes_require_authentication() -> None:
|
|
client = _build_app(authenticated=False)
|
|
|
|
response = client.get("/api/crm/tasks")
|
|
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == "Missing or malformed Authorization header."
|
|
|
|
|
|
def test_canonical_crm_task_write_routes_require_authentication() -> None:
|
|
client = _build_app(authenticated=False)
|
|
|
|
response = client.patch(
|
|
"/api/crm/tasks/33333333-3333-3333-3333-333333333333",
|
|
json={"status": "done"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == "Missing or malformed Authorization header."
|
|
|
|
|
|
def test_canonical_crm_lead_stage_write_routes_require_authentication() -> None:
|
|
client = _build_app(authenticated=False)
|
|
|
|
response = client.patch(
|
|
"/api/crm/leads/22222222-2222-2222-2222-222222222222/stage",
|
|
json={"status": "qualified"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == "Missing or malformed Authorization header."
|
|
|
|
|
|
def test_canonical_crm_opportunity_write_routes_require_authentication() -> None:
|
|
client = _build_app(authenticated=False)
|
|
|
|
response = client.patch(
|
|
"/api/crm/opportunities/55555555-5555-5555-5555-555555555555",
|
|
json={"stage": "negotiation"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == "Missing or malformed Authorization header."
|
|
|
|
|
|
def test_canonical_crm_import_upload_requires_authentication() -> None:
|
|
client = _build_app(authenticated=False)
|
|
|
|
response = client.post(
|
|
"/api/crm/imports",
|
|
params={"source_system": "csv_upload"},
|
|
files={"file": ("contacts.csv", io.BytesIO(b"name,phone\nAmina,+9715000\n"), "text/csv")},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == "Missing or malformed Authorization header."
|
|
|
|
|
|
def test_canonical_crm_contacts_can_be_read_when_authenticated(monkeypatch) -> None:
|
|
client = _build_app(authenticated=True)
|
|
|
|
async def fake_get_contact_list(
|
|
conn: Any,
|
|
tenant_id: str,
|
|
search: str | None = None,
|
|
buyer_type: str | None = None,
|
|
status: str | None = None,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> dict[str, Any]:
|
|
assert tenant_id == "tenant_alpha"
|
|
assert search is None
|
|
assert buyer_type is None
|
|
assert status is None
|
|
return {
|
|
"contacts": [
|
|
{
|
|
"person_id": "11111111-1111-1111-1111-111111111111",
|
|
"full_name": "Amina Rahman",
|
|
"primary_email": "amina@example.com",
|
|
"primary_phone": "+971500000001",
|
|
"buyer_type": "high_intent",
|
|
"lead_id": "22222222-2222-2222-2222-222222222222",
|
|
"legacy_li_id": None,
|
|
"lead_status": "qualified",
|
|
"budget_band": "AED 12M",
|
|
"urgency": "high",
|
|
"primary_interest": "Marina Penthouse",
|
|
"intent_score": 0.94,
|
|
"engagement_score": 0.91,
|
|
"urgency_score": 0.88,
|
|
"interaction_count": 6,
|
|
"last_interaction_at": "2026-04-22T10:00:00+00:00",
|
|
"pending_tasks": 1,
|
|
"created_at": "2026-04-21T10:00:00+00:00",
|
|
}
|
|
],
|
|
"total": 1,
|
|
"limit": limit,
|
|
"offset": offset,
|
|
}
|
|
|
|
monkeypatch.setattr(routes_crm_imports, "get_contact_list", fake_get_contact_list)
|
|
|
|
response = client.get("/api/crm/contacts")
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()["data"]
|
|
assert payload["total"] == 1
|
|
assert payload["contacts"][0]["full_name"] == "Amina Rahman"
|