test(core): cover session pool growth, DRM locks, playlist encoding, and hybrid L6

tests/core/test_dovi_info_frame.py

@@ -0,0 +1,90 @@
+"""Unit tests for dovi.info_frame, which parses one frame of RPU metadata out of
+`dovi_tool info` output. The tests never run dovi_tool; they mock subprocess.run."""
+
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from unshackle.core.utils import dovi
+
+pytestmark = pytest.mark.unit
+
+RPU_JSON = (
+    "Parsing RPU file...\n"
+    '{"vdr_dm_data": {"cmv29_metadata": {"ext_metadata_blocks": '
+    '[{"Level6": {"max_content_light_level": 588}}]}}}'
+)
+
+
+class FakeCompleted:
+    def __init__(self, returncode: int, stdout: str = "", stderr: str = "") -> None:
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+@pytest.fixture(autouse=True)
+def fake_tool(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(dovi.binaries, "DoviTool", Path("dovi_tool"))
+    monkeypatch.setattr(dovi, "log_tool_run", lambda *a, **k: None)
+
+
+def patch_run(monkeypatch: pytest.MonkeyPatch, result: FakeCompleted) -> list[list[str]]:
+    calls: list[list[str]] = []
+
+    def fake_run(args: list[str], **kwargs: Any) -> FakeCompleted:
+        calls.append(args)
+        return result
+
+    monkeypatch.setattr(subprocess, "run", fake_run)
+    return calls
+
+
+def test_parses_json_after_a_preamble(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    calls = patch_run(monkeypatch, FakeCompleted(0, RPU_JSON))
+
+    info = dovi.info_frame(tmp_path / "RPU.bin")
+
+    blocks = info["vdr_dm_data"]["cmv29_metadata"]["ext_metadata_blocks"]
+    assert blocks[0]["Level6"]["max_content_light_level"] == 588
+    assert calls[0][1:] == ["info", "-i", str(tmp_path / "RPU.bin"), "-f", "0"]
+
+
+def test_frame_number_is_passed_through(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    calls = patch_run(monkeypatch, FakeCompleted(0, RPU_JSON))
+
+    dovi.info_frame(tmp_path / "RPU.bin", frame=12)
+
+    assert calls[0][-2:] == ["-f", "12"]
+
+
+def test_non_zero_return_code_raises(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    patch_run(monkeypatch, FakeCompleted(1, "", "boom"))
+
+    with pytest.raises(RuntimeError, match="dovi_tool info failed"):
+        dovi.info_frame(tmp_path / "RPU.bin")
+
+
+def test_stdout_without_json_raises(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    patch_run(monkeypatch, FakeCompleted(0, "Parsing RPU file...\nno json here\n"))
+
+    with pytest.raises(RuntimeError, match="no JSON"):
+        dovi.info_frame(tmp_path / "RPU.bin")
+
+
+def test_broken_json_raises(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    patch_run(monkeypatch, FakeCompleted(0, "Parsing RPU file...\n{not: valid}"))
+
+    with pytest.raises(RuntimeError, match="invalid JSON"):
+        dovi.info_frame(tmp_path / "RPU.bin")
+
+
+def test_missing_binary_raises(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    monkeypatch.setattr(dovi.binaries, "DoviTool", None)
+
+    with pytest.raises(EnvironmentError):
+        dovi.info_frame(tmp_path / "RPU.bin")

tests/core/test_drm_lock.py

@@ -0,0 +1,70 @@
+"""Unit tests for dl.drm_lock, the per-key-set lock that makes tracks sharing KIDs
+take turns so the second one reads the content keys out of the cache."""
+
+from __future__ import annotations
+
+from threading import Lock
+from typing import Any, Optional
+from uuid import UUID
+
+import pytest
+
+from unshackle.commands.dl import dl
+
+pytestmark = pytest.mark.unit
+
+KID_A = UUID("11111111111111111111111111111111")
+KID_B = UUID("22222222222222222222222222222222")
+
+
+class FakeDRM:
+    def __init__(self, kids: Optional[list[Any]] = None, content_id: Optional[str] = None) -> None:
+        self.kids = kids or []
+        self.content_id = content_id
+
+
+class NamelessDRM:
+    """No kids and no content_id, so the class name is the only key left."""
+
+    pass
+
+
+@pytest.fixture(autouse=True)
+def clean_locks() -> Any:
+    dl.DRM_LOCKS.clear()
+    yield
+    dl.DRM_LOCKS.clear()
+
+
+def test_same_kid_set_shares_one_lock() -> None:
+    lock = dl.drm_lock(FakeDRM(kids=[KID_A, KID_B]))
+    assert isinstance(lock, type(Lock()))
+    assert dl.drm_lock(FakeDRM(kids=[KID_A, KID_B])) is lock
+
+
+def test_kid_order_does_not_matter() -> None:
+    first = dl.drm_lock(FakeDRM(kids=[KID_A, KID_B]))
+    assert dl.drm_lock(FakeDRM(kids=[KID_B, KID_A])) is first
+
+
+def test_different_kids_get_different_locks() -> None:
+    assert dl.drm_lock(FakeDRM(kids=[KID_A])) is not dl.drm_lock(FakeDRM(kids=[KID_B]))
+    assert len(dl.DRM_LOCKS) == 2
+
+
+def test_no_kids_falls_back_to_content_id() -> None:
+    lock = dl.drm_lock(FakeDRM(content_id="abc"))
+    assert dl.drm_lock(FakeDRM(content_id="abc")) is lock
+    assert dl.drm_lock(FakeDRM(content_id="def")) is not lock
+    assert set(dl.DRM_LOCKS) == {"abc", "def"}
+
+
+def test_no_kids_and_no_content_id_falls_back_to_class_name() -> None:
+    lock = dl.drm_lock(NamelessDRM())
+    assert dl.drm_lock(NamelessDRM()) is lock
+    assert set(dl.DRM_LOCKS) == {"NamelessDRM"}
+
+
+def test_empty_content_id_still_falls_back_to_class_name() -> None:
+    dl.drm_lock(FakeDRM(content_id=""))
+    assert set(dl.DRM_LOCKS) == {"FakeDRM"}

tests/core/test_hybrid_l6.py

@@ -0,0 +1,205 @@
+"""Unit tests for the Hybrid Level 6 path: the FFprobe HDR10 metadata probe and the
+rule that decides when the RPU L6 block gets overwritten. Nothing runs FFprobe or
+dovi_tool; the tests mock subprocess.run and the dovi wrappers."""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import logging
+import subprocess
+from pathlib import Path
+from typing import Any, Iterator, Optional
+
+import pytest
+
+from unshackle.core.config import config
+from unshackle.core.tracks import hybrid as hybrid_mod
+from unshackle.core.tracks.hybrid import Hybrid
+
+pytestmark = pytest.mark.unit
+
+MASTERING = {
+    "side_data_type": "Mastering display metadata",
+    "min_luminance": "50/10000",
+    "max_luminance": "10000000/10000",
+}
+CLL = {"side_data_type": "Content light level metadata", "max_content": 588, "max_average": 148}
+
+
+class FakeCompleted:
+    def __init__(self, returncode: int, stdout: str = "", stderr: str = "") -> None:
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+@pytest.fixture(autouse=True)
+def quiet(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    monkeypatch.setattr(config.directories, "temp", tmp_path)
+    monkeypatch.setattr(hybrid_mod, "log_event", lambda *a, **k: None)
+
+    @contextlib.contextmanager
+    def no_status(*args: Any, **kwargs: Any) -> Iterator[None]:
+        yield
+
+    monkeypatch.setattr(hybrid_mod.console, "status", no_status)
+
+
+def make_hybrid() -> Hybrid:
+    """Build a Hybrid without __init__, which would demux real video tracks."""
+    obj = Hybrid.__new__(Hybrid)
+    obj.log = logging.getLogger("hybrid-test")
+    obj.rpu_file = "RPU.bin"
+    return obj
+
+
+def patch_probe_run(monkeypatch: pytest.MonkeyPatch, result: FakeCompleted) -> list[list[str]]:
+    calls: list[list[str]] = []
+
+    def fake_run(args: list[str], **kwargs: Any) -> FakeCompleted:
+        calls.append([str(a) for a in args])
+        return result
+
+    monkeypatch.setattr(subprocess, "run", fake_run)
+    return calls
+
+
+def probe_output(*side_data: dict) -> str:
+    return json.dumps({"frames": [{"side_data_list": list(side_data)}]})
+
+
+def test_probe_reads_mastering_display_and_light_level(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = patch_probe_run(monkeypatch, FakeCompleted(0, probe_output(MASTERING, CLL)))
+
+    assert make_hybrid().probe_hdr_metadata() == (1000, 50, 588, 148)
+    assert "-show_frames" in calls[0]
+    assert "frame_side_data_list" in calls[0]
+
+
+def test_probe_min_luminance_uses_integer_math(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Float rounding would turn 0.0003 nits into 2 units instead of 3."""
+    mastering = dict(MASTERING, min_luminance="3/10000")
+    patch_probe_run(monkeypatch, FakeCompleted(0, probe_output(mastering, CLL)))
+
+    assert make_hybrid().probe_hdr_metadata()[1] == 3
+
+
+def test_probe_falls_back_on_failed_ffprobe(monkeypatch: pytest.MonkeyPatch) -> None:
+    patch_probe_run(monkeypatch, FakeCompleted(1, "", "boom"))
+
+    assert make_hybrid().probe_hdr_metadata() == (1000, 1, 0, 0)
+
+
+def test_probe_falls_back_on_garbage_output(monkeypatch: pytest.MonkeyPatch) -> None:
+    patch_probe_run(monkeypatch, FakeCompleted(0, "not json at all"))
+
+    assert make_hybrid().probe_hdr_metadata() == (1000, 1, 0, 0)
+
+
+def rpu_info(max_cll: int, max_fall: int, max_mdl: int, min_mdl: int) -> dict:
+    return {
+        "vdr_dm_data": {
+            "cmv29_metadata": {
+                "ext_metadata_blocks": [
+                    {
+                        "Level6": {
+                            "max_content_light_level": max_cll,
+                            "max_frame_average_light_level": max_fall,
+                            "max_display_mastering_luminance": max_mdl,
+                            "min_display_mastering_luminance": min_mdl,
+                        }
+                    }
+                ]
+            }
+        }
+    }
+
+
+def patch_level6(
+    monkeypatch: pytest.MonkeyPatch,
+    info: dict,
+    probe: Optional[tuple[int, int, int, int]],
+) -> list[Path]:
+    """Mock the RPU read, the HDR10 probe, and the editor. Returns the editor call log."""
+    edits: list[Path] = []
+
+    monkeypatch.setattr(hybrid_mod.dovi, "info_frame", lambda *a, **k: info)
+
+    def fake_probe(self: Hybrid) -> tuple[int, int, int, int]:
+        if probe is None:
+            raise AssertionError("probe_hdr_metadata must not run when the RPU L6 block is populated")
+        return probe
+
+    monkeypatch.setattr(Hybrid, "probe_hdr_metadata", fake_probe)
+
+    def fake_editor(source: Path, json_spec: Path, output: Path, **kwargs: Any) -> bytes:
+        edits.append(json_spec)
+        return b""
+
+    monkeypatch.setattr(hybrid_mod.dovi, "editor", fake_editor)
+    return edits
+
+
+def test_zero_light_levels_are_filled_from_the_probe(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    edits = patch_level6(monkeypatch, rpu_info(0, 0, 1000, 1), (1000, 50, 588, 148))
+
+    obj = make_hybrid()
+    obj.level_6()
+
+    assert len(edits) == 1
+    written = json.loads(edits[0].read_text())["level6"]
+    assert written["max_content_light_level"] == 588
+    assert written["max_frame_average_light_level"] == 148
+    assert written["max_display_mastering_luminance"] == 1000
+    # The RPU mastering display wins over the probed 50; only unset values come from the probe.
+    assert written["min_display_mastering_luminance"] == 1
+    assert obj.rpu_file == "RPU_L6.bin"
+
+
+def test_populated_light_levels_skip_the_probe_and_the_edit(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    edits = patch_level6(monkeypatch, rpu_info(580, 538, 1000, 1), None)
+
+    obj = make_hybrid()
+    obj.level_6()
+
+    assert edits == []
+    assert obj.rpu_file == "RPU.bin"
+
+
+def test_probe_without_light_levels_leaves_the_rpu_alone(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    edits = patch_level6(monkeypatch, rpu_info(0, 0, 1000, 1), (1000, 1, 0, 0))
+
+    obj = make_hybrid()
+    obj.level_6()
+
+    assert edits == []
+    assert obj.rpu_file == "RPU.bin"
+
+
+def test_level_6_returns_early_when_the_edited_rpu_exists(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    (tmp_path / "RPU_L6.bin").write_bytes(b"")
+
+    def boom(*args: Any, **kwargs: Any) -> None:
+        raise AssertionError("level_6 must not read the RPU when RPU_L6.bin is already there")
+
+    monkeypatch.setattr(hybrid_mod.dovi, "info_frame", boom)
+
+    obj = make_hybrid()
+    obj.level_6()
+
+    assert obj.rpu_file == "RPU_L6.bin"
+
+
+def test_level_5_returns_early_when_the_cropped_rpu_exists(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    (tmp_path / "RPU_L5.bin").write_bytes(b"")
+
+    def boom(*args: Any, **kwargs: Any) -> None:
+        raise AssertionError("level_5 must not run ffprobe when RPU_L5.bin is already there")
+
+    monkeypatch.setattr(subprocess, "run", boom)
+
+    obj = make_hybrid()
+    obj.level_5(tmp_path / "HDR10.hevc")
+
+    assert obj.rpu_file == "RPU_L5.bin"

tests/core/test_playlist_encoding.py

@@ -0,0 +1,76 @@
+"""Unit tests for M3U8 response decoding. Reading `.text` on a response with no encoding
+makes requests run chardet over the whole body, so both playlist fetch paths set utf-8
+first. A charset the server did send must survive."""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+import pytest
+import requests
+
+from unshackle.core.manifests.hls import HLS
+from unshackle.core.tracks import Video
+
+pytestmark = pytest.mark.unit
+
+MASTER = b'#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=1000000,RESOLUTION=1280x720,CODECS="avc1.4d401f"\nvideo/720.m3u8\n'
+MEDIA = b"#EXTM3U\n#EXT-X-TARGETDURATION:6\n#EXTINF:6.0,\nseg1.ts\n#EXT-X-ENDLIST\n"
+
+
+def make_response(body: bytes, url: str, encoding: Optional[str]) -> requests.Response:
+    res = requests.Response()
+    res.status_code = 200
+    res._content = body
+    res.url = url
+    res.encoding = encoding
+    return res
+
+
+class FakeSession(requests.Session):
+    def __init__(self, res: requests.Response) -> None:
+        super().__init__()
+        self.res = res
+
+    def get(self, url: str, **kwargs: Any) -> requests.Response:  # type: ignore[override]
+        return self.res
+
+
+def test_hls_from_url_sets_utf8_when_server_sent_no_charset() -> None:
+    res = make_response(MASTER, "https://example.com/master.m3u8", None)
+    hls = HLS.from_url("https://example.com/master.m3u8", FakeSession(res))
+
+    assert res.encoding == "utf-8"
+    assert hls.manifest.is_variant
+    assert len(hls.manifest.playlists) == 1
+
+
+def test_hls_from_url_keeps_a_charset_the_server_sent() -> None:
+    res = make_response(MASTER, "https://example.com/master.m3u8", "iso-8859-1")
+    HLS.from_url("https://example.com/master.m3u8", FakeSession(res))
+
+    assert res.encoding == "iso-8859-1"
+
+
+def make_track(res: requests.Response) -> Video:
+    track = Video.__new__(Video)
+    track.id = "test"
+    track.url = "https://example.com/video/720.m3u8"
+    track.session = FakeSession(res)
+    track.drm = None
+    track.needs_drm_loading = True
+    return track
+
+
+def test_load_drm_from_playlist_sets_utf8_when_server_sent_no_charset() -> None:
+    res = make_response(MEDIA, "https://example.com/video/720.m3u8", None)
+
+    assert make_track(res).load_drm_from_playlist() is False
+    assert res.encoding == "utf-8"
+
+
+def test_load_drm_from_playlist_keeps_a_charset_the_server_sent() -> None:
+    res = make_response(MEDIA, "https://example.com/video/720.m3u8", "iso-8859-1")
+
+    make_track(res).load_drm_from_playlist()
+    assert res.encoding == "iso-8859-1"

tests/core/test_session_pool.py

@@ -0,0 +1,62 @@
+"""Unit tests for grow_session_pool, which enlarges a shared requests session's
+connection pool before download threads start. The tests open no sockets and check
+only the adapter mounts."""
+
+from __future__ import annotations
+
+import pytest
+import requests
+from requests.adapters import HTTPAdapter
+
+from unshackle.core.service import TimeoutHTTPAdapter, grow_session_pool
+
+pytestmark = pytest.mark.unit
+
+
+class FakeRnetSession:
+    """Stands in for RnetSession, which is not a requests.Session, so grow_session_pool skips it."""
+
+    def get_adapter(self, url: str) -> None:
+        raise AssertionError("grow_session_pool must not touch a non-requests session")
+
+
+def test_small_pool_is_remounted_on_both_schemes() -> None:
+    session = requests.Session()
+    old = session.get_adapter("https://x")
+    assert old._pool_maxsize < 64
+
+    grow_session_pool(session, 64)
+
+    https = session.get_adapter("https://x")
+    http = session.get_adapter("http://x")
+    assert isinstance(https, TimeoutHTTPAdapter)
+    assert https is http
+    assert https is not old
+    assert https._pool_maxsize == 64
+    assert https._pool_block is True
+    assert https.max_retries is old.max_retries
+
+
+def test_large_enough_pool_is_left_alone() -> None:
+    session = requests.Session()
+    session.mount("https://", HTTPAdapter(pool_connections=64, pool_maxsize=64))
+    old = session.get_adapter("https://x")
+
+    grow_session_pool(session, 64)
+
+    assert session.get_adapter("https://x") is old
+
+
+def test_equal_size_pool_is_left_alone() -> None:
+    session = requests.Session()
+    session.mount("https://", HTTPAdapter(pool_maxsize=32))
+    old = session.get_adapter("https://x")
+
+    grow_session_pool(session, 32)
+
+    assert session.get_adapter("https://x") is old
+
+
+@pytest.mark.parametrize("session", [object(), FakeRnetSession()])
+def test_non_requests_session_is_ignored(session: object) -> None:
+    grow_session_pool(session, 64)