"""Milestone 387 C5: why the posts are invisible, in the roster's own words. A3 gave a tier-gated source a count — "47 posts you can't see". The roster can turn that into a reason, and the whole risk of this step is that a reason is a much stronger claim than a count. So most of what follows pins refusals: * an unrecognised status says NOTHING rather than guessing lapsed; * a campaign absent from the roster says nothing, because absence is not evidence (the same discipline as `test_post_is_gated_only_on_explicit_false`); * a stale roster degrades to the bare count rather than asserting last week's reason as today's; * a current FREE follower is not told they used to be a patron. And the one that matters most: no fetch path can read the roster at all. That is asserted structurally, not by inspection — the roster explains a skip that already happened, and must never be able to cause one. """ from __future__ import annotations import ast from datetime import timedelta from pathlib import Path import pytest from backend.app.models import Artist, Source from backend.app.services.gallery_dl import ErrorType from backend.app.services.membership_roster import ( GATED_FREE, GATED_LAPSED, GATED_TIER, ROSTER_STALE_AFTER, gated_reason, gated_reasons_for_sources, ) from tests.roster_builders import membership as _membership from tests.roster_builders import synced as _synced # --- the pure mapping: status word -> the words we are entitled to say ------- # # No database, so these run in the unit lane. `gated_reason` is where the claim # is actually decided; everything below it is plumbing. def test_a_former_patron_is_told_the_membership_ended(): assert gated_reason("patreon", "former_patron") == GATED_LAPSED def test_an_active_patron_is_told_the_tier_does_not_reach_these_posts(): assert gated_reason("patreon", "active_patron") == GATED_TIER def test_a_current_free_follower_is_not_told_they_used_to_be_a_patron(): """`free` and `lapsed` are different states and must not share a sentence. Both make `has_paid_access` False, which is why the reason is derived from the status axis first rather than from that boolean: "you're not a patron any more" is a false statement about someone who never was one. """ assert gated_reason("patreon", "active_patron", is_free_member=True) == GATED_FREE assert gated_reason("patreon", "former_patron", is_free_member=True) == GATED_LAPSED @pytest.mark.parametrize("status", [None, "declined_patron", "", "wat"]) def test_an_unrecognised_status_says_nothing(status): """Unknown is not lapsed. `declined_patron` is in here on purpose: it is the plausible-looking word the C0 capture proved is NOT in the `patron_status` vocabulary. If someone adds it to MEMBERSHIP_STATUS on the strength of the request filter, this test is where that shows up. """ assert gated_reason("patreon", status) is None def test_a_platform_that_has_never_been_characterised_says_nothing(): """SubscribeStar and FANBOX (D1) inherit silence, not a Patreon guess.""" assert gated_reason("subscribestar", "active_patron") is None # --- the join, against the database ---------------------------------------- async def _artist(db, name="Maewix"): a = Artist(name=name, slug=name.lower().replace(" ", "")) db.add(a) await db.flush() return a async def _gated_source(db, artist, *, url="https://www.patreon.com/maewix", platform="patreon", overrides=None): s = Source( artist_id=artist.id, platform=platform, url=url, enabled=True, error_type=ErrorType.TIER_LIMITED, config_overrides=overrides, ) db.add(s) await db.flush() return s @pytest.mark.integration @pytest.mark.asyncio async def test_a_matched_active_membership_explains_the_gate(db): artist = await _artist(db) source = await _gated_source(db, artist) await _membership(db) await _synced(db) assert await gated_reasons_for_sources(db, [source]) == {source.id: GATED_TIER} @pytest.mark.integration @pytest.mark.asyncio async def test_a_lapsed_membership_explains_the_gate(db): artist = await _artist(db) source = await _gated_source(db, artist) await _membership(db, status="former_patron") await _synced(db) assert await gated_reasons_for_sources(db, [source]) == {source.id: GATED_LAPSED} @pytest.mark.integration @pytest.mark.asyncio async def test_a_campaign_absent_from_the_roster_adds_no_words(db): """The third case in the step: absence is not evidence. The sweep succeeded an hour ago and this creator is simply not in it. That could mean the subscription lapsed — or that the sweep's pagination is incomplete, or that the creator renamed. The count stands alone. """ artist = await _artist(db) source = await _gated_source(db, artist) await _membership(db, campaign="somebody-else", url="https://www.patreon.com/someoneelse", details={"campaign": {"vanity": "someoneelse"}}) await _synced(db) assert await gated_reasons_for_sources(db, [source]) == {} @pytest.mark.integration @pytest.mark.asyncio async def test_a_stale_roster_degrades_to_the_bare_count(db): """Same data as the passing case, one stale sync state. A reason is a claim about NOW. Last week's roster cannot make it. """ artist = await _artist(db) source = await _gated_source(db, artist) await _membership(db, status="former_patron") await _synced(db, ago=ROSTER_STALE_AFTER + timedelta(hours=1)) assert await gated_reasons_for_sources(db, [source]) == {} @pytest.mark.integration @pytest.mark.asyncio async def test_a_never_synced_platform_degrades_to_the_bare_count(db): artist = await _artist(db) source = await _gated_source(db, artist) await _membership(db, status="former_patron") # No MembershipSync row at all. assert await gated_reasons_for_sources(db, [source]) == {} @pytest.mark.integration @pytest.mark.asyncio async def test_one_platforms_fresh_sweep_does_not_vouch_for_anothers(db): """Freshness is per platform, so a working Patreon sweep cannot lend its credibility to a SubscribeStar roster that has never run (D1's future).""" artist = await _artist(db) patreon = await _gated_source(db, artist) other = await _gated_source( db, artist, platform="subscribestar", url="https://subscribestar.adult/maewix", ) await _membership(db, status="former_patron") await _membership(db, platform="subscribestar", campaign="s1", status="former_patron", url="https://subscribestar.adult/maewix", details={"campaign": {"vanity": "maewix"}}) await _synced(db) # patreon only reasons = await gated_reasons_for_sources(db, [patreon, other]) assert reasons == {patreon.id: GATED_LAPSED} @pytest.mark.integration @pytest.mark.asyncio async def test_the_cached_campaign_id_wins_over_a_stale_url(db): """Inherited from `match_kind` via the shared join — asserted here so C5 keeps identity in step with C4 rather than quietly growing its own.""" artist = await _artist(db) source = await _gated_source( db, artist, url="https://www.patreon.com/old-handle", overrides={"patreon_campaign_id": "c1"}, ) await _membership(db, campaign="c1", url="https://www.patreon.com/new-handle", details={"campaign": {"vanity": "new-handle"}}, status="former_patron") await _synced(db) assert await gated_reasons_for_sources(db, [source]) == {source.id: GATED_LAPSED} @pytest.mark.integration @pytest.mark.asyncio async def test_no_sources_asks_the_database_nothing(db): assert await gated_reasons_for_sources(db, []) == {} @pytest.mark.integration @pytest.mark.asyncio async def test_the_source_list_carries_the_reason_to_the_ui(db): """End to end through `SourceService.list`, because the field being on the record is not the same as the field reaching the payload.""" from backend.app.services.source_service import SourceService artist = await _artist(db) source = await _gated_source(db, artist) await _membership(db, status="former_patron") await _synced(db) await db.commit() records = await SourceService(db).list() row = next(r.to_dict() for r in records if r.id == source.id) assert row["gated_reason"] == GATED_LAPSED @pytest.mark.integration @pytest.mark.asyncio async def test_a_source_that_is_not_gated_gets_no_reason(db): """The roster annotates a gated state; it never puts one on a healthy row. A source walking fine whose creator the operator stopped paying for has nothing to explain — and saying "you're not a patron any more" beside a healthy row would be the roster asserting inaccessibility on its own. """ from backend.app.services.source_service import SourceService artist = await _artist(db) source = Source( artist_id=artist.id, platform="patreon", enabled=True, url="https://www.patreon.com/maewix", error_type=None, ) db.add(source) await db.flush() await _membership(db, status="former_patron") await _synced(db) await db.commit() records = await SourceService(db).list() row = next(r.to_dict() for r in records if r.id == source.id) assert row["gated_reason"] is None assert row["tier_gated_count"] is None # --- the line that must not be crossed -------------------------------------- _APP = Path(__file__).resolve().parents[1] / "backend" / "app" # The modules that FETCH: the walk, the downloaders, the per-platform clients # and ingesters. Named as ROOTS of an import walk rather than as the place the # assertion looks — the check follows their transitive first-party imports, so # it still holds when the logic inside them moves (rule #167). _FETCH_ROOTS = [ "services/download_service.py", "services/ingest_core.py", "services/native_ingest_common.py", "services/gallery_dl.py", "services/refetch_service.py", ] # Reading ANY of these is reading the roster. _ROSTER_MODULES = {"services.membership_roster", "services.membership_reconcile"} def _first_party_imports(path: Path) -> set[str]: """Every `backend.app.*` module this file imports, as a dotted path relative to `backend/app` — absolute and relative forms both.""" tree = ast.parse(path.read_text()) here = path.relative_to(_APP).with_suffix("").parts if here and here[-1] == "__init__": # A package's `__init__` IS the package, so `from .x import y` inside it # resolves one level shallower than the file path suggests. Getting this # wrong silently under-resolves every relative import in every package # and would make the guard below unable to fail. here = here[:-1] out: set[str] = set() for node in ast.walk(tree): if isinstance(node, ast.ImportFrom): if node.level: # `from ..models import X` -> walk up from this module's package base = list(here[: len(here) - node.level]) mod = base + (node.module.split(".") if node.module else []) elif node.module and node.module.startswith("backend.app."): mod = node.module[len("backend.app."):].split(".") else: continue out.add(".".join(mod)) # `from .membership_roster import x` and # `from . import membership_roster` must both resolve to the module. for alias in node.names: out.add(".".join([*mod, alias.name])) elif isinstance(node, ast.Import): for alias in node.names: if alias.name.startswith("backend.app."): out.add(alias.name[len("backend.app."):]) return out def _reachable_from(roots: list[str]) -> set[str]: seen: set[str] = set() queue = [r[:-3].replace("/", ".") for r in roots] while queue: mod = queue.pop() if mod in seen: continue seen.add(mod) for candidate in (_APP / Path(*mod.split(".")) / "__init__.py", _APP / Path(*mod.split(".")).with_suffix(".py")): if candidate.is_file(): queue.extend(_first_party_imports(candidate) - seen) break return seen def test_the_fetch_roots_all_exist(): """Falsifiability guard for the guard below. A reachability check passes trivially if its roots resolve to nothing, so a renamed module would silently turn the real assertion into a no-op that still reads as coverage. """ missing = [r for r in _FETCH_ROOTS if not (_APP / r).is_file()] assert not missing, f"fetch roots moved or were renamed: {missing}" # The falsifiability controls. FC runs no local verification (rule #85), so a # guard cannot be falsified by hand before it is committed — it has to carry its # own proof that it is capable of failing, and re-prove it on every run. These # two assert the walker finds roster imports that ARE there, one direct and one # through a hop, so the real assertion below can never pass merely because the # walk resolved nothing. def test_the_walk_sees_a_direct_roster_import(): """`source_service` legitimately reads the roster — it is what annotates the Subscriptions rows. If this stops holding the walker has gone blind, not the dependency gone away.""" assert "services.membership_roster" in _reachable_from(["services/source_service.py"]) def test_the_walk_follows_more_than_one_hop(): """api/sources.py -> services/source_service.py -> services/membership_roster.py. A one-hop walker would pass the guard below on a fetch path that reaches the roster through any intermediate module, which is the likeliest way this ever actually regresses. """ reachable = _reachable_from(["api/sources.py"]) assert "services.source_service" in reachable assert "services.membership_roster" in reachable def test_no_fetch_path_can_read_the_roster(): """The roster explains a skip. It must never be able to cause one. Entitled-tier data says which tiers the account holds, not which posts those tiers unlock — `current_user_can_view` is the only per-post truth. A fetch path that could consult the roster could decide not to fetch something the operator is paying for, silently, and that is the worst failure available in this milestone. Import reachability rather than a call-site scan: a module that cannot reach the roster cannot consult it, and unlike a grep for a function name this keeps holding when the functions are renamed. """ reachable = _reachable_from(_FETCH_ROOTS) assert not (reachable & _ROSTER_MODULES), ( "a fetch path can now reach the membership roster: " f"{sorted(reachable & _ROSTER_MODULES)}. The roster may annotate a " "gated state, never produce one — see membership_roster.gated_reason." )