siftlode/backend/tests/test_plex_playback_class.py

"""The playback-class filter (E7): filter the Plex feed by Direct Play / Direct Stream / Transcode·HW /
Transcode·SW. Covers the shared classifier + Stream parsing (unit), the sync derivation + codec-change
reset, the bit-depth backfill, and the /library filter + /facets counts (DB lane).

The HW/SW split is derived from the stored `transcode_hw` facet (codec at sync, refined by the bit-depth
backfill). These tests pin that a 10-bit/HDR HEVC lands in SW even though its codec is HW-decodable —
the exact case the runtime VAAPI path (plex.stream) also special-cases.
"""

import re
from pathlib import Path

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, User
from app.plex import sync
from fastapi.testclient import TestClient


class TestTranscodeHwCapable:
    """The single classifier both the catalog facet and (by codec) the runtime encoder choice key off."""

    def test_non_transcode_items_are_none(self):
        assert sync.transcode_hw_capable("direct", "h264") is None
        assert sync.transcode_hw_capable("remux", "h264") is None
        assert sync.transcode_hw_capable(None, "hevc") is None

    def test_hw_decodable_8bit_sdr_is_hw(self):
        assert sync.transcode_hw_capable("transcode", "hevc") is True
        assert sync.transcode_hw_capable("transcode", "vp9", bit_depth=8) is True

    def test_non_decodable_codec_is_sw(self):
        # AV1 / MPEG-4 ASP (XviD) can't be HW-decoded on Gen 9.5 → software regardless of bit depth.
        assert sync.transcode_hw_capable("transcode", "av1") is False
        assert sync.transcode_hw_capable("transcode", "mpeg4", bit_depth=8) is False

    def test_10bit_or_hdr_hevc_is_sw(self):
        # A HW-decodable codec that is 10-bit or HDR still breaks the VAAPI nv12 pipeline → software.
        assert sync.transcode_hw_capable("transcode", "hevc", bit_depth=10) is False
        assert sync.transcode_hw_capable("transcode", "hevc", bit_depth=8, hdr=True) is False


class TestVideoStreamProps:
    def test_reads_bit_depth_and_hdr_from_the_video_stream(self):
        meta = {
            "Media": [
                {
                    "Part": [
                        {
                            "Stream": [
                                {"streamType": 2, "codec": "eac3"},  # audio — skipped
                                {"streamType": 1, "bitDepth": 10, "colorTrc": "smpte2084"},
                            ]
                        }
                    ]
                }
            ]
        }
        assert sync._video_stream_props(meta) == (10, True)

    def test_8bit_sdr_stream(self):
        meta = {"Media": [{"Part": [{"Stream": [{"streamType": 1, "bitDepth": 8}]}]}]}
        assert sync._video_stream_props(meta) == (8, False)

    def test_absent_stream_is_none_false(self):
        assert sync._video_stream_props({}) == (None, False)
        assert sync._video_stream_props({"Media": [{"Part": [{}]}]}) == (None, False)


class TestSyncDerivation:
    """`_apply_item` sets transcode_hw codec-only at sync (bit depth unknown from the listing), and a
    changed video codec resets the probe watermark so a stale bit-depth can't linger."""

    def _meta(self, vcodec, *, key="/parts/1/file.mkv", file="/m.mkv"):
        return {
            "title": "X",
            "Media": [{"videoCodec": vcodec, "Part": [{"file": file, "key": key}]}],
        }

    def test_codec_only_derivation_at_sync(self):
        row = PlexItem(rating_key="x", kind="movie")
        sync._apply_item(row, lib_id=1, kind="movie", meta=self._meta("hevc"))
        assert row.playable == "transcode" and row.transcode_hw is True  # decodable, 8-bit assumed
        row2 = PlexItem(rating_key="y", kind="movie")
        sync._apply_item(row2, lib_id=1, kind="movie", meta=self._meta("av1"))
        assert row2.transcode_hw is False  # av1 not HW-decodable

    def test_probed_value_survives_resync_when_codec_unchanged(self):
        row = PlexItem(rating_key="x", kind="movie")
        sync._apply_item(row, lib_id=1, kind="movie", meta=self._meta("hevc"))
        # Simulate the backfill downgrading this 10-bit HEVC to software.
        row.transcode_hw = False
        row.video_bit_depth = 10
        row.media_probed_at = sync.datetime.now(sync.timezone.utc)
        sync._apply_item(row, lib_id=1, kind="movie", meta=self._meta("hevc"))
        assert row.transcode_hw is False  # NOT re-derived back to True
        assert row.video_bit_depth == 10 and row.media_probed_at is not None

    def test_codec_change_resets_the_probe(self):
        row = PlexItem(rating_key="x", kind="movie")
        sync._apply_item(row, lib_id=1, kind="movie", meta=self._meta("hevc"))
        row.transcode_hw = False
        row.video_bit_depth = 10
        row.media_probed_at = sync.datetime.now(sync.timezone.utc)
        # The file was replaced with an h264 remux — the stale 10-bit probe must be cleared.
        sync._apply_item(row, lib_id=1, kind="movie", meta=self._meta("h264"))
        assert row.video_bit_depth is None and row.media_probed_at is None

    def test_same_codec_reencode_with_a_new_part_resets_the_probe(self):
        # A same-codec bit-depth re-encode (e.g. 8-bit→10-bit HEVC) keeps codec 'hevc' but replaces the
        # physical file, so its Plex Part key changes. Codec-only reset would miss this and keep the stale
        # 8-bit/HW verdict forever (the backfill only re-probes media_probed_at IS NULL rows) — the
        # part-key move must reset the watermark so the new bit depth is re-probed.
        row = PlexItem(rating_key="x", kind="movie")
        sync._apply_item(row, lib_id=1, kind="movie", meta=self._meta("hevc", key="/parts/1/f.mkv"))
        row.transcode_hw = True
        row.video_bit_depth = 8
        row.media_probed_at = sync.datetime.now(sync.timezone.utc)
        # Re-encoded in place: same codec 'hevc', but a new Part key (new file).
        sync._apply_item(row, lib_id=1, kind="movie", meta=self._meta("hevc", key="/parts/2/f.mkv"))
        assert row.video_bit_depth is None and row.media_probed_at is None  # re-probe scheduled


def test_migration_codec_set_matches_hw_decodable_codecs():
    # Migration 0077 hardcodes the HW-decodable codec set in raw SQL and its comment mandates it stay
    # identical to plex.sync.HW_DECODABLE_CODECS. A comment isn't enforcement (house rule) — pin the
    # equality so adding/removing a codec from the frozenset without editing the migration fails here
    # (else already-migrated prod rows keep a wrong transcode_hw the runtime derivation wouldn't catch).
    mig = (
        Path(__file__).resolve().parent.parent
        / "alembic"
        / "versions"
        / "0077_plex_playback_class.py"
    )
    m = re.search(r"lower\(codec_video\)\s+IN\s+\(([^)]*)\)", mig.read_text())
    assert m, "could not find the codec IN(...) list in migration 0077"
    codecs = {c.strip().strip("'\"").lower() for c in m.group(1).split(",")}
    assert codecs == {c.lower() for c in sync.HW_DECODABLE_CODECS}


class _FakePlex:
    """Stands in for PlexClient.metadata — returns a canned full-metadata dict per rating_key."""

    def __init__(self, by_key):
        self.by_key = by_key

    def metadata(self, rating_key, markers=False):
        return self.by_key.get(rating_key)


def test_probe_media_batch_refines_hw_to_sw_for_10bit(db):
    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
    db.add(lib)
    db.flush()
    hevc8 = PlexItem(
        rating_key="h8",
        library_id=lib.id,
        kind="movie",
        title="8bit",
        playable="transcode",
        codec_video="hevc",
        transcode_hw=True,
    )
    hevc10 = PlexItem(
        rating_key="h10",
        library_id=lib.id,
        kind="movie",
        title="10bit",
        playable="transcode",
        codec_video="hevc",
        transcode_hw=True,
    )
    xvid = PlexItem(
        rating_key="xv",
        library_id=lib.id,
        kind="movie",
        title="xvid",
        playable="transcode",
        codec_video="mpeg4",
        transcode_hw=False,
    )
    db.add_all([hevc8, hevc10, xvid])
    db.commit()
    plex = _FakePlex(
        {
            "h8": {"Media": [{"Part": [{"Stream": [{"streamType": 1, "bitDepth": 8}]}]}]},
            "h10": {"Media": [{"Part": [{"Stream": [{"streamType": 1, "bitDepth": 10}]}]}]},
        }
    )
    probed = sync._probe_media_batch(db, plex)
    db.refresh(hevc8)
    db.refresh(hevc10)
    db.refresh(xvid)
    # Only the two HW-decodable HEVC items are probed; the XviD is SW-by-codec and never fetched.
    assert probed == 2
    assert hevc8.video_bit_depth == 8 and hevc8.transcode_hw is True and hevc8.media_probed_at
    assert hevc10.video_bit_depth == 10 and hevc10.transcode_hw is False  # downgraded to software
    assert xvid.media_probed_at is None  # not in the work-list
    # A second run has nothing left to do (watermarks set).
    assert sync._probe_media_batch(db, plex) == 0


def test_probe_stamps_even_when_plex_has_no_stream(db):
    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
    db.add(lib)
    db.flush()
    item = PlexItem(
        rating_key="ns",
        library_id=lib.id,
        kind="movie",
        title="nostream",
        playable="transcode",
        codec_video="hevc",
        transcode_hw=True,
    )
    db.add(item)
    db.commit()
    plex = _FakePlex({"ns": {"Media": [{"Part": [{}]}]}})  # no Stream array
    assert sync._probe_media_batch(db, plex) == 1
    db.refresh(item)
    # Probed (watermark set so it isn't re-fetched forever); bit depth unknown → stays codec-HW.
    assert item.media_probed_at is not None and item.video_bit_depth is None
    assert item.transcode_hw is True


def test_probe_media_batch_is_bounded_and_oldest_first(db):
    # The per-run bound (limit) caps Plex load; oldest-id-first means successive runs cover the whole
    # backlog without starving any item. Seed 3 eligible items, probe with limit=2.
    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
    db.add(lib)
    db.flush()
    items = [
        PlexItem(
            rating_key=f"h{i}",
            library_id=lib.id,
            kind="movie",
            title=f"m{i}",
            playable="transcode",
            codec_video="hevc",
            transcode_hw=True,
        )
        for i in range(3)
    ]
    db.add_all(items)  # inserted in order → ascending ids
    db.commit()
    plex = _FakePlex(
        {
            f"h{i}": {"Media": [{"Part": [{"Stream": [{"streamType": 1, "bitDepth": 8}]}]}]}
            for i in range(3)
        }
    )
    assert sync._probe_media_batch(db, plex, limit=2) == 2  # exactly the bound
    for it in items:
        db.refresh(it)
    # The two LOWEST-id rows were probed; the third waits for the next run.
    assert items[0].media_probed_at is not None and items[1].media_probed_at is not None
    assert items[2].media_probed_at is None
    # The next run continues with the remaining item.
    assert sync._probe_media_batch(db, plex, limit=2) == 1
    db.refresh(items[2])
    assert items[2].media_probed_at is not None


class _RaisingPlex:
    """metadata() raises for `bad_key`, returns a canned dict otherwise — to exercise the transient
    fetch-failure branch (leave the watermark NULL so the item retries next run)."""

    def __init__(self, by_key, bad_key):
        self.by_key = by_key
        self.bad_key = bad_key

    def metadata(self, rating_key, markers=False):
        if rating_key == self.bad_key:
            raise RuntimeError("plex timeout")
        return self.by_key.get(rating_key)


def test_probe_media_batch_leaves_a_transient_failure_unstamped_to_retry(db):
    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
    db.add(lib)
    db.flush()
    good = PlexItem(
        rating_key="g",
        library_id=lib.id,
        kind="movie",
        title="good",
        playable="transcode",
        codec_video="hevc",
        transcode_hw=True,
    )
    bad = PlexItem(
        rating_key="b",
        library_id=lib.id,
        kind="movie",
        title="bad",
        playable="transcode",
        codec_video="hevc",
        transcode_hw=True,
    )
    db.add_all([good, bad])
    db.commit()
    plex = _RaisingPlex(
        {"g": {"Media": [{"Part": [{"Stream": [{"streamType": 1, "bitDepth": 10}]}]}]}}, bad_key="b"
    )
    # The raise on 'b' must not abort the batch: 'g' is still probed and counted; 'b' stays NULL.
    assert sync._probe_media_batch(db, plex) == 1
    db.refresh(good)
    db.refresh(bad)
    assert good.media_probed_at is not None and good.transcode_hw is False  # 10-bit → SW
    assert bad.media_probed_at is None  # transient failure → retried next run, not stamped


def _seed_classes(db):
    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
    db.add(lib)
    db.flush()
    db.add_all(
        [
            PlexItem(
                rating_key="d1", library_id=lib.id, kind="movie", title="Direct", playable="direct"
            ),
            PlexItem(
                rating_key="r1", library_id=lib.id, kind="movie", title="Remux", playable="remux"
            ),
            PlexItem(
                rating_key="thw",
                library_id=lib.id,
                kind="movie",
                title="HW",
                playable="transcode",
                codec_video="hevc",
                transcode_hw=True,
            ),
            PlexItem(
                rating_key="tsw",
                library_id=lib.id,
                kind="movie",
                title="SW",
                playable="transcode",
                codec_video="mpeg4",
                transcode_hw=False,
            ),
        ]
    )
    return lib


def _client(db, user, monkeypatch, override_dep):
    monkeypatch.setattr(state, "_configured_cache", True)
    override_dep(get_db, lambda: db)
    override_dep(current_user, lambda: user)
    return TestClient(app)


def test_library_filters_by_playback_class(db, monkeypatch, override_dep):
    _seed_classes(db)
    user = User(email="pclass@example.com")
    db.add(user)
    db.commit()
    client = _client(db, user, monkeypatch, override_dep)
    # Software-only transcodes.
    sw = client.get("/api/plex/library?scope=movie&playable=transcode_sw")
    assert sw.status_code == 200, sw.text
    assert {i["id"] for i in sw.json()["items"]} == {"tsw"}
    # HW transcode + Direct Play, OR-combined.
    both = client.get("/api/plex/library?scope=movie&playable=transcode_hw,direct")
    assert {i["id"] for i in both.json()["items"]} == {"thw", "d1"}
    # The HW card carries the transcode_hw flag for the 4-way badge.
    card = next(i for i in both.json()["items"] if i["id"] == "thw")
    assert card["transcode_hw"] is True
    # No filter → everything.
    allm = client.get("/api/plex/library?scope=movie")
    assert {i["id"] for i in allm.json()["items"]} == {"d1", "r1", "thw", "tsw"}


def test_facets_count_playback_classes_in_severity_order(db, monkeypatch, override_dep):
    _seed_classes(db)
    user = User(email="pclassfacet@example.com")
    db.add(user)
    db.commit()
    client = _client(db, user, monkeypatch, override_dep)
    resp = client.get("/api/plex/facets?scope=movie")
    assert resp.status_code == 200, resp.text
    playable = resp.json()["playable"]
    # Fixed good→heavy order (not by count), one per seeded class.
    assert [p["value"] for p in playable] == ["direct", "remux", "transcode_hw", "transcode_sw"]
    assert {p["value"]: p["count"] for p in playable} == {
        "direct": 1,
        "remux": 1,
        "transcode_hw": 1,
        "transcode_sw": 1,
    }


def test_unprobed_transcode_folds_into_sw(db, monkeypatch, override_dep):
    # transcode_hw NULL is the COMMON initial state (before the bit-depth backfill runs). The query's
    # CASE has no explicit NULL arm — it relies on the `playable=='transcode'` else folding NULL in with
    # software. Pin that fold for BOTH the filter and the facet, so a reorder can't silently drop the
    # item from the feed (an honest-state violation).
    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
    db.add(lib)
    db.flush()
    db.add(
        PlexItem(
            rating_key="tnull",
            library_id=lib.id,
            kind="movie",
            title="Unprobed",
            playable="transcode",
            codec_video="hevc",
            transcode_hw=None,
        )
    )
    user = User(email="pnull@example.com")
    db.add(user)
    db.commit()
    client = _client(db, user, monkeypatch, override_dep)
    sw = client.get("/api/plex/library?scope=movie&playable=transcode_sw")
    assert {i["id"] for i in sw.json()["items"]} == {"tnull"}  # NULL → SW bucket
    hw = client.get("/api/plex/library?scope=movie&playable=transcode_hw")
    assert hw.json()["items"] == []  # never HW
    facets = client.get("/api/plex/facets?scope=movie")
    counts = {p["value"]: p["count"] for p in facets.json()["playable"]}
    assert counts.get("transcode_sw") == 1 and counts.get("transcode_hw", 0) == 0


def test_facet_playable_self_excludes_the_active_selection(db, monkeypatch, override_dep):
    # A selected playable class must NOT narrow the playable facet's OWN counts (facet self-exclusion),
    # otherwise picking one class would collapse the list and you couldn't OR-add a second. Selecting
    # 'direct' must still return all four classes at their full, unnarrowed counts.
    _seed_classes(db)
    user = User(email="pself@example.com")
    db.add(user)
    db.commit()
    client = _client(db, user, monkeypatch, override_dep)
    resp = client.get("/api/plex/facets?scope=movie&playable=direct")
    assert resp.status_code == 200, resp.text
    counts = {p["value"]: p["count"] for p in resp.json()["playable"]}
    assert counts == {"direct": 1, "remux": 1, "transcode_hw": 1, "transcode_sw": 1}


def test_active_playable_narrows_the_other_facets(db, monkeypatch, override_dep):
    # The flip side of self-exclusion: an active playable class DOES narrow the OTHER facet groups (so
    # e.g. the year range reflects only the selected class). Seed a Direct movie (2000) and a
    # software-transcode movie (2020); selecting playable=direct must collapse the year facet to 2000.
    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
    db.add(lib)
    db.flush()
    db.add_all(
        [
            PlexItem(
                rating_key="d2000",
                library_id=lib.id,
                kind="movie",
                title="Direct2000",
                playable="direct",
                year=2000,
            ),
            PlexItem(
                rating_key="s2020",
                library_id=lib.id,
                kind="movie",
                title="SW2020",
                playable="transcode",
                codec_video="mpeg4",
                transcode_hw=False,
                year=2020,
            ),
        ]
    )
    user = User(email="pnarrow@example.com")
    db.add(user)
    db.commit()
    client = _client(db, user, monkeypatch, override_dep)
    wide = client.get("/api/plex/facets?scope=movie").json()
    narrow = client.get("/api/plex/facets?scope=movie&playable=direct").json()
    assert (wide["year_min"], wide["year_max"]) == (2000, 2020)  # both classes span the range
    assert (narrow["year_min"], narrow["year_max"]) == (2000, 2000)  # narrowed to the Direct subset