438 lines
15 KiB
Python
438 lines
15 KiB
Python
import json
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
import services.sales_service.app as sales_module
|
|
from services.sales_service.automation_worker import SalesAutomationWorker
|
|
from services.shared.core import new_id, utc_now_iso
|
|
from services.shared.db import get_session
|
|
from services.shared.sales_sql_models import (
|
|
SalesAutomationTaskRow,
|
|
SalesDealRow,
|
|
SalesEscalationRow,
|
|
SalesInvoiceRow,
|
|
SalesPipelineStageRow,
|
|
)
|
|
from services.shared.sql_models import EventOutboxRow
|
|
|
|
|
|
def _headers(tenant_id: str = "tenant_worker") -> dict[str, str]:
|
|
return {"X-User": "admin", "X-Role": "admin", "X-Tenant-ID": tenant_id}
|
|
|
|
|
|
def _deal_payload(seed: str = "worker", **overrides: object) -> dict:
|
|
payload = {
|
|
"stage_id": "new_qualified_lead",
|
|
"scenario_type": "quick_sale",
|
|
"priority": 3,
|
|
"title": f"Automation deal {seed}",
|
|
"need_summary": "Need automation coverage.",
|
|
"document_required": False,
|
|
"preferred_channel": "telegram",
|
|
"current_channel": "telegram",
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
def _create_deal(client: TestClient, tenant_id: str = "tenant_worker", **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, stage_code: str, tenant_id: str = "tenant_worker") -> dict:
|
|
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
|
|
return response.json()
|
|
|
|
|
|
def _prepare_invoice(client: TestClient, deal_id: str, tenant_id: str = "tenant_worker") -> dict:
|
|
_change_stage(client, deal_id, "active_text_communication", tenant_id)
|
|
_change_stage(client, deal_id, "need_confirmed", 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 _create_task(
|
|
*,
|
|
tenant_id: str,
|
|
deal_id: str,
|
|
task_type: str,
|
|
payload: dict | None = None,
|
|
run_at: str | None = None,
|
|
status: str = "pending",
|
|
retry_count: int = 0,
|
|
max_retries: int = 3,
|
|
) -> str:
|
|
session = get_session()
|
|
try:
|
|
now = utc_now_iso()
|
|
row = SalesAutomationTaskRow(
|
|
task_id=new_id("tsk"),
|
|
tenant_id=tenant_id,
|
|
deal_id=deal_id,
|
|
task_type=task_type,
|
|
payload_json=json.dumps(payload or {}, ensure_ascii=False),
|
|
run_at=run_at or now,
|
|
status=status,
|
|
retry_count=retry_count,
|
|
max_retries=max_retries,
|
|
locked_at=None,
|
|
locked_by=None,
|
|
completed_at=None,
|
|
failed_at=None,
|
|
last_error=None,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(row)
|
|
session.commit()
|
|
return row.task_id
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _task(task_id: str) -> SalesAutomationTaskRow:
|
|
session = get_session()
|
|
try:
|
|
return session.execute(select(SalesAutomationTaskRow).where(SalesAutomationTaskRow.task_id == task_id)).scalar_one()
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _deal(deal_id: str, tenant_id: str) -> SalesDealRow:
|
|
session = get_session()
|
|
try:
|
|
return session.execute(
|
|
select(SalesDealRow).where(SalesDealRow.deal_id == deal_id, SalesDealRow.tenant_id == tenant_id)
|
|
).scalar_one()
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _invoice(invoice_id: str, tenant_id: str) -> SalesInvoiceRow:
|
|
session = get_session()
|
|
try:
|
|
return session.execute(
|
|
select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == invoice_id, SalesInvoiceRow.tenant_id == tenant_id)
|
|
).scalar_one()
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _stage_code(deal_id: str, tenant_id: str) -> str:
|
|
session = get_session()
|
|
try:
|
|
deal = session.execute(
|
|
select(SalesDealRow).where(SalesDealRow.deal_id == deal_id, SalesDealRow.tenant_id == tenant_id)
|
|
).scalar_one()
|
|
stage = session.execute(
|
|
select(SalesPipelineStageRow).where(
|
|
SalesPipelineStageRow.stage_id == deal.stage_id,
|
|
SalesPipelineStageRow.tenant_id == tenant_id,
|
|
)
|
|
).scalar_one()
|
|
return stage.code
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _events(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_worker_claims_due_pending_task():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client)
|
|
task_id = _create_task(tenant_id="tenant_worker", deal_id=deal["deal_id"], task_type="follow_up_customer")
|
|
worker = SalesAutomationWorker(worker_id="worker-claim")
|
|
session = get_session()
|
|
try:
|
|
claimed = worker.claim_pending_tasks(session)
|
|
assert [task.task_id for task in claimed] == [task_id]
|
|
assert claimed[0].status == "running"
|
|
assert claimed[0].locked_by == "worker-claim"
|
|
finally:
|
|
session.rollback()
|
|
session.close()
|
|
|
|
|
|
def test_worker_does_not_claim_future_task():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client)
|
|
_create_task(
|
|
tenant_id="tenant_worker",
|
|
deal_id=deal["deal_id"],
|
|
task_type="follow_up_customer",
|
|
run_at="2999-01-01T00:00:00+00:00",
|
|
)
|
|
worker = SalesAutomationWorker(worker_id="worker-future")
|
|
session = get_session()
|
|
try:
|
|
assert worker.claim_pending_tasks(session) == []
|
|
finally:
|
|
session.rollback()
|
|
session.close()
|
|
|
|
|
|
def test_worker_marks_task_completed_on_success():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client)
|
|
task_id = _create_task(tenant_id="tenant_worker", deal_id=deal["deal_id"], task_type="follow_up_customer")
|
|
|
|
assert SalesAutomationWorker(worker_id="worker-complete").run_once() == 1
|
|
|
|
task = _task(task_id)
|
|
assert task.status == "completed"
|
|
assert task.completed_at
|
|
|
|
|
|
def test_worker_retries_failed_task():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client)
|
|
task_id = _create_task(tenant_id="tenant_worker", deal_id=deal["deal_id"], task_type="unknown_task")
|
|
|
|
SalesAutomationWorker(worker_id="worker-retry").run_once()
|
|
|
|
task = _task(task_id)
|
|
assert task.status == "pending"
|
|
assert task.retry_count == 1
|
|
assert task.last_error
|
|
|
|
|
|
def test_worker_marks_task_failed_after_max_retries():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client)
|
|
task_id = _create_task(
|
|
tenant_id="tenant_worker",
|
|
deal_id=deal["deal_id"],
|
|
task_type="unknown_task",
|
|
max_retries=1,
|
|
)
|
|
|
|
SalesAutomationWorker(worker_id="worker-failed").run_once()
|
|
|
|
task = _task(task_id)
|
|
assert task.status == "failed"
|
|
assert task.retry_count == 1
|
|
assert task.failed_at
|
|
|
|
|
|
def test_mark_invoice_overdue_updates_invoice_status():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client)
|
|
invoice = _prepare_invoice(client, deal["deal_id"])
|
|
task_id = _create_task(
|
|
tenant_id="tenant_worker",
|
|
deal_id=deal["deal_id"],
|
|
task_type="mark_invoice_overdue",
|
|
payload={"invoice_id": invoice["invoice_id"]},
|
|
)
|
|
|
|
SalesAutomationWorker(worker_id="worker-overdue").run_once()
|
|
|
|
assert _task(task_id).status == "completed"
|
|
assert _invoice(invoice["invoice_id"], "tenant_worker").status == "overdue"
|
|
|
|
|
|
def test_mark_invoice_overdue_transitions_deal_via_state_machine():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client, "tenant_worker_stage")
|
|
invoice = _prepare_invoice(client, deal["deal_id"], "tenant_worker_stage")
|
|
_create_task(
|
|
tenant_id="tenant_worker_stage",
|
|
deal_id=deal["deal_id"],
|
|
task_type="mark_invoice_overdue",
|
|
payload={"invoice_id": invoice["invoice_id"]},
|
|
)
|
|
|
|
SalesAutomationWorker(worker_id="worker-overdue-stage").run_once()
|
|
|
|
assert _stage_code(deal["deal_id"], "tenant_worker_stage") == "payment_overdue"
|
|
assert _events("deal.stage_changed", "tenant_worker_stage")[-1]["to_stage_code"] == "payment_overdue"
|
|
|
|
|
|
def test_send_invoice_reminder_creates_follow_up():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client, "tenant_worker_reminder")
|
|
invoice = _prepare_invoice(client, deal["deal_id"], "tenant_worker_reminder")
|
|
_create_task(
|
|
tenant_id="tenant_worker_reminder",
|
|
deal_id=deal["deal_id"],
|
|
task_type="send_invoice_reminder",
|
|
payload={"invoice_id": invoice["invoice_id"]},
|
|
)
|
|
|
|
SalesAutomationWorker(worker_id="worker-reminder").run_once()
|
|
|
|
session = get_session()
|
|
try:
|
|
follow_up = session.execute(
|
|
select(SalesAutomationTaskRow).where(
|
|
SalesAutomationTaskRow.tenant_id == "tenant_worker_reminder",
|
|
SalesAutomationTaskRow.deal_id == deal["deal_id"],
|
|
SalesAutomationTaskRow.task_type == "follow_up_customer",
|
|
)
|
|
).scalars().all()
|
|
assert len(follow_up) == 1
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_follow_up_customer_updates_next_action():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client, "tenant_worker_follow")
|
|
_create_task(tenant_id="tenant_worker_follow", deal_id=deal["deal_id"], task_type="follow_up_customer")
|
|
|
|
SalesAutomationWorker(worker_id="worker-follow").run_once()
|
|
|
|
assert _deal(deal["deal_id"], "tenant_worker_follow").next_action_type == "follow_up_customer"
|
|
|
|
|
|
def test_recommend_channel_switch_records_recommendation():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client, "tenant_worker_switch")
|
|
task_id = _create_task(
|
|
tenant_id="tenant_worker_switch",
|
|
deal_id=deal["deal_id"],
|
|
task_type="recommend_channel_switch",
|
|
payload={"recommended_from_channel": "text", "recommended_to_channel": "voice", "reason": "no_reply_in_text"},
|
|
)
|
|
|
|
SalesAutomationWorker(worker_id="worker-switch").run_once()
|
|
|
|
payload = json.loads(_task(task_id).payload_json)
|
|
assert payload["automation_result"]["recommended_to_channel"] == "voice"
|
|
assert _events("deal.channel_switch_recommended", "tenant_worker_switch")
|
|
|
|
|
|
def test_escalate_to_human_creates_escalation():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client, "tenant_worker_escalation")
|
|
_create_task(
|
|
tenant_id="tenant_worker_escalation",
|
|
deal_id=deal["deal_id"],
|
|
task_type="escalate_to_human",
|
|
payload={"reason": "low_ai_confidence", "assigned_to_user_id": "operator-1"},
|
|
)
|
|
|
|
SalesAutomationWorker(worker_id="worker-escalation").run_once()
|
|
|
|
session = get_session()
|
|
try:
|
|
rows = session.execute(
|
|
select(SalesEscalationRow).where(
|
|
SalesEscalationRow.tenant_id == "tenant_worker_escalation",
|
|
SalesEscalationRow.deal_id == deal["deal_id"],
|
|
)
|
|
).scalars().all()
|
|
assert len(rows) == 1
|
|
assert rows[0].assigned_to_user_id == "operator-1"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_escalate_to_human_transitions_deal_to_support():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client, "tenant_worker_support")
|
|
_create_task(
|
|
tenant_id="tenant_worker_support",
|
|
deal_id=deal["deal_id"],
|
|
task_type="escalate_to_human",
|
|
payload={"reason": "customer_requested_human"},
|
|
)
|
|
|
|
SalesAutomationWorker(worker_id="worker-support").run_once()
|
|
|
|
assert _stage_code(deal["deal_id"], "tenant_worker_support") == "transferred_to_support"
|
|
|
|
|
|
def test_post_sale_transfer_publishes_event():
|
|
client = TestClient(sales_module.app)
|
|
tenant_id = "tenant_worker_post_sale"
|
|
deal = _create_deal(client, tenant_id)
|
|
invoice = _prepare_invoice(client, deal["deal_id"], tenant_id)
|
|
payment = client.post(
|
|
"/api/v1/payments/webhook",
|
|
json={
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"payment_provider": "manual",
|
|
"external_payment_id": "worker-paid-1",
|
|
"amount": 1000,
|
|
"currency": "KZT",
|
|
"status": "success",
|
|
},
|
|
headers=_headers(tenant_id),
|
|
)
|
|
assert payment.status_code == 200
|
|
_create_task(tenant_id=tenant_id, deal_id=deal["deal_id"], task_type="post_sale_transfer")
|
|
|
|
SalesAutomationWorker(worker_id="worker-post-sale").run_once()
|
|
|
|
assert _events("deal.transferred_post_sale", tenant_id)
|
|
assert _stage_code(deal["deal_id"], tenant_id) == "transferred_to_execution"
|
|
|
|
|
|
def test_worker_is_tenant_scoped():
|
|
client = TestClient(sales_module.app)
|
|
deal_a = _create_deal(client, "tenant_worker_a")
|
|
deal_b = _create_deal(client, "tenant_worker_b")
|
|
invoice_b = _prepare_invoice(client, deal_b["deal_id"], "tenant_worker_b")
|
|
task_id = _create_task(
|
|
tenant_id="tenant_worker_a",
|
|
deal_id=deal_a["deal_id"],
|
|
task_type="mark_invoice_overdue",
|
|
payload={"invoice_id": invoice_b["invoice_id"]},
|
|
)
|
|
|
|
SalesAutomationWorker(worker_id="worker-tenant").run_once()
|
|
|
|
assert _task(task_id).status == "pending"
|
|
assert _invoice(invoice_b["invoice_id"], "tenant_worker_b").status == "sent"
|
|
|
|
|
|
def test_worker_is_idempotent_for_completed_task():
|
|
client = TestClient(sales_module.app)
|
|
deal = _create_deal(client, "tenant_worker_done")
|
|
task_id = _create_task(
|
|
tenant_id="tenant_worker_done",
|
|
deal_id=deal["deal_id"],
|
|
task_type="follow_up_customer",
|
|
status="completed",
|
|
)
|
|
|
|
assert SalesAutomationWorker(worker_id="worker-done").run_once() == 0
|
|
|
|
task = _task(task_id)
|
|
assert task.status == "completed"
|
|
assert _events("deal.follow_up_ready", "tenant_worker_done") == []
|