feat(manifests): add dts-hd, mpeg-h and lcevc codecs and a manifest corpus

Fix DASH SegmentTimeline negative @r and Pattern, TTML sidecar sets with no contentType, HLS CHANNELS identifiers and Dolby channel masks; test against 408 public DASH/HLS/ISM manifests.

tests/core/test_dash_segment_numbering.py

@@ -61,3 +61,31 @@ def test_timeline_numbers_follow_start_number(start: int, count: int) -> None:
 @pytest.mark.parametrize(("start", "seconds", "count"), [(0, "1M36.720", 17), (1, "1M36.720", 17), (2642, "30", 5)])
 def test_duration_numbers_follow_start_number(start: int, seconds: str, count: int) -> None:
     assert _numbers(DURATION_MPD.format(start=start, seconds=seconds)) == list(range(start, start + count))
+
+
+TIME_MPD = """
+<MPD xmlns="urn:mpeg:dash:schema:mpd:2011" type="static" mediaPresentationDuration="PT10S">
+  <Period id="0" start="PT0S" duration="PT10S">
+    <AdaptationSet mimeType="audio/mp4">
+      <SegmentTemplate timescale="10" media="a/$Time$.mp4">
+        <SegmentTimeline>{timeline}</SegmentTimeline>
+      </SegmentTemplate>
+      <Representation id="a" bandwidth="1"/>
+    </AdaptationSet>
+  </Period>
+</MPD>
+"""
+
+
+@pytest.mark.parametrize(
+    ("timeline", "starts"),
+    [
+        # a negative @r repeats to the end of the Period, or to the next S@t
+        ('<S t="0" d="20" r="-1"/>', [0, 20, 40, 60, 80]),
+        ('<S t="0" d="20" r="-1"/><S t="60" d="40"/>', [0, 20, 40, 60]),
+        # an S with @p cycles the Pattern durations from index @pE
+        ('<Pattern id="1"><P d="10" r="1"/><P d="30"/></Pattern><S t="0" r="4" p="1" pE="1"/>', [0, 10, 40, 50, 60]),
+    ],
+)
+def test_timeline_repeat_and_pattern_forms(timeline: str, starts: list[int]) -> None:
+    assert _numbers(TIME_MPD.format(timeline=timeline)) == starts

tests/core/test_manifest_corpus.py

@@ -0,0 +1,106 @@
+"""Parse every public test manifest in tests/fixtures/manifests (refresh: scripts/fetch_manifest_corpus.py)."""
+
+from __future__ import annotations
+
+import json
+from collections import Counter
+from pathlib import Path
+from typing import Any, Union
+from unittest.mock import MagicMock
+
+import pytest
+
+from unshackle.core.manifests import DASH, HLS, ISM
+from unshackle.core.tracks import Audio, Tracks, Video
+from unshackle.core.utils.xml import load_xml
+
+CORPUS = Path(__file__).resolve().parent.parent / "fixtures" / "manifests"
+INDEX: list[dict[str, Any]] = json.loads((CORPUS / "index.json").read_text(encoding="utf-8"))
+
+# file -> reason; strict, so a fix that makes one parse fails here until its entry is removed
+KNOWN_FAILURES: dict[str, str] = {}
+
+
+def cases(entries: list[dict[str, Any]]) -> list[Any]:
+    return [
+        pytest.param(e, id=e["file"], marks=pytest.mark.xfail(reason=KNOWN_FAILURES[e["file"]], strict=True))
+        if e["file"] in KNOWN_FAILURES
+        else pytest.param(e, id=e["file"])
+        for e in entries
+    ]
+
+
+def parse(entry: dict[str, Any]) -> Tracks:
+    """Parse a fixture the way from_url does: XML from bytes, M3U8 decoded as UTF-8."""
+    data = (CORPUS / entry["file"]).read_bytes()
+    parser: Union[DASH, HLS, ISM]
+    if entry["kind"] == "hls":
+        parser = HLS.from_text(data.decode("utf-8"), entry["url"])
+    else:
+        parser = (DASH if entry["kind"] == "dash" else ISM)(load_xml(data), entry["url"])
+    return parser.to_tracks("en")
+
+
+def most_per_set(tracks: list[Any]) -> int:
+    return max(Counter(id(t.data["dash"]["adaptation_set"]) for t in tracks).values(), default=0)
+
+
+class NeedsNetwork(Exception):
+    pass
+
+
+def test_index_lists_every_known_failure() -> None:
+    assert INDEX
+    assert set(KNOWN_FAILURES) <= {e["file"] for e in INDEX}
+
+
+@pytest.mark.parametrize("entry", cases(INDEX))
+def test_manifest_parses_to_tracks(entry: dict[str, Any]) -> None:
+    tracks = parse(entry)
+
+    assert tracks.videos or tracks.audio or tracks.subtitles
+    for track in tracks:
+        # HLS CODECS is only a SHOULD (RFC 8216 4.3.4.2), so an HLS track can honestly have none
+        assert track.codec is not None or entry["kind"] == "hls", track
+        assert track.language, track
+
+    expect = entry.get("expect")
+    if expect:
+        assert most_per_set(tracks.videos) == expect["max_num_video"]
+        assert most_per_set(tracks.audio) == expect["max_num_audio"]
+        for codec_type, codecs, found in (
+            (Video.Codec, expect["video_codec"], tracks.videos),
+            (Audio.Codec, expect["audio_codec"], tracks.audio),
+        ):
+            if codecs:
+                assert {codec_type.from_codecs(c) for c in codecs.split(",")} <= {t.codec for t in found}
+
+
+@pytest.mark.parametrize(
+    "entry",
+    [
+        e
+        for e in INDEX
+        if e["kind"] == "dash"
+        and e["file"] not in KNOWN_FAILURES
+        and b'type="dynamic"' not in (CORPUS / e["file"]).read_bytes()
+    ],
+    ids=lambda e: e["file"],
+)
+def test_dash_segments_and_drm_resolve_offline(entry: dict[str, Any]) -> None:
+    """Build the DRM and the segment list of every track; a track whose segment index must be fetched is skipped."""
+    session = MagicMock()
+    session.get.side_effect = NeedsNetwork
+    for track in parse(entry):
+        dash = track.data["dash"]
+        manifest, adaptation_set, representation = dash["manifest"], dash["adaptation_set"], dash["representation"]
+        DASH.get_drm(representation.findall("ContentProtection") + adaptation_set.findall("ContentProtection"))
+        period = dash["period"] if dash.get("period") is not None else manifest.find("Period")
+        try:
+            _, segments, _, _, _ = DASH.get_period_segments(
+                period, adaptation_set, representation, manifest, track, track.url, session, probe_kid=False
+            )
+        except NeedsNetwork:
+            continue
+        assert segments, track
+        assert all(url.startswith(("http://", "https://")) for url, _ in segments), track

tests/core/test_mp4_container_routing.py

@@ -19,7 +19,7 @@ import pytest
 from unshackle.core import binaries
 from unshackle.core.tracks import Audio, Subtitle, Tracks, Video
 from unshackle.core.tracks.track import (
-    has_dts_uhd_sample_entry,
+    has_mp4_only_audio_entry,
     iter_top_level_boxes,
     strip_duplicate_init_boxes,
 )
@@ -36,9 +36,9 @@ def write_mp4(path: Path, sample_entry: bytes) -> Path:
     return path
 
 
-@pytest.mark.parametrize("fourcc", [b"dtsx", b"dtsy"])
-def test_dts_uhd_sample_entries_are_detected(tmp_path: Path, fourcc: bytes) -> None:
-    assert has_dts_uhd_sample_entry(write_mp4(tmp_path / "a.mp4", fourcc)) is True
+@pytest.mark.parametrize("fourcc", [b"dtsx", b"dtsy", b"mha1", b"mha2", b"mhm1", b"mhm2"])
+def test_dts_uhd_and_mpeg_h_sample_entries_are_detected(tmp_path: Path, fourcc: bytes) -> None:
+    assert has_mp4_only_audio_entry(write_mp4(tmp_path / "a.mp4", fourcc)) is True
 
 
 @pytest.mark.parametrize("fourcc", [b"dtsc", b"dtse", b"dtsh", b"dtsl", b"ec-3", b"mp4a"])
@@ -47,13 +47,13 @@ def test_other_dts_profiles_stay_in_matroska(tmp_path: Path, fourcc: bytes) -> N
     DTS:X Profile 1 is DTS-HD Master Audio underneath and Matroska stores it as
     A_DTS, so it must not be pulled out of Matroska by the DTS:X container rule.
     """
-    assert has_dts_uhd_sample_entry(write_mp4(tmp_path / "a.mp4", fourcc)) is False
+    assert has_mp4_only_audio_entry(write_mp4(tmp_path / "a.mp4", fourcc)) is False
 
 
 def test_a_file_with_no_moov_is_not_detected(tmp_path: Path) -> None:
     path = tmp_path / "a.mp4"
     path.write_bytes(box(b"ftyp", b"isom") + box(b"mdat", b"dtsx" * 8))
-    assert has_dts_uhd_sample_entry(path) is False, "a 4CC in mdat is not a sample entry"
+    assert has_mp4_only_audio_entry(path) is False, "a 4CC in mdat is not a sample entry"
 
 
 def build_tracks(tmp_path: Path, audio_entry: bytes) -> Tracks: