siftlode/backend/tests/test_plex_client.py

"""plex/client — the one door to the Plex server (it was 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="<html>login</html>"),
        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"

    # Measured on the real server: a paged request carries `totalSize` even when the
    # page is empty, past the end, or the section holds nothing — `size: 0, totalSize: 0`.
    plex.routes["/library/sections/7/all"] = {"Metadata": [], "size": 0, "totalSize": 0}
    assert plex.section_items("7", 2) == ([], 0)
    # No total at all is not "0": the caller must not read it as a complete, empty section.
    plex.routes["/library/sections/7/all"] = {"Metadata": []}
    assert plex.section_items("7", 2) == ([], None)
    # …and `size` is THIS page's count, never the section's: one full page is not the whole section.
    plex.routes["/library/sections/7/all"] = {"Metadata": [{"ratingKey": "1"}], "size": 1}
    assert plex.section_items("7", 2) == ([{"ratingKey": "1"}], None)


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