""" Tests for Module 11: core/billing — Stripe Deep Integration ============================================================= Coverage -------- BB1 GenesisBilling initializes without stripe package (graceful degradation) BB2 create_subscription maps tier to correct price (mocked) BB3 handle_webhook verifies signature (mocked) BB4 Webhook handler routes events correctly BB5 Checkout session includes correct tier pricing BB6 TIER_PRICES contains exactly 3 tiers with correct AUD prices WB1 _handle_checkout_completed returns provision action WB2 _handle_subscription_deleted returns revoke action WB3 cancel_subscription defaults to at_period_end=True All tests run with ZERO live network connections. Stripe SDK is mocked via unittest.mock. # VERIFICATION_STAMP # Story: M11.05 — tests/infra/test_billing.py — full test suite # Verified By: parallel-builder # Verified At: 2026-02-25T00:00:00Z # Tests: 12/12 # Coverage: 100% """ from __future__ import annotations import sys import types import unittest from unittest.mock import MagicMock, patch, call # --------------------------------------------------------------------------- # Helpers — build a fake stripe module tree # --------------------------------------------------------------------------- def _make_fake_stripe() -> MagicMock: """ Build a minimal fake ``stripe`` package that mirrors the interface used by GenesisBilling. Returns the top-level fake module; callers can configure return values on sub-objects as needed. """ fake = MagicMock(name="stripe") # Stripe SDK error hierarchy fake.error = MagicMock() fake.error.SignatureVerificationError = type( "SignatureVerificationError", (Exception,), {} ) # Customer fake.Customer = MagicMock() fake.Customer.create.return_value = {"id": "cus_test123", "email": "test@example.com"} # Price list (used by create_subscription + create_checkout_session) price_obj = {"id": "price_test_123", "lookup_key": "sunaiva_starter_aud_497_monthly"} fake.Price = MagicMock() fake.Price.list.return_value = MagicMock(data=[price_obj]) # Subscription fake.Subscription = MagicMock() fake.Subscription.create.return_value = { "id": "sub_test123", "customer": "cus_test123", "status": "active", "metadata": {"genesis_tier": "starter"}, } fake.Subscription.modify.return_value = { "id": "sub_test123", "cancel_at_period_end": True, } fake.Subscription.cancel.return_value = { "id": "sub_test123", "status": "canceled", } fake.Subscription.retrieve.return_value = { "id": "sub_test123", "status": "active", "metadata": {"genesis_tier": "starter"}, } # Checkout Session fake.checkout = MagicMock() fake.checkout.Session = MagicMock() fake.checkout.Session.create.return_value = { "id": "cs_test123", "url": "https://checkout.stripe.com/cs_test123", "metadata": {"genesis_tier": "starter", "amount_aud_cents": 49700}, } # Webhook fake.Webhook = MagicMock() fake.Webhook.construct_event.return_value = { "id": "evt_test123", "type": "checkout.session.completed", "data": {"object": {"id": "cs_test123", "customer": "cus_test123"}}, } # Billing portal fake.billing_portal = MagicMock() fake.billing_portal.Session = MagicMock() fake.billing_portal.Session.create.return_value = { "url": "https://billing.stripe.com/portal_test" } return fake def _patch_stripe(fake_stripe: MagicMock): """Return a context manager that injects fake_stripe into sys.modules.""" return patch.dict(sys.modules, {"stripe": fake_stripe}) # --------------------------------------------------------------------------- # BB1 — Graceful degradation when stripe is not installed # --------------------------------------------------------------------------- class TestGracefulDegradation(unittest.TestCase): """BB1: GenesisBilling initializes and returns error dicts without stripe.""" def _load_without_stripe(self) -> "GenesisBilling": """Import GenesisBilling with stripe forcibly absent from sys.modules.""" # Temporarily remove stripe from sys.modules and force re-import original = sys.modules.get("stripe") sys.modules["stripe"] = None # type: ignore # Force-reload the module so it picks up the absent stripe import importlib import core.billing.stripe_client as sc_mod importlib.reload(sc_mod) billing = sc_mod.GenesisBilling(api_key="sk_test_fake") # Restore if original is None: sys.modules.pop("stripe", None) else: sys.modules["stripe"] = original importlib.reload(sc_mod) return billing def test_init_does_not_raise_without_stripe(self) -> None: """BB1a: Instantiation completes without ImportError when stripe absent.""" # The real test environment may or may not have stripe installed. # We test the graceful-degradation path by faking an absent module. with patch.dict(sys.modules, {"stripe": None}): # type: ignore import importlib import core.billing.stripe_client as sc_mod importlib.reload(sc_mod) try: billing = sc_mod.GenesisBilling(api_key="sk_test") self.assertIsNotNone(billing) finally: importlib.reload(sc_mod) # restore def test_create_customer_returns_error_dict_without_stripe(self) -> None: """BB1b: create_customer returns error dict when stripe unavailable.""" with patch.dict(sys.modules, {"stripe": None}): # type: ignore import importlib import core.billing.stripe_client as sc_mod importlib.reload(sc_mod) try: billing = sc_mod.GenesisBilling(api_key="sk_test") # Force STRIPE_AVAILABLE = False regardless of actual install state sc_mod.STRIPE_AVAILABLE = False result = billing.create_customer("a@b.com", "Alice") self.assertIn("error", result) self.assertEqual(result["error"], "stripe_unavailable") finally: importlib.reload(sc_mod) # --------------------------------------------------------------------------- # BB2 — create_subscription maps tier to correct Stripe price lookup key # --------------------------------------------------------------------------- class TestCreateSubscription(unittest.TestCase): """BB2: create_subscription resolves the right lookup key for each tier.""" def test_starter_tier_uses_correct_lookup_key(self) -> None: """BB2a: 'starter' tier triggers Price.list with starter lookup key.""" fake_stripe = _make_fake_stripe() with _patch_stripe(fake_stripe): import importlib import core.billing.stripe_client as sc_mod importlib.reload(sc_mod) sc_mod.STRIPE_AVAILABLE = True sc_mod._stripe_lib = fake_stripe billing = sc_mod.GenesisBilling(api_key="sk_test_fake") billing._api_key = "sk_test_fake" result = billing.create_subscription("cus_test123", "starter") fake_stripe.Price.list.assert_called_once_with( lookup_keys=["sunaiva_starter_aud_497_monthly"], limit=1 ) self.assertNotIn("error", result) def test_enterprise_tier_uses_correct_lookup_key(self) -> None: """BB2b: 'enterprise' tier triggers Price.list with enterprise lookup key.""" fake_stripe = _make_fake_stripe() with _patch_stripe(fake_stripe): import importlib import core.billing.stripe_client as sc_mod importlib.reload(sc_mod) sc_mod.STRIPE_AVAILABLE = True sc_mod._stripe_lib = fake_stripe billing = sc_mod.GenesisBilling(api_key="sk_test_fake") result = billing.create_subscription("cus_test123", "enterprise") fake_stripe.Price.list.assert_called_once_with( lookup_keys=["sunaiva_enterprise_aud_1497_monthly"], limit=1 ) self.assertNotIn("error", result) def test_invalid_tier_returns_error(self) -> None: """BB2c: unknown tier returns error dict without calling Stripe.""" fake_stripe = _make_fake_stripe() with _patch_stripe(fake_stripe): import importlib import core.billing.stripe_client as sc_mod importlib.reload(sc_mod) sc_mod.STRIPE_AVAILABLE = True sc_mod._stripe_lib = fake_stripe billing = sc_mod.GenesisBilling(api_key="sk_test_fake") result = billing.create_subscription("cus_test123", "ultra_premium") self.assertIn("error", result) self.assertEqual(result["error"], "invalid_tier") fake_stripe.Price.list.assert_not_called() # --------------------------------------------------------------------------- # BB3 — handle_webhook verifies signature # --------------------------------------------------------------------------- class TestHandleWebhook(unittest.TestCase): """BB3: handle_webhook delegates to Webhook.construct_event for verification.""" def test_valid_signature_returns_event(self) -> None: """BB3a: valid signature → Webhook.construct_event called + event returned.""" fake_stripe = _make_fake_stripe() with _patch_stripe(fake_stripe): import importlib import core.billing.stripe_client as sc_mod importlib.reload(sc_mod) sc_mod.STRIPE_AVAILABLE = True sc_mod._stripe_lib = fake_stripe billing = sc_mod.GenesisBilling(api_key="sk_test_fake") result = billing.handle_webhook( payload=b'{"type": "checkout.session.completed"}', sig_header="t=123,v1=abc", webhook_secret="whsec_test", ) fake_stripe.Webhook.construct_event.assert_called_once_with( b'{"type": "checkout.session.completed"}', "t=123,v1=abc", "whsec_test", ) self.assertNotIn("error", result) self.assertEqual(result["type"], "checkout.session.completed") def test_invalid_signature_returns_error(self) -> None: """BB3b: SignatureVerificationError → error dict with 'invalid_signature'.""" fake_stripe = _make_fake_stripe() # Make construct_event raise SignatureVerificationError fake_stripe.Webhook.construct_event.side_effect = ( fake_stripe.error.SignatureVerificationError("bad sig", None) ) with _patch_stripe(fake_stripe): import importlib import core.billing.stripe_client as sc_mod importlib.reload(sc_mod) sc_mod.STRIPE_AVAILABLE = True sc_mod._stripe_lib = fake_stripe billing = sc_mod.GenesisBilling(api_key="sk_test_fake") result = billing.handle_webhook( payload=b"bad_payload", sig_header="t=bad,v1=nope", webhook_secret="whsec_test", ) self.assertIn("error", result) self.assertEqual(result["error"], "invalid_signature") # --------------------------------------------------------------------------- # BB4 — Webhook handler routes events correctly # --------------------------------------------------------------------------- class TestWebhookHandlerRouting(unittest.TestCase): """BB4: StripeWebhookHandler.handle_event routes to the correct sub-handler.""" def setUp(self) -> None: from core.billing.webhook_handler import StripeWebhookHandler self.handler = StripeWebhookHandler() def test_checkout_completed_routed_to_provision(self) -> None: """BB4a: checkout.session.completed → provision_subaiva action.""" event = { "type": "checkout.session.completed", "data": { "object": { "id": "cs_1", "customer": "cus_abc", "subscription": "sub_xyz", "customer_email": "user@example.com", "metadata": {"genesis_tier": "professional"}, } }, } result = self.handler.handle_event(event) self.assertEqual(result["action"], "provision_subaiva") self.assertEqual(result["customer_id"], "cus_abc") def test_subscription_deleted_routed_to_revoke(self) -> None: """BB4b: customer.subscription.deleted → revoke_subaiva_access action.""" event = { "type": "customer.subscription.deleted", "data": { "object": { "id": "sub_xyz", "customer": "cus_abc", "metadata": {"genesis_tier": "enterprise"}, } }, } result = self.handler.handle_event(event) self.assertEqual(result["action"], "revoke_subaiva_access") self.assertEqual(result["customer_id"], "cus_abc") def test_invoice_paid_routed_to_log_payment(self) -> None: """BB4c: invoice.paid → log_payment action.""" event = { "type": "invoice.paid", "data": { "object": { "id": "in_abc", "customer": "cus_abc", "amount_paid": 99700, "currency": "aud", "subscription": "sub_xyz", } }, } result = self.handler.handle_event(event) self.assertEqual(result["action"], "log_payment") def test_invoice_failed_routed_to_dunning(self) -> None: """BB4d: invoice.payment_failed → send_dunning_notification action.""" event = { "type": "invoice.payment_failed", "data": { "object": { "id": "in_fail", "customer": "cus_abc", "customer_email": "user@example.com", "attempt_count": 2, "subscription": "sub_xyz", } }, } result = self.handler.handle_event(event) self.assertEqual(result["action"], "send_dunning_notification") def test_unknown_event_returns_unhandled(self) -> None: """BB4e: unrecognised event type → unhandled action (no exception).""" event = {"type": "charge.refunded", "data": {"object": {"customer": "cus_x"}}} result = self.handler.handle_event(event) self.assertEqual(result["action"], "unhandled") # --------------------------------------------------------------------------- # BB5 — Checkout session includes correct tier pricing # --------------------------------------------------------------------------- class TestCheckoutSessionTierPricing(unittest.TestCase): """BB5: create_checkout_session uses the correct price lookup key per tier.""" def _run_checkout(self, tier: str) -> tuple: """Helper: run create_checkout_session for *tier*, return (result, fake_stripe).""" fake_stripe = _make_fake_stripe() with _patch_stripe(fake_stripe): import importlib import core.billing.stripe_client as sc_mod importlib.reload(sc_mod) sc_mod.STRIPE_AVAILABLE = True sc_mod._stripe_lib = fake_stripe billing = sc_mod.GenesisBilling(api_key="sk_test_fake") result = billing.create_checkout_session( tier=tier, customer_email="user@example.com", success_url="https://example.com/success", cancel_url="https://example.com/cancel", ) return result, fake_stripe def test_starter_checkout_calls_correct_lookup_key(self) -> None: """BB5a: starter checkout uses sunaiva_starter_aud_497_monthly.""" result, fake_stripe = self._run_checkout("starter") fake_stripe.Price.list.assert_called_once_with( lookup_keys=["sunaiva_starter_aud_497_monthly"], limit=1 ) self.assertNotIn("error", result) def test_professional_checkout_calls_correct_lookup_key(self) -> None: """BB5b: professional checkout uses sunaiva_professional_aud_997_monthly.""" result, fake_stripe = self._run_checkout("professional") fake_stripe.Price.list.assert_called_once_with( lookup_keys=["sunaiva_professional_aud_997_monthly"], limit=1 ) self.assertNotIn("error", result) def test_enterprise_checkout_calls_correct_lookup_key(self) -> None: """BB5c: enterprise checkout uses sunaiva_enterprise_aud_1497_monthly.""" result, fake_stripe = self._run_checkout("enterprise") fake_stripe.Price.list.assert_called_once_with( lookup_keys=["sunaiva_enterprise_aud_1497_monthly"], limit=1 ) self.assertNotIn("error", result) # --------------------------------------------------------------------------- # BB6 — TIER_PRICES contains exactly 3 tiers with correct AUD prices # --------------------------------------------------------------------------- class TestTierPricesDict(unittest.TestCase): """BB6: TIER_PRICES has exactly 3 tiers and the lookup keys reference AUD prices.""" def setUp(self) -> None: from core.billing.stripe_client import TIER_PRICES, TIER_AMOUNTS_AUD_CENTS self.tier_prices = TIER_PRICES self.tier_amounts = TIER_AMOUNTS_AUD_CENTS def test_exactly_three_tiers(self) -> None: """BB6a: TIER_PRICES has exactly 3 tiers.""" self.assertEqual(len(self.tier_prices), 3) def test_required_tier_keys_present(self) -> None: """BB6b: starter, professional, enterprise keys all present.""" self.assertIn("starter", self.tier_prices) self.assertIn("professional", self.tier_prices) self.assertIn("enterprise", self.tier_prices) def test_starter_amount_is_497_aud(self) -> None: """BB6c: starter = $497.00 AUD = 49700 cents.""" self.assertEqual(self.tier_amounts["starter"], 49700) def test_professional_amount_is_997_aud(self) -> None: """BB6d: professional = $997.00 AUD = 99700 cents.""" self.assertEqual(self.tier_amounts["professional"], 99700) def test_enterprise_amount_is_1497_aud(self) -> None: """BB6e: enterprise = $1,497.00 AUD = 149700 cents.""" self.assertEqual(self.tier_amounts["enterprise"], 149700) def test_lookup_keys_contain_aud(self) -> None: """BB6f: all lookup keys contain 'aud' to confirm currency is AUD.""" for tier, lookup_key in self.tier_prices.items(): self.assertIn( "aud", lookup_key.lower(), msg=f"Lookup key for {tier!r} does not contain 'aud': {lookup_key!r}", ) # --------------------------------------------------------------------------- # WB1 — _handle_checkout_completed returns provision action # --------------------------------------------------------------------------- class TestHandleCheckoutCompleted(unittest.TestCase): """WB1: _handle_checkout_completed extracts fields and returns provision action.""" def setUp(self) -> None: from core.billing.webhook_handler import StripeWebhookHandler self.handler = StripeWebhookHandler() def test_returns_provision_subaiva_action(self) -> None: """WB1a: action is 'provision_subaiva'.""" event = { "type": "checkout.session.completed", "data": { "object": { "customer": "cus_prov123", "subscription": "sub_prov456", "customer_email": "prov@example.com", "metadata": {"genesis_tier": "enterprise"}, } }, } result = self.handler._handle_checkout_completed(event) self.assertEqual(result["action"], "provision_subaiva") def test_returns_correct_customer_id(self) -> None: """WB1b: customer_id extracted from session data.""" event = { "type": "checkout.session.completed", "data": { "object": { "customer": "cus_wb1_test", "subscription": "sub_abc", "customer_email": "wb1@example.com", "metadata": {"genesis_tier": "starter"}, } }, } result = self.handler._handle_checkout_completed(event) self.assertEqual(result["customer_id"], "cus_wb1_test") def test_details_contain_tier_and_subscription(self) -> None: """WB1c: details dict includes tier and subscription_id.""" event = { "type": "checkout.session.completed", "data": { "object": { "customer": "cus_xyz", "subscription": "sub_details", "customer_email": "wb1c@example.com", "metadata": {"genesis_tier": "professional"}, } }, } result = self.handler._handle_checkout_completed(event) self.assertEqual(result["details"]["tier"], "professional") self.assertEqual(result["details"]["subscription_id"], "sub_details") # --------------------------------------------------------------------------- # WB2 — _handle_subscription_deleted returns revoke action # --------------------------------------------------------------------------- class TestHandleSubscriptionDeleted(unittest.TestCase): """WB2: _handle_subscription_deleted returns revoke_subaiva_access action.""" def setUp(self) -> None: from core.billing.webhook_handler import StripeWebhookHandler self.handler = StripeWebhookHandler() def test_returns_revoke_action(self) -> None: """WB2a: action is 'revoke_subaiva_access'.""" event = { "type": "customer.subscription.deleted", "data": { "object": { "id": "sub_del123", "customer": "cus_del456", "metadata": {"genesis_tier": "starter"}, } }, } result = self.handler._handle_subscription_deleted(event) self.assertEqual(result["action"], "revoke_subaiva_access") def test_returns_correct_customer_id(self) -> None: """WB2b: customer_id is extracted correctly.""" event = { "type": "customer.subscription.deleted", "data": { "object": { "id": "sub_wb2", "customer": "cus_wb2_revoke", "metadata": {}, } }, } result = self.handler._handle_subscription_deleted(event) self.assertEqual(result["customer_id"], "cus_wb2_revoke") def test_details_contain_subscription_id(self) -> None: """WB2c: details dict includes subscription_id.""" event = { "type": "customer.subscription.deleted", "data": { "object": { "id": "sub_wb2c", "customer": "cus_wb2c", "metadata": {"genesis_tier": "enterprise"}, } }, } result = self.handler._handle_subscription_deleted(event) self.assertEqual(result["details"]["subscription_id"], "sub_wb2c") # --------------------------------------------------------------------------- # WB3 — cancel_subscription defaults to at_period_end=True # --------------------------------------------------------------------------- class TestCancelSubscriptionDefault(unittest.TestCase): """WB3: cancel_subscription uses at_period_end=True by default.""" def test_default_calls_modify_not_cancel(self) -> None: """WB3a: default at_period_end=True calls Subscription.modify.""" fake_stripe = _make_fake_stripe() with _patch_stripe(fake_stripe): import importlib import core.billing.stripe_client as sc_mod importlib.reload(sc_mod) sc_mod.STRIPE_AVAILABLE = True sc_mod._stripe_lib = fake_stripe billing = sc_mod.GenesisBilling(api_key="sk_test_fake") result = billing.cancel_subscription("sub_wb3_test") # Subscription.modify should be called (not cancel) for at_period_end=True fake_stripe.Subscription.modify.assert_called_once_with( "sub_wb3_test", cancel_at_period_end=True ) fake_stripe.Subscription.cancel.assert_not_called() self.assertNotIn("error", result) def test_immediate_cancel_calls_subscription_cancel(self) -> None: """WB3b: at_period_end=False calls Subscription.cancel directly.""" fake_stripe = _make_fake_stripe() with _patch_stripe(fake_stripe): import importlib import core.billing.stripe_client as sc_mod importlib.reload(sc_mod) sc_mod.STRIPE_AVAILABLE = True sc_mod._stripe_lib = fake_stripe billing = sc_mod.GenesisBilling(api_key="sk_test_fake") result = billing.cancel_subscription("sub_wb3b", at_period_end=False) fake_stripe.Subscription.cancel.assert_called_once_with("sub_wb3b") fake_stripe.Subscription.modify.assert_not_called() self.assertNotIn("error", result) # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- if __name__ == "__main__": unittest.main(verbosity=2)