"""Serving a master playlist for inputstream.adaptive, so quality can be chosen
from Kodi's player OSD instead of before playback starts.
Kodi only exposes a video-stream picker when ISA is driving the stream, and ISA
needs a MASTER (several variants) rather than the single variant we serve today.
This was tried and reverted once (org labs #2): ISA fetched the variant playlist
without a Referer and the CDN returned 403. Re-measured 2026-08-10 -- master,
variant and segment all return 200/206 with no headers whatsoever -- so the
blocker is gone.
Two properties the reverted attempt established and this must keep:
- 2160p never reaches the client. vidcore lists it first at ~15.8Mbps and the
Pi's V4L2 decoder cannot decode it; an adaptive client on a fast link would
climb straight into it.
- The manifest is rewritten, the video bytes are never proxied.
Run: .venv/bin/python3 -m pytest test_vidcore_adaptive.py -q
"""
import pytest
import addon
MASTER_URL = "https://cdn.example/vd/TOKEN/master.m3u8"
MASTER = """#EXTM3U
#EXT-X-STREAM-INF:PROGRAM-ID=1,BANDWIDTH=15800000,RESOLUTION=3840x2160,CODECS="avc1.640033"
sd/52/index-s2160p-v1-a1.m3u8
#EXT-X-STREAM-INF:PROGRAM-ID=1,BANDWIDTH=6047207,RESOLUTION=1920x1080,CODECS="avc1.640028"
sd/52/index-s1080p-v1-a1.m3u8
#EXT-X-STREAM-INF:PROGRAM-ID=1,BANDWIDTH=3049124,RESOLUTION=1280x720,CODECS="avc1.4d401f"
sd/52/index-s720p-v1-a1.m3u8
#EXT-X-STREAM-INF:PROGRAM-ID=1,BANDWIDTH=1287760,RESOLUTION=852x480,CODECS="avc1.4d401e"
sd/52/index-s480p-v1-a1.m3u8
"""
MEDIA = "#EXTM3U\n#EXT-X-PLAYLIST-TYPE:VOD\n#EXTINF:6.0,\nseg-1.ts\n#EXT-X-ENDLIST\n"
HIGH_ONLY = """#EXTM3U
#EXT-X-STREAM-INF:BANDWIDTH=15800000,RESOLUTION=3840x2160
only-2160.m3u8
"""
class _Resp:
def __init__(self, text):
self.text = text
def raise_for_status(self):
pass
class _Body:
text = MASTER
_body = _Body()
@pytest.fixture(autouse=True)
def no_network(monkeypatch):
_body.text = MASTER
monkeypatch.setattr(addon, "_variant_cache", {})
monkeypatch.setattr(addon.requests, "get", lambda *a, **k: _Resp(_body.text))
yield
_body.text = MASTER
def _parse(body):
"""-> [(inf_line, uri)] pairs."""
lines = [l for l in body.splitlines() if l.strip()]
assert lines[0] == "#EXTM3U", body
return [(lines[i], lines[i + 1]) for i in range(1, len(lines) - 1, 2)]
# --- the manifest we hand to ISA -------------------------------------------
def test_master_keeps_every_variant_at_or_below_the_cap():
pairs = _parse(addon._master_playlist(MASTER_URL))
assert len(pairs) == 3
assert [u for _, u in pairs] == [
"https://cdn.example/vd/TOKEN/sd/52/index-s1080p-v1-a1.m3u8",
"https://cdn.example/vd/TOKEN/sd/52/index-s720p-v1-a1.m3u8",
"https://cdn.example/vd/TOKEN/sd/52/index-s480p-v1-a1.m3u8",
]
def test_4k_is_stripped():
body = addon._master_playlist(MASTER_URL)
assert "2160" not in body
assert "15800000" not in body
def test_uris_are_absolute():
"""We serve this body ourselves, so a relative URI would resolve against
OUR host and 404 -- the bytes live on the CDN."""
for _, uri in _parse(addon._master_playlist(MASTER_URL)):
assert uri.startswith("https://cdn.example/")
def test_stream_inf_attributes_are_preserved():
"""ISA picks by BANDWIDTH/RESOLUTION; dropping them loses the ladder."""
body = addon._master_playlist(MASTER_URL)
assert "BANDWIDTH=6047207" in body and "RESOLUTION=1920x1080" in body
assert 'CODECS="avc1.640028"' in body
def test_a_media_playlist_yields_no_master():
_body.text = MEDIA
assert addon._master_playlist(MASTER_URL) is None
def test_fetch_failure_yields_no_master(monkeypatch):
def boom(*a, **k):
raise RuntimeError("connection reset")
monkeypatch.setattr(addon.requests, "get", boom)
assert addon._master_playlist(MASTER_URL) is None
def test_never_returns_an_empty_ladder():
"""If everything is above the cap, keep the lowest rather than serve a
master with no variants at all."""
_body.text = HIGH_ONLY
pairs = _parse(addon._master_playlist(MASTER_URL))
assert len(pairs) == 1
assert pairs[0][1].endswith("only-2160.m3u8")
# --- the route --------------------------------------------------------------
@pytest.fixture
def client(monkeypatch, tmp_path):
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: MASTER_URL)
monkeypatch.setattr(addon, "_prefetch_vidcore", lambda *a, **k: None)
# The route checks for a local copy BEFORE resolving, so without these the
# test depends on the developer's disk: tmdb:138502 S02E07 is a real file in
# the real mp4-cache on the machine this was written on, and h=720 found it
# and redirected to /mp4/ instead of the variant. Working out the library
# path also reaches TMDB over the network, from a unit test.
monkeypatch.setattr(addon, "MP4_DIR", str(tmp_path / "cache"))
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path / "lib"))
monkeypatch.setattr(addon, "_library_meta", lambda *a, **k: None)
monkeypatch.setattr(addon, "_ensure_ingest", lambda *a, **k: None)
return addon.app.test_client()
def test_auto_serves_the_manifest_itself(client):
r = client.get("/vidcore/series/tmdb:138502:2:7.m3u8?h=auto")
assert r.status_code == 200
assert "mpegurl" in r.headers["Content-Type"]
body = r.get_data(as_text=True)
assert body.startswith("#EXTM3U")
assert "2160" not in body
assert len(_parse(body)) == 3
def test_a_fixed_height_still_redirects(client):
"""The existing rows must be untouched by this."""
r = client.get("/vidcore/series/tmdb:138502:2:7.m3u8?h=720")
assert r.status_code == 302
assert r.headers["Location"].endswith("index-s720p-v1-a1.m3u8")
def test_auto_falls_back_to_a_variant_redirect(client):
"""If the ladder can't be built, still play something rather than 404."""
_body.text = MEDIA
r = client.get("/vidcore/series/tmdb:138502:2:7.m3u8?h=auto")
assert r.status_code == 302
def test_auto_works_for_movies(client):
r = client.get("/vidcore/movie/tmdb:1078605.m3u8?h=auto")
assert r.status_code == 200
assert r.get_data(as_text=True).startswith("#EXTM3U")
def test_unresolvable_title_still_404s(client, monkeypatch):
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: None)
assert client.get("/vidcore/series/tmdb:1:1:1.m3u8?h=auto").status_code == 404
# --- the rows offered ------------------------------------------------------
def test_no_adaptive_row_is_offered(monkeypatch):
"""ISA cannot play this CDN's audio: video and audio are muxed into one
fMP4 track set and ISA's HLS reader feeds only the first track, so the aac
decoder opens and never receives a sample. Offering the row would offer a
silent stream. The ?h=auto endpoint stays for a future non-muxed source."""
monkeypatch.setattr(addon, '_prefetch_vidcore', lambda *a, **k: None)
with addon.app.test_request_context('/stream/series/fxl:1:2:7.json'):
out = []
addon._append_vidcore_row(out, 'series', 'tmdb:1', 2, 7)
assert len(out) == len(addon.VIDCORE_HEIGHTS)
assert not any('adaptive=1' in r['url'] or 'h=auto' in r['url'] for r in out)
assert all('S02E07' in r['title'] for r in out)