"""Plex's service and queries below the HTTP layer, one use case each. The routes are thin callers now, and their own tests stay the contract a client sees. These hold what each read and write does on its own: that a user's watch state is theirs alone (another user's state is the same answer as none), that a write leaves the commit to its caller, and the rule each one carries (an explicit empty key list asks for nothing, a kept collection is editable and holds its seed). DB-backed (real Postgres). """ from datetime import datetime, timezone from sqlalchemy import select from app.plex import library, queries, service from app.plex.models import PlexCollection, PlexItem, PlexLibrary, PlexShow, PlexState from app.plex.query import LibraryFilters NOW = datetime(2024, 6, 1, tzinfo=timezone.utc) def _lib(db, key="1", kind="movie", **kw): lib = PlexLibrary(plex_key=key, title=kw.pop("title", f"L{key}"), kind=kind, enabled=True, **kw) db.add(lib) db.commit() return lib def _movie(db, lib, rk, **kw): it = PlexItem(rating_key=rk, library_id=lib.id, kind="movie", title=kw.pop("title", rk), **kw) db.add(it) db.commit() return it def _state(db, user, it, **kw): st = PlexState(user_id=user.id, item_id=it.id, **kw) db.add(st) db.commit() return st def _uncommitted(db): return bool(db.new or db.dirty or db.deleted) or db.in_transaction() def _page(db, user, **kw): args = dict( scope="both", q=None, field="all", sort="title", sort_dir="asc", show="all", lang="en", offset=0, limit=40, f=LibraryFilters(), ) args.update(kw) return library.library_page(db, user, **args) # --- whose a watch state is ---------------------------------------------------------------- def test_a_users_watch_state_is_theirs_alone(db, make_user): owner, stranger = make_user(), make_user() lib = _lib(db) it = _movie(db, lib, "m1") _state(db, owner, it, status="watched") assert queries.state(db, owner, it).status == "watched" assert queries.state(db, stranger, it) is None assert queries.states(db, stranger, [it.id]) == {} assert set(queries.states(db, owner, [it.id])) == {it.id} assert queries.states(db, owner, []) == {} def test_the_grid_shows_the_callers_state_and_hides_only_what_they_hid(db, make_user): me, other = make_user(), make_user() lib = _lib(db) seen = _movie(db, lib, "seen") hid = _movie(db, lib, "hid") _state(db, other, seen, status="watched") _state(db, other, hid, status="hidden") _state(db, me, seen, status="new", position_seconds=90) total, rows, _ = _page(db, me) assert total == 2 assert {r.rk: (r.status, r.position_seconds) for r in rows}["seen"][1] == 90 assert [r.rk for r in _page(db, me, show="watched")[1]] == [] assert [r.rk for r in _page(db, other, show="watched")[1]] == ["seen"] assert [r.rk for r in _page(db, other)[1]] == ["seen"] # their hidden one is gone, for them def test_the_facets_count_what_the_caller_would_see(db, make_user): me, other = make_user(), make_user() lib = _lib(db) _movie(db, lib, "a", genres=["Drama"], year=2001) hid = _movie(db, lib, "b", genres=["Horror"], year=1990) _state(db, other, hid, status="hidden") mine = library.facets(db, me, scope="both", show="all", lang="en", f=LibraryFilters()) theirs = library.facets(db, other, scope="both", show="all", lang="en", f=LibraryFilters()) assert [g["value"] for g in mine["genres"]] == ["Drama", "Horror"] assert [g["value"] for g in theirs["genres"]] == ["Drama"] assert (theirs["year_min"], mine["year_min"]) == (2001, 1990) def test_an_episode_search_joins_the_callers_state(db, make_user): me, other = make_user(), make_user() lib = _lib(db, "2", kind="show") sh = PlexShow(rating_key="s1", library_id=lib.id, title="Show") db.add(sh) db.commit() ep = PlexItem(rating_key="e1", library_id=lib.id, kind="episode", title="Pilot", show_id=sh.id) db.add(ep) db.commit() _state(db, other, ep, status="watched") _, _, eps = _page(db, me, q="pilot") assert [(e.rating_key, s, show.rating_key) for e, s, show in eps] == [("e1", None, "s1")] def test_a_shows_rollup_is_the_callers(db, make_user): me, other = make_user(), make_user() lib = _lib(db, "2", kind="show") sh = PlexShow(rating_key="s1", library_id=lib.id, title="Show") db.add(sh) db.commit() ep = PlexItem(rating_key="e1", library_id=lib.id, kind="episode", title="E", show_id=sh.id) db.add(ep) db.commit() _state(db, other, ep, status="watched", watched_at=NOW) (row,) = queries.show_rollups(db, me.id, [sh.id]) assert (row.total, row.watched, row.resumed_at) == (1, 0, None) assert queries.show_rollups(db, me.id, []) == [] # --- libraries, people, a show --------------------------------------------------------------- def test_the_browsable_libraries_leave_an_overlay_out_and_count_their_titles(db): en = _lib(db, "1", title="Movies") _lib(db, "9", title="Filmek", overlay_of=en.id) _movie(db, en, "m1") _movie(db, en, "m2") db.add(PlexItem(rating_key="x", library_id=en.id, kind="episode", title="not a movie")) db.commit() assert [(lib.plex_key, n) for lib, n in queries.browsable_libraries(db, "en")] == [("1", 2)] def test_a_filmography_is_cast_or_directors_and_movies_only(db): lib = _lib(db) _movie(db, lib, "acted", cast_names=["Ann"], year=2000) _movie(db, lib, "directed", directors=["Ann"], year=2010) _movie(db, lib, "other", cast_names=["Bob"]) db.add( PlexItem(rating_key="ep", library_id=lib.id, kind="episode", title="E", cast_names=["Ann"]) ) db.commit() rows = queries.filmography(db, "movie", "Ann", [lib.id], "en", limit=10) assert [r.rating_key for r in rows] == ["directed", "acted"] assert queries.filmography(db, "show", "Ann", [lib.id], "en", limit=10) == [] assert queries.cast_count(db, "Ann", [lib.id], []) == 1 # the director credit is not cast assert queries.cast_count(db, "Ann", [], []) == 0 def test_a_shows_episodes_and_their_keys_are_in_season_order(db): lib = _lib(db, "2", kind="show") sh = PlexShow(rating_key="s1", library_id=lib.id, title="Show") db.add(sh) db.commit() for rk, season, ep in (("b", 2, 1), ("a", 1, 2), ("c", 1, 1)): db.add( PlexItem( rating_key=rk, library_id=lib.id, kind="episode", title=rk, show_id=sh.id, season_number=season, episode_number=ep, ) ) db.commit() assert [e.rating_key for e in queries.episodes(db, sh)] == ["c", "a", "b"] assert queries.episode_keys(db, sh.id) == ["c", "a", "b"] assert queries.show(db, "s1\x00") is None # a NUL names no row # --- collections ----------------------------------------------------------------------------- def test_a_collection_list_narrows_by_library_name_and_keys(db): lib = _lib(db) off = _lib(db, "3", title="Off") off.enabled = False db.add_all( [ PlexCollection(rating_key="c1", library_id=lib.id, title="Bond 100%", child_count=5), PlexCollection(rating_key="c2", library_id=lib.id, title="Pixar", child_count=9), PlexCollection(rating_key="c3", library_id=off.id, title="Hidden"), ] ) db.commit() keys = lambda rows: [c.rating_key for c in rows] # noqa: E731 assert keys(queries.collections_page(db, None, None, None, "en")) == ["c2", "c1"] assert keys(queries.collections_page(db, "3", None, None, "en")) == ["c3"] assert keys(queries.collections_page(db, "404", None, None, "en")) == [] assert keys(queries.collections_page(db, None, "0%", None, "en")) == ["c1"] # % is literal assert keys(queries.collections_page(db, None, "%", None, "en")) == ["c1"] assert keys(queries.collections_page(db, None, None, ["c1"], "en")) == ["c1"] assert keys(queries.collections_page(db, None, None, [], "en")) == [] # asks for none assert queries.collection_named(db, lib.id, "PIXAR").rating_key == "c2" assert queries.collection_named(db, off.id, "pixar") is None def test_a_strip_lists_one_librarys_members_of_a_collection(db): lib, other = _lib(db), _lib(db, "4") _movie(db, lib, "in", collection_keys=["c1"], year=2002) _movie(db, lib, "early", collection_keys=["c1"], year=1999) _movie(db, other, "elsewhere", collection_keys=["c1"]) _movie(db, lib, "out", collection_keys=["c2"]) rows = queries.library_movies_in(db, lib.id, "c1", "en", limit=10) assert [r.rating_key for r in rows] == ["early", "in"] movies, shows = queries.collection_members(db, "c1", "en") assert ([m.rating_key for m in movies], shows) == (["early", "in", "elsewhere"], []) # --- writes ---------------------------------------------------------------------------------- def test_a_movie_enrichment_is_stamped_and_left_to_the_caller_to_commit(db): lib = _lib(db) it = _movie(db, lib, "m1") meta = {"Role": [{"tag": "Ann", "role": "Lead"}], "tagline": "Big."} service.enrich_movie(db, it, meta) assert it.people_enriched_at is not None assert "Ann" in (it.cast_names or []) assert _uncommitted(db) db.rollback() db.expire_all() assert db.get(PlexItem, it.id).people_enriched_at is None def test_a_collection_kept_after_a_failed_read_is_editable_and_holds_its_seed(db): lib = _lib(db) seed = _movie(db, lib, "seed", collection_keys=["old"]) service.keep_created_collection(db, lib, "new", "Mine", "seed") service.keep_created_collection(db, lib, "new2", "Mine 2", "seed") db.commit() db.expire_all() col = queries.collection(db, "new") assert (col.editable, col.child_count, col.title) == (True, 1, "Mine") assert db.get(PlexItem, seed.id).collection_keys == ["old", "new", "new2"] def test_marking_a_collection_editable_is_left_to_the_caller(db): lib = _lib(db) db.add(PlexCollection(rating_key="c1", library_id=lib.id, title="C")) db.commit() col = queries.collection(db, "c1") service.set_collection_editable(col, True) assert _uncommitted(db) db.commit() db.expire_all() assert db.scalar(select(PlexCollection.editable)) is True