FabledCurator/tests/test_membership_sync.py

"""Milestone 387 C3: the roster sweep, and the state that makes it readable.

The failure this step must not have is an empty roster that looks like an
answer. Three situations produce zero rows in `platform_membership` — the
account subscribes to nothing, the sweep never ran, the sweep failed — and
telling the operator "you are tracking 12 sources you no longer subscribe to"
is correct in the first and catastrophic in the other two. So most of what
follows pins the difference between "we know" and "we do not know".
"""
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select

from backend.app.models import MembershipSync, PlatformMembership
from backend.app.services.membership_roster import (
    ROSTER_STALE_AFTER,
    get_sync_state,
    roster_is_fresh,
    sync_platform,
)

pytestmark = pytest.mark.integration


@dataclass
class FakeMembership:
    """Stands in for patreon_client.Membership — the sweep only needs the
    attribute names, and depending on the real class here would couple this
    test to a client it is not testing."""

    campaign_id: str
    display_name: str = "Creator"
    url: str = "https://www.patreon.com/creator"
    status: str = "active_patron"
    is_free_member: bool = False
    tier_names: list = None
    amount_cents: int | None = 500
    currency: str = "USD"
    details: dict = None


def _fetch(items):
    async def fetch():
        return items
    return fetch


def _raises(exc):
    async def fetch():
        raise exc
    return fetch


# --- freshness: the question C4 actually asks ------------------------------


def test_a_never_synced_roster_is_not_fresh():
    """NULL last_success_at means NEVER, and never is not zero."""
    assert roster_is_fresh(None) is False
    assert roster_is_fresh(MembershipSync(platform="patreon")) is False


def test_a_stale_roster_is_not_fresh():
    old = MembershipSync(
        platform="patreon",
        last_success_at=datetime.now(UTC) - ROSTER_STALE_AFTER - timedelta(hours=1),
    )
    assert roster_is_fresh(old) is False


def test_a_recent_roster_is_fresh():
    recent = MembershipSync(
        platform="patreon", last_success_at=datetime.now(UTC) - timedelta(hours=1),
    )
    assert roster_is_fresh(recent) is True


def test_a_failing_sync_goes_stale_even_though_it_keeps_trying():
    """The gap between attempt and success IS the signal. A sweep hammering a
    broken credential every day must not look healthy because it ran recently.
    """
    state = MembershipSync(
        platform="patreon",
        last_attempt_at=datetime.now(UTC),
        last_success_at=datetime.now(UTC) - ROSTER_STALE_AFTER - timedelta(days=1),
        last_error_type="PatreonAuthError",
    )
    assert roster_is_fresh(state) is False


# --- the sweep -------------------------------------------------------------


@pytest.mark.asyncio
async def test_a_successful_sweep_writes_the_roster_and_records_success(db):
    result = await sync_platform(
        db, platform="patreon",
        fetch=_fetch([FakeMembership("111"), FakeMembership("222")]),
    )
    assert result == {"platform": "patreon", "ok": True, "count": 2}

    rows = (await db.execute(select(PlatformMembership))).scalars().all()
    assert {r.external_campaign_id for r in rows} == {"111", "222"}

    state = await get_sync_state(db, "patreon")
    assert state.last_success_at is not None
    assert state.last_attempt_at is not None
    assert state.last_count == 2
    assert state.last_error_type is None


@pytest.mark.asyncio
async def test_a_genuinely_empty_roster_is_a_success_not_a_silence(db):
    """Zero memberships with a RECENT success is the one case where zero is an
    answer — and it has to be distinguishable from the other two."""
    result = await sync_platform(db, platform="patreon", fetch=_fetch([]))
    assert result["ok"] is True
    state = await get_sync_state(db, "patreon")
    assert state.last_success_at is not None
    assert state.last_count == 0
    assert roster_is_fresh(state) is True


@pytest.mark.asyncio
async def test_a_failed_sweep_leaves_the_previous_roster_intact(db):
    """THE property. An empty roster written over a good one is the worst
    outcome available here — it reads downstream as 'cancel everything'."""
    await sync_platform(
        db, platform="patreon", fetch=_fetch([FakeMembership("111")]),
    )
    before = (await db.execute(select(PlatformMembership))).scalars().all()
    assert len(before) == 1

    result = await sync_platform(
        db, platform="patreon", fetch=_raises(RuntimeError("boom")),
    )
    assert result["ok"] is False

    after = (await db.execute(select(PlatformMembership))).scalars().all()
    assert len(after) == 1, "a failure must not remove anything"
    assert after[0].external_campaign_id == "111"


@pytest.mark.asyncio
async def test_a_failure_records_the_error_and_does_not_advance_success(db):
    await sync_platform(db, platform="patreon", fetch=_fetch([FakeMembership("111")]))
    db.expunge_all()
    first = await get_sync_state(db, "patreon")
    success_at = first.last_success_at

    await sync_platform(
        db, platform="patreon", fetch=_raises(ValueError("drifted")),
    )
    db.expunge_all()
    state = await get_sync_state(db, "patreon")
    assert state.last_error_type == "ValueError"
    assert "drifted" in state.last_error_message
    assert state.last_success_at == success_at, "a failure must not look like a sync"
    assert state.last_attempt_at >= success_at, "but the ATTEMPT must be recorded"


@pytest.mark.asyncio
async def test_a_later_success_clears_the_previous_error(db):
    """A stale error beside a fresh success would read as 'still broken'."""
    await sync_platform(db, platform="patreon", fetch=_raises(RuntimeError("x")))
    db.expunge_all()
    assert (await get_sync_state(db, "patreon")).last_error_type == "RuntimeError"

    await sync_platform(db, platform="patreon", fetch=_fetch([FakeMembership("1")]))
    db.expunge_all()
    state = await get_sync_state(db, "patreon")
    assert state.last_error_type is None
    assert state.last_error_message is None


@pytest.mark.asyncio
async def test_the_sweep_never_raises_so_one_platform_cannot_abort_the_others(db):
    """A sweep is a background job: any escape kills the run for every OTHER
    platform too, so an ordinary failure is caught however exotic."""
    class WeirdError(Exception):
        pass

    result = await sync_platform(
        db, platform="patreon", fetch=_raises(WeirdError("unexpected")),
    )
    assert result["ok"] is False
    assert result["error"] == "WeirdError"


@pytest.mark.asyncio
async def test_a_base_exception_is_not_swallowed(db):
    """KeyboardInterrupt/SystemExit are BaseException, so `except Exception`
    lets them through — which is correct: a sweep that cannot be stopped is
    worse than one that fails."""
    with pytest.raises(KeyboardInterrupt):
        await sync_platform(
            db, platform="patreon", fetch=_raises(KeyboardInterrupt("stop")),
        )


@pytest.mark.asyncio
async def test_celerys_soft_time_limit_is_not_swallowed_either(db):
    """The one that actually needed code. SoftTimeLimitExceeded is an ORDINARY
    Exception subclass, so the broad catch would have swallowed the worker's
    request to stop and let the sweep run on into the HARD limit, where it is
    SIGKILLed mid-transaction. Caught in review, not in production."""
    from celery.exceptions import SoftTimeLimitExceeded

    assert issubclass(SoftTimeLimitExceeded, Exception), (
        "if this ever becomes a BaseException the explicit re-raise is dead code"
    )
    with pytest.raises(SoftTimeLimitExceeded):
        await sync_platform(
            db, platform="patreon", fetch=_raises(SoftTimeLimitExceeded()),
        )


@pytest.mark.asyncio
async def test_re_running_preserves_first_seen_and_updates_the_rest(db):
    """Recovery is 'run it again' — which only works because the write is an
    upsert that never moves first_seen_at (C1)."""
    await sync_platform(
        db, platform="patreon",
        fetch=_fetch([FakeMembership("111", display_name="Old", amount_cents=500)]),
    )
    db.expunge_all()
    original = (await db.execute(select(PlatformMembership))).scalar_one()
    first_seen = original.first_seen_at
    db.expunge_all()

    await sync_platform(
        db, platform="patreon",
        fetch=_fetch([FakeMembership("111", display_name="New", amount_cents=1500)]),
    )
    db.expunge_all()
    row = (await db.execute(select(PlatformMembership))).scalar_one()
    assert row.first_seen_at == first_seen
    assert row.display_name == "New"
    assert row.amount_cents == 1500


@pytest.mark.asyncio
async def test_a_membership_that_disappears_is_kept_not_deleted(db):
    """Disappearance is the signal C4 reads. Deleting the row would destroy it
    at exactly the moment it became interesting."""
    await sync_platform(
        db, platform="patreon",
        fetch=_fetch([FakeMembership("111"), FakeMembership("222")]),
    )
    await sync_platform(
        db, platform="patreon", fetch=_fetch([FakeMembership("111")]),
    )
    rows = (await db.execute(select(PlatformMembership))).scalars().all()
    assert {r.external_campaign_id for r in rows} == {"111", "222"}


@pytest.mark.asyncio
async def test_the_free_member_flag_survives_into_details(db):
    """`is_free_member` is a second axis the roster's columns do not model
    (C0 correction 2), so it rides in details rather than being dropped."""
    await sync_platform(
        db, platform="patreon",
        fetch=_fetch([FakeMembership("111", is_free_member=True, status="former_patron")]),
    )
    row = (await db.execute(select(PlatformMembership))).scalar_one()
    assert row.details["is_free_member"] is True
    assert row.status == "former_patron"


@pytest.mark.asyncio
async def test_platforms_keep_separate_sync_state(db):
    await sync_platform(db, platform="patreon", fetch=_fetch([FakeMembership("1")]))
    await sync_platform(db, platform="subscribestar", fetch=_raises(RuntimeError("x")))
    db.expunge_all()

    assert (await get_sync_state(db, "patreon")).last_error_type is None
    assert (await get_sync_state(db, "subscribestar")).last_error_type == "RuntimeError"
    assert roster_is_fresh(await get_sync_state(db, "patreon")) is True
    assert roster_is_fresh(await get_sync_state(db, "subscribestar")) is False