westfarn
d01f3a2696
Add account_deleted, subscription_started, and subscription_updated event types. Soft-delete, Checkout/Backer assign, and Stripe portal lifecycle webhooks write audit rows visible on the user admin.
239 lines
9.2 KiB
Python
239 lines
9.2 KiB
Python
"""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()
|