fix(manifests): keep service drm, rebase carried vtt cues, ship single-track mpds

DASH no longer drops service-set DRM when the manifest has none, and the WebVTT rebase allows a cue that starts one segment early. --remote also sends single-track manifests matched by track id.

tests/core/test_dash_clearkey_licensing.py

@@ -56,11 +56,8 @@ class RecordingLicence:
             drm.content_keys[kid] = KEY.hex()
 
 
-def test_download_track_licenses_clearkey_cenc(tmp_path, monkeypatch) -> None:
-    track = DASH.from_text(MPD, MANIFEST_URL).to_tracks().videos[0]
-    # KID normally comes from the init segment; short-circuit the probe (offline).
-    monkeypatch.setattr(track, "get_key_id", lambda *args, **kwargs: KID)
-
+def licence_only(track: Any, tmp_path: Any) -> RecordingLicence:
+    """Run download_track as far as licensing and return the licence callback it used."""
     licence = RecordingLicence()
     ctx = DownloadContext(
         save_path=tmp_path / "video.mp4",
@@ -69,12 +66,20 @@ def test_download_track_licenses_clearkey_cenc(tmp_path, monkeypatch) -> None:
         license_widevine=licence,
         cdm=None,
     )
-
     DOWNLOAD_LICENCE_ONLY.set()
     try:
         DASH.download_track(track=track, ctx=ctx)
     finally:
         DOWNLOAD_LICENCE_ONLY.clear()
+    return licence
+
+
+def test_download_track_licenses_clearkey_cenc(tmp_path, monkeypatch) -> None:
+    track = DASH.from_text(MPD, MANIFEST_URL).to_tracks().videos[0]
+    # KID normally comes from the init segment; short-circuit the probe (offline).
+    monkeypatch.setattr(track, "get_key_id", lambda *args, **kwargs: KID)
+
+    licence = licence_only(track, tmp_path)
 
     assert len(licence.calls) == 1, "ClearKeyCENC track was never licensed"
     drm, track_kid = licence.calls[0]
@@ -82,3 +87,17 @@ def test_download_track_licenses_clearkey_cenc(tmp_path, monkeypatch) -> None:
     assert track_kid == KID
     # with content keys populated, decrypt has what it needs
     assert drm.content_keys == {KID: KEY.hex()}
+
+
+def test_download_track_keeps_service_drm_when_manifest_has_none(tmp_path, monkeypatch) -> None:
+    """A service-set non-Widevine DRM survives a manifest with no ContentProtection, and is licensed."""
+    service_drm = [ClearKeyCENC([KID], LAURL)]
+    bare_mpd = MPD.split("<ContentProtection")[0] + MPD.split("</ContentProtection>")[1]
+    track = DASH.from_text(bare_mpd, MANIFEST_URL).to_tracks().videos[0]
+    track.drm = service_drm
+    monkeypatch.setattr(track, "get_key_id", lambda *args, **kwargs: KID)
+
+    licence = licence_only(track, tmp_path)
+
+    assert track.drm == service_drm
+    assert len(licence.calls) == 1

tests/remote/unit/test_track_manifest_fragments.py

@@ -72,10 +72,46 @@ def test_detached_adaptation_set_round_trips() -> None:
     assert client_track.descriptor == Subtitle.Descriptor.DASH
 
 
+def test_single_track_manifest_round_trips_by_id() -> None:
+    """A manifest that holds only its own track ships as a fragment, so the client needs no attribute match."""
+    manifest = etree.fromstring(MPD.encode())
+    period = manifest.find("Period")
+    adaptation_set = period.find("AdaptationSet")
+    track = Subtitle(
+        id_="abc123",
+        url="https://cdn.example/manifest.mpd",
+        codec=Subtitle.Codec.WebVTT,
+        language=Language.get("en"),
+        sdh=True,
+        descriptor=Subtitle.Descriptor.DASH,
+        data={
+            "dash": {
+                "manifest": manifest,
+                "period": period,
+                "adaptation_set": adaptation_set,
+                "representation": adaptation_set.find("Representation"),
+            }
+        },
+    )
+    client_track = Subtitle(
+        id_="abc123",
+        url=track.url,
+        codec=Subtitle.Codec.WebVTT,
+        language=Language.get("en"),
+        sdh=True,
+        descriptor=Subtitle.Descriptor.URL,
+    )
+    resolve_track_manifests(Tracks([client_track]), extract_track_manifests(Tracks([track])))
+
+    assert client_track.descriptor == Subtitle.Descriptor.DASH
+    assert client_track.data["dash"]["representation_id"] == "sub-0"
+
+
 def test_in_tree_adaptation_set_is_not_shipped() -> None:
     manifest = etree.fromstring(MPD.encode())
     period = manifest.find("Period")
     adaptation_set = period.find("AdaptationSet")
+    adaptation_set.append(etree.fromstring(b'<Representation id="sub-1" bandwidth="2000"/>'))
     track = Subtitle(
         id_="abc123",
         url="https://cdn.example/manifest.mpd",

tests/tracks/test_webvtt_merge.py

@@ -126,6 +126,18 @@ def test_rebase_moves_live_clock_cues_to_zero():
     assert "00:05:29.600 --> 00:05:30.129 line:0.01%\nText" in out
 
 
+def test_rebase_moves_cues_when_the_first_is_carried_over_from_before_the_offset():
+    vtt = "WEBVTT\n\n03:59:43.500 --> 03:59:45.000\nCarried\n\n04:03:32.598 --> 04:03:32.698\nNext\n"
+    out = rebase_webvtt(vtt, timestamp_ms("03:59:44.000"), 2002)
+    assert "00:00:00.000 --> 00:00:01.000\nCarried" in out
+    assert "00:03:48.598 --> 00:03:48.698\nNext" in out
+
+
+def test_rebase_leaves_a_later_period_with_a_small_offset_alone():
+    vtt = "WEBVTT\n\n00:00:05.000 --> 00:00:12.000\nText\n"
+    assert rebase_webvtt(vtt, 10_000, 2000) == vtt
+
+
 def test_rebase_leaves_presentation_time_cues_alone():
     # a second pass, or a VOD period with a non-zero offset, must not shift again
     vtt = "WEBVTT\n\n00:05:29.600 --> 00:05:30.129\nText\n"