Merge branch 'dev' into feat/downloader-speed

# Conflicts:
#	unshackle/core/downloaders/requests.py
#	unshackle/core/manifests/dash.py

tests/core/test_cdm_detect.py

@@ -0,0 +1,208 @@
+"""CDM detection and load-time DRM stamping for local and remote CDMs.
+
+Covers the predicates in ``unshackle.core.cdm.detect`` (playready/widevine, local/remote,
+and the remote combinations) and the loader stamping that sets a remote CDM's DRM type at
+load time instead of guessing it on each call. Services such as AMZN route to a device
+profile off that type, so a wrong classification here is what produces a Downgrade.Hd
+license denial.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from unshackle.core.cdm import loader as loader_mod
+from unshackle.core.cdm.detect import (
+    cdm_location,
+    is_local_cdm,
+    is_playready_cdm,
+    is_remote_cdm,
+    is_remote_playready_cdm,
+    is_remote_widevine_cdm,
+    is_widevine_cdm,
+)
+from unshackle.core.cdm.loader import _stamp_remote
+
+pytestmark = pytest.mark.unit
+
+REPO = Path(__file__).resolve().parents[2]
+PRD_DIR = REPO / "unshackle" / "PRDs"
+
+
+def test_stamped_remote_playready():
+    cdm = SimpleNamespace(drm="playready", is_remote_cdm=True)
+    assert is_playready_cdm(cdm)
+    assert not is_widevine_cdm(cdm)
+    assert is_remote_cdm(cdm)
+    assert is_remote_playready_cdm(cdm)
+    assert not is_remote_widevine_cdm(cdm)
+    assert cdm_location(cdm) == "remote"
+
+
+def test_stamped_remote_widevine():
+    cdm = SimpleNamespace(drm="widevine", is_remote_cdm=True)
+    assert is_widevine_cdm(cdm)
+    assert not is_playready_cdm(cdm)
+    assert is_remote_widevine_cdm(cdm)
+    assert not is_remote_playready_cdm(cdm)
+    assert cdm_location(cdm) == "remote"
+
+
+def test_drm_stamp_beats_conflicting_is_playready_attr():
+    # The stamped drm takes priority over the wrapper's own is_playready guess.
+    cdm = SimpleNamespace(drm="playready", is_playready=False, is_remote_cdm=True)
+    assert is_playready_cdm(cdm)
+    assert not is_widevine_cdm(cdm)
+
+
+def test_drm_stamp_case_insensitive():
+    cdm = SimpleNamespace(drm="PlayReady", is_remote_cdm=True)
+    assert is_playready_cdm(cdm)
+
+
+def test_none_classifies_as_nothing():
+    for fn in (
+        is_playready_cdm,
+        is_widevine_cdm,
+        is_remote_cdm,
+        is_local_cdm,
+        is_remote_playready_cdm,
+        is_remote_widevine_cdm,
+    ):
+        assert fn(None) is False
+    assert cdm_location(None) == "unknown"
+
+
+@pytest.mark.parametrize("device_name", ["SL2", "SL3", "SL2000", "SL3000", "sl3000"])
+def test_decryptlabs_sl_names_are_playready(device_name):
+    from unshackle.core.cdm.decrypt_labs_remote_cdm import DecryptLabsRemoteCDM
+
+    cdm = DecryptLabsRemoteCDM(secret="x", device_name=device_name)
+    assert cdm.is_playready
+    assert is_playready_cdm(cdm)
+    assert is_remote_cdm(cdm)
+    assert is_remote_playready_cdm(cdm)
+    assert not is_local_cdm(cdm)
+
+
+@pytest.mark.parametrize("device_name", ["ChromeCDM", "L1", "L2"])
+def test_decryptlabs_widevine_names(device_name):
+    from unshackle.core.cdm.decrypt_labs_remote_cdm import DecryptLabsRemoteCDM
+
+    cdm = DecryptLabsRemoteCDM(secret="x", device_name=device_name)
+    assert not cdm.is_playready
+    assert is_widevine_cdm(cdm)
+    assert is_remote_widevine_cdm(cdm)
+
+
+def test_decryptlabs_device_type_forces_playready():
+    from unshackle.core.cdm.decrypt_labs_remote_cdm import DecryptLabsRemoteCDM
+
+    cdm = DecryptLabsRemoteCDM(secret="x", device_name="ChromeCDM", device_type="PLAYREADY")
+    assert cdm.is_playready
+
+
+@pytest.mark.parametrize(
+    ("device", "expect_pr"),
+    [
+        ({"name": "SL3000"}, True),
+        ({"name": "SL2"}, True),
+        ({"name": "ChromeCDM"}, False),
+        ({"type": "PLAYREADY", "name": "ChromeCDM"}, True),
+    ],
+)
+def test_custom_remote_cdm_classification(device, expect_pr):
+    from unshackle.core.cdm.custom_remote_cdm import CustomRemoteCDM
+
+    cdm = CustomRemoteCDM(host="http://example.invalid", device=device)
+    assert cdm.is_playready is expect_pr
+    assert is_playready_cdm(cdm) is expect_pr
+    assert is_remote_cdm(cdm)
+
+
+def test_stamp_remote_sets_attrs_and_returns_same_object():
+    obj = SimpleNamespace()
+    out = _stamp_remote(obj, "playready")
+    assert out is obj
+    assert obj.drm == "playready"
+    assert obj.is_remote_cdm is True
+
+
+def test_stamp_remote_is_best_effort_on_unsettable_object():
+    class Slotted:
+        __slots__ = ()
+
+    obj = Slotted()
+    out = _stamp_remote(obj, "widevine")  # must not raise
+    assert out is obj
+    assert not hasattr(obj, "drm")
+
+
+def test_load_remote_decryptlabs_playready_is_stamped():
+    cdm_api = {"name": "dl-pr", "type": "decrypt_labs", "secret": "x", "device_name": "SL3000"}
+    cdm = loader_mod._load_remote_cdm(dict(cdm_api), "dl-pr", "AMZN", None)
+    assert cdm.drm == "playready"
+    assert cdm.is_remote_cdm is True
+    assert is_remote_playready_cdm(cdm)
+
+
+def test_load_remote_decryptlabs_widevine_is_stamped():
+    cdm_api = {"name": "dl-wv", "type": "decrypt_labs", "secret": "x", "device_name": "ChromeCDM"}
+    cdm = loader_mod._load_remote_cdm(dict(cdm_api), "dl-wv", "AMZN", None)
+    assert cdm.drm == "widevine"
+    assert is_remote_widevine_cdm(cdm)
+
+
+def test_load_remote_native_playready_is_stamped(monkeypatch):
+    import pyplayready.remote.remotecdm as prmod
+
+    class FakePlayReadyRemote:
+        def __init__(self, **kwargs):
+            self.kwargs = kwargs
+
+    monkeypatch.setattr(prmod, "RemoteCdm", FakePlayReadyRemote)
+    cdm_api = {"name": "pr", "Device Type": "PLAYREADY", "host": "h", "secret": "s", "device_name": "d"}
+    cdm = loader_mod._load_remote_cdm(dict(cdm_api), "pr", "AMZN", None)
+    assert isinstance(cdm, FakePlayReadyRemote)
+    assert cdm.drm == "playready"
+    assert cdm.is_remote_cdm is True
+    assert is_remote_playready_cdm(cdm)
+
+
+def test_load_remote_native_widevine_is_stamped(monkeypatch):
+    import pywidevine.remotecdm as wvmod
+
+    class FakeWidevineRemote:
+        def __init__(self, **kwargs):
+            self.kwargs = kwargs
+
+    monkeypatch.setattr(wvmod, "RemoteCdm", FakeWidevineRemote)
+    cdm_api = {"name": "wv", "Device Type": "ANDROID", "host": "h", "secret": "s", "device_name": "d"}
+    cdm = loader_mod._load_remote_cdm(dict(cdm_api), "wv", "AMZN", None)
+    assert isinstance(cdm, FakeWidevineRemote)
+    assert cdm.drm == "widevine"
+    assert is_remote_widevine_cdm(cdm)
+
+
+@pytest.fixture
+def local_playready_cdm():
+    prds = sorted(PRD_DIR.glob("*.prd"))
+    if not prds:
+        pytest.skip(f"no local .prd present in {PRD_DIR}")
+    from pyplayready.cdm import Cdm
+    from pyplayready.device import Device
+
+    return Cdm.from_device(Device.load(str(prds[0])))
+
+
+def test_local_playready_is_playready_and_local(local_playready_cdm):
+    cdm = local_playready_cdm
+    assert is_playready_cdm(cdm)
+    assert not is_widevine_cdm(cdm)
+    assert is_local_cdm(cdm)
+    assert not is_remote_cdm(cdm)
+    assert not is_remote_playready_cdm(cdm)
+    assert cdm_location(cdm) == "local"

tests/core/test_clearkey_cenc.py

@@ -16,6 +16,7 @@ from typing import Any, Optional
 from uuid import UUID
 
 import pytest
+
 # lxml.etree: XML parser used to build ContentProtection fixtures for DASH.get_drm
 from lxml import etree
 

tests/core/test_debug_logging.py

@@ -9,8 +9,13 @@ from pathlib import Path
 
 import pytest
 
-from unshackle.core.utilities import (close_debug_logger, get_debug_logger, init_debug_logger, log_event,
-                                      timed_operation)
+from unshackle.core.utilities import (
+    close_debug_logger,
+    get_debug_logger,
+    init_debug_logger,
+    log_event,
+    timed_operation,
+)
 from unshackle.core.utils.redact import redact_all, redact_path, redact_text, redact_url, safe_display_url
 from unshackle.core.utils.subprocess import log_tool_run
 

tests/core/test_ism_init.py

@@ -12,11 +12,24 @@ import struct
 
 import pytest
 
-from unshackle.core.manifests.ism_init import (NAL_START_CODE, PIFF_SENC_UUID, box, build_avcc, build_dec3,
-                                               build_hvcc, build_init_segment, full_box,
-                                               parse_codec_private_data_colour, parse_hevc_sps_format,
-                                               read_per_sample_iv_size, read_track_id, remove_emulation_prevention,
-                                               split_nal_units, synthesize_aac_codec_private_data)
+from unshackle.core.manifests.ism_init import (
+    NAL_START_CODE,
+    PIFF_SENC_UUID,
+    box,
+    build_avcc,
+    build_dec3,
+    build_hvcc,
+    build_init_segment,
+    full_box,
+    parse_codec_private_data_colour,
+    parse_codec_private_data_vui,
+    parse_hevc_sps_format,
+    read_per_sample_iv_size,
+    read_track_id,
+    remove_emulation_prevention,
+    split_nal_units,
+    synthesize_aac_codec_private_data,
+)
 
 # Real CodecPrivateData taken from a Smooth Streaming manifest.
 VIDEO_HEVC_CPD = (
@@ -472,6 +485,15 @@ def test_parse_colour_absent_or_unknown_returns_none():
     assert parse_codec_private_data_colour("HVC1", b"\x00\x00\x00\x01\x42") is None
 
 
+def test_parse_vui_fps():
+    assert parse_codec_private_data_vui("HVC1", bytes.fromhex(VIDEO_HEVC_CPD))[1] == pytest.approx(24000 / 1001)
+    assert parse_codec_private_data_vui("HVC1", bytes.fromhex(VIDEO_HEVC_PQ_CPD))[1] == 25.0
+    # VUI timing can be present even when colour info is absent.
+    assert parse_codec_private_data_vui("HVC1", bytes.fromhex(VIDEO_HEVC10_CPD)) == (None, 25.0)
+    # AVC is deliberately untrusted for fps (field-based VUI timing).
+    assert parse_codec_private_data_vui("H264", bytes.fromhex(VIDEO_AVC_CPD)) == (None, None)
+
+
 def test_hvcc_profile_tier_level_is_nonzero():
     # De-emulated PTL must yield real profile/level, not the off-by-one garbage.
     hvcc = build_hvcc(bytes.fromhex(VIDEO_HEVC_CPD))

tests/core/test_ism_parse.py

@@ -0,0 +1,43 @@
+import pytest
+
+from unshackle.core.manifests.ism import ISM
+from unshackle.core.tracks import Audio
+
+MANIFEST = """<?xml version="1.0"?>
+<SmoothStreamingMedia MajorVersion="2" MinorVersion="0" Duration="60000000" TimeScale="10000000">
+  <StreamIndex Type="audio" Name="audio" Language="en"
+               Url="QualityLevels({Bitrate})/Fragments(audio={start_time})">
+    <QualityLevel Index="0" Bitrate="128000" AudioTag="255" SamplingRate="48000"
+                  Channels="2" BitsPerSample="16" CodecPrivateData="1190"/>
+    <c t="0" d="20000000" r="2"/>
+    <c/>
+  </StreamIndex>
+</SmoothStreamingMedia>
+"""
+
+
+def test_placeholder_spellings_repeat_and_audiotag_fallback() -> None:
+    tracks = ISM.from_text(MANIFEST, "https://cdn.example/x.ism/manifest").to_tracks()
+    (audio,) = tracks.audio
+    segments = audio.data["ism"]["segments"]
+    assert segments == [
+        "https://cdn.example/x.ism/QualityLevels(128000)/Fragments(audio=0)",
+        "https://cdn.example/x.ism/QualityLevels(128000)/Fragments(audio=20000000)",
+        "https://cdn.example/x.ism/QualityLevels(128000)/Fragments(audio=40000000)",
+    ]
+    assert audio.codec == Audio.Codec.AAC  # AudioTag 255 means AACL
+
+
+def test_missing_language_without_fallback_raises() -> None:
+    # MS-SSTR Language is optional; video StreamIndexes commonly omit it.
+    no_lang = MANIFEST.replace(' Language="en"', "")
+    with pytest.raises(ValueError, match="fallback language"):
+        ISM.from_text(no_lang, "https://cdn.example/x.ism/manifest").to_tracks()
+    tracks = ISM.from_text(no_lang, "https://cdn.example/x.ism/manifest").to_tracks(language="en")
+    assert str(tracks.audio[0].language) == "en"
+
+
+def test_live_manifest_rejected() -> None:
+    live = MANIFEST.replace('Duration="60000000"', 'Duration="60000000" IsLive="TRUE"', 1)
+    with pytest.raises(ValueError, match="Live"):
+        ISM.from_text(live, "https://cdn.example/x.ism/manifest").to_tracks()

tests/core/test_ism_range.py

@@ -4,11 +4,11 @@ undecidable from the manifest; the post-mux bitstream probe names it."""
 
 from __future__ import annotations
 
+from tests.core.test_ism_init import VIDEO_HEVC10_CPD, VIDEO_HEVC_DV_CPD, VIDEO_HEVC_PQ_CPD, VIDEO_HEVC_SDR_CPD
 from unshackle.core.manifests import ISM
 from unshackle.core.tracks import Video
 
-from tests.core.test_ism_init import (VIDEO_HEVC10_CPD, VIDEO_HEVC_DV_CPD, VIDEO_HEVC_HLG_CPD,  # isort: skip
-                                      VIDEO_HEVC_PQ_CPD, VIDEO_HEVC_SDR_CPD)
+from tests.core.test_ism_init import VIDEO_HEVC_HLG_CPD  # isort: skip
 
 
 def manifest_xml(cpd: str, fourcc: str = "HVC1") -> str:

tests/core/test_music_display.py

@@ -12,8 +12,14 @@ from typing import Any, Optional
 from rich.console import Console, RenderableType
 from rich.panel import Panel
 
-from unshackle.core.music.display import (MusicHeaderInfo, TrackRow, format_track_detail_quality, render_album_header,
-                                          render_artwork_preview, render_track_panel)
+from unshackle.core.music.display import (
+    MusicHeaderInfo,
+    TrackRow,
+    format_track_detail_quality,
+    render_album_header,
+    render_artwork_preview,
+    render_track_panel,
+)
 from unshackle.core.titles.music import Song
 
 
@@ -97,8 +103,20 @@ def test_format_track_detail_quality_passthrough() -> None:
 
 def test_render_track_panel_returns_panel() -> None:
     rows = [
-        TrackRow(song=make_song(track=1, name="First"), quality_label="FLAC 16-bit/44.1kHz", layout="Stereo", duration_str="3:00", cd=True),
-        TrackRow(song=make_song(track=2, name="Second"), quality_label="OGG 320", layout="Stereo", duration_str="4:00", hires=True),
+        TrackRow(
+            song=make_song(track=1, name="First"),
+            quality_label="FLAC 16-bit/44.1kHz",
+            layout="Stereo",
+            duration_str="3:00",
+            cd=True,
+        ),
+        TrackRow(
+            song=make_song(track=2, name="Second"),
+            quality_label="OGG 320",
+            layout="Stereo",
+            duration_str="4:00",
+            hires=True,
+        ),
     ]
     panel = render_track_panel(rows, total=len(rows))
     assert isinstance(panel, Panel)
@@ -112,7 +130,9 @@ def test_render_track_panel_returns_panel() -> None:
 
 
 def test_render_track_panel_singular_label() -> None:
-    rows = [TrackRow(song=make_song(name="Only"), quality_label="FLAC 16-bit/44.1kHz", layout="Stereo", duration_str="3:00")]
+    rows = [
+        TrackRow(song=make_song(name="Only"), quality_label="FLAC 16-bit/44.1kHz", layout="Stereo", duration_str="3:00")
+    ]
     text = render_to_text(render_track_panel(rows, total=1))
     # The count node reads "1 Track" (singular); the panel title is always
     # "Available Tracks", so we only check the count node phrasing here.

tests/core/test_music_extract.py

@@ -6,9 +6,17 @@ from typing import Any, Optional
 
 import pytest
 
-from unshackle.core.music.extract import (build_music_from_songs, classify_release_kind, dedupe_track_options,
-                                          duration_seconds, first_number, first_text, format_duration, format_names,
-                                          year_from_value)
+from unshackle.core.music.extract import (
+    build_music_from_songs,
+    classify_release_kind,
+    dedupe_track_options,
+    duration_seconds,
+    first_number,
+    first_text,
+    format_duration,
+    format_names,
+    year_from_value,
+)
 from unshackle.core.music.models import MusicTrackOption
 from unshackle.core.titles.music import Music, Song
 

tests/core/test_service_dl_overrides.py

@@ -115,9 +115,7 @@ def hand_built_ctx(body: dict[str, Any]) -> click.Context:
     ctx = click.Context(parsed.command)
     ctx.params = {"repack": body.get("repack", False)}
     for name in ctx.params:
-        ctx.set_parameter_source(
-            name, ParameterSource.COMMANDLINE if name in body else ParameterSource.DEFAULT
-        )
+        ctx.set_parameter_source(name, ParameterSource.COMMANDLINE if name in body else ParameterSource.DEFAULT)
     return ctx
 
 

tests/core/test_service_repo.py

@@ -240,9 +240,7 @@ def test_refresh_reports_discarded_local_edits(monkeypatch, tmp_path):
 def test_changed_services_up_to_date(monkeypatch):
     monkeypatch.setattr(service_repo.binaries, "Git", "git")
     # same HEAD before/after → no diff computed at all
-    monkeypatch.setattr(
-        service_repo.subprocess, "run", lambda *a, **k: pytest.fail("no git diff when HEAD unchanged")
-    )
+    monkeypatch.setattr(service_repo.subprocess, "run", lambda *a, **k: pytest.fail("no git diff when HEAD unchanged"))
     assert service_repo._changed_services(Path("/x"), "same", "same") == []