371 lines
13 KiB
Python
371 lines
13 KiB
Python
import json
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
import services.sales_service.app as sales_module
|
|
from services.shared.db import get_session
|
|
from services.shared.sql_models import EventOutboxRow
|
|
from services.shared.sales_sql_models import SalesStageHistoryRow
|
|
|
|
|
|
def _headers(tenant_id: str = "tenant_state", role: str = "admin") -> dict[str, str]:
|
|
return {"X-User": "admin", "X-Role": role, "X-Tenant-ID": tenant_id}
|
|
|
|
|
|
def _deal_payload(seed: str = "state", **overrides: object) -> dict:
|
|
payload = {
|
|
"stage_id": "new_qualified_lead",
|
|
"scenario_type": "quick_sale",
|
|
"priority": 3,
|
|
"title": f"State machine deal {seed}",
|
|
"need_summary": "Need a managed sales flow.",
|
|
"document_required": False,
|
|
"preferred_channel": "telegram",
|
|
"current_channel": "telegram",
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
def _create_deal(client: TestClient, tenant_id: str = "tenant_state", **overrides: object) -> dict:
|
|
response = client.post("/api/v1/deals", json=_deal_payload(tenant_id, **overrides), headers=_headers(tenant_id))
|
|
assert response.status_code == 200
|
|
return response.json()
|
|
|
|
|
|
def _change_stage(
|
|
client: TestClient,
|
|
deal_id: str,
|
|
target_stage_code: str,
|
|
tenant_id: str = "tenant_state",
|
|
*,
|
|
reason: str | None = None,
|
|
metadata: dict | None = None,
|
|
force: bool = False,
|
|
role: str = "admin",
|
|
) -> dict:
|
|
payload = {"target_stage_code": target_stage_code, "metadata": metadata or {}, "force": force}
|
|
if reason is not None:
|
|
payload["reason"] = reason
|
|
response = client.post(f"/api/v1/deals/{deal_id}/change-stage", json=payload, headers=_headers(tenant_id, role))
|
|
assert response.status_code == 200
|
|
return response.json()
|
|
|
|
|
|
def _prepare_need_confirmed(client: TestClient, deal_id: str, tenant_id: str = "tenant_state") -> None:
|
|
_change_stage(client, deal_id, "active_text_communication", tenant_id, reason="text started")
|
|
_change_stage(client, deal_id, "need_confirmed", tenant_id, reason="need confirmed")
|
|
|
|
|
|
def _prepare_invoice(client: TestClient, deal_id: str, tenant_id: str = "tenant_state") -> dict:
|
|
_prepare_need_confirmed(client, deal_id, tenant_id)
|
|
created = client.post(
|
|
f"/api/v1/deals/{deal_id}/invoices",
|
|
json={"amount": 1000, "currency": "KZT", "due_date": "2026-05-15"},
|
|
headers=_headers(tenant_id),
|
|
)
|
|
assert created.status_code == 200
|
|
sent = client.post(f"/api/v1/invoices/{created.json()['invoice_id']}/send", headers=_headers(tenant_id))
|
|
assert sent.status_code == 200
|
|
return sent.json()
|
|
|
|
|
|
def _stage_id(client: TestClient, tenant_id: str, code: str) -> str:
|
|
pipelines = client.get("/api/v1/pipelines", headers=_headers(tenant_id))
|
|
assert pipelines.status_code == 200
|
|
default_pipeline = next(item for item in pipelines.json() if item["is_default"])
|
|
stages = client.get(f"/api/v1/pipelines/{default_pipeline['pipeline_id']}/stages", headers=_headers(tenant_id))
|
|
assert stages.status_code == 200
|
|
return next(item["stage_id"] for item in stages.json() if item["code"] == code)
|
|
|
|
|
|
def _event_payloads(event_type: str, tenant_id: str) -> list[dict]:
|
|
session = get_session()
|
|
try:
|
|
rows = session.execute(
|
|
select(EventOutboxRow)
|
|
.where(EventOutboxRow.producer_service == "sales-service", EventOutboxRow.event_type == event_type)
|
|
.order_by(EventOutboxRow.id.asc())
|
|
).scalars().all()
|
|
payloads = []
|
|
for row in rows:
|
|
envelope = json.loads(row.payload_json or "{}")
|
|
payload = envelope.get("payload") if isinstance(envelope.get("payload"), dict) else {}
|
|
if payload.get("tenant_id") == tenant_id:
|
|
payloads.append(payload)
|
|
return payloads
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_valid_stage_transition_succeeds():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client)
|
|
|
|
changed = _change_stage(client, deal["deal_id"], "warm_lead", reason="lead warmed")
|
|
|
|
assert changed["stage"]["code"] == "warm_lead"
|
|
|
|
|
|
def test_invalid_stage_transition_fails():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client)
|
|
|
|
response = client.post(
|
|
f"/api/v1/deals/{deal['deal_id']}/change-stage",
|
|
json={"target_stage_code": "offer_sent", "reason": "skip pipeline"},
|
|
headers=_headers(),
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["error"] == "invalid_stage_transition"
|
|
|
|
|
|
def test_change_stage_missing_deal_returns_state_machine_error():
|
|
client = TestClient(sales_module.app)
|
|
|
|
response = client.post(
|
|
"/api/v1/deals/sde_missing/change-stage",
|
|
json={"target_stage_code": "warm_lead", "reason": "missing deal"},
|
|
headers=_headers(),
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
assert response.json()["error"] == "deal_not_found"
|
|
|
|
|
|
def test_cannot_jump_from_new_lead_to_paid():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client)
|
|
|
|
response = client.post(
|
|
f"/api/v1/deals/{deal['deal_id']}/change-stage",
|
|
json={"target_stage_code": "paid", "reason": "jump"},
|
|
headers=_headers(),
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["error"] == "invalid_stage_transition"
|
|
|
|
|
|
def test_cannot_set_stage_from_another_tenant():
|
|
client = TestClient(sales_module.app)
|
|
foreign_stage_id = _stage_id(client, "tenant_state_foreign", "warm_lead")
|
|
deal = _create_deal(client, "tenant_state_owner")
|
|
|
|
response = client.post(
|
|
f"/api/v1/deals/{deal['deal_id']}/change-stage",
|
|
json={"target_stage_id": foreign_stage_id, "reason": "foreign stage"},
|
|
headers=_headers("tenant_state_owner"),
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert response.json()["error"] == "cross_tenant_stage_forbidden"
|
|
|
|
|
|
def test_cannot_set_stage_from_another_pipeline():
|
|
client = TestClient(sales_module.app)
|
|
tenant_id = "tenant_state_pipeline"
|
|
deal = _create_deal(client, tenant_id)
|
|
pipeline = client.post(
|
|
"/api/v1/pipelines",
|
|
json={"code": "enterprise_sales", "name": "Enterprise Sales"},
|
|
headers=_headers(tenant_id),
|
|
)
|
|
assert pipeline.status_code == 200
|
|
stage = client.post(
|
|
f"/api/v1/pipelines/{pipeline.json()['pipeline_id']}/stages",
|
|
json={"code": "enterprise_warm", "name": "Enterprise Warm", "category": "entry", "sort_order": 10},
|
|
headers=_headers(tenant_id),
|
|
)
|
|
assert stage.status_code == 200
|
|
|
|
response = client.post(
|
|
f"/api/v1/deals/{deal['deal_id']}/change-stage",
|
|
json={"target_stage_id": stage.json()["stage_id"], "reason": "wrong pipeline"},
|
|
headers=_headers(tenant_id),
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["error"] == "invalid_stage_pipeline"
|
|
|
|
|
|
def test_offer_sent_requires_offer():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client)
|
|
_prepare_need_confirmed(client, deal["deal_id"])
|
|
_change_stage(client, deal["deal_id"], "offer_preparing", reason="prepare offer")
|
|
|
|
response = client.post(
|
|
f"/api/v1/deals/{deal['deal_id']}/change-stage",
|
|
json={"target_stage_code": "offer_sent", "reason": "send missing offer"},
|
|
headers=_headers(),
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["error"] == "offer_required_before_offer_sent"
|
|
|
|
|
|
def test_invoice_sent_requires_invoice():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client)
|
|
_prepare_need_confirmed(client, deal["deal_id"])
|
|
_change_stage(client, deal["deal_id"], "invoice_preparing", reason="prepare invoice")
|
|
|
|
response = client.post(
|
|
f"/api/v1/deals/{deal['deal_id']}/change-stage",
|
|
json={"target_stage_code": "invoice_sent", "reason": "send missing invoice"},
|
|
headers=_headers(),
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["error"] == "invoice_required_before_invoice_sent"
|
|
|
|
|
|
def test_won_requires_payment_when_payment_required():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client, payment_required=True)
|
|
escalated = client.post(
|
|
f"/api/v1/deals/{deal['deal_id']}/escalate",
|
|
json={"escalation_type": "manual_review", "reason": "Need human review", "severity": "medium"},
|
|
headers=_headers(),
|
|
)
|
|
assert escalated.status_code == 200
|
|
|
|
response = client.post(
|
|
f"/api/v1/deals/{deal['deal_id']}/change-stage",
|
|
json={"target_stage_code": "won", "reason": "manual win"},
|
|
headers=_headers(),
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["error"] == "payment_required_before_won"
|
|
|
|
|
|
def test_lost_requires_reason():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client)
|
|
|
|
response = client.post(
|
|
f"/api/v1/deals/{deal['deal_id']}/change-stage",
|
|
json={"target_stage_code": "lost"},
|
|
headers=_headers(),
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["error"] == "lost_reason_required"
|
|
|
|
|
|
def test_successful_transition_writes_stage_history():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client, "tenant_state_history")
|
|
|
|
_change_stage(client, deal["deal_id"], "warm_lead", "tenant_state_history", reason="lead warmed")
|
|
|
|
session = get_session()
|
|
try:
|
|
rows = session.execute(
|
|
select(SalesStageHistoryRow).where(
|
|
SalesStageHistoryRow.tenant_id == "tenant_state_history",
|
|
SalesStageHistoryRow.deal_id == deal["deal_id"],
|
|
)
|
|
).scalars().all()
|
|
assert rows
|
|
assert rows[-1].reason == "lead warmed"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_successful_transition_publishes_deal_stage_changed():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client, "tenant_state_event")
|
|
|
|
_change_stage(client, deal["deal_id"], "warm_lead", "tenant_state_event", reason="lead warmed")
|
|
|
|
payload = _event_payloads("deal.stage_changed", "tenant_state_event")[-1]
|
|
assert payload["deal_id"] == deal["deal_id"]
|
|
assert payload["from_stage_code"] == "new_qualified_lead"
|
|
assert payload["to_stage_code"] == "warm_lead"
|
|
|
|
|
|
def test_won_transition_sets_status_and_closed_at():
|
|
client = TestClient(sales_module.app)
|
|
tenant_id = "tenant_state_won"
|
|
deal = _create_deal(client, tenant_id)
|
|
invoice = _prepare_invoice(client, deal["deal_id"], tenant_id)
|
|
|
|
response = client.post(
|
|
"/api/v1/payments/webhook",
|
|
json={
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"payment_provider": "manual",
|
|
"external_payment_id": "state-paid-1",
|
|
"amount": 1000,
|
|
"currency": "KZT",
|
|
"status": "success",
|
|
},
|
|
headers=_headers(tenant_id),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
refreshed = client.get(f"/api/v1/deals/{deal['deal_id']}", headers=_headers(tenant_id))
|
|
assert refreshed.status_code == 200
|
|
assert refreshed.json()["status"] == "won"
|
|
assert refreshed.json()["closed_at"]
|
|
assert refreshed.json()["stage"]["code"] == "won"
|
|
|
|
|
|
def test_lost_transition_sets_status_and_closed_at():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client, "tenant_state_lost")
|
|
|
|
changed = _change_stage(client, deal["deal_id"], "lost", "tenant_state_lost", reason="not interested")
|
|
|
|
assert changed["status"] == "lost"
|
|
assert changed["closed_at"]
|
|
assert changed["lost_reason"] == "not interested"
|
|
|
|
|
|
def test_payment_webhook_transitions_deal_to_paid_and_won():
|
|
client = TestClient(sales_module.app)
|
|
tenant_id = "tenant_state_payment"
|
|
deal = _create_deal(client, tenant_id)
|
|
invoice = _prepare_invoice(client, deal["deal_id"], tenant_id)
|
|
|
|
response = client.post(
|
|
"/api/v1/payments/webhook",
|
|
json={
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"payment_provider": "manual",
|
|
"external_payment_id": "state-paid-2",
|
|
"amount": 1000,
|
|
"currency": "KZT",
|
|
"status": "success",
|
|
},
|
|
headers=_headers(tenant_id),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
workspace = client.get(f"/api/v1/deals/{deal['deal_id']}/workspace", headers=_headers(tenant_id))
|
|
assert workspace.status_code == 200
|
|
stage_codes = [item["to_stage"]["code"] for item in workspace.json()["stage_history"]]
|
|
assert "paid" in stage_codes
|
|
assert "won" in stage_codes
|
|
|
|
|
|
def test_force_transition_requires_system_or_admin_actor():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client, "tenant_state_force")
|
|
|
|
response = client.post(
|
|
f"/api/v1/deals/{deal['deal_id']}/change-stage",
|
|
json={"target_stage_code": "warm_lead", "reason": "operator force", "force": True},
|
|
headers=_headers("tenant_state_force", role="operator"),
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert response.json()["error"] == "force_transition_forbidden"
|