"""plex/client — the one door to the Plex server (E32 S2: 31% covered). What matters here is what the callers cannot see: a failure must surface as PlexError (never as an empty answer), the paged reads must stop where they should, and the write URIs must be well formed. The network is an httpx MockTransport; the configuration is stubbed. """ import httpx import pytest from app.plex import client as plex_client from app.plex.client import PlexClient, PlexError, PlexNotConfigured @pytest.fixture def plex(monkeypatch): """A PlexClient on a stub server; set `routes[path] = handler-or-json` per test.""" routes: dict = {} seen: list[httpx.Request] = [] def transport(request: httpx.Request) -> httpx.Response: seen.append(request) answer = routes.get(request.url.path) if answer is None: return httpx.Response(404) if callable(answer): return answer(request) return httpx.Response(200, json={"MediaContainer": answer}) real = httpx.Client monkeypatch.setattr( plex_client.httpx, "Client", lambda **kw: real(transport=httpx.MockTransport(transport), **kw), ) conf = {"plex_server_url": "http://plex.test:32400/", "plex_server_token": "tok"} monkeypatch.setattr(plex_client.sysconfig, "get_str", lambda db, key: conf[key]) c = PlexClient(None) # type: ignore[arg-type] # the stubbed config never reads the session c.routes, c.seen = routes, seen # type: ignore[attr-defined] yield c c.close() def test_an_unconfigured_server_is_refused_up_front(monkeypatch): monkeypatch.setattr(plex_client.sysconfig, "get_str", lambda db, key: "") with pytest.raises(PlexNotConfigured): PlexClient(None) # type: ignore[arg-type] def test_reads_carry_the_token_and_unwrap_the_container(plex): plex.routes["/library/sections"] = {"Directory": [{"key": "1", "type": "movie"}]} assert plex.sections() == [{"key": "1", "type": "movie"}] req = plex.seen[-1] assert str(req.url) == "http://plex.test:32400/library/sections" assert req.headers["x-plex-token"] == "tok" assert req.headers["accept"] == "application/json" @pytest.mark.parametrize( "answer", [ lambda r: httpx.Response(500), lambda r: httpx.Response(200, text="login"), lambda r: (_ for _ in ()).throw(httpx.ConnectError("down", request=r)), ], ids=["http-500", "not-json", "unreachable"], ) def test_a_failed_read_is_a_plex_error_never_an_empty_answer(plex, answer): plex.routes["/library/sections"] = answer with pytest.raises(PlexError): plex.sections() def test_a_section_page_asks_for_guids_and_reports_the_server_total(plex): plex.routes["/library/sections/7/all"] = {"Metadata": [{"ratingKey": "1"}], "totalSize": 250} items, total = plex.section_items("7", 1, 200, 50) assert items == [{"ratingKey": "1"}] and total == 250 params = plex.seen[-1].url.params assert params["includeGuids"] == "1" and params["type"] == "1" assert params["X-Plex-Container-Start"] == "200" and params["X-Plex-Container-Size"] == "50" plex.routes["/library/sections/7/all"] = {"Metadata": [], "size": 0} assert plex.section_items("7", 2) == ([], 0) def test_related_flattens_the_hubs_and_degrades_to_empty(plex): plex.routes["/library/metadata/5/related"] = { "Hub": [{"Metadata": [{"ratingKey": "a"}]}, {"Metadata": [{"ratingKey": "b"}]}, {}] } assert [m["ratingKey"] for m in plex.related("5")] == ["a", "b"] plex.routes["/library/metadata/5/related"] = lambda r: httpx.Response(500) assert plex.related("5") == [] def test_metadata_is_the_first_item_or_none(plex): plex.routes["/library/metadata/9"] = {"Metadata": [{"ratingKey": "9", "title": "X"}]} assert plex.metadata("9", markers=True) == {"ratingKey": "9", "title": "X"} assert plex.seen[-1].url.params["includeMarkers"] == "1" plex.routes["/library/metadata/9"] = {} assert plex.metadata("9") is None def _history(rows_by_start: dict[int, list[dict]]): def answer(request: httpx.Request) -> httpx.Response: start = int(request.url.params["X-Plex-Container-Start"]) return httpx.Response( 200, json={"MediaContainer": {"Metadata": rows_by_start.get(start, [])}} ) return answer def test_watch_history_pages_until_the_cutoff_and_keeps_only_the_owner(plex): plex.routes["/status/sessions/history/all"] = _history( { 0: [ {"ratingKey": "1", "viewedAt": 500, "accountID": 2}, {"ratingKey": "2", "viewedAt": 400, "accountID": 1}, ], 2: [ {"ratingKey": "3", "viewedAt": 300, "accountID": 1}, {"ratingKey": "4", "viewedAt": 100, "accountID": 1}, ], 4: [{"ratingKey": "never", "viewedAt": 50, "accountID": 1}], } ) rows = plex.watch_history(min_viewed_at=200, account_id=1, size=2) assert [r["ratingKey"] for r in rows] == ["2", "3"] # page 2 reached, the cutoff stops it assert [int(r.url.params["X-Plex-Container-Start"]) for r in plex.seen] == [0, 2] def test_watch_history_stops_on_a_short_page_and_at_the_page_cap(plex): plex.routes["/status/sessions/history/all"] = _history({0: [{"ratingKey": "1", "viewedAt": 9}]}) assert [r["ratingKey"] for r in plex.watch_history(size=2)] == ["1"] assert len(plex.seen) == 1 full = [{"ratingKey": "r", "viewedAt": 9}, {"ratingKey": "r", "viewedAt": 9}] plex.routes["/status/sessions/history/all"] = lambda r: httpx.Response( 200, json={"MediaContainer": {"Metadata": full}} ) plex.seen.clear() assert len(plex.watch_history(size=2, max_pages=3)) == 6 assert len(plex.seen) == 3 def test_a_collection_is_created_with_a_well_formed_uri(plex): plex.routes["/identity"] = {"machineIdentifier": "abc"} plex.routes["/library/collections"] = {"Metadata": [{"ratingKey": "c1"}]} assert plex.create_collection("3", "Best of: 90s", "42") == {"ratingKey": "c1"} params = plex.seen[-1].url.params assert plex.seen[-1].method == "POST" assert params["title"] == "Best of: 90s" and params["sectionId"] == "3" assert params["uri"] == "server://abc/com.plexapp.plugins.library/library/metadata/42" plex.routes["/library/collections"] = {} with pytest.raises(PlexError): plex.create_collection("3", "Empty", "42") assert [r.url.path for r in plex.seen].count("/identity") == 1 # cached def test_a_server_without_a_machine_id_refuses_the_write(plex): """No machineIdentifier used to become `server:///…` in the write URI — a malformed request Plex answers in its own way. It is refused here, named (E32 S2).""" plex.routes["/identity"] = {} plex.routes["/library/collections"] = {"Metadata": [{"ratingKey": "c1"}]} with pytest.raises(PlexError, match="machine"): plex.create_collection("3", "T", "42") assert all(r.url.path == "/identity" for r in plex.seen) def test_watch_state_writes_target_the_library_item(plex): for path in ("/:/scrobble", "/:/unscrobble", "/:/timeline"): plex.routes[path] = lambda r: httpx.Response(200) # Plex answers with an empty body plex.scrobble("42") assert plex.seen[-1].url.params["key"] == "42" plex.unscrobble("42") assert plex.seen[-1].url.path == "/:/unscrobble" plex.set_timeline("42", 61_500.7, 120_000) # type: ignore[arg-type] p = plex.seen[-1].url.params assert p["key"] == "/library/metadata/42" and p["time"] == "61500" and p["state"] == "stopped" plex.routes["/:/scrobble"] = lambda r: httpx.Response(401) with pytest.raises(PlexError): plex.scrobble("42") def test_an_image_is_resized_by_plex_when_a_box_is_given(plex): plex.routes["/photo/:/transcode"] = lambda r: httpx.Response( 200, content=b"small", headers={"Content-Type": "image/webp"} ) plex.routes["/library/metadata/1/thumb/2"] = lambda r: httpx.Response(200, content=b"full") assert plex.image_bytes("/library/metadata/1/thumb/2", 300, 450) == (b"small", "image/webp") assert plex.seen[-1].url.params["url"] == "/library/metadata/1/thumb/2" body, ctype = plex.image_bytes("/library/metadata/1/thumb/2") assert body == b"full" and ctype == "image/jpeg" with pytest.raises(PlexError): plex.raw_get("/library/streams/404")