peter
b1832afd18
Mechanical, whole-tree `ruff format` pass — no behaviour change. Recorded in .git-blame-ignore-revs (next commit) so git blame skips it.
217 lines
7.7 KiB
Python
217 lines
7.7 KiB
Python
"""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
|