chat_backend/llm_be/finance/tests/test_webhooks.py

"""Tests for Stripe webhook verification and ledger upserts."""

from unittest.mock import patch

from django.test import override_settings
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase

from chat_backend.tests.factories import make_company, make_user
from finance.models import Invoice, Payment
from finance.services.webhooks import (
    dispatch_stripe_event,
    handle_checkout_session_completed,
    handle_invoice_paid,
    handle_invoice_payment_failed,
)


class WebhookHandlerUnitTestCase(APITestCase):
    def setUp(self):
        self.company = make_company()
        self.user = make_user(company=self.company)

    def test_checkout_session_completed_creates_invoice_and_payment(self):
        session = {
            "id": "cs_test_completed",
            "metadata": {"user_id": str(self.user.pk)},
            "customer": "cus_123",
            "subscription": "sub_123",
            "payment_intent": "pi_123",
            "payment_status": "paid",
            "amount_total": 1000,
            "currency": "usd",
            "customer_email": self.user.email,
        }
        invoice = handle_checkout_session_completed(session)
        self.assertIsNotNone(invoice)
        self.assertEqual(invoice.status, Invoice.Status.PAID)
        self.assertEqual(invoice.amount_paid, 1000)
        self.assertEqual(invoice.stripe_subscription_id, "sub_123")
        payment = Payment.objects.get(stripe_payment_intent_id="pi_123")
        self.assertEqual(payment.status, Payment.Status.SUCCEEDED)
        self.assertEqual(payment.invoice_id, invoice.pk)

    def test_checkout_session_completed_is_idempotent(self):
        session = {
            "id": "cs_test_idem",
            "metadata": {"user_id": str(self.user.pk)},
            "payment_status": "paid",
            "amount_total": 1000,
            "currency": "usd",
            "payment_intent": "pi_idem",
        }
        handle_checkout_session_completed(session)
        handle_checkout_session_completed(session)
        self.assertEqual(
            Invoice.objects.filter(stripe_checkout_session_id="cs_test_idem").count(),
            1,
        )
        self.assertEqual(
            Payment.objects.filter(stripe_payment_intent_id="pi_idem").count(),
            1,
        )

    def test_invoice_paid_upserts(self):
        stripe_invoice = {
            "id": "in_paid_1",
            "metadata": {"user_id": str(self.user.pk)},
            "customer": "cus_1",
            "subscription": "sub_1",
            "amount_due": 1000,
            "amount_paid": 1000,
            "currency": "usd",
            "status": "paid",
            "payment_intent": "pi_paid_1",
            "charge": "ch_paid_1",
            "period_start": 1_700_000_000,
            "period_end": 1_700_259_200,
            "hosted_invoice_url": "https://invoice.stripe.com/i/test",
            "status_transitions": {"paid_at": 1_700_000_100},
        }
        invoice = handle_invoice_paid(stripe_invoice)
        self.assertEqual(invoice.status, Invoice.Status.PAID)
        self.assertEqual(invoice.stripe_invoice_id, "in_paid_1")
        payment = Payment.objects.get(stripe_payment_intent_id="pi_paid_1")
        self.assertEqual(payment.stripe_charge_id, "ch_paid_1")
        self.assertEqual(payment.status, Payment.Status.SUCCEEDED)

    def test_invoice_payment_failed(self):
        stripe_invoice = {
            "id": "in_fail_1",
            "metadata": {"user_id": str(self.user.pk)},
            "amount_due": 1000,
            "amount_paid": 0,
            "currency": "usd",
            "payment_intent": "pi_fail_1",
        }
        invoice = handle_invoice_payment_failed(stripe_invoice)
        self.assertEqual(invoice.status, Invoice.Status.PAYMENT_FAILED)
        payment = Payment.objects.get(stripe_payment_intent_id="pi_fail_1")
        self.assertEqual(payment.status, Payment.Status.FAILED)

    def test_dispatch_ignores_unknown_events(self):
        result = dispatch_stripe_event(
            {"type": "customer.created", "data": {"object": {}}}
        )
        self.assertIsNone(result)


class StripeWebhookViewTestCase(APITestCase):
    def setUp(self):
        self.url = reverse("finance_stripe_webhook")
        self.company = make_company()
        self.user = make_user(company=self.company)

    @override_settings(STRIPE_WEBHOOK_SECRET="")
    def test_missing_webhook_secret_returns_503(self):
        response = self.client.post(
            self.url,
            data=b"{}",
            content_type="application/json",
        )
        self.assertEqual(response.status_code, status.HTTP_503_SERVICE_UNAVAILABLE)

    @override_settings(STRIPE_WEBHOOK_SECRET="whsec_test")
    @patch("finance.views.stripe.Webhook.construct_event")
    def test_invalid_signature_returns_400(self, mock_construct):
        import stripe

        mock_construct.side_effect = stripe.SignatureVerificationError(
            "bad sig", "sig_header"
        )
        response = self.client.post(
            self.url,
            data=b"{}",
            content_type="application/json",
            HTTP_STRIPE_SIGNATURE="t=1,v1=bad",
        )
        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

    @override_settings(STRIPE_WEBHOOK_SECRET="whsec_test")
    @patch("finance.views.dispatch_stripe_event")
    @patch("finance.views.stripe.Webhook.construct_event")
    def test_valid_event_dispatched(self, mock_construct, mock_dispatch):
        mock_construct.return_value = {
            "id": "evt_1",
            "type": "checkout.session.completed",
            "data": {"object": {"id": "cs_x"}},
        }
        response = self.client.post(
            self.url,
            data=b'{"id":"evt_1"}',
            content_type="application/json",
            HTTP_STRIPE_SIGNATURE="t=1,v1=good",
        )
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertTrue(response.data["received"])
        mock_dispatch.assert_called_once()