From 330197637515d7b90baa2e4b41e211d480d4d5dd Mon Sep 17 00:00:00 2001 From: npeter83 Date: Tue, 18 Aug 2026 20:48:43 +0200 Subject: [PATCH] =?UTF-8?q?test(plex):=20E3=20S1=20characterization=20net?= =?UTF-8?q?=20=E2=80=94=20grid=E2=9F=B7facets=20watch-state=20parity?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Coverage-first safety net for the E3 S1 Plex god-module split. The movie/show watch-state predicates are currently hand-duplicated in unified_library (the grid) and facets.wsq (the facet aggregation); this pins the invariant the facets docstring promises — with no metadata filter active, a facet's genre / content-rating counts equal the same counts over the grid's visible items — for every (scope × show) combination, deriving the expected counts from the grid's own response. Locks the two branch-sets together before they merge into one shared helper, so the merge is provably behaviour-equivalent. Mutation-verified: dropping the facets hidden-exclusion branch makes both tests fail (Drama 3≠2, R 2≠1). Green on the current code (348 DB-lane pass). --- backend/tests/test_plex_facets_grid_parity.py | 148 ++++++++++++++++++ 1 file changed, 148 insertions(+) create mode 100644 backend/tests/test_plex_facets_grid_parity.py diff --git a/backend/tests/test_plex_facets_grid_parity.py b/backend/tests/test_plex_facets_grid_parity.py new file mode 100644 index 00000000..8809bf20 --- /dev/null +++ b/backend/tests/test_plex_facets_grid_parity.py @@ -0,0 +1,148 @@ +"""E3 S1 characterization net — /library grid ⟷ /facets equivalence across watch-state scopes. + +The Plex god-module split (E3 S1) extracts the movie/show watch-state predicates that are currently +HAND-DUPLICATED in two places: `unified_library` (the grid) and `facets.wsq` (the facet aggregation). +The facets docstring promises the facets "reflect exactly the visible result set" — i.e. with NO +metadata filter active (only the `show` state scope), a facet's genre/content-rating counts must equal +the same counts taken over the grid's visible items. This test derives the expected counts from the +grid's own response and asserts /facets matches, for every (scope × show) combination. + +That locks the two duplicated branch-sets together BEFORE they are merged into one shared helper, so +the merge is provably behaviour-equivalent (and a drift between grid and facets fails here). DB lane +(`db` fixture) — skipped without Postgres. +""" +from collections import Counter +from datetime import datetime, timezone + +from app.models import ( + PlexItem, + PlexLibrary, + PlexSeason, + PlexShow, + PlexState, + User, +) + +SCOPES = ("movie", "show", "both") +SHOW_STATES = ("all", "watched", "in_progress", "unwatched") + + +def _dt(day: int) -> datetime: + return datetime(2024, 1, day, tzinfo=timezone.utc) + + +def _seed(db): + """A corpus spanning every watch state, each row carrying genres + a content rating so the facet + aggregation has something to count. Movies get per-item state directly; a show's state is the + rollup of its episodes' state (new / in-progress / fully watched).""" + movie_lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True) + show_lib = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True) + db.add_all([movie_lib, show_lib]) + db.flush() + + # Movies — new / in-progress / watched / hidden, with overlapping genres. + m_new = PlexItem(rating_key="m_new", library_id=movie_lib.id, kind="movie", title="New Movie", + year=2001, added_at=_dt(1), genres=["Drama"], content_rating="PG") + m_prog = PlexItem(rating_key="m_prog", library_id=movie_lib.id, kind="movie", title="Prog Movie", + year=2002, added_at=_dt(2), genres=["Drama", "Comedy"], content_rating="PG-13") + m_watch = PlexItem(rating_key="m_watch", library_id=movie_lib.id, kind="movie", title="Watched Movie", + year=2003, added_at=_dt(3), genres=["Comedy"], content_rating="R") + m_hidden = PlexItem(rating_key="m_hidden", library_id=movie_lib.id, kind="movie", title="Hidden Movie", + year=2004, added_at=_dt(4), genres=["Drama"], content_rating="R") + + # Shows — new / in-progress / watched (state comes from the episodes below). + sh_new = PlexShow(rating_key="sh_new", library_id=show_lib.id, title="New Show", year=2005, + added_at=_dt(5), child_count=1, genres=["Sci-Fi"], content_rating="TV-14") + sh_prog = PlexShow(rating_key="sh_prog", library_id=show_lib.id, title="Prog Show", year=2006, + added_at=_dt(6), child_count=1, genres=["Sci-Fi", "Drama"], content_rating="TV-MA") + sh_watch = PlexShow(rating_key="sh_watch", library_id=show_lib.id, title="Watched Show", year=2007, + added_at=_dt(7), child_count=1, genres=["Drama"], content_rating="TV-PG") + db.add_all([m_new, m_prog, m_watch, m_hidden, sh_new, sh_prog, sh_watch]) + db.flush() + + def _episode(rk, show, day, n): + se = PlexSeason(rating_key=f"se_{rk}", show_id=show.id, season_number=1) + db.add(se) + db.flush() + ep = PlexItem(rating_key=rk, library_id=show_lib.id, kind="episode", title=f"Ep {rk}", + show_id=show.id, season_id=se.id, season_number=1, episode_number=n, added_at=_dt(day)) + db.add(ep) + db.flush() + return ep + + _episode("e_new", sh_new, 8, 1) # no state -> show "new" + e_prog = _episode("e_prog", sh_prog, 9, 1) # in-progress episode -> show "in_progress" + _episode("e_prog2", sh_prog, 10, 2) # second episode, untouched (so watched < total) + e_watch = _episode("e_watch", sh_watch, 11, 1) # watched episode -> show "watched" + + user = User(email="plexparity@example.com") + db.add(user) + db.flush() + db.add_all([ + PlexState(user_id=user.id, item_id=m_prog.id, position_seconds=120), + PlexState(user_id=user.id, item_id=m_watch.id, status="watched"), + PlexState(user_id=user.id, item_id=m_hidden.id, status="hidden"), + PlexState(user_id=user.id, item_id=e_prog.id, position_seconds=90), + PlexState(user_id=user.id, item_id=e_watch.id, status="watched"), + ]) + db.commit() + # rating_key -> (table, genres, content_rating) so the test can total the grid's visible set. + meta = {} + for it in (m_new, m_prog, m_watch, m_hidden): + meta[it.rating_key] = (it.genres, it.content_rating) + for sh in (sh_new, sh_prog, sh_watch): + meta[sh.rating_key] = (sh.genres, sh.content_rating) + return {"user": user, "meta": meta} + + +def _facet_counts(body): + genres = {g["value"]: g["count"] for g in body["genres"]} + crs = {c["value"]: c["count"] for c in body["content_ratings"]} + return genres, crs + + +def test_facets_mirror_the_grid_visible_set(db, as_user): + """For every (scope, show) the facet genre / content-rating counts equal the counts over the + grid's visible items — the invariant the extracted shared watch-state predicate must preserve.""" + seed = _seed(db) + meta = seed["meta"] + client = as_user(seed["user"]) + + for scope in SCOPES: + for show in SHOW_STATES: + grid = client.get(f"/api/plex/library?scope={scope}&show={show}&limit=100") + assert grid.status_code == 200, grid.text + ids = [i["id"] for i in grid.json()["items"]] + + exp_genres: Counter = Counter() + exp_crs: Counter = Counter() + for rk in ids: + genres, cr = meta[rk] + for g in genres or []: + exp_genres[g] += 1 + if cr is not None: + exp_crs[cr] += 1 + + facets = client.get(f"/api/plex/facets?scope={scope}&show={show}") + assert facets.status_code == 200, facets.text + got_genres, got_crs = _facet_counts(facets.json()) + + ctx = f"scope={scope} show={show} grid_ids={ids}" + assert got_genres == dict(exp_genres), f"genre facet != grid ({ctx})" + assert got_crs == dict(exp_crs), f"content-rating facet != grid ({ctx})" + + +def test_hidden_movie_absent_from_grid_and_facets(db, as_user): + """A hidden movie must not appear in the default grid NOR inflate its facets — the single most + important equivalence between the two duplicated 'exclude hidden' branches.""" + seed = _seed(db) + client = as_user(seed["user"]) + + grid = client.get("/api/plex/library?scope=movie&show=all&limit=100") + assert "m_hidden" not in [i["id"] for i in grid.json()["items"]] + + # Only the hidden movie carries content_rating "R" among the *visible* movies? No — m_watch is + # also "R". The hidden one must not add to the "R" count: visible R = {m_watch} = 1. + facets = client.get("/api/plex/facets?scope=movie&show=all") + crs = {c["value"]: c["count"] for c in facets.json()["content_ratings"]} + assert crs.get("R") == 1, crs