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 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):
        return {"title": "X", "Media": [{"videoCodec": vcodec, "Part": [{"file": "/m.mkv"}]}]}

    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


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 _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):
    monkeypatch.setattr(state, "_configured_cache", True)
    app.dependency_overrides[get_db] = lambda: db
    app.dependency_overrides[current_user] = lambda: user
    return TestClient(app)


def test_library_filters_by_playback_class(db, monkeypatch):
    _seed_classes(db)
    user = User(email="pclass@example.com")
    db.add(user)
    db.commit()
    client = _client(db, user, monkeypatch)
    try:
        # 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"}
    finally:
        app.dependency_overrides.clear()


def test_facets_count_playback_classes_in_severity_order(db, monkeypatch):
    _seed_classes(db)
    user = User(email="pclassfacet@example.com")
    db.add(user)
    db.commit()
    client = _client(db, user, monkeypatch)
    try:
        resp = client.get("/api/plex/facets?scope=movie")
    finally:
        app.dependency_overrides.clear()
    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,
    }