peter/siftlode
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

test(plex): characterization net for the db.query->select conversion

Browse Source 
Locks the observable behaviour of the previously-untested Plex query hot
paths before the 2.0 conversion, so it can be proven behaviour-equivalent:
- unified_library: both/movie/show scope, added-desc ordering, the four
  watch-state branches, search rank + episode grouping, pagination.
- rollup.state_map: new / in_progress / watched rollup.
- sync upsert idempotency: _sync_movies + _sync_shows run twice = no dup
  rows, in-place update, episode->show linking.

Green on the current 1.x code (9 passed). The conversion commits must keep
these identical.
This commit is contained in:
peter
2026-08-16 05:45:05 +02:00
parent f38833f104
commit ecbf5737b6
1 changed files with 199 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_plex_query_conversion.py
+199
View File
@@ -0,0 +1,199 @@
"""S1 characterization net for the Plex-package db.query() -> SQLAlchemy 2.0 select() conversion.
These lock the OBSERVABLE behaviour of the previously-untested hot query paths — the unified_library
union builder (+ its watch-state branches, ordering, search ranking, episode grouping), the rollup
aggregate, and the sync upsert idempotency — so the mechanical 1.x -> 2.0 conversion can be proven
behaviour-equivalent: the exact same assertions must pass before and after the conversion. DB lane
(`db` fixture) — skipped without Postgres.
"""
from datetime import datetime, timezone
from fastapi.testclient import TestClient
from sqlalchemy import delete, select
import app.state as state
from app.auth import current_user
from app.db import get_db
from app.main import app
from app.models import PlexItem, PlexLibrary, PlexSeason, PlexShow, PlexState, User
from app.plex import rollup, sync
def _dt(day: int) -> datetime:
return datetime(2024, 1, day, tzinfo=timezone.utc)
def _seed(db):
"""A minimal cross-library corpus: 2 movies, 1 show with 2 episodes, and a per-user watch-state
where movie m1 is watched and episode e1 is in-progress (so the show rolls up to in_progress)."""
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()
m1 = PlexItem(rating_key="m1", library_id=movie_lib.id, kind="movie",
title="Alpha Movie", year=2001, added_at=_dt(1))
m2 = PlexItem(rating_key="m2", library_id=movie_lib.id, kind="movie",
title="Bravo Movie", year=2002, added_at=_dt(2))
show = PlexShow(rating_key="s1", library_id=show_lib.id, title="Charlie Show",
year=2003, added_at=_dt(3), child_count=1)
db.add_all([m1, m2, show])
db.flush()
season = PlexSeason(rating_key="se1", show_id=show.id, season_number=1)
db.add(season)
db.flush()
e1 = PlexItem(rating_key="e1", library_id=show_lib.id, kind="episode", title="Charlie Pilot",
show_id=show.id, season_id=season.id, season_number=1, episode_number=1, added_at=_dt(4))
e2 = PlexItem(rating_key="e2", library_id=show_lib.id, kind="episode", title="Charlie Finale",
show_id=show.id, season_id=season.id, season_number=1, episode_number=2, added_at=_dt(5))
db.add_all([e1, e2])
user = User(email="plexbrowse@example.com")
db.add(user)
db.flush()
db.add_all([
PlexState(user_id=user.id, item_id=m1.id, status="watched"),
PlexState(user_id=user.id, item_id=e1.id, position_seconds=120),
])
db.commit()
return {"user": user, "movie_lib": movie_lib, "show_lib": show_lib, "show": show}
def _get(db, user, monkeypatch, path):
monkeypatch.setattr(state, "_configured_cache", True) # the request gate 503s an unconfigured app
app.dependency_overrides[get_db] = lambda: db
app.dependency_overrides[current_user] = lambda: user
try:
return TestClient(app).get(path) # no context manager: skip lifespan (scheduler/HLS sweep)
finally:
app.dependency_overrides.clear()
class TestUnifiedLibrary:
def test_both_scope_unions_movies_and_shows_ordered_by_added(self, db, monkeypatch):
s = _seed(db)
r = _get(db, s["user"], monkeypatch, "/api/plex/library?scope=both")
assert r.status_code == 200, r.text
body = r.json()
assert body["total"] == 3 # 2 movies + 1 show, unioned
# default sort = added_at desc: show (day 3) > bravo (day 2) > alpha (day 1)
assert [i["id"] for i in body["items"]] == ["s1", "m2", "m1"]
show_item = next(i for i in body["items"] if i["type"] == "show")
assert show_item["status"] == "in_progress" # e1 in-progress -> show rollup
assert show_item["season_count"] == 1
assert next(i for i in body["items"] if i["id"] == "m1")["status"] == "watched"
def test_scope_movie_only(self, db, monkeypatch):
s = _seed(db)
body = _get(db, s["user"], monkeypatch, "/api/plex/library?scope=movie").json()
assert body["total"] == 2
assert all(i["type"] == "movie" for i in body["items"])
def test_scope_show_only(self, db, monkeypatch):
s = _seed(db)
body = _get(db, s["user"], monkeypatch, "/api/plex/library?scope=show").json()
assert [i["id"] for i in body["items"]] == ["s1"]
assert body["items"][0]["season_count"] == 1
def test_watch_state_branches(self, db, monkeypatch):
s = _seed(db)
# watched: only movie m1 (the show is in_progress, not fully watched)
watched = _get(db, s["user"], monkeypatch, "/api/plex/library?show=watched").json()
assert [i["id"] for i in watched["items"]] == ["m1"]
# in_progress: only the show (no movie is in progress)
inprog = _get(db, s["user"], monkeypatch, "/api/plex/library?show=in_progress").json()
assert [i["id"] for i in inprog["items"]] == ["s1"]
# unwatched: only movie m2 (no state). The show is EXCLUDED — "unwatched" for a show means
# 0 watched AND 0 in-progress episodes, and e1 is in-progress. (Locks the actual branch.)
unwatched = _get(db, s["user"], monkeypatch, "/api/plex/library?show=unwatched").json()
assert [i["id"] for i in unwatched["items"]] == ["m2"]
def test_search_matches_by_rank_and_groups_episodes(self, db, monkeypatch):
s = _seed(db)
body = _get(db, s["user"], monkeypatch, "/api/plex/library?q=charlie").json()
# only the show title matches in the item grid; the movies don't contain "charlie"
assert [i["id"] for i in body["items"]] == ["s1"]
assert body["total"] == 1
# matching episodes come back grouped in the separate episodes list
assert sorted(e["title"] for e in body["episodes"]) == ["Charlie Finale", "Charlie Pilot"]
def test_pagination_caps_items_not_total(self, db, monkeypatch):
s = _seed(db)
body = _get(db, s["user"], monkeypatch, "/api/plex/library?limit=2&offset=0").json()
assert body["total"] == 3
assert [i["id"] for i in body["items"]] == ["s1", "m2"] # first page of the added-desc order
class TestRollup:
def test_state_map_new_inprogress_watched(self, db):
s = _seed(db)
show, user = s["show"], s["user"]
# e1 in-progress, e2 new -> in_progress
assert rollup.state_map(db, user.id, [show.id]) == {show.id: "in_progress"}
# empty input short-circuits
assert rollup.state_map(db, user.id, []) == {}
# a fresh user with no state -> new
other = User(email="rollup-other@example.com")
db.add(other)
db.commit()
assert rollup.state_map(db, other.id, [show.id]) == {show.id: "new"}
# mark every episode watched for `user` -> watched
db.execute(delete(PlexState).where(PlexState.user_id == user.id))
for e in db.execute(select(PlexItem).where(PlexItem.kind == "episode")).scalars():
db.add(PlexState(user_id=user.id, item_id=e.id, status="watched"))
db.commit()
assert rollup.state_map(db, user.id, [show.id]) == {show.id: "watched"}
def _fake_paginate(by_type):
def _p(_plex, _key, typ):
return list(by_type.get(typ, []))
return _p
class TestSyncUpsertIdempotency:
def test_movies_upsert_in_place_no_duplicates(self, db, monkeypatch):
lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
db.add(lib)
db.commit()
metas = [
{"ratingKey": "100", "title": "First", "year": 2000, "addedAt": 1700000000},
{"ratingKey": "101", "title": "Second", "year": 2001, "addedAt": 1700000001},
]
monkeypatch.setattr(sync, "_paginate", _fake_paginate({sync._TYPE_MOVIE: metas}))
sync._sync_movies(db, object(), lib, {"movies": 0})
db.commit()
# second run with one title changed: same rows, updated in place (upsert, not re-insert)
metas[0]["title"] = "First Renamed"
sync._sync_movies(db, object(), lib, {"movies": 0})
db.commit()
rows = db.execute(
select(PlexItem).where(PlexItem.library_id == lib.id, PlexItem.kind == "movie")
).scalars().all()
assert {r.rating_key for r in rows} == {"100", "101"} # no duplicate rows
assert next(r for r in rows if r.rating_key == "100").title == "First Renamed"
def test_shows_upsert_links_episodes_and_is_idempotent(self, db, monkeypatch):
lib = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
db.add(lib)
db.commit()
by_type = {
sync._TYPE_SHOW: [
{"ratingKey": "s1", "title": "Show One", "childCount": 1, "addedAt": 1700000000}
],
sync._TYPE_SEASON: [
{"ratingKey": "se1", "parentRatingKey": "s1", "title": "Season 1", "index": 1}
],
sync._TYPE_EPISODE: [
{"ratingKey": "e1", "title": "Ep One", "grandparentRatingKey": "s1",
"parentRatingKey": "se1", "parentIndex": 1, "index": 1, "addedAt": 1700000001}
],
}
monkeypatch.setattr(sync, "_paginate", _fake_paginate(by_type))
for _ in range(2): # run twice — must be idempotent
sync._sync_shows(db, object(), lib, {"shows": 0, "seasons": 0, "episodes": 0})
db.commit()
shows = db.execute(select(PlexShow).where(PlexShow.library_id == lib.id)).scalars().all()
assert {sh.rating_key for sh in shows} == {"s1"} # no duplicate show
ep = db.execute(select(PlexItem).where(PlexItem.kind == "episode")).scalars().one()
assert ep.rating_key == "e1"
assert ep.show_id == shows[0].id # episode linked to its show
assert (ep.season_number, ep.episode_number) == (1, 1)