siftlode/backend/tests/test_plex_admin_routes.py

"""routes/plex/admin — the connection test and the owner's watch-sync link (it was 37% covered).

The connection test is the admin's one early warning that the media mount or path map is wrong, and
it must not quote the server's URL back in a toast. The watch link runs the one-time "Plex is master"
import exactly once. Plex itself is a stub.
"""

from sqlalchemy import select

import app.routes.plex.admin as admin_routes
from app import sysconfig
from app.plex.models import PlexLink
from app.plex.client import PlexError, PlexNotConfigured


class _Plex:
    def __init__(self, *, sections=None, leaf=None, fail=None):
        self._sections = sections or []
        self._leaf = leaf
        self._fail = fail
        self.section_items_calls: list[str] = []

    def __enter__(self):
        if self._fail:
            raise self._fail
        return self

    def __exit__(self, *exc):
        return False

    def server_info(self):
        return {"friendlyName": "Basement", "version": "1.40"}

    def sections(self):
        return self._sections

    def section_items(self, key, *_args):
        self.section_items_calls.append(key)
        return ([self._leaf] if self._leaf else []), 1


SECTIONS = [
    {"key": 1, "title": "Movies", "type": "movie", "count": 10},
    {"key": 2, "title": "Shows", "type": "show", "count": 3},
    {"key": 3, "title": "Music", "type": "artist", "count": 99},
]
LEAF = {"Media": [{"Part": [{"file": "/data/movies/a.mkv"}]}]}


def _use(monkeypatch, plex):
    monkeypatch.setattr(admin_routes, "PlexClient", lambda db: plex)


def test_the_connection_test_lists_movie_and_show_sections_and_probes_one_file(
    db, as_user, make_user, monkeypatch, tmp_path
):
    (tmp_path / "a.mkv").write_bytes(b"x")
    sysconfig.set_value(db, "plex_path_map", f"/data/movies={tmp_path}")
    sysconfig.set_value(db, "plex_libraries", "4")
    db.commit()
    # Only section 4 is enabled; the movie section listed first (1) must not be the one sampled.
    other = {"key": 4, "title": "Kids", "type": "movie", "count": 2}
    plex = _Plex(sections=[*SECTIONS, other], leaf=LEAF)
    _use(monkeypatch, plex)
    r = as_user(make_user(role="admin")).post("/api/plex/test")
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["server_name"] == "Basement"
    assert [s["key"] for s in body["sections"]] == ["1", "2", "4"]  # music is not ours
    assert body["media_check"] == {
        "checked": True,
        "ok": True,
        "plex_path": "/data/movies/a.mkv",
        "local_path": str(tmp_path / "a.mkv"),
    }
    assert plex.section_items_calls == ["4"]  # only an enabled movie section is sampled


def test_a_wrong_mount_is_reported_and_an_empty_library_is_not_checked(
    db, as_user, make_user, monkeypatch, tmp_path
):
    sysconfig.set_value(db, "plex_path_map", f"/data/movies={tmp_path}")
    db.commit()
    client = as_user(make_user(role="admin"))
    _use(monkeypatch, _Plex(sections=SECTIONS, leaf=LEAF))  # the file is not on the mount
    check = client.post("/api/plex/test").json()["media_check"]
    assert check["checked"] is True and check["ok"] is False

    _use(monkeypatch, _Plex(sections=SECTIONS))
    assert client.post("/api/plex/test").json()["media_check"] == {"checked": False}


def test_a_failed_connection_says_so_without_quoting_the_server(
    db, as_user, make_user, monkeypatch
):
    client = as_user(make_user(role="admin"))
    _use(monkeypatch, _Plex(fail=PlexNotConfigured()))
    r = client.post("/api/plex/test")
    assert r.status_code == 400
    assert r.json()["detail"]["reason"] == "plex_not_configured"

    _use(monkeypatch, _Plex(fail=PlexError("GET http://10.0.0.5:32400/secret refused")))
    r = client.post("/api/plex/test")
    assert r.status_code == 422
    assert r.json()["detail"]["reason"] == "plex_unreachable"
    assert "10.0.0.5" not in r.text


def test_plex_admin_routes_refuse_a_regular_user(db, as_user, make_user, monkeypatch):
    _use(monkeypatch, _Plex(sections=SECTIONS))
    client = as_user(make_user())
    for path in ("/api/plex/test", "/api/plex/sync", "/api/plex/watch/import"):
        assert client.post(path).status_code == 403, path
    assert client.post("/api/plex/watch/link", json={"enabled": True}).status_code == 403


def test_the_watch_link_imports_once_and_needs_plex_on(db, as_user, make_user, monkeypatch):
    imports: list[int] = []

    def fake_import(db_, link):
        imports.append(link.user_id)
        link.initial_import_done = True
        db_.commit()
        return {"imported": 7}

    monkeypatch.setattr(admin_routes.plex_watch, "import_owner_watch_state", fake_import)
    owner = make_user(role="admin")
    client = as_user(owner)

    r = client.post("/api/plex/watch/link", json={"enabled": True})
    assert r.status_code == 409 and r.json()["detail"]["reason"] == "plex_disabled"
    assert client.post("/api/plex/watch/import").json()["detail"]["reason"] == "watch_sync_off"

    sysconfig.set_value(db, "plex_enabled", True)
    db.commit()
    first = client.post("/api/plex/watch/link", json={"enabled": True})
    assert first.status_code == 200
    assert first.json()["import"] == {"imported": 7}
    assert first.json()["linked"] and first.json()["sync_enabled"] and first.json()["uses_admin"]

    assert client.post("/api/plex/watch/link", json={"enabled": False}).json()["import"] is None
    again = client.post("/api/plex/watch/link", json={"enabled": True})
    assert again.json()["import"] is None  # the one-time import does not run twice
    assert imports == [owner.id]

    status = client.get("/api/plex/watch/link")
    assert status.status_code == 200
    assert status.json()["initial_import_done"] is True
    links = db.execute(select(PlexLink).where(PlexLink.user_id == owner.id)).scalars().all()
    assert len(links) == 1  # toggling never creates a second link

    rerun = client.post("/api/plex/watch/import")
    assert rerun.status_code == 200 and rerun.json()["import"] == {"imported": 7}
    assert imports == [owner.id, owner.id]