westfarn
e1e086a474
## Summary - Rename `finance` → **`monetization`** Django app (keep `finance_*` tables via `label = "finance"`) - Add `services/stripe.py` + `services/revenuecat.py`; RevenueCat webhook upserts **subscription + Invoice/Payment** (billing history parity with Stripe) - Mount `/api/monetization/` + keep `/api/finance/` alias - Extend `Source`/`Provider` with `revenuecat`; product→plan mapping via `revenuecat_product_id` / `REVENUECAT_PRODUCT_PLAN_MAP` Closes #68. Companion to [chat_web_app#100](ai_ml_operations/chat_web_app#100). ## Test plan - [x] `manage.py test monetization.tests` (54 OK) - [x] Smoke `chat_backend.tests.test_views_documents` + `test_oauth` - [ ] Deploy: set `REVENUECAT_WEBHOOK_SECRET`; point RC webhook at `/api/finance/webhooks/revenuecat/` - [ ] Map store product IDs on `SubscriptionPlan.revenuecat_product_id` (or env JSON map) - [ ] Sandbox INITIAL_PURCHASE → subscription `source=revenuecat` + invoice in `/finance/invoices/`Reviewed-on: #69
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 monetization.models import Invoice, Payment, UserSubscription
|
|
from monetization.services.plans import assign_plan_from_stripe, seed_subscription_plans
|
|
from monetization.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("monetization.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("monetization.views.dispatch_stripe_event")
|
|
@patch("monetization.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()
|