"""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 chat_backend.models import UserAuthEvent from finance.models import Invoice, Payment, UserSubscription from finance.services.plans import assign_plan_from_stripe, seed_subscription_plans from finance.services.webhooks import ( dispatch_stripe_event, handle_checkout_session_completed, handle_customer_subscription_deleted, handle_customer_subscription_updated, 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) def test_checkout_assigns_plan_from_metadata(self): seed_subscription_plans(update_existing=False) session = { "id": "cs_test_plan_meta", "metadata": {"user_id": str(self.user.pk), "plan_slug": "founders"}, "customer": "cus_meta", "subscription": "sub_meta", "payment_intent": "pi_meta", "payment_status": "paid", "amount_total": 1000, "currency": "usd", } handle_checkout_session_completed(session) sub = UserSubscription.objects.get(user=self.user) self.assertEqual(sub.plan.slug, "founders") self.assertEqual(sub.source, UserSubscription.Source.STRIPE) self.assertEqual(sub.status, UserSubscription.Status.ACTIVE) started = UserAuthEvent.objects.get( user=self.user, event_type=UserAuthEvent.EventType.SUBSCRIPTION_STARTED, ) self.assertIn("founders", started.detail) def test_subscription_updated_sets_cancel_at_period_end(self): seed_subscription_plans(update_existing=False) assign_plan_from_stripe( self.user, plan_slug="founders", stripe_subscription_id="sub_cancel", ) result = handle_customer_subscription_updated( { "id": "sub_cancel", "status": "active", "cancel_at_period_end": True, "current_period_end": 1_700_259_200, "metadata": {"user_id": str(self.user.pk), "plan_slug": "founders"}, } ) self.assertIsNotNone(result) sub = UserSubscription.objects.get(user=self.user) self.assertTrue(sub.cancel_at_period_end) self.assertEqual(sub.status, UserSubscription.Status.ACTIVE) self.assertIsNotNone(sub.current_period_end) updated = UserAuthEvent.objects.filter( user=self.user, event_type=UserAuthEvent.EventType.SUBSCRIPTION_UPDATED, ).latest("created") self.assertIn("cancel_at_period_end=True", updated.detail) def test_subscription_deleted_marks_canceled(self): seed_subscription_plans(update_existing=False) assign_plan_from_stripe( self.user, plan_slug="founders", stripe_subscription_id="sub_gone", ) result = handle_customer_subscription_deleted( { "id": "sub_gone", "status": "canceled", "current_period_end": 1_700_259_200, "metadata": {"user_id": str(self.user.pk)}, } ) self.assertIsNotNone(result) sub = UserSubscription.objects.get(user=self.user) self.assertEqual(sub.status, UserSubscription.Status.CANCELED) self.assertFalse(sub.cancel_at_period_end) updated = UserAuthEvent.objects.filter( user=self.user, event_type=UserAuthEvent.EventType.SUBSCRIPTION_UPDATED, ).latest("created") self.assertIn("status=canceled", updated.detail) 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()