"""DB-backed tests for the referral-conversion path: when a referred user becomes paid, both parties should get REFERRAL_CREDIT_DAYS of extra credit, the Referral row should be stamped, and the operation should be idempotent so renewal events don't double-credit. Also exercises the Stripe-webhook integration to confirm the credit actually lands when a subscription.created event fires for a referred user.""" from __future__ import annotations import asyncio import hashlib import hmac import json import time from datetime import datetime, timedelta, timezone import pytest # --------------------------------------------------------------------------- # Shared scaffolding # --------------------------------------------------------------------------- def _build_session_factory(tmp_path): """Spin up a fresh in-memory schema and return (engine, factory). Mirrors test_stripe_billing._build_app's seeding strategy but skips the FastAPI app — most conversion tests only need the session factory.""" from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app import db as db_mod from app.db import Base engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/conv.db") factory = async_sessionmaker(engine, expire_on_commit=False) db_mod._engine = engine db_mod._session_factory = factory async def _seed(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) asyncio.run(_seed()) return factory async def _add_pair(factory, *, referrer_id=1, referred_id=2): """Insert a referrer + referred user pair and a linking Referral row. Returns nothing — tests re-fetch via the factory.""" from app.db import utcnow from app.models import Referral, User async with factory() as s: s.add(User(id=referrer_id, email=f"r{referrer_id}@x", tier="paid", referral_code="REFREFXX")) s.add(User(id=referred_id, email=f"u{referred_id}@x", tier="free", referred_by_user_id=referrer_id)) s.add(Referral(referrer_user_id=referrer_id, referred_user_id=referred_id, created_at=utcnow())) await s.commit() # --------------------------------------------------------------------------- # convert_referral happy path # --------------------------------------------------------------------------- def test_first_conversion_credits_both_parties(tmp_path): """Calling convert_referral on a freshly-paid referred user should extend credit_until by REFERRAL_CREDIT_DAYS for BOTH the buyer and the referrer, and stamp converted_at + credited_at.""" from app.models import Referral, User from app.services.referral_service import ( REFERRAL_CREDIT_DAYS, convert_referral, ) factory = _build_session_factory(tmp_path) asyncio.run(_add_pair(factory)) async def _run(): async with factory() as s: referred = await s.get(User, 2) ref = await convert_referral(s, referred) assert ref is not None assert ref.converted_at is not None assert ref.credited_at is not None await s.commit() # Re-open a fresh session so we read committed state, not the # session-cached version. async with factory() as s: referrer = await s.get(User, 1) referred = await s.get(User, 2) now = datetime.now(timezone.utc) # Both windows should sit ~REFERRAL_CREDIT_DAYS in the # future (allow 1 day slack for clock + rounding). for u in (referrer, referred): assert u.credit_until is not None cu = u.credit_until if cu.tzinfo is None: cu = cu.replace(tzinfo=timezone.utc) delta_days = (cu - now).total_seconds() / 86400 assert REFERRAL_CREDIT_DAYS - 1 <= delta_days <= REFERRAL_CREDIT_DAYS + 1 asyncio.run(_run()) def test_idempotent_on_repeat_call(tmp_path): """A second convert_referral call (e.g. from a duplicate webhook or renewal event) must NOT extend credit a second time. The Referral row is already stamped, so we should early-return unchanged.""" from app.models import User from app.services.referral_service import convert_referral factory = _build_session_factory(tmp_path) asyncio.run(_add_pair(factory)) async def _run(): async with factory() as s: referred = await s.get(User, 2) await convert_referral(s, referred) await s.commit() # Snapshot credit_until after first conversion. async with factory() as s: referrer = await s.get(User, 1) referred = await s.get(User, 2) first_referrer_credit = referrer.credit_until first_referred_credit = referred.credit_until # Second call — should no-op. async with factory() as s: referred = await s.get(User, 2) ref2 = await convert_referral(s, referred) assert ref2 is not None # we still return the row await s.commit() async with factory() as s: referrer = await s.get(User, 1) referred = await s.get(User, 2) assert referrer.credit_until == first_referrer_credit assert referred.credit_until == first_referred_credit asyncio.run(_run()) def test_no_referral_row_returns_none(tmp_path): """A user signing up directly (no inviter) has no Referral row. convert_referral must return None and touch nothing.""" from app.models import User from app.services.referral_service import convert_referral factory = _build_session_factory(tmp_path) async def _seed_orphan(): async with factory() as s: s.add(User(id=9, email="lone@x", tier="free")) await s.commit() asyncio.run(_seed_orphan()) async def _run(): async with factory() as s: user = await s.get(User, 9) result = await convert_referral(s, user) assert result is None assert user.credit_until is None asyncio.run(_run()) def test_credit_stacks_from_existing_window(tmp_path): """If the user already has a future credit_until (admin grant, prior referral), the new credit should extend from THAT anchor — not from now. Mirrors cli.grant_credit's stacking semantics.""" from app.models import User from app.services.referral_service import ( REFERRAL_CREDIT_DAYS, convert_referral, ) factory = _build_session_factory(tmp_path) asyncio.run(_add_pair(factory)) # Pre-load 30 days of credit on the referred user. existing = datetime.now(timezone.utc) + timedelta(days=30) async def _preload(): async with factory() as s: u = await s.get(User, 2) u.credit_until = existing await s.commit() asyncio.run(_preload()) async def _run(): async with factory() as s: referred = await s.get(User, 2) await convert_referral(s, referred) await s.commit() async with factory() as s: referred = await s.get(User, 2) cu = referred.credit_until if cu.tzinfo is None: cu = cu.replace(tzinfo=timezone.utc) # Expected: existing + REFERRAL_CREDIT_DAYS days, not now + days. expected = existing + timedelta(days=REFERRAL_CREDIT_DAYS) delta_seconds = abs((cu - expected).total_seconds()) assert delta_seconds < 60, ( f"new credit anchored at now, not existing window: " f"got {cu}, expected ~{expected}" ) asyncio.run(_run()) def test_deleted_referrer_does_not_crash(tmp_path): """If the referrer's User row has been deleted, the referred user should still be credited and the Referral still stamped — we just skip the missing referrer.""" from app.models import Referral, User from app.services.referral_service import convert_referral factory = _build_session_factory(tmp_path) async def _seed(): from app.db import utcnow async with factory() as s: # Referrer with FK SET NULL — we don't delete the row, we # instead create a Referral pointing at a non-existent id # to simulate a deleted referrer. s.add(User(id=2, email="u2@x", tier="free")) s.add(Referral(referrer_user_id=999, # nonexistent referred_user_id=2, created_at=utcnow())) await s.commit() asyncio.run(_seed()) async def _run(): async with factory() as s: referred = await s.get(User, 2) ref = await convert_referral(s, referred) await s.commit() assert ref is not None assert ref.converted_at is not None # Referred still got their credit even though referrer is gone. assert referred.credit_until is not None asyncio.run(_run()) # --------------------------------------------------------------------------- # Stripe-webhook integration # --------------------------------------------------------------------------- _WEBHOOK_SECRET = "whsec_dummy_test_secret_for_unit_tests" def _stripe_sig(body: bytes, secret: str, ts: int | None = None) -> str: ts = ts if ts is not None else int(time.time()) signed = f"{ts}.{body.decode('utf-8')}" mac = hmac.new(secret.encode("utf-8"), signed.encode("utf-8"), hashlib.sha256).hexdigest() return f"t={ts},v1={mac}" def _build_webhook_app(tmp_path): """Same as the helper in test_stripe_billing but pre-seeds a referrer+referred pair so the webhook can drive a full conversion.""" from fastapi import FastAPI from fastapi.testclient import TestClient from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app import db as db_mod from app.config import get_settings from app.db import Base, utcnow from app.models import Referral, User from app.routers import stripe_billing as stripe_router s = get_settings() s.STRIPE_API_KEY = "sk_test_dummy" # type: ignore[misc] s.STRIPE_WEBHOOK_SECRET = _WEBHOOK_SECRET # type: ignore[misc] s.STRIPE_PRICE_MONTHLY = "price_test_monthly_xxxxxxxxxxxxxxxxxxxx" # type: ignore[misc] s.STRIPE_PRICE_ANNUAL = "price_test_annual_xxxxxxxxxxxxxxxxxxxxx" # type: ignore[misc] engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/conv-int.db") factory = async_sessionmaker(engine, expire_on_commit=False) db_mod._engine = engine db_mod._session_factory = factory async def _seed(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) async with factory() as session: session.add(User(id=1, email="referrer@x", tier="paid", referral_code="REFCODE1")) session.add(User(id=2, email="buyer@x", tier="free", stripe_customer_id="cus_test_referred", referred_by_user_id=1)) session.add(Referral(referrer_user_id=1, referred_user_id=2, created_at=utcnow())) await session.commit() asyncio.run(_seed()) app = FastAPI() app.include_router(stripe_router.router) return TestClient(app), factory def _post_webhook(client, body: dict): body.setdefault("object", "event") raw = json.dumps(body).encode("utf-8") sig = _stripe_sig(raw, _WEBHOOK_SECRET) return client.post( "/api/stripe/webhook", content=raw, headers={"stripe-signature": sig, "content-type": "application/json"}, ) def test_subscription_active_event_triggers_referral_conversion(tmp_path): """End-to-end: fire a customer.subscription.created event for the referred user. The webhook should flip their tier AND extend credit_until on both parties via convert_referral.""" from app.models import Referral, User from app.services.referral_service import REFERRAL_CREDIT_DAYS client, factory = _build_webhook_app(tmp_path) body = { "id": "evt_conv_1", "type": "customer.subscription.created", "data": {"object": { "id": "sub_test_1", "customer": "cus_test_referred", "status": "active", "trial_end": None, }}, } r = _post_webhook(client, body) assert r.status_code == 200, r.text assert r.json()["status"] == "ok" async def _check(): async with factory() as s: referred = await s.get(User, 2) referrer = await s.get(User, 1) assert referred.tier == "paid" now = datetime.now(timezone.utc) for u in (referred, referrer): assert u.credit_until is not None, ( f"user {u.id} got no credit" ) cu = u.credit_until if cu.tzinfo is None: cu = cu.replace(tzinfo=timezone.utc) delta_days = (cu - now).total_seconds() / 86400 assert REFERRAL_CREDIT_DAYS - 1 <= delta_days <= REFERRAL_CREDIT_DAYS + 1, ( f"user {u.id} credit days={delta_days} not ~{REFERRAL_CREDIT_DAYS}" ) # Referral row stamped. from sqlalchemy import select ref = (await s.execute( select(Referral).where(Referral.referred_user_id == 2) )).scalar_one() assert ref.converted_at is not None assert ref.credited_at is not None asyncio.run(_check()) def test_renewal_event_does_not_double_credit(tmp_path): """A subscription.updated event firing later (renewal, status change) on an already-paid user must NOT re-trigger conversion. The first_paid_transition guard inside _grant_paid should skip the convert_referral lookup; convert_referral itself is also idempotent as a second line of defence.""" from app.models import User client, factory = _build_webhook_app(tmp_path) # First event: created → conversion fires. _post_webhook(client, { "id": "evt_first", "type": "customer.subscription.created", "data": {"object": { "id": "sub_test_1", "customer": "cus_test_referred", "status": "active", }}, }) async def _snapshot(): async with factory() as s: referred = await s.get(User, 2) referrer = await s.get(User, 1) return referred.credit_until, referrer.credit_until before = asyncio.run(_snapshot()) # Second event: updated → must not extend credit further. r = _post_webhook(client, { "id": "evt_second", "type": "customer.subscription.updated", "data": {"object": { "id": "sub_test_1", "customer": "cus_test_referred", "status": "active", }}, }) assert r.status_code == 200 after = asyncio.run(_snapshot()) assert before == after, ( f"renewal event extended credit: before={before}, after={after}" )