siftlode/backend/tests/test_plex_facets_grid_parity.py

"""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