"""Preview the stream a running download is copying (labs#13).
Watch it, judge the quality, then let the download finish or Cancel. The
preview plays the SAME source the job copies -- not a second resolve -- because
the job already holds the URL. (The .part itself is unplayable: +faststart
writes the moov atom last.)
Measured 2026-10-04 on stortron (Inception, moon.zenoak.top):
- vidcore variant playlists are REFERER-GATED: 403 bare, 200 with
`Referer: https://vidcore.net/` -- 6/6 on one egress. CORS is `*`. A
browser cannot set Referer, so vidcore previews go through a relay here
that adds it.
- HdHub hands out direct files (no referer). Chromium plays h264 .mkv as
well as .mp4 (aac audible; eac3 silent, as for Watch), so the browser
plays those straight from the source.
The relay is NOT an open proxy: it only fetches from the host of the live
job's own playlist.
Run: .venv/bin/python3 -m pytest test_web_preview.py -q
"""
import os
import threading
from urllib.parse import quote
import pytest
import addon
KEY = "series:tmdb:1:2:6:1080"
VARIANT = "https://cdn.example/vd/abc/sd/720.m3u8"
PLAYLIST = """#EXTM3U
#EXT-X-VERSION:7
#EXT-X-MAP:URI="init-720.mp4"
#EXTINF:6.0,
seg-1.html?token=a
#EXTINF:6.0,
https://cdn.example/vd/abc/sd/seg-2.html?token=b
#EXT-X-ENDLIST
"""
@pytest.fixture(autouse=True)
def isolated(tmp_path, monkeypatch):
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path / "lib"))
monkeypatch.setattr(addon, "INGEST_QUEUE_FILE", str(tmp_path / "q.json"))
monkeypatch.setattr(addon, "_job_info", {})
monkeypatch.setattr(addon, "_mp4_jobs", {})
monkeypatch.setattr(addon, "_enough_space", lambda: True)
yield
@pytest.fixture
def client():
return addon.app.test_client()
def _live(preview=None, phase="downloading"):
addon._job_info[KEY] = {
"phase": phase, "source": "vidcore", "dest": None, "started": 0,
"kind": "series", "ident": "tmdb:1", "season": 2, "episode": 6,
"height": 1080, "label": "S02E06", "title": "Show"}
if preview:
# A copy: the relay records its allow-list on the preview dict, and a
# shared module-level dict let one test's allow-list pass another.
addon._job_info[KEY]["preview"] = dict(preview)
addon._mp4_jobs[KEY] = "running"
HLS = {"type": "hls", "url": VARIANT, "height": 720, "bandwidth": 3121132}
FILE = {"type": "file", "url": "https://pub-x.r2.dev/Show.S02E06.mkv?X-Amz=1",
"filename": "Show.S02E06.1080p.mkv", "height": 1080, "size": 2_000_000_000,
"probe": {"format": {"duration": "1800"}, "streams": [
{"codec_type": "video", "codec_name": "h264", "height": 1080},
{"codec_type": "audio", "codec_name": "eac3", "tags": {"language": "eng"}}]}}
class FakeResp:
def __init__(self, body=b"", status=200, headers=None):
self.content = body
self.text = body.decode() if isinstance(body, bytes) else body
self.status_code = status
self.headers = headers or {}
def raise_for_status(self):
if self.status_code >= 400:
raise addon.requests.HTTPError(str(self.status_code))
def iter_content(self, chunk_size=1):
yield self.content
def close(self):
pass
# --- the job records what it copies -----------------------------------------
META = {"tmdb_id": "1", "title": "Show", "year": "2020",
"episode_title": "Six", "runtime": 30}
def _ingest_env(monkeypatch, tmp_path):
monkeypatch.setattr(addon, "_library_meta", lambda *a, **k: dict(META))
monkeypatch.setattr(addon, "_library_has", lambda *a, **k: False)
monkeypatch.setattr(addon, "_fetch_subtitles_for", lambda *a, **k: None)
monkeypatch.setattr(addon, "_notify_jellyfin", lambda *a, **k: None)
addon._job_info[KEY] = {}
def test_a_vidcore_job_records_the_variant_it_copies(monkeypatch, tmp_path):
_ingest_env(monkeypatch, tmp_path)
master = "https://cdn.example/vd/abc/ma"
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: master)
monkeypatch.setattr(addon, "_variants", lambda m: [
(1080, 6152556, "sd/1080.m3u8", ""), (720, 3121132, "sd/720.m3u8", "")])
monkeypatch.setattr(addon, "_cap_variant", lambda m, h=None: VARIANT)
seen = {}
def remux(url, dest):
seen.update(addon._job_info[KEY].get("preview") or {})
os.makedirs(os.path.dirname(dest), exist_ok=True)
open(dest, "wb").write(b"x")
monkeypatch.setattr(addon, "_remux_to_mp4", remux)
addon._run_ingest("series", "tmdb:1", 2, 6, 1080)
assert seen == {"type": "hls", "url": VARIANT, "height": 720,
"bandwidth": 3121132}
def test_an_hdhub_job_records_the_file_it_copies(monkeypatch, tmp_path):
_ingest_env(monkeypatch, tmp_path)
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: None)
monkeypatch.setattr(addon, "_hdhub_enabled", lambda: True)
f = addon.HdHubFile(url=FILE["url"], filename=FILE["filename"], height=1080,
size=FILE["size"], host="pub-x.r2.dev", english_only=True)
monkeypatch.setattr(addon, "_hdhub_pick", lambda *a, **k: (f, FILE["probe"]))
seen = {}
def copy(url, probe, dest):
seen.update(addon._job_info[KEY].get("preview") or {})
os.makedirs(os.path.dirname(dest), exist_ok=True)
open(dest, "wb").write(b"x")
monkeypatch.setattr(addon, "_copy_progressive", copy)
addon._run_ingest("series", "tmdb:1", 2, 6, 1080)
assert seen["type"] == "file" and seen["url"] == FILE["url"]
assert seen["filename"] == FILE["filename"] and seen["probe"] == FILE["probe"]
# --- the page ----------------------------------------------------------------
def test_an_hls_preview_plays_through_the_relay_with_hls_js(client):
_live(HLS)
body = client.get(f"/web/preview?key={quote(KEY)}").get_data(as_text=True)
assert "hls.js/1.6.16/hls.min.js" in body
assert "/web/relay/playlist?key=" + quote(KEY, safe="") in body
assert "720p" in body and "3.1 Mbps" in body
def test_a_file_preview_plays_the_source_directly(client):
_live(FILE)
body = client.get(f"/web/preview?key={quote(KEY)}").get_data(as_text=True)
assert "src='https://pub-x.r2.dev/Show.S02E06.mkv?X-Amz=1'" in body
assert "hls.min.js" not in body
assert "without sound" in body # eac3 first track
assert "Show.S02E06.1080p.mkv" in body
def test_the_preview_page_has_cancel_and_does_not_auto_refresh(client):
"""A refresh would restart the video under the viewer."""
_live(HLS)
body = client.get(f"/web/preview?key={quote(KEY)}").get_data(as_text=True)
assert "action='/web/cancel'" in body and f"value='{KEY}'" in body
assert "http-equiv=refresh" not in body
def test_before_the_source_is_known_the_page_waits_and_refreshes(client):
_live(None, phase="resolving")
body = client.get(f"/web/preview?key={quote(KEY)}").get_data(as_text=True)
assert "resolving" in body and "http-equiv=refresh" in body
assert "<video" not in body
def test_a_finished_job_sends_you_to_the_title(client):
addon._mp4_jobs[KEY] = "done"
r = client.get(f"/web/preview?key={quote(KEY)}")
assert r.status_code == 303
assert r.headers["Location"].endswith("/web/title/series/1?s=2")
def test_an_unknown_job_is_404(client):
assert client.get("/web/preview?key=nope").status_code == 404
def test_the_downloads_page_links_a_previewable_job(client):
_live(HLS)
body = client.get("/web/downloads").get_data(as_text=True)
assert f"/web/preview?key={quote(KEY, safe='')}" in body
def test_the_title_page_links_a_previewable_job(client, monkeypatch):
monkeypatch.setattr(addon, "_tmdb_get", lambda path, params=None: {
"tv/1": {"name": "Show", "first_air_date": "2020-01-01",
"seasons": [{"season_number": 2}]},
"tv/1/season/2": {"episodes": [
{"episode_number": 6, "name": "Six", "air_date": "2020-01-01"}]},
"tv/1/season/2/episode/6": {"name": "Six", "runtime": 30}}[path])
_live(HLS)
body = client.get("/web/title/series/1?s=2").get_data(as_text=True)
assert f"/web/preview?key={quote(KEY, safe='')}" in body
# --- the relay ---------------------------------------------------------------
def test_the_playlist_is_fetched_with_the_vidcore_referer_and_rewritten(client, monkeypatch):
_live(HLS)
calls = []
def get(url, headers=None, timeout=None, stream=False):
calls.append((url, headers))
return FakeResp(PLAYLIST.encode())
monkeypatch.setattr(addon.requests, "get", get)
r = client.get(f"/web/relay/playlist?key={quote(KEY)}")
assert r.status_code == 200
assert r.mimetype == "application/vnd.apple.mpegurl"
assert calls[0][0] == VARIANT
assert calls[0][1]["Referer"] == "https://vidcore.net/"
body = r.get_data(as_text=True)
k = quote(KEY, safe="")
assert f'URI="/web/relay/seg?key={k}&u=' + quote("https://cdn.example/vd/abc/sd/init-720.mp4", safe="") + '"' in body
assert f"/web/relay/seg?key={k}&u=" + quote("https://cdn.example/vd/abc/sd/seg-1.html?token=a", safe="") in body
assert f"/web/relay/seg?key={k}&u=" + quote("https://cdn.example/vd/abc/sd/seg-2.html?token=b", safe="") in body
assert "#EXT-X-ENDLIST" in body
def test_a_segment_is_relayed_with_the_referer_and_range(client, monkeypatch):
_live(HLS)
calls = []
def get(url, headers=None, timeout=None, stream=False):
calls.append((url, headers))
if url == VARIANT:
return FakeResp(PLAYLIST.encode())
return FakeResp(b"MOOF", status=206,
headers={"Content-Type": "text/html",
"Content-Range": "bytes 0-3/10"})
monkeypatch.setattr(addon.requests, "get", get)
client.get(f"/web/relay/playlist?key={quote(KEY)}") # as hls.js does first
calls.clear()
u = "https://cdn.example/vd/abc/sd/seg-1.html?token=a"
r = client.get(f"/web/relay/seg?key={quote(KEY)}&u={quote(u, safe='')}",
headers={"Range": "bytes=0-3"})
assert r.status_code == 206
assert r.get_data() == b"MOOF"
assert r.headers["Content-Range"] == "bytes 0-3/10"
assert calls[0][0] == u
assert calls[0][1]["Referer"] == "https://vidcore.net/"
assert calls[0][1]["Range"] == "bytes=0-3"
def test_the_relay_is_not_an_open_proxy(client, monkeypatch):
_live(HLS)
monkeypatch.setattr(addon.requests, "get",
lambda *a, **k: pytest.fail("fetched a foreign host"))
u = "https://evil.example/x"
# Never listed by any playlist, so never fetched.
assert client.get(f"/web/relay/seg?key={quote(KEY)}&u={quote(u, safe='')}").status_code == 403
def test_the_relay_needs_a_live_hls_job(client, monkeypatch):
monkeypatch.setattr(addon.requests, "get",
lambda *a, **k: pytest.fail("fetched without a job"))
assert client.get(f"/web/relay/playlist?key={quote(KEY)}").status_code == 404
_live(FILE)
assert client.get(f"/web/relay/playlist?key={quote(KEY)}").status_code == 404
def test_an_upstream_error_is_passed_on_not_crashed(client, monkeypatch):
_live(HLS)
monkeypatch.setattr(addon.requests, "get",
lambda *a, **k: FakeResp(b"no", status=403))
assert client.get(f"/web/relay/playlist?key={quote(KEY)}").status_code == 502
def test_segments_on_a_different_cdn_host_are_relayed(client, monkeypatch):
"""Measured live 2026-10-04: the playlist came from one host and its
segments from another (paperlake.top). The relay allows exactly the URLs
the job's own relayed playlist listed -- still not an open proxy."""
_live(HLS)
other = "https://paperlake.top/vd/x/seg-9.html?token=z"
body = PLAYLIST.replace("seg-1.html?token=a", other)
calls = []
def get(url, headers=None, timeout=None, stream=False):
calls.append(url)
return FakeResp(body.encode() if url == VARIANT else b"SEG", status=200)
monkeypatch.setattr(addon.requests, "get", get)
assert client.get(f"/web/relay/playlist?key={quote(KEY)}").status_code == 200
r = client.get(f"/web/relay/seg?key={quote(KEY)}&u={quote(other, safe='')}")
assert r.status_code == 200 and r.get_data() == b"SEG"
stranger = "https://paperlake.top/vd/x/not-listed.html"
assert client.get(f"/web/relay/seg?key={quote(KEY)}&u={quote(stranger, safe='')}").status_code == 403
def test_a_segment_before_any_playlist_is_refused(client, monkeypatch):
_live(HLS)
monkeypatch.setattr(addon.requests, "get",
lambda *a, **k: pytest.fail("fetched an unlisted URL"))
u = "https://cdn.example/vd/abc/sd/seg-1.html?token=a"
assert client.get(f"/web/relay/seg?key={quote(KEY)}&u={quote(u, safe='')}").status_code == 403