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()