fix(dash): end segmentbase media range at the last byte of the file

The range ran one byte past the end, so a collapsed SegmentBase download gained a trailing zero byte. An unknown "*" total now sends an open-ended range.

tests/core/test_dash_collapse_integration.py

@@ -9,6 +9,8 @@ localhost server, asserting that:
 - a SegmentList whose ranges collapse fetches the parent resource whole in one direct
   (Range-less) request and writes a byte-correct file at ``save_path``, leaving no
   segment dir or ``.!dev`` artifacts, and
+- a SegmentBase track writes the parent byte for byte, when the server sends its total
+  length and when it sends an unknown (``*``) one, and
 - a track with mixed segment URLs still takes the per-segment download-and-merge path, and
 - that same segmented path raises ``DownloadCancelled`` when a sibling track's failure has
   already set ``DOWNLOAD_CANCELLED``, instead of treating the segments the cancel left
@@ -57,10 +59,12 @@ class _Handler(BaseHTTPRequestHandler):
             self.end_headers()
             return
         if rng:
-            start, end = (int(x) for x in rng.removeprefix("bytes=").split("-"))
+            first, _, last = rng.removeprefix("bytes=").partition("-")
+            start, end = int(first), int(last) if last else len(body) - 1
             chunk = body[start : end + 1]
             self.send_response(206)
-            self.send_header("Content-Range", f"bytes {start}-{end}/{len(body)}")
+            total = "*" if server.unknown_total else len(body)
+            self.send_header("Content-Range", f"bytes {start}-{end}/{total}")
             body = chunk
         else:
             self.send_response(200)
@@ -78,6 +82,7 @@ def server():
     srv.lock = threading.Lock()
     srv.requests = []
     srv.routes = {}
+    srv.unknown_total = False
     thread = threading.Thread(target=srv.serve_forever, daemon=True)
     thread.start()
     yield srv
@@ -126,9 +131,8 @@ def sidx_box(sizes: list[int]) -> bytes:
     return struct.pack(">I", 8 + len(body)) + b"sidx" + body
 
 
-def ranged_mpd(srv, ranges: list[tuple[int, int]], init: str = f'range="0-{INIT_LEN - 1}"') -> str:
-    # SegmentList of byte ranges on one media.mp4, with the given Initialization attributes
-    segment_urls = "\n".join(f'          <SegmentURL mediaRange="{s}-{e}"/>' for s, e in ranges)
+def media_mpd(srv, segments: str) -> str:
+    # one Representation on media.mp4, described by the given segment element
     return f"""<?xml version="1.0"?>
 <MPD xmlns="urn:mpeg:dash:schema:mpd:2011" type="static" profiles="urn:mpeg:dash:profile:isoff-on-demand:2011">
   <BaseURL>{base(srv)}</BaseURL>
@@ -136,16 +140,35 @@ def ranged_mpd(srv, ranges: list[tuple[int, int]], init: str = f'range="0-{INIT_
     <AdaptationSet contentType="video" mimeType="video/mp4" lang="en">
       <Representation id="v0" codecs="avc1.640028" bandwidth="1000000" width="1920" height="1080">
         <BaseURL>media.mp4</BaseURL>
-        <SegmentList timescale="1">
-          <Initialization {init}/>
-{segment_urls}
-        </SegmentList>
+{segments}
       </Representation>
     </AdaptationSet>
   </Period>
 </MPD>"""
 
 
+def ranged_mpd(srv, ranges: list[tuple[int, int]], init: str = f'range="0-{INIT_LEN - 1}"') -> str:
+    # SegmentList of byte ranges on one media.mp4, with the given Initialization attributes
+    segment_urls = "\n".join(f'          <SegmentURL mediaRange="{s}-{e}"/>' for s, e in ranges)
+    return media_mpd(
+        srv,
+        f"""        <SegmentList timescale="1">
+          <Initialization {init}/>
+{segment_urls}
+        </SegmentList>""",
+    )
+
+
+def segment_base_mpd(srv) -> str:
+    # the media range comes from the init response, not the manifest
+    return media_mpd(
+        srv,
+        f"""        <SegmentBase>
+          <Initialization range="0-{INIT_LEN - 1}"/>
+        </SegmentBase>""",
+    )
+
+
 def make_track(mpd: str, srv):
     return DASH.from_text(mpd, url=f"{base(srv)}manifest.mpd").to_tracks(language="en").videos[0]
 
@@ -204,6 +227,27 @@ def test_collapse_keeps_the_sidx_between_init_and_first_segment(server, tmp_path
     assert [path for path, rng in server.requests if rng is None] == ["/media.mp4"]
 
 
+def test_segment_base_collapse_ends_at_the_last_byte(server, tmp_path):
+    server.routes["/media.mp4"] = PARENT
+    ctx = make_ctx(tmp_path, "segment_base.mp4")
+
+    DASH.download_track(make_track(segment_base_mpd(server), server), ctx)
+
+    # the media range ends at total - 1, so the truncate after the collapse adds no byte
+    assert ctx.save_path.read_bytes() == PARENT
+
+
+def test_segment_base_unknown_total_fetches_the_rest_open_ended(server, tmp_path):
+    server.routes["/media.mp4"] = PARENT
+    server.unknown_total = True
+    ctx = make_ctx(tmp_path, "unknown_total.mp4")
+
+    DASH.download_track(make_track(segment_base_mpd(server), server), ctx)
+
+    assert ctx.save_path.read_bytes() == PARENT
+    assert ("/media.mp4", f"bytes={INIT_LEN}-") in server.requests
+
+
 def test_leading_gap_that_is_not_a_sidx_takes_segmented_path(server, tmp_path):
     # a filtered period cut from the same file leaves a gap of media, not an index
     gap = struct.pack(">I", 1000) + b"moof" + bytes(992)

tests/core/test_dash_multi_period_init.py

@@ -185,6 +185,6 @@ def test_segment_base_range_survives_a_skipped_probe(server, probes):
         probe_kid=False,
     )
 
-    assert segments == [(f"{base(server)}p1.mp4", f"{INIT_LEN}-{len(MEDIA)}")]
+    assert segments == [(f"{base(server)}p1.mp4", f"{INIT_LEN}-{len(MEDIA) - 1}")]
     assert track_kid is None
     assert not probes

unshackle/core/manifests/dash.py

@@ -27,6 +27,7 @@ from requests import Session
 from unshackle.core import binaries
 from unshackle.core.config import config
 from unshackle.core.constants import DOWNLOAD_ALL_DRM, DOWNLOAD_CANCELLED, DOWNLOAD_LICENCE_ONLY, AnyTrack
+from unshackle.core.downloaders.requests import parse_content_range
 from unshackle.core.drm import DRM_T, ClearKeyCENC, PlayReady, Widevine
 from unshackle.core.drm.segment_decrypt import SegmentDecrypter, can_use
 from unshackle.core.drm.verify import decrypt_track
@@ -1104,9 +1105,10 @@ class DASH:
                 init_data = res.content
                 if probe_kid:
                     track_kid = track.get_key_id(init_data)
-                total_size = res.headers.get("Content-Range", "").split("/")[-1]
-                if total_size:
-                    media_range = f"{len(init_data)}-{total_size}"
+                content_range = parse_content_range(res.headers.get("Content-Range"))
+                if content_range:
+                    _, init_end, total = content_range
+                    media_range = f"{init_end + 1}-{'' if total is None else total - 1}"
 
             subsegments: list[tuple[int, int, int]] = []
             index_range = segment_base.get("indexRange")