read.markets/tests/test_referral_conversion.py

"""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}"
    )