westfarn
74e8a4ad71
The app confirms a StoreKit signed transaction or Play purchase token. The backend checks it with Apple or Google before writing the ledger, and the RevenueCat webhook no longer grants access.
129 lines
4.7 KiB
Python
129 lines
4.7 KiB
Python
"""Tests for RevenueCat webhook auth, ledger upserts, and plan assignment."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from django.test import TestCase
|
|
|
|
from chat_backend.tests.factories import make_user
|
|
from monetization.models import Invoice, Payment, SubscriptionPlan, UserSubscription
|
|
from monetization.services.plans import seed_subscription_plans
|
|
from monetization.services.revenuecat import (
|
|
RevenueCatWebhookAuthError,
|
|
dispatch_revenuecat_event,
|
|
verify_revenuecat_authorization,
|
|
)
|
|
|
|
|
|
def _rc_event(**overrides):
|
|
base = {
|
|
"id": "rc_evt_1",
|
|
"type": "INITIAL_PURCHASE",
|
|
"app_user_id": "1",
|
|
"product_id": "hesychia_founders_monthly",
|
|
"store": "PLAY_STORE",
|
|
"price": 10.0,
|
|
"currency": "USD",
|
|
"purchased_at_ms": 1_700_000_000_000,
|
|
"expiration_at_ms": 1_702_592_000_000,
|
|
"transaction_id": "GPA.1234",
|
|
"original_transaction_id": "GPA.1234",
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
class RevenueCatAuthTests(TestCase):
|
|
def test_bearer_token_ok(self):
|
|
verify_revenuecat_authorization(
|
|
authorization_header="Bearer secret-token",
|
|
expected_secret="secret-token",
|
|
)
|
|
|
|
def test_raw_token_ok(self):
|
|
verify_revenuecat_authorization(
|
|
authorization_header="secret-token",
|
|
expected_secret="secret-token",
|
|
)
|
|
|
|
def test_bad_token(self):
|
|
with self.assertRaises(RevenueCatWebhookAuthError):
|
|
verify_revenuecat_authorization(
|
|
authorization_header="Bearer nope",
|
|
expected_secret="secret-token",
|
|
)
|
|
|
|
|
|
class RevenueCatDispatchTests(TestCase):
|
|
def setUp(self):
|
|
seed_subscription_plans(update_existing=False)
|
|
self.user = make_user(email="rc@example.com")
|
|
founders = SubscriptionPlan.objects.get(slug="founders")
|
|
founders.revenuecat_product_id = "hesychia_founders_monthly"
|
|
founders.save(update_fields=["revenuecat_product_id", "last_modified"])
|
|
|
|
def test_initial_purchase_assigns_plan_and_ledger(self):
|
|
event = _rc_event(app_user_id=str(self.user.pk))
|
|
dispatch_revenuecat_event({"api_version": "1.0", "event": event})
|
|
|
|
sub = UserSubscription.objects.get(user=self.user)
|
|
self.assertEqual(sub.source, UserSubscription.Source.REVENUECAT)
|
|
self.assertEqual(sub.status, UserSubscription.Status.ACTIVE)
|
|
self.assertEqual(sub.plan.slug, "founders")
|
|
self.assertEqual(sub.revenuecat_original_transaction_id, "GPA.1234")
|
|
|
|
invoice = Invoice.objects.get(revenuecat_event_id="rc_evt_1")
|
|
self.assertEqual(invoice.provider, Invoice.Provider.REVENUECAT)
|
|
self.assertEqual(invoice.status, Invoice.Status.PAID)
|
|
self.assertEqual(invoice.amount_paid, 1000)
|
|
self.assertEqual(invoice.revenuecat_store, "PLAY_STORE")
|
|
self.assertIn("PLAY_STORE", invoice.description)
|
|
|
|
payment = Payment.objects.get(revenuecat_transaction_id="GPA.1234")
|
|
self.assertEqual(payment.provider, Payment.Provider.REVENUECAT)
|
|
self.assertEqual(payment.status, Payment.Status.SUCCEEDED)
|
|
self.assertEqual(payment.invoice_id, invoice.pk)
|
|
|
|
def test_idempotent_replay(self):
|
|
event = _rc_event(app_user_id=str(self.user.pk))
|
|
dispatch_revenuecat_event({"event": event})
|
|
dispatch_revenuecat_event({"event": event})
|
|
self.assertEqual(Invoice.objects.filter(user=self.user).count(), 1)
|
|
self.assertEqual(Payment.objects.filter(user=self.user).count(), 1)
|
|
|
|
def test_cancellation_sets_cancel_at_period_end(self):
|
|
dispatch_revenuecat_event(
|
|
{"event": _rc_event(app_user_id=str(self.user.pk))}
|
|
)
|
|
dispatch_revenuecat_event(
|
|
{
|
|
"event": _rc_event(
|
|
id="rc_evt_cancel",
|
|
type="CANCELLATION",
|
|
app_user_id=str(self.user.pk),
|
|
transaction_id="GPA.999",
|
|
)
|
|
}
|
|
)
|
|
sub = UserSubscription.objects.get(user=self.user)
|
|
self.assertTrue(sub.cancel_at_period_end)
|
|
self.assertEqual(sub.status, UserSubscription.Status.ACTIVE)
|
|
|
|
def test_expiration_cancels(self):
|
|
dispatch_revenuecat_event(
|
|
{"event": _rc_event(app_user_id=str(self.user.pk))}
|
|
)
|
|
dispatch_revenuecat_event(
|
|
{
|
|
"event": _rc_event(
|
|
id="rc_evt_exp",
|
|
type="EXPIRATION",
|
|
app_user_id=str(self.user.pk),
|
|
transaction_id="GPA.exp",
|
|
)
|
|
}
|
|
)
|
|
sub = UserSubscription.objects.get(user=self.user)
|
|
self.assertEqual(sub.status, UserSubscription.Status.CANCELED)
|
|
|
|
|