mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-06-10 03:02:09 +00:00
d109fe63eb
-r HYBRID alone muxed the HDR10/HDR10+ base layer as a standalone output because only the ingredient DV track was flagged hybrid_base_only. The inverse was also broken: HDR10/HDR10+/DV tracks never entered the standalone-deliverable pool, so -r HYBRID,HDR10P only delivered the standalone HDR10+ by accident of the first bug. - Add Tracks.partition_hybrid_videos: ingredient ranges (HDR10/HDR10+/DV) enter the deliverable pool only when their range is explicitly requested alongside HYBRID; replaces the duplicated filter in dl.py. - Add Tracks.flag_hybrid_ingredients: any track in the hybrid selection but not in the deliverable selection is flagged hybrid_base_only; replaces and generalises the DV-only dv_is_deliverable special case.
263 lines
12 KiB
Python
263 lines
12 KiB
Python
"""Track-selection tests for HYBRID + DV behaviour.
|
|
|
|
Covers the selection primitives that back `-r ...,DV,HYBRID` downloads:
|
|
|
|
- ``Tracks.select_hybrid`` — picks the HDR base layer per resolution and the
|
|
single lowest DV track used as a hybrid ingredient.
|
|
- ``Tracks.merge_video_selections`` — de-duplicates the ingredient/deliverable
|
|
overlap so a DV track that is chosen as both is not muxed/downloaded twice.
|
|
- ``Tracks.partition_hybrid_videos`` — splits the ladder into hybrid-ingredient
|
|
candidates and the standalone-deliverable pool; HDR10/HDR10+/DV only enter
|
|
the pool when their range was explicitly requested alongside HYBRID.
|
|
- ``Tracks.flag_hybrid_ingredients`` — marks ingredient-only tracks with
|
|
``hybrid_base_only`` so the standalone mux loop skips them.
|
|
|
|
The remaining ``dl`` glue (the Cartesian deliverable product and the mux loop)
|
|
is orchestration inside the Click command; these tests lock down the pure
|
|
units it relies on plus the documented end-state of a realistic ATV-style ladder.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from unshackle.core.tracks import Tracks, Video
|
|
|
|
|
|
def make_video(track_id: str, *, range_: Video.Range, height: int, bitrate: int, codec: Video.Codec) -> Video:
|
|
return Video(
|
|
id_=track_id,
|
|
url=f"https://example.test/{track_id}.m3u8",
|
|
language="en",
|
|
codec=codec,
|
|
range_=range_,
|
|
width=int(height * 16 / 9),
|
|
height=height,
|
|
bitrate=bitrate,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def ladder() -> list[Video]:
|
|
"""Mirrors the reported ATV ladder: HDR10+, DV and SDR at multiple resolutions."""
|
|
H = Video.Codec.HEVC
|
|
A = Video.Codec.AVC
|
|
return [
|
|
make_video("hdr10p-2160", range_=Video.Range.HDR10P, height=2160, bitrate=25_516_000, codec=H),
|
|
make_video("hdr10p-1080", range_=Video.Range.HDR10P, height=1080, bitrate=9_096_000, codec=H),
|
|
make_video("dv-2160", range_=Video.Range.DV, height=2160, bitrate=25_511_000, codec=H),
|
|
make_video("dv-1080", range_=Video.Range.DV, height=1080, bitrate=9_152_000, codec=H),
|
|
make_video("dv-360", range_=Video.Range.DV, height=360, bitrate=1_328_000, codec=H),
|
|
make_video("sdr-2160", range_=Video.Range.SDR, height=2160, bitrate=21_501_000, codec=H),
|
|
make_video("sdr-1080-avc", range_=Video.Range.SDR, height=1080, bitrate=10_793_000, codec=A),
|
|
make_video("sdr-1080-hevc", range_=Video.Range.SDR, height=1080, bitrate=5_768_000, codec=H),
|
|
]
|
|
|
|
|
|
def ids(tracks: list[Video]) -> set[str]:
|
|
return {t.id for t in tracks}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# select_hybrid
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_select_hybrid_picks_base_per_resolution_and_lowest_dv(ladder: list[Video]) -> None:
|
|
chosen = list(filter(Tracks().select_hybrid(ladder, [2160, 1080]), ladder))
|
|
# HDR10+ base at each requested resolution, plus the single lowest DV ingredient.
|
|
assert ids(chosen) == {"hdr10p-2160", "hdr10p-1080", "dv-360"}
|
|
|
|
|
|
def test_select_hybrid_ingredient_is_lowest_dv_regardless_of_quality(ladder: list[Video]) -> None:
|
|
# Even when only 2160 is requested, the ingredient is the globally lowest DV.
|
|
chosen = list(filter(Tracks().select_hybrid(ladder, [2160]), ladder))
|
|
assert ids(chosen) == {"hdr10p-2160", "dv-360"}
|
|
|
|
|
|
def test_select_hybrid_prefers_hdr10p_over_hdr10() -> None:
|
|
H = Video.Codec.HEVC
|
|
tracks = [
|
|
make_video("hdr10-2160", range_=Video.Range.HDR10, height=2160, bitrate=20_000_000, codec=H),
|
|
make_video("hdr10p-2160", range_=Video.Range.HDR10P, height=2160, bitrate=20_000_000, codec=H),
|
|
make_video("dv-360", range_=Video.Range.DV, height=360, bitrate=1_000_000, codec=H),
|
|
]
|
|
chosen = list(filter(Tracks().select_hybrid(tracks, [2160]), tracks))
|
|
assert ids(chosen) == {"hdr10p-2160", "dv-360"}
|
|
|
|
|
|
def test_select_hybrid_base_picks_highest_bitrate_then_worst_flips() -> None:
|
|
H = Video.Codec.HEVC
|
|
tracks = [
|
|
make_video("hdr10p-2160-lo", range_=Video.Range.HDR10P, height=2160, bitrate=18_000_000, codec=H),
|
|
make_video("hdr10p-2160-hi", range_=Video.Range.HDR10P, height=2160, bitrate=25_000_000, codec=H),
|
|
make_video("dv-360", range_=Video.Range.DV, height=360, bitrate=1_000_000, codec=H),
|
|
]
|
|
best = list(filter(Tracks().select_hybrid(tracks, [2160]), tracks))
|
|
assert ids(best) == {"hdr10p-2160-hi", "dv-360"}
|
|
|
|
worst = list(filter(Tracks().select_hybrid(tracks, [2160], worst=True), tracks))
|
|
assert ids(worst) == {"hdr10p-2160-lo", "dv-360"}
|
|
|
|
|
|
def test_select_hybrid_no_dv_selects_only_base() -> None:
|
|
H = Video.Codec.HEVC
|
|
tracks = [make_video("hdr10p-2160", range_=Video.Range.HDR10P, height=2160, bitrate=20_000_000, codec=H)]
|
|
chosen = list(filter(Tracks().select_hybrid(tracks, [2160]), tracks))
|
|
assert ids(chosen) == {"hdr10p-2160"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# merge_video_selections (the dedup fix)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_merge_dedups_shared_ingredient_and_deliverable() -> None:
|
|
H = Video.Codec.HEVC
|
|
dv = make_video("dv-1080", range_=Video.Range.DV, height=1080, bitrate=9_000_000, codec=H)
|
|
hdr = make_video("hdr10p-2160", range_=Video.Range.HDR10P, height=2160, bitrate=20_000_000, codec=H)
|
|
sdr = make_video("sdr-2160", range_=Video.Range.SDR, height=2160, bitrate=21_000_000, codec=H)
|
|
|
|
# dv is both the hybrid ingredient and an explicit DV deliverable (same object).
|
|
merged = Tracks.merge_video_selections([hdr, dv], [sdr, dv])
|
|
assert [t.id for t in merged] == ["hdr10p-2160", "dv-1080", "sdr-2160"]
|
|
|
|
|
|
def test_merge_preserves_order_and_keeps_distinct_dv() -> None:
|
|
H = Video.Codec.HEVC
|
|
ingredient = make_video("dv-360", range_=Video.Range.DV, height=360, bitrate=1_000_000, codec=H)
|
|
deliverable = make_video("dv-2160", range_=Video.Range.DV, height=2160, bitrate=25_000_000, codec=H)
|
|
|
|
merged = Tracks.merge_video_selections([ingredient], [deliverable])
|
|
assert [t.id for t in merged] == ["dv-360", "dv-2160"]
|
|
|
|
|
|
def test_merge_dedup_uses_track_identity_by_id() -> None:
|
|
# Tracks compare equal by id; merge must treat same-id tracks as one.
|
|
H = Video.Codec.HEVC
|
|
a = make_video("same", range_=Video.Range.DV, height=1080, bitrate=9_000_000, codec=H)
|
|
b = make_video("same", range_=Video.Range.DV, height=1080, bitrate=9_000_000, codec=H)
|
|
assert a == b
|
|
assert len(Tracks.merge_video_selections([a], [b])) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# partition_hybrid_videos
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_partition_hybrid_only_keeps_ingredients_out_of_pool(ladder: list[Video]) -> None:
|
|
"""`-r HYBRID`: HDR10+/DV are ingredients only; pool holds just SDR."""
|
|
candidates, pool = Tracks.partition_hybrid_videos(ladder, [])
|
|
assert ids(candidates) == {"hdr10p-2160", "hdr10p-1080", "dv-2160", "dv-1080", "dv-360"}
|
|
assert ids(pool) == {"sdr-2160", "sdr-1080-avc", "sdr-1080-hevc"}
|
|
|
|
|
|
def test_partition_admits_hdr10p_to_pool_when_requested(ladder: list[Video]) -> None:
|
|
"""`-r HYBRID,HDR10P`: HDR10+ tracks become standalone deliverable candidates."""
|
|
candidates, pool = Tracks.partition_hybrid_videos(ladder, [Video.Range.HDR10P])
|
|
assert {"hdr10p-2160", "hdr10p-1080"} <= ids(pool)
|
|
assert not any(t.range == Video.Range.DV for t in pool)
|
|
# Candidates are unaffected by the requested ranges.
|
|
assert ids(candidates) == {"hdr10p-2160", "hdr10p-1080", "dv-2160", "dv-1080", "dv-360"}
|
|
|
|
|
|
def test_partition_admits_dv_and_hdr10p_when_both_requested(ladder: list[Video]) -> None:
|
|
"""`-r HYBRID,HDR10P,DV`: both ranges enter the deliverable pool."""
|
|
_, pool = Tracks.partition_hybrid_videos(ladder, [Video.Range.HDR10P, Video.Range.DV])
|
|
assert {"hdr10p-2160", "hdr10p-1080", "dv-2160", "dv-1080", "dv-360"} <= ids(pool)
|
|
|
|
|
|
def test_partition_hdr10_requested_does_not_admit_hdr10p() -> None:
|
|
H = Video.Codec.HEVC
|
|
tracks = [
|
|
make_video("hdr10-2160", range_=Video.Range.HDR10, height=2160, bitrate=20_000_000, codec=H),
|
|
make_video("hdr10p-2160", range_=Video.Range.HDR10P, height=2160, bitrate=20_000_000, codec=H),
|
|
]
|
|
_, pool = Tracks.partition_hybrid_videos(tracks, [Video.Range.HDR10])
|
|
assert ids(pool) == {"hdr10-2160"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# flag_hybrid_ingredients
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def flagged(tracks: list[Video]) -> set[str]:
|
|
return {t.id for t in tracks if t.hybrid_base_only}
|
|
|
|
|
|
def test_flag_hybrid_only_flags_base_and_ingredient_dv(ladder: list[Video]) -> None:
|
|
"""`-r HYBRID`: no deliverables, so the base and the ingredient DV are both
|
|
skipped by the standalone mux loop — only the hybrid output remains."""
|
|
hybrid_selected = list(filter(Tracks().select_hybrid(ladder, [1080]), ladder))
|
|
Tracks.flag_hybrid_ingredients(hybrid_selected, [])
|
|
assert flagged(ladder) == {"hdr10p-1080", "dv-360"}
|
|
|
|
|
|
def test_flag_hybrid_plus_hdr10p_keeps_base_deliverable(ladder: list[Video]) -> None:
|
|
"""`-r HYBRID,HDR10P`: the base is also an explicit deliverable, only the
|
|
ingredient DV is skipped — hybrid + standalone HDR10+ are muxed."""
|
|
hybrid_selected = list(filter(Tracks().select_hybrid(ladder, [1080]), ladder))
|
|
base = next(t for t in ladder if t.id == "hdr10p-1080")
|
|
Tracks.flag_hybrid_ingredients(hybrid_selected, [base])
|
|
assert flagged(ladder) == {"dv-360"}
|
|
|
|
|
|
def test_flag_hybrid_plus_hdr10p_and_dv_keeps_both_deliverables(ladder: list[Video]) -> None:
|
|
"""`-r HYBRID,HDR10P,DV`: best DV is a deliverable, lowest DV stays
|
|
ingredient-only — hybrid + HDR10+ + DV are muxed."""
|
|
hybrid_selected = list(filter(Tracks().select_hybrid(ladder, [1080]), ladder))
|
|
base = next(t for t in ladder if t.id == "hdr10p-1080")
|
|
best_dv = next(t for t in ladder if t.id == "dv-1080")
|
|
Tracks.flag_hybrid_ingredients(hybrid_selected, [base, best_dv])
|
|
assert flagged(ladder) == {"dv-360"}
|
|
|
|
|
|
def test_flag_single_dv_rendition_as_deliverable_stays_unflagged() -> None:
|
|
"""`-r HYBRID,DV` with one DV rendition: the same track is ingredient and
|
|
deliverable, so it must still be muxed standalone."""
|
|
H = Video.Codec.HEVC
|
|
tracks = [
|
|
make_video("hdr10p-1080", range_=Video.Range.HDR10P, height=1080, bitrate=9_000_000, codec=H),
|
|
make_video("dv-1080", range_=Video.Range.DV, height=1080, bitrate=9_000_000, codec=H),
|
|
]
|
|
hybrid_selected = list(filter(Tracks().select_hybrid(tracks, [1080]), tracks))
|
|
dv = next(t for t in tracks if t.id == "dv-1080")
|
|
Tracks.flag_hybrid_ingredients(hybrid_selected, [dv])
|
|
assert flagged(tracks) == {"hdr10p-1080"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# documented end-state for the reported command
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_hybrid_plus_dv_deliverable_end_state(ladder: list[Video]) -> None:
|
|
"""`-r SDR,HDR10P,DV,HYBRID -q 2160,1080`: hybrid base + lowest DV ingredient,
|
|
de-duplicated against the best-DV-per-resolution deliverables and SDR."""
|
|
quality = [2160, 1080]
|
|
|
|
hybrid_selected = list(filter(Tracks().select_hybrid(ladder, quality), ladder))
|
|
|
|
# Deliverables: best DV and SDR per requested resolution (what the dl Cartesian picks).
|
|
def best_per_res(range_: Video.Range, codec: Video.Codec | None = None) -> list[Video]:
|
|
out = []
|
|
for res in quality:
|
|
cands = [t for t in ladder if t.range == range_ and t.height == res and (codec is None or t.codec == codec)]
|
|
if cands:
|
|
out.append(max(cands, key=lambda t: t.bitrate))
|
|
return out
|
|
|
|
dv_deliverable = best_per_res(Video.Range.DV)
|
|
sdr_deliverable = [t for t in ladder if t.range == Video.Range.SDR]
|
|
|
|
final = Tracks.merge_video_selections(hybrid_selected, dv_deliverable + sdr_deliverable)
|
|
|
|
# No duplicates, and the lowest DV ingredient coexists with the DV deliverables.
|
|
assert len(final) == len({t.id for t in final})
|
|
assert "dv-360" in ids(final) # ingredient retained for hybrid build
|
|
assert {"dv-2160", "dv-1080"} <= ids(final) # standalone DV deliverables
|
|
assert {"hdr10p-2160", "hdr10p-1080"} <= ids(final)
|
|
assert {"sdr-2160", "sdr-1080-avc", "sdr-1080-hevc"} <= ids(final)
|