FabledCurator/tests/test_membership_roster.py

"""Milestone 387 C1: the learned membership roster.

The load-bearing property is that `first_seen_at` survives every re-observation
— it is what makes a membership's later DISAPPEARANCE readable as a lapse
rather than indistinguishable from a creator FC never knew about. Everything
else is last-writer-wins, including status, because a membership that goes from
active to former has to move.
"""
import pytest
from sqlalchemy import select

from backend.app.models import PlatformMembership
from backend.app.services.membership_roster import touch_membership
from backend.app.services.native_ingest_common import MEMBERSHIP_STATUS, has_paid_access

pytestmark = pytest.mark.integration


async def _row(db, platform="patreon", campaign="c1"):
    return (await db.execute(
        select(PlatformMembership).where(
            PlatformMembership.platform == platform,
            PlatformMembership.external_campaign_id == campaign,
        )
    )).scalar_one()


@pytest.mark.asyncio
async def test_touch_inserts_a_new_membership(db):
    await touch_membership(
        db, platform="patreon", external_campaign_id="c-new",
        display_name="Maewix Studios", url="https://patreon.com/maewix",
        status="some_platform_word", tier_names=["Sketches"],
        amount_cents=500, currency="USD", details={"raw": 1},
    )
    await db.commit()

    row = await _row(db, campaign="c-new")
    assert row.display_name == "Maewix Studios"
    assert row.status == "some_platform_word"
    assert row.tier_names == ["Sketches"]
    assert row.amount_cents == 500
    assert row.details == {"raw": 1}
    assert row.first_seen_at is not None


@pytest.mark.asyncio
async def test_re_observation_preserves_first_seen_and_advances_last_seen(db):
    """The whole point of the table.

    Committed between touches deliberately: `func.now()` is the TRANSACTION
    timestamp in Postgres, so two touches in one transaction would share a
    last_seen_at and this test would pass without proving anything. (That
    sharing is correct for a sweep — every row it touches is one observation —
    but it makes an in-transaction assertion vacuous.)
    """
    await touch_membership(
        db, platform="patreon", external_campaign_id="c-again", status="active_ish",
    )
    await db.commit()
    original = await _row(db, campaign="c-again")
    first_seen, first_last_seen = original.first_seen_at, original.last_seen_at
    db.expunge_all()

    await touch_membership(
        db, platform="patreon", external_campaign_id="c-again", status="former_ish",
    )
    await db.commit()

    row = await _row(db, campaign="c-again")
    assert row.first_seen_at == first_seen, "first_seen_at must never move"
    assert row.last_seen_at >= first_last_seen
    # Status is last-writer-wins: a lapse has to be able to overwrite.
    assert row.status == "former_ish"


@pytest.mark.asyncio
async def test_re_observation_overwrites_the_mutable_fields(db):
    """A creator who renames, retiers or changes price must not leave the
    roster asserting the old value — every column except first_seen_at moves."""
    await touch_membership(
        db, platform="patreon", external_campaign_id="c-mut",
        display_name="Old Name", amount_cents=500, tier_names=["Cheap"],
        details={"v": 1},
    )
    await db.commit()
    db.expunge_all()

    await touch_membership(
        db, platform="patreon", external_campaign_id="c-mut",
        display_name="New Name", amount_cents=1500, tier_names=["Pricey"],
        details={"v": 2},
    )
    await db.commit()

    row = await _row(db, campaign="c-mut")
    assert row.display_name == "New Name"
    assert row.amount_cents == 1500
    assert row.tier_names == ["Pricey"]
    assert row.details == {"v": 2}


@pytest.mark.asyncio
async def test_the_same_campaign_id_on_two_platforms_is_two_rows(db):
    """The key is (platform, external_campaign_id). Nothing stops two platforms
    minting the same opaque id, and collapsing them would merge one creator's
    membership into another's."""
    await touch_membership(db, platform="patreon", external_campaign_id="shared-id")
    await touch_membership(db, platform="subscribestar", external_campaign_id="shared-id")
    await db.commit()

    rows = (await db.execute(
        select(PlatformMembership).where(
            PlatformMembership.external_campaign_id == "shared-id"
        )
    )).scalars().all()
    assert {r.platform for r in rows} == {"patreon", "subscribestar"}


@pytest.mark.asyncio
async def test_a_free_follow_keeps_absent_distinct_from_zero(db):
    """"Free" and "we don't know what this costs" are different answers, and
    the reconciliation UI has to be able to tell them apart."""
    await touch_membership(
        db, platform="patreon", external_campaign_id="c-free", status="free_ish",
    )
    await db.commit()

    row = await _row(db, campaign="c-free")
    assert row.amount_cents is None
    assert row.tier_names is None


# --- has_paid_access: unknown must never read as "lost access" -------------


def test_unknown_status_is_unknown_not_false():
    """The dangerous case. False means "the operator lost access", which C4
    turns into an offer to disable the source. Asserting that from a word this
    code simply has not been taught would tell them to cancel a subscription
    they are still paying for."""
    assert has_paid_access("patreon", "a_word_nobody_characterised_yet") is None


def test_absent_status_is_unknown():
    assert has_paid_access("patreon", None) is None


def test_unknown_platform_is_unknown():
    assert has_paid_access("a-platform-with-no-mapping", "active") is None


def test_the_status_map_contains_only_characterised_values():
    """Guards project rule 130 at the one place it is easiest to break.

    Every entry must come from a characterised response, never from API docs or
    a plausible-looking guess. This started life asserting the map was EMPTY;
    C0 then captured Patreon's real `/api/members` response (Scribe note #3886)
    and this assertion is the confirmation step — updated with the capture, not
    ahead of it.

    `declined_patron` is absent ON PURPOSE and must stay absent until a capture
    shows it in `patron_status`. It appears in the request's
    `filter[membership_type]`, and the capture proved that filter is a
    different vocabulary from the attribute: a row the filter selected as
    `free_member` came back as `former_patron`, a word the filter does not
    contain. Adding it because it "obviously" belongs is precisely the guess
    this test exists to stop.
    """
    # subscribestar added with its own capture (note #3989, 2026-09-13). Its
    # "words" are the two table identifiers, because the page carries no
    # per-row status at all.
    assert MEMBERSHIP_STATUS == {
        "patreon": {"active_patron": True, "former_patron": False},
        "subscribestar": {"active_subscriptions": True, "cancelled_subscriptions": False},
    }


def test_a_free_member_does_not_count_as_paid_access():
    """The second axis. Patreon expresses a free follow as a boolean beside
    `patron_status`, so a CURRENT membership can still be one nobody pays for —
    and reporting that as paid access would hide it from C4 forever."""
    assert has_paid_access("patreon", "active_patron") is True
    assert has_paid_access("patreon", "active_patron", is_free_member=True) is False


def test_the_free_flag_cannot_rescue_a_lapsed_membership():
    """False from the status is terminal: not-free does not mean still-paying."""
    assert has_paid_access("patreon", "former_patron") is False
    assert has_paid_access("patreon", "former_patron", is_free_member=False) is False


def test_an_unknown_status_stays_unknown_whatever_the_free_flag_says():
    """The free flag refines a KNOWN answer; it never manufactures one."""
    assert has_paid_access("patreon", "declined_patron") is None
    assert has_paid_access("patreon", "declined_patron", is_free_member=True) is None


def test_the_map_is_consulted_once_it_has_entries(monkeypatch):
    """The map is empty today, so exercise the lookup with a stand-in — proving
    the plumbing works without pretending to know a real platform's word."""
    monkeypatch.setitem(
        MEMBERSHIP_STATUS, "testplat", {"paying": True, "lapsed": False},
    )
    assert has_paid_access("testplat", "paying") is True
    assert has_paid_access("testplat", "lapsed") is False
    assert has_paid_access("testplat", "something_else") is None