chat_backend/llm_be/monetization/tests/test_revenuecat.py

"""Tests for RevenueCat webhook auth, ledger upserts, and plan assignment."""

from __future__ import annotations

from django.test import TestCase, override_settings
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APIClient

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)


@override_settings(REVENUECAT_WEBHOOK_SECRET="test-rc-secret")
class RevenueCatWebhookViewTests(TestCase):
    def setUp(self):
        seed_subscription_plans(update_existing=False)
        self.client = APIClient()
        self.url = reverse("finance_revenuecat_webhook")
        self.user = make_user(email="rcview@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_missing_secret_config(self):
        with override_settings(REVENUECAT_WEBHOOK_SECRET=""):
            response = self.client.post(
                self.url, {"event": _rc_event()}, format="json"
            )
        self.assertEqual(response.status_code, status.HTTP_503_SERVICE_UNAVAILABLE)

    def test_unauthorized(self):
        response = self.client.post(
            self.url,
            {"event": _rc_event(app_user_id=str(self.user.pk))},
            format="json",
            HTTP_AUTHORIZATION="Bearer wrong",
        )
        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

    def test_success(self):
        response = self.client.post(
            self.url,
            {"api_version": "1.0", "event": _rc_event(app_user_id=str(self.user.pk))},
            format="json",
            HTTP_AUTHORIZATION="Bearer test-rc-secret",
        )
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertTrue(
            UserSubscription.objects.filter(
                user=self.user,
                source=UserSubscription.Source.REVENUECAT,
                status=UserSubscription.Status.ACTIVE,
            ).exists()
        )
        self.assertTrue(
            Invoice.objects.filter(
                user=self.user, provider=Invoice.Provider.REVENUECAT
            ).exists()
        )