"""Founding-price campaign: slot cap, lifecycle, checkout wiring, webhook confirm/release.""" from datetime import datetime, timedelta, timezone from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock from uuid import uuid4 import pytest from sqlalchemy.exc import ProgrammingError from src.core.exceptions import ValidationError from src.domain.billing.campaign_service import ( RESERVATION_TTL, PricingCampaignService, email_subject_key, provision_campaign_price, workspace_subject_key, ) from src.domain.billing.service import BillingService, STRIPE_EVENT_HANDLERS NOW = datetime(2026, 9, 21, 12, 0, tzinfo=timezone.utc) class _NullSavepoint: async def __aenter__(self): return self async def __aexit__(self, *exc): return False def missing_table_error(): return ProgrammingError( "SELECT ... FROM pricing_campaigns", {}, Exception('relation "pricing_campaigns" does not exist') ) class FakeCampaignRepo: """In-memory stand-in mirroring PricingCampaignRepository semantics.""" def __init__(self, max_redemptions=2, is_active=True): self.campaign = SimpleNamespace( id=uuid4(), slug="founding", name="Founding Customer", max_redemptions=max_redemptions, is_active=is_active, starts_at=None, ends_at=None, ) self.prices = [] self.redemptions = [] self.locked = 0 self.error = None def add_price(self, plan, interval, minor, stripe_id=None): row = SimpleNamespace( id=uuid4(), campaign_id=self.campaign.id, subscription_plan_id=plan.id, interval=interval, stripe_price_id=stripe_id or f"price_f_{plan.slug}_{interval}", price_minor=minor, currency="usd", ) self.prices.append(row) return row def savepoint(self): return _NullSavepoint() async def get_by_slug(self, slug): if self.error is not None: raise self.error return self.campaign if slug == self.campaign.slug else None async def lock_campaign(self, campaign_id): self.locked += 1 async def get_price(self, campaign_id, plan_id, interval): return next( (p for p in self.prices if p.subscription_plan_id == plan_id and p.interval == interval), None ) async def list_prices(self, campaign_id): return list(self.prices) async def count_live(self, campaign_id, now): return sum( 1 for r in self.redemptions if r.status == "confirmed" or (r.status == "reserved" and (r.expires_at is None or r.expires_at > now)) ) async def get_live_for_subject(self, campaign_id, subject_key): return next( (r for r in self.redemptions if r.subject_key == subject_key and r.status in ("reserved", "confirmed")), None ) async def create_redemption(self, *, campaign_id, campaign_price_id, subject_key, workspace_id, expires_at): row = SimpleNamespace( id=uuid4(), campaign_id=campaign_id, campaign_price_id=campaign_price_id, subject_key=subject_key, workspace_id=workspace_id, status="reserved", stripe_checkout_session_id=None, stripe_subscription_id=None, expires_at=expires_at, confirmed_at=None, ) self.redemptions.append(row) return row async def get_redemption(self, redemption_id): return next((r for r in self.redemptions if r.id == redemption_id), None) async def get_redemption_by_session(self, session_id): return next((r for r in self.redemptions if r.stripe_checkout_session_id == session_id), None) async def create_campaign(self, *, slug, name, max_redemptions): raise NotImplementedError async def create_price(self, *, campaign_id, subscription_plan_id, interval, stripe_price_id, price_minor, currency): row = SimpleNamespace( id=uuid4(), campaign_id=campaign_id, subscription_plan_id=subscription_plan_id, interval=interval, stripe_price_id=stripe_price_id, price_minor=price_minor, currency=currency, ) self.prices.append(row) return row async def flush(self): return None def plan(slug="base"): return SimpleNamespace( id=uuid4(), slug=slug, name=slug.title(), currency="usd", stripe_product_id=f"prod_{slug}" ) def make(max_redemptions=2): repo = FakeCampaignRepo(max_redemptions) base, pro = plan("base"), plan("pro") repo.add_price(base, "month", 2000) repo.add_price(base, "year", 20000) repo.add_price(pro, "month", 6000) return repo, PricingCampaignService(repo), base, pro # --- slot lifecycle ------------------------------------------------------------------------- @pytest.mark.asyncio async def test_reserve_returns_campaign_price_and_takes_a_slot(): repo, svc, base, _ = make() res = await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) assert res.stripe_price_id == "price_f_base_month" assert res.price_minor == 2000 assert repo.locked == 1 assert await repo.count_live(repo.campaign.id, NOW) == 1 @pytest.mark.asyncio async def test_cap_is_one_pool_shared_across_plans_and_intervals(): repo, svc, base, pro = make(max_redemptions=2) assert await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) assert await svc.reserve(plan=pro, interval="month", subject_key="ws:b", now=NOW) # Pool is exhausted even though this is a different plan/interval with its own price. assert await svc.reserve(plan=base, interval="year", subject_key="ws:c", now=NOW) is None @pytest.mark.asyncio async def test_retry_reuses_the_same_slot_and_can_switch_price(): repo, svc, base, pro = make(max_redemptions=1) first = await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) again = await svc.reserve(plan=pro, interval="month", subject_key="ws:a", now=NOW) assert again.redemption_id == first.redemption_id assert again.stripe_price_id == "price_f_pro_month" assert len(repo.redemptions) == 1 assert await svc.reserve(plan=base, interval="month", subject_key="ws:b", now=NOW) is None @pytest.mark.asyncio async def test_plan_or_interval_without_campaign_price_gets_catalog_price(): repo, svc, base, pro = make() assert await svc.reserve(plan=pro, interval="year", subject_key="ws:a", now=NOW) is None assert await svc.reserve(plan=plan("enterprise"), interval="month", subject_key="ws:a", now=NOW) is None assert repo.redemptions == [] @pytest.mark.asyncio async def test_closed_campaign_gives_no_price(): repo, svc, base, _ = make() repo.campaign.is_active = False assert await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) is None repo.campaign.is_active = True repo.campaign.ends_at = NOW - timedelta(seconds=1) assert await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) is None repo.campaign.ends_at = None repo.campaign.starts_at = NOW + timedelta(days=1) assert await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) is None @pytest.mark.asyncio async def test_expired_reservation_stops_counting_and_can_be_reclaimed(): repo, svc, base, _ = make(max_redemptions=1) assert await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) later = NOW + RESERVATION_TTL + timedelta(minutes=1) assert await svc.reserve(plan=base, interval="month", subject_key="ws:b", now=later) # The abandoned subject's lapsed row must not skip the cap now that ws:b holds the slot. assert await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=later) is None @pytest.mark.asyncio async def test_lapsed_reservation_can_be_revived_when_a_slot_is_free(): repo, svc, base, _ = make(max_redemptions=1) first = await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) later = NOW + RESERVATION_TTL + timedelta(minutes=1) revived = await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=later) assert revived.redemption_id == first.redemption_id assert len(repo.redemptions) == 1 assert await repo.count_live(repo.campaign.id, later) == 1 @pytest.mark.asyncio async def test_confirm_is_permanent_idempotent_and_blocks_a_second_founding_price(): repo, svc, base, _ = make(max_redemptions=5) res = await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) await svc.attach_session(res.redemption_id, "cs_1") ws = uuid4() assert await svc.confirm(redemption_id=res.redemption_id, session_id="cs_1", subscription_id="sub_1", workspace_id=ws) assert await svc.confirm(redemption_id=res.redemption_id, session_id="cs_1", subscription_id="sub_1", workspace_id=ws) row = repo.redemptions[0] assert (row.status, row.expires_at, row.stripe_subscription_id, row.workspace_id) == ( "confirmed", None, "sub_1", ws, ) # A returning founder who cancelled and re-subscribes pays catalog. assert await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) is None @pytest.mark.asyncio async def test_confirm_unknown_redemption_is_a_noop(): _, svc, _, _ = make() assert not await svc.confirm(redemption_id=uuid4(), session_id="cs_x", subscription_id=None) @pytest.mark.asyncio async def test_release_frees_the_slot_only_for_the_current_session(): repo, svc, base, _ = make(max_redemptions=1) res = await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) await svc.attach_session(res.redemption_id, "cs_old") # Re-targeted checkout points the slot at a new session; the old one expiring must not free it. again = await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) await svc.attach_session(again.redemption_id, "cs_new") assert not await svc.release_session("cs_old") assert await svc.reserve(plan=base, interval="month", subject_key="ws:b", now=NOW) is None assert await svc.release_session("cs_new") assert await svc.reserve(plan=base, interval="month", subject_key="ws:b", now=NOW) @pytest.mark.asyncio async def test_release_does_not_undo_a_confirmed_slot(): repo, svc, base, _ = make() res = await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) await svc.attach_session(res.redemption_id, "cs_1") await svc.confirm(redemption_id=res.redemption_id, session_id="cs_1", subscription_id="sub_1") assert not await svc.release_session("cs_1") assert repo.redemptions[0].status == "confirmed" # --- display -------------------------------------------------------------------------------- @pytest.mark.asyncio async def test_offers_show_prices_and_spots_remaining(): repo, svc, base, pro = make(max_redemptions=3) await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) offers = await svc.offers_for_plans([base, pro, plan("free")], now=NOW) assert set(offers) == {base.id, pro.id} assert (offers[base.id].price_monthly, offers[base.id].price_yearly) == (2000, 20000) assert (offers[pro.id].price_monthly, offers[pro.id].price_yearly) == (6000, None) assert offers[base.id].spots_remaining == 2 @pytest.mark.asyncio async def test_offers_disappear_when_sold_out_or_subject_already_a_founder(): repo, svc, base, _ = make(max_redemptions=1) res = await svc.reserve(plan=base, interval="month", subject_key="ws:a", now=NOW) assert await svc.offers_for_plans([base], now=NOW) == {} repo.campaign.max_redemptions = 5 await svc.confirm(redemption_id=res.redemption_id, session_id=None, subscription_id="sub_1") assert await svc.offers_for_plans([base], subject_key="ws:a", now=NOW) == {} assert await svc.offers_for_plans([base], subject_key="ws:other", now=NOW) # --- provisioning --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_provision_creates_price_and_registers_non_current_version_once(): repo = FakeCampaignRepo() base = plan("base") stripe = AsyncMock() stripe.create_price.return_value = {"id": "price_new", "unit_amount": 2000} versions = AsyncMock() row, created = await provision_campaign_price( repo=repo, price_version_repo=versions, stripe=stripe, campaign=repo.campaign, plan=base, interval="month", amount_minor=2000, ) assert created and row.stripe_price_id == "price_new" assert versions.create.await_args.kwargs["is_current"] is False again, created = await provision_campaign_price( repo=repo, price_version_repo=versions, stripe=stripe, campaign=repo.campaign, plan=base, interval="month", amount_minor=2000, ) assert not created and again is row stripe.create_price.assert_awaited_once() with pytest.raises(ValidationError, match="already has a campaign price"): await provision_campaign_price( repo=repo, price_version_repo=versions, stripe=stripe, campaign=repo.campaign, plan=base, interval="month", amount_minor=2500, ) # --- checkout + webhook wiring -------------------------------------------------------------- def billing_service(repo, existing_sub=None): catalog = MagicMock() base = plan("base") catalog.get_plan_by_slug = AsyncMock(return_value=base) catalog.resolve_stripe_price_id_for_plan.return_value = "price_catalog" svc = BillingService( customer_repo=AsyncMock(), subscription_repo=AsyncMock(), invoice_repo=AsyncMock(), event_repo=AsyncMock(), usage_repo=AsyncMock(), purchase_repo=AsyncMock(), ledger_repo=AsyncMock(), catalog=catalog, stripe_client=AsyncMock(), campaign_service=PricingCampaignService(repo), ) svc._subscription_repo.get_latest_for_workspace.return_value = existing_sub svc._customer_repo.get_by_workspace.return_value = SimpleNamespace(stripe_customer_id="cus_1") svc._stripe.create_subscription_checkout.return_value = ("https://stripe.test/c", "cs_1") return svc, base async def start(svc, workspace_id, plan_slug="base", interval="month", key=None): return await svc.start_subscription_checkout( workspace_id, uuid4(), "a@b.co", "WS", plan_slug, interval, "https://app.test/ok", "https://app.test/no", key, ) @pytest.mark.asyncio async def test_checkout_uses_campaign_price_and_tags_session_metadata(): repo = FakeCampaignRepo() svc, base = billing_service(repo) repo.add_price(base, "month", 2000, stripe_id="price_founding") ws = uuid4() await start(svc, ws, key="idem-1") args = svc._stripe.create_subscription_checkout.await_args.args assert args[1] == "price_founding" metadata = args[4] row = repo.redemptions[0] assert metadata["campaign_redemption_id"] == str(row.id) assert metadata["campaign_slug"] == "founding" assert row.subject_key == workspace_subject_key(ws) assert row.stripe_checkout_session_id == "cs_1" # A price-specific key stops a replayed key from returning a session at the old price. assert args[5] == "idem-1:price_founding" @pytest.mark.asyncio async def test_checkout_falls_back_to_catalog_price_when_sold_out(): repo = FakeCampaignRepo(max_redemptions=0) svc, base = billing_service(repo) repo.add_price(base, "month", 2000) await start(svc, uuid4(), key="idem-1") args = svc._stripe.create_subscription_checkout.await_args.args assert args[1] == "price_catalog" assert "campaign_redemption_id" not in args[4] assert args[5] == "idem-1" @pytest.mark.asyncio async def test_plan_change_for_paying_workspace_uses_catalog_price_and_takes_no_slot(): repo = FakeCampaignRepo() svc, base = billing_service(repo, SimpleNamespace(stripe_subscription_id="sub_1", status="active")) repo.add_price(base, "month", 2000) svc._stripe.create_subscription_update_portal.return_value = "https://portal" result = await start(svc, uuid4()) assert result["checkout_url"] == "https://portal" svc._stripe.create_subscription_update_portal.assert_awaited_once() assert svc._stripe.create_subscription_update_portal.await_args.args[2] == "price_catalog" assert repo.redemptions == [] # The paying workspace is not shown a campaign price it could not get... assert await svc.get_plan_offers(uuid4(), [base]) == {} # ...while a workspace with no paid subscription (or a cancelled one) is. for no_paid_sub in (None, SimpleNamespace(stripe_subscription_id="sub_1", status="canceled")): svc._subscription_repo.get_latest_for_workspace.return_value = no_paid_sub assert base.id in await svc.get_plan_offers(uuid4(), [base]) @pytest.mark.asyncio async def test_checkout_completed_confirms_slot_even_before_workspace_exists(): repo = FakeCampaignRepo() svc, base = billing_service(repo) repo.add_price(base, "month", 2000) reservation = await svc.reserve_campaign_price(base, "month", email_subject_key("New@Site.com")) await svc._campaigns.attach_session(reservation.redemption_id, "cs_pre") svc._customer_repo.get_by_stripe_id.return_value = None # pre-registration: no workspace yet event = {"object": { "id": "cs_pre", "customer": "cus_9", "subscription": "sub_9", "payment_status": "paid", "metadata": { "purchase_type": "subscription", "campaign_redemption_id": str(reservation.redemption_id), }, }} await STRIPE_EVENT_HANDLERS["checkout.session.completed"](svc, event, "checkout.session.completed") row = repo.redemptions[0] assert row.status == "confirmed" and row.stripe_subscription_id == "sub_9" assert row.subject_key == "email:new@site.com" @pytest.mark.asyncio async def test_unpaid_checkout_does_not_confirm_slot(): repo = FakeCampaignRepo() svc, base = billing_service(repo) repo.add_price(base, "month", 2000) reservation = await svc.reserve_campaign_price(base, "month", "ws:a") svc._customer_repo.get_by_stripe_id.return_value = None event = {"object": { "id": "cs_1", "customer": "cus_1", "payment_status": "unpaid", "metadata": {"campaign_redemption_id": str(reservation.redemption_id)}, }} await STRIPE_EVENT_HANDLERS["checkout.session.completed"](svc, event, "checkout.session.completed") assert repo.redemptions[0].status == "reserved" @pytest.mark.asyncio async def test_expired_checkout_session_releases_slot(): repo = FakeCampaignRepo(max_redemptions=1) svc, base = billing_service(repo) repo.add_price(base, "month", 2000) await start(svc, uuid4()) assert repo.redemptions[0].status == "reserved" await STRIPE_EVENT_HANDLERS["checkout.session.expired"]( svc, {"object": {"id": "cs_1"}}, "checkout.session.expired" ) assert repo.redemptions[0].status == "released" assert await svc._campaigns.reserve(plan=base, interval="month", subject_key="ws:new") @pytest.mark.asyncio async def test_unmigrated_campaign_tables_fall_back_to_catalog_pricing(): """Code can deploy before migration 168; pricing pages and checkout must not break.""" repo = FakeCampaignRepo() svc, base = billing_service(repo) repo.add_price(base, "month", 2000) repo.error = missing_table_error() assert await svc.get_plan_offers(uuid4(), [base]) == {} assert await svc._campaigns.offers_for_plans([base]) == {} await start(svc, uuid4()) args = svc._stripe.create_subscription_checkout.await_args.args assert args[1] == "price_catalog" assert "campaign_redemption_id" not in args[4] @pytest.mark.asyncio async def test_unrelated_database_errors_are_not_swallowed(): repo = FakeCampaignRepo() _, svc, base, _ = make() repo.error = ProgrammingError("SELECT 1 FROM other_table", {}, Exception("syntax error")) svc = PricingCampaignService(repo) with pytest.raises(ProgrammingError): await svc.offers_for_plans([base]) with pytest.raises(ProgrammingError): await svc.reserve(plan=base, interval="month", subject_key="ws:a") @pytest.mark.asyncio async def test_service_without_campaigns_is_unchanged(): svc, base = billing_service(FakeCampaignRepo()) svc._campaigns = None await start(svc, uuid4()) assert svc._stripe.create_subscription_checkout.await_args.args[1] == "price_catalog" assert await svc.get_plan_offers(uuid4(), [base]) == {}