from fastapi.testclient import TestClient import services.sales_service.app as sales_module from services.shared.core import new_id, utc_now_iso from services.shared.db import get_session from services.shared.sales_sql_models import TenantIntegrationRow def _headers(tenant_id: str | None) -> dict[str, str]: headers = {"X-User": "admin", "X-Role": "admin"} if tenant_id: headers["X-Tenant-ID"] = tenant_id return headers def _lead_payload(seed: str) -> dict: return { "source_type": "crm", "source_channel": "telegram", "full_name": f"Buyer {seed}", "company_name": "Tenant QA", "phone": f"+7700{seed[-7:]}", "email": f"{seed}@test.local", "lead_temperature": "hot", "lead_score": 80, "initial_need_summary": "Need an offer.", "preferred_channel": "telegram", "assigned_agent_type": "text_ai", "status": "new_qualified_lead", "priority": 3, "title": f"Lead {seed}", } def _deal_payload(seed: str) -> dict: return { "stage_id": "new_qualified_lead", "scenario_type": "quick_sale", "priority": 3, "title": f"Standalone deal {seed}", "need_summary": "Need sales follow-up.", "preferred_channel": "telegram", "current_channel": "telegram", } def _create_lead_and_deal(client: TestClient, tenant_id: str) -> tuple[dict, str]: seed = new_id("ten").replace("_", "") created = client.post("/api/v1/leads", json=_lead_payload(seed), headers=_headers(tenant_id)) assert created.status_code == 200 lead = created.json() deals = client.get("/api/v1/deals", headers=_headers(tenant_id)) assert deals.status_code == 200 deal_id = next(item["deal_id"] for item in deals.json() if item["lead_id"] == lead["lead_id"]) return lead, deal_id def _prepare_invoice_deal(client: TestClient, deal_id: str, tenant_id: str) -> None: updated = client.patch( f"/api/v1/deals/{deal_id}", json={"document_required": False}, headers=_headers(tenant_id), ) assert updated.status_code == 200 for stage_code in ["active_text_communication", "need_confirmed"]: response = client.post( f"/api/v1/deals/{deal_id}/change-stage", json={"target_stage_code": stage_code, "reason": f"test.{stage_code}"}, headers=_headers(tenant_id), ) assert response.status_code == 200 def _create_integration(*, tenant_id: str, provider_type: str, provider_name: str, provider_account_id: str) -> None: session = get_session() try: now = utc_now_iso() session.add( TenantIntegrationRow( integration_id=new_id("tin"), tenant_id=tenant_id, provider_type=provider_type, provider_name=provider_name, provider_account_id=provider_account_id, external_identifier=None, settings_json="{}", is_active=True, created_at=now, updated_at=now, ) ) session.commit() finally: session.close() def test_tenant_cannot_read_foreign_lead_or_deal(): client = TestClient(sales_module.app) lead_b, deal_b = _create_lead_and_deal(client, "tenant_b") assert client.get(f"/api/v1/leads/{lead_b['lead_id']}", headers=_headers("tenant_a")).status_code == 404 assert client.get(f"/api/v1/deals/{deal_b}", headers=_headers("tenant_a")).status_code == 404 assert client.get(f"/api/v1/deals/{deal_b}/workspace", headers=_headers("tenant_a")).status_code == 404 def test_tenant_cannot_access_foreign_invoice_or_payment(): client = TestClient(sales_module.app) _, deal_b = _create_lead_and_deal(client, "tenant_b") _prepare_invoice_deal(client, deal_b, "tenant_b") invoice = client.post( f"/api/v1/deals/{deal_b}/invoices", json={"amount": 1000, "currency": "KZT", "line_items": [{"name": "Service", "amount": 1000}]}, headers=_headers("tenant_b"), ) assert invoice.status_code == 200 invoice_id = invoice.json()["invoice_id"] payment = client.post( "/api/v1/payments/webhook", json={ "deal_id": deal_b, "invoice_id": invoice_id, "payment_provider": "manual", "external_payment_id": new_id("ext"), "amount": 1000, "currency": "KZT", "status": "success", }, headers=_headers("tenant_b"), ) assert payment.status_code == 200 payment_id = payment.json()["payment_id"] assert client.get(f"/api/v1/invoices/{invoice_id}", headers=_headers("tenant_a")).status_code == 404 assert client.get(f"/api/v1/deals/{deal_b}/payments", headers=_headers("tenant_a")).status_code == 404 assert ( client.post( f"/api/v1/payments/{payment_id}/reconcile", json={"status": "failed", "failure_reason": "foreign tenant", "metadata": {}}, headers=_headers("tenant_a"), ).status_code == 404 ) def test_tenant_cannot_write_foreign_message_or_stage(): client = TestClient(sales_module.app) _, deal_b = _create_lead_and_deal(client, "tenant_b") message = client.post( "/api/v1/messages/outbound", json={"deal_id": deal_b, "channel_provider": "telegram", "sender_type": "human", "body": "Hello"}, headers=_headers("tenant_a"), ) assert message.status_code == 404 stage = client.post( f"/api/v1/deals/{deal_b}/change-stage", json={"stage_id": "offer_sent", "reason": "foreign tenant attempt"}, headers=_headers("tenant_a"), ) assert stage.status_code == 404 def test_provider_webhook_resolves_tenant_mapping(): client = TestClient(sales_module.app) _create_integration( tenant_id="tenant_webhook", provider_type="message", provider_name="telegram", provider_account_id="tg-account-tenant-webhook", ) webhook = client.post( "/api/v1/messages/inbound-webhook", json={ "phone": "+77009990001", "channel_provider": "telegram", "sender_id": "tg-user", "body": "Need pricing", "metadata": {"provider_account_id": "tg-account-tenant-webhook"}, }, headers={}, ) assert webhook.status_code == 200 tenant_rows = client.get("/api/v1/deals", headers=_headers("tenant_webhook")) assert tenant_rows.status_code == 200 assert any(item["tenant_id"] == "tenant_webhook" for item in tenant_rows.json()) other_rows = client.get("/api/v1/deals", headers=_headers("tenant_other")) assert other_rows.status_code == 200 assert other_rows.json() == [] def test_create_sales_objects_requires_tenant_context(): client = TestClient(sales_module.app) lead = client.post("/api/v1/leads", json=_lead_payload("missingtenant"), headers=_headers(None)) assert lead.status_code == 400 deal = client.post("/api/v1/deals", json=_deal_payload("missingtenant"), headers=_headers(None)) assert deal.status_code == 400