205 lines
7.0 KiB
Python
205 lines
7.0 KiB
Python
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
|