551 lines
20 KiB
Python
551 lines
20 KiB
Python
import hashlib
|
|
import hmac
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
import services.sales_service.app as sales_module
|
|
from services.sales_service.payment_webhooks import register_payment_provider_adapter
|
|
from services.shared.core import new_id, utc_now_iso
|
|
from services.shared.db import get_session
|
|
from services.shared.sql_models import EventOutboxRow
|
|
from services.shared.sales_sql_models import (
|
|
SalesAutomationTaskRow,
|
|
SalesInvoiceRow,
|
|
SalesPaymentRow,
|
|
SalesPaymentWebhookEventRow,
|
|
TenantIntegrationRow,
|
|
)
|
|
|
|
|
|
def _headers(tenant_id: str = "tenant_payment_webhooks") -> dict[str, str]:
|
|
return {"X-User": "admin", "X-Role": "admin", "X-Tenant-ID": tenant_id}
|
|
|
|
|
|
def _lead_payload(seed: str) -> dict:
|
|
return {
|
|
"source_type": "website",
|
|
"source_channel": "webchat",
|
|
"full_name": f"Payment Buyer {seed}",
|
|
"company_name": "Payments QA",
|
|
"phone": f"+7700{seed[-7:]}",
|
|
"email": f"{seed}@payments.test",
|
|
"lead_temperature": "warm",
|
|
"lead_score": 70,
|
|
"initial_need_summary": "Need invoice payment coverage.",
|
|
"preferred_channel": "telegram",
|
|
"assigned_agent_type": "text_ai",
|
|
"status": "new_qualified_lead",
|
|
"priority": 3,
|
|
"title": f"Payment Lead {seed}",
|
|
}
|
|
|
|
|
|
def _create_lead_and_deal(client: TestClient, tenant_id: str = "tenant_payment_webhooks") -> tuple[dict, dict]:
|
|
seed = new_id("paywh").replace("_", "")
|
|
response = client.post("/api/v1/leads", json=_lead_payload(seed), headers=_headers(tenant_id))
|
|
assert response.status_code == 200
|
|
lead = response.json()
|
|
deals = client.get("/api/v1/deals", headers=_headers(tenant_id))
|
|
assert deals.status_code == 200
|
|
deal = next(item for item in deals.json() if item["lead_id"] == lead["lead_id"])
|
|
return lead, deal
|
|
|
|
|
|
def _change_stage(client: TestClient, deal_id: str, stage_code: str, tenant_id: str = "tenant_payment_webhooks") -> None:
|
|
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 _prepare_invoice_deal(client: TestClient, deal_id: str, tenant_id: str = "tenant_payment_webhooks") -> None:
|
|
updated = client.patch(
|
|
f"/api/v1/deals/{deal_id}",
|
|
json={"document_required": False},
|
|
headers=_headers(tenant_id),
|
|
)
|
|
assert updated.status_code == 200
|
|
_change_stage(client, deal_id, "active_text_communication", tenant_id)
|
|
_change_stage(client, deal_id, "need_confirmed", tenant_id)
|
|
|
|
|
|
def _create_invoice(
|
|
client: TestClient,
|
|
deal_id: str,
|
|
tenant_id: str = "tenant_payment_webhooks",
|
|
*,
|
|
amount: float = 150000,
|
|
currency: str = "KZT",
|
|
) -> dict:
|
|
_prepare_invoice_deal(client, deal_id, tenant_id)
|
|
response = client.post(
|
|
f"/api/v1/deals/{deal_id}/invoices",
|
|
json={"amount": amount, "currency": currency, "due_date": "2026-05-15"},
|
|
headers=_headers(tenant_id),
|
|
)
|
|
assert response.status_code == 200
|
|
return response.json()
|
|
|
|
|
|
def _create_payment_integration(
|
|
*,
|
|
tenant_id: str,
|
|
provider: str = "kaspi",
|
|
provider_account_id: str = "merchant-1",
|
|
secret: str = "whsec_test",
|
|
) -> None:
|
|
session = get_session()
|
|
try:
|
|
now = utc_now_iso()
|
|
session.add(
|
|
TenantIntegrationRow(
|
|
integration_id=new_id("pin"),
|
|
tenant_id=tenant_id,
|
|
provider_type="payment",
|
|
provider_name=provider,
|
|
provider_account_id=provider_account_id,
|
|
external_identifier=None,
|
|
settings_json=json.dumps({"webhook_secret": secret, "webhook_replay_window_seconds": 600}),
|
|
is_active=True,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _signed_headers(raw_body: bytes, *, provider: str = "kaspi", provider_account_id: str = "merchant-1", secret: str = "whsec_test", timestamp: int | None = None) -> dict[str, str]:
|
|
ts = str(timestamp if timestamp is not None else int(datetime.now(timezone.utc).timestamp()))
|
|
digest = hmac.new(secret.encode("utf-8"), f"{ts}.".encode("utf-8") + raw_body, hashlib.sha256).hexdigest()
|
|
return {
|
|
"Content-Type": "application/json",
|
|
"X-Payment-Provider": provider,
|
|
"X-Provider-Account-Id": provider_account_id,
|
|
"X-Webhook-Timestamp": ts,
|
|
"X-Webhook-Signature": digest,
|
|
}
|
|
|
|
|
|
def _post_signed_webhook(client: TestClient, payload: dict, *, secret: str = "whsec_test", timestamp: int | None = None):
|
|
raw_body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
|
|
return client.post(
|
|
"/api/v1/payments/webhook",
|
|
content=raw_body,
|
|
headers=_signed_headers(raw_body, secret=secret, timestamp=timestamp),
|
|
)
|
|
|
|
|
|
def _event(event_type: str, tenant_id: str) -> EventOutboxRow | None:
|
|
session = get_session()
|
|
try:
|
|
rows = session.execute(
|
|
select(EventOutboxRow).where(EventOutboxRow.event_type == event_type).order_by(EventOutboxRow.id.desc())
|
|
).scalars().all()
|
|
for row in rows:
|
|
payload = json.loads(row.payload_json or "{}").get("payload", {})
|
|
if payload.get("tenant_id") == tenant_id:
|
|
return row
|
|
return None
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _webhook_event(external_event_id: str) -> SalesPaymentWebhookEventRow:
|
|
session = get_session()
|
|
try:
|
|
row = session.execute(
|
|
select(SalesPaymentWebhookEventRow).where(SalesPaymentWebhookEventRow.external_event_id == external_event_id)
|
|
).scalar_one()
|
|
session.expunge(row)
|
|
return row
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_payment_webhook_logs_processed_event_hash_and_normalized_payload():
|
|
tenant_id = "tenant_payment_webhook_log"
|
|
client = TestClient(sales_module.app)
|
|
_create_payment_integration(tenant_id=tenant_id)
|
|
_, deal = _create_lead_and_deal(client, tenant_id)
|
|
invoice = _create_invoice(client, deal["deal_id"], tenant_id)
|
|
payload = {
|
|
"external_event_id": "evt-log-1",
|
|
"external_payment_id": "ext-log-1",
|
|
"event_type": "payment.succeeded",
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"amount": 150000,
|
|
"currency": "KZT",
|
|
"status": "succeeded",
|
|
}
|
|
raw_body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
|
|
|
|
response = client.post("/api/v1/payments/webhook", content=raw_body, headers=_signed_headers(raw_body))
|
|
|
|
assert response.status_code == 200
|
|
event = _webhook_event("evt-log-1")
|
|
assert event.event_status == "processed"
|
|
assert event.signature_status == "valid"
|
|
assert event.raw_payload_hash == hashlib.sha256(raw_body).hexdigest()
|
|
normalized = json.loads(event.normalized_payload_json)
|
|
assert normalized["external_payment_id"] == "ext-log-1"
|
|
assert normalized["result"]["payment_id"] == response.json()["payment_id"]
|
|
|
|
|
|
def test_payment_webhook_rejects_invalid_signature_before_business_logic():
|
|
tenant_id = "tenant_payment_bad_signature"
|
|
client = TestClient(sales_module.app)
|
|
_create_payment_integration(tenant_id=tenant_id)
|
|
_, deal = _create_lead_and_deal(client, tenant_id)
|
|
invoice = _create_invoice(client, deal["deal_id"], tenant_id)
|
|
payload = {
|
|
"external_event_id": "evt-bad-signature-1",
|
|
"external_payment_id": "ext-bad-signature-1",
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"amount": 150000,
|
|
"currency": "KZT",
|
|
"status": "succeeded",
|
|
}
|
|
raw_body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
|
|
headers = _signed_headers(raw_body)
|
|
headers["X-Webhook-Signature"] = "sha256=invalid"
|
|
|
|
response = client.post("/api/v1/payments/webhook", content=raw_body, headers=headers)
|
|
|
|
assert response.status_code == 403
|
|
event = _webhook_event("evt-bad-signature-1")
|
|
assert event.event_status == "rejected"
|
|
assert event.signature_status == "invalid"
|
|
session = get_session()
|
|
try:
|
|
assert session.execute(select(SalesPaymentRow).where(SalesPaymentRow.external_payment_id == "ext-bad-signature-1")).scalar_one_or_none() is None
|
|
stored_invoice = session.execute(select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == invoice["invoice_id"])).scalar_one()
|
|
assert stored_invoice.status == "draft"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_payment_webhook_rejects_expired_replay_timestamp():
|
|
tenant_id = "tenant_payment_expired"
|
|
client = TestClient(sales_module.app)
|
|
_create_payment_integration(tenant_id=tenant_id)
|
|
_, deal = _create_lead_and_deal(client, tenant_id)
|
|
invoice = _create_invoice(client, deal["deal_id"], tenant_id)
|
|
payload = {
|
|
"external_event_id": "evt-expired-1",
|
|
"external_payment_id": "ext-expired-1",
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"amount": 150000,
|
|
"currency": "KZT",
|
|
"status": "succeeded",
|
|
}
|
|
expired_ts = int((datetime.now(timezone.utc) - timedelta(minutes=20)).timestamp())
|
|
|
|
response = _post_signed_webhook(client, payload, timestamp=expired_ts)
|
|
|
|
assert response.status_code == 400
|
|
event = _webhook_event("evt-expired-1")
|
|
assert event.event_status == "rejected"
|
|
assert event.signature_status == "expired"
|
|
|
|
|
|
def test_payment_webhook_duplicate_external_event_is_idempotent():
|
|
tenant_id = "tenant_payment_duplicate_event"
|
|
client = TestClient(sales_module.app)
|
|
_create_payment_integration(tenant_id=tenant_id)
|
|
_, deal = _create_lead_and_deal(client, tenant_id)
|
|
invoice = _create_invoice(client, deal["deal_id"], tenant_id)
|
|
payload = {
|
|
"external_event_id": "evt-dupe-1",
|
|
"external_payment_id": "ext-dupe-1",
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"amount": 150000,
|
|
"currency": "KZT",
|
|
"status": "succeeded",
|
|
}
|
|
|
|
first = _post_signed_webhook(client, payload)
|
|
second = _post_signed_webhook(client, payload)
|
|
|
|
assert first.status_code == 200
|
|
assert second.status_code == 200
|
|
assert second.json()["status"] == "duplicate"
|
|
session = get_session()
|
|
try:
|
|
payments = session.execute(select(SalesPaymentRow).where(SalesPaymentRow.external_payment_id == "ext-dupe-1")).scalars().all()
|
|
events = session.execute(select(SalesPaymentWebhookEventRow).where(SalesPaymentWebhookEventRow.external_event_id == "evt-dupe-1")).scalars().all()
|
|
assert len(payments) == 1
|
|
assert len(events) == 1
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_payment_webhook_same_external_payment_does_not_double_paid_sum():
|
|
tenant_id = "tenant_payment_duplicate_payment"
|
|
client = TestClient(sales_module.app)
|
|
_create_payment_integration(tenant_id=tenant_id)
|
|
_, deal = _create_lead_and_deal(client, tenant_id)
|
|
invoice = _create_invoice(client, deal["deal_id"], tenant_id)
|
|
|
|
first = _post_signed_webhook(
|
|
client,
|
|
{
|
|
"external_event_id": "evt-payment-1",
|
|
"external_payment_id": "ext-payment-1",
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"amount": 150000,
|
|
"currency": "KZT",
|
|
"status": "succeeded",
|
|
},
|
|
)
|
|
second = _post_signed_webhook(
|
|
client,
|
|
{
|
|
"external_event_id": "evt-payment-2",
|
|
"external_payment_id": "ext-payment-1",
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"amount": 150000,
|
|
"currency": "KZT",
|
|
"status": "succeeded",
|
|
},
|
|
)
|
|
|
|
assert first.status_code == 200
|
|
assert second.status_code == 200
|
|
session = get_session()
|
|
try:
|
|
payments = session.execute(select(SalesPaymentRow).where(SalesPaymentRow.external_payment_id == "ext-payment-1")).scalars().all()
|
|
stored_invoice = session.execute(select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == invoice["invoice_id"])).scalar_one()
|
|
assert len(payments) == 1
|
|
assert payments[0].amount == 150000
|
|
assert stored_invoice.status == "paid"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_payment_webhook_rejects_currency_mismatch_without_invoice_update():
|
|
tenant_id = "tenant_payment_currency_mismatch"
|
|
client = TestClient(sales_module.app)
|
|
_create_payment_integration(tenant_id=tenant_id)
|
|
_, deal = _create_lead_and_deal(client, tenant_id)
|
|
invoice = _create_invoice(client, deal["deal_id"], tenant_id, currency="KZT")
|
|
|
|
response = _post_signed_webhook(
|
|
client,
|
|
{
|
|
"external_event_id": "evt-currency-1",
|
|
"external_payment_id": "ext-currency-1",
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"amount": 150000,
|
|
"currency": "USD",
|
|
"status": "succeeded",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
event = _webhook_event("evt-currency-1")
|
|
assert event.event_status == "failed"
|
|
session = get_session()
|
|
try:
|
|
stored_invoice = session.execute(select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == invoice["invoice_id"])).scalar_one()
|
|
assert stored_invoice.status == "draft"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_partial_payment_sets_invoice_partially_paid():
|
|
tenant_id = "tenant_payment_partial"
|
|
client = TestClient(sales_module.app)
|
|
_create_payment_integration(tenant_id=tenant_id)
|
|
_, deal = _create_lead_and_deal(client, tenant_id)
|
|
invoice = _create_invoice(client, deal["deal_id"], tenant_id)
|
|
|
|
response = _post_signed_webhook(
|
|
client,
|
|
{
|
|
"external_event_id": "evt-partial-1",
|
|
"external_payment_id": "ext-partial-1",
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"amount": 50000,
|
|
"currency": "KZT",
|
|
"status": "partial",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
session = get_session()
|
|
try:
|
|
stored_invoice = session.execute(select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == invoice["invoice_id"])).scalar_one()
|
|
assert stored_invoice.status == "partially_paid"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_paid_invoice_cancels_payment_related_automation_tasks():
|
|
tenant_id = "tenant_payment_cancels_tasks"
|
|
client = TestClient(sales_module.app)
|
|
_create_payment_integration(tenant_id=tenant_id)
|
|
_, deal = _create_lead_and_deal(client, tenant_id)
|
|
invoice = _create_invoice(client, deal["deal_id"], tenant_id)
|
|
sent = client.post(f"/api/v1/invoices/{invoice['invoice_id']}/send", headers=_headers(tenant_id))
|
|
assert sent.status_code == 200
|
|
|
|
response = _post_signed_webhook(
|
|
client,
|
|
{
|
|
"external_event_id": "evt-cancel-tasks-1",
|
|
"external_payment_id": "ext-cancel-tasks-1",
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"amount": 150000,
|
|
"currency": "KZT",
|
|
"status": "succeeded",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
session = get_session()
|
|
try:
|
|
tasks = session.execute(
|
|
select(SalesAutomationTaskRow).where(
|
|
SalesAutomationTaskRow.deal_id == deal["deal_id"],
|
|
SalesAutomationTaskRow.task_type.in_(["send_invoice_reminder", "mark_invoice_overdue"]),
|
|
)
|
|
).scalars().all()
|
|
assert tasks
|
|
assert {task.status for task in tasks} == {"canceled"}
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
class _FakePaidAdapter:
|
|
def get_status(self, external_payment_id: str, *, context: dict | None = None):
|
|
return {"status": "captured", "metadata": {"reconciled": True}}
|
|
|
|
|
|
def test_reconcile_uses_provider_adapter_and_safe_payment_flow():
|
|
tenant_id = "tenant_payment_reconcile"
|
|
client = TestClient(sales_module.app)
|
|
_create_payment_integration(tenant_id=tenant_id)
|
|
register_payment_provider_adapter("kaspi", _FakePaidAdapter())
|
|
_, deal = _create_lead_and_deal(client, tenant_id)
|
|
invoice = _create_invoice(client, deal["deal_id"], tenant_id)
|
|
created = client.post(
|
|
"/api/v1/payments/webhook",
|
|
json={
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"payment_provider": "manual",
|
|
"external_payment_id": "ext-reconcile-1",
|
|
"amount": 150000,
|
|
"currency": "KZT",
|
|
"status": "pending",
|
|
},
|
|
headers=_headers(tenant_id),
|
|
)
|
|
assert created.status_code == 200
|
|
payment_id = created.json()["payment_id"]
|
|
session = get_session()
|
|
try:
|
|
payment = session.execute(select(SalesPaymentRow).where(SalesPaymentRow.payment_id == payment_id)).scalar_one()
|
|
payment.payment_provider = "kaspi"
|
|
payment.updated_at = utc_now_iso()
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
response = client.post(f"/api/v1/payments/{payment_id}/reconcile", headers=_headers(tenant_id))
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "success"
|
|
session = get_session()
|
|
try:
|
|
stored_invoice = session.execute(select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == invoice["invoice_id"])).scalar_one()
|
|
assert stored_invoice.status == "paid"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_reconcile_manual_payment_is_idempotent_without_adapter():
|
|
tenant_id = "tenant_payment_manual_reconcile"
|
|
client = TestClient(sales_module.app)
|
|
_, deal = _create_lead_and_deal(client, tenant_id)
|
|
invoice = _create_invoice(client, deal["deal_id"], tenant_id, amount=50000)
|
|
created = client.post(
|
|
"/api/v1/payments/webhook",
|
|
json={
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"payment_provider": "manual",
|
|
"external_payment_id": "manual-reconcile-1",
|
|
"amount": 50000,
|
|
"currency": "KZT",
|
|
"status": "success",
|
|
},
|
|
headers=_headers(tenant_id),
|
|
)
|
|
assert created.status_code == 200
|
|
payment_id = created.json()["payment_id"]
|
|
|
|
response = client.post(f"/api/v1/payments/{payment_id}/reconcile", headers=_headers(tenant_id))
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "success"
|
|
|
|
|
|
def test_payment_webhook_publishes_payment_and_invoice_events_once_for_duplicate_payment():
|
|
tenant_id = "tenant_payment_events_once"
|
|
client = TestClient(sales_module.app)
|
|
_create_payment_integration(tenant_id=tenant_id)
|
|
_, deal = _create_lead_and_deal(client, tenant_id)
|
|
invoice = _create_invoice(client, deal["deal_id"], tenant_id)
|
|
|
|
for index in range(2):
|
|
response = _post_signed_webhook(
|
|
client,
|
|
{
|
|
"external_event_id": f"evt-events-once-{index}",
|
|
"external_payment_id": "ext-events-once",
|
|
"deal_id": deal["deal_id"],
|
|
"invoice_id": invoice["invoice_id"],
|
|
"amount": 150000,
|
|
"currency": "KZT",
|
|
"status": "succeeded",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
assert _event("payment.received", tenant_id) is not None
|
|
assert _event("invoice.paid", tenant_id) is not None
|
|
session = get_session()
|
|
try:
|
|
payment_events = []
|
|
invoice_events = []
|
|
for row in session.execute(select(EventOutboxRow)).scalars().all():
|
|
envelope = json.loads(row.payload_json or "{}")
|
|
payload = envelope.get("payload", {})
|
|
if payload.get("tenant_id") != tenant_id:
|
|
continue
|
|
if row.event_type == "payment.received":
|
|
payment_events.append(row)
|
|
if row.event_type == "invoice.paid":
|
|
invoice_events.append(row)
|
|
assert len(payment_events) == 1
|
|
assert len(invoice_events) == 1
|
|
finally:
|
|
session.close()
|