mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-12 12:05:07 +00:00
test(core): cover session pool growth, DRM locks, playlist encoding, and hybrid L6
This commit is contained in:
imSp4rky
@@ -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")
|
||||
@@ -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"}
|
||||
@@ -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"
|
||||
@@ -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"
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user