"""Watch a library copy in the browser, with what it actually is (labs#13).
The point is judging quality, so the page states the real numbers: width x
height (1920x800 is not "800p"), codec, frame rate, video bitrate, and every
audio track -- flagging the ones a browser plays as silence.
That flag is MEASURED, not assumed (Chromium, 2026-10-04): canPlayType for
ec-3 and ac-3 is "", an eac3-only file decodes 0 audio bytes, and an eac3 +
aac file ALSO decodes 0 -- the browser plays the first audio track and does
not fall back to a later one it could decode.
/web/file/ is a separate route from /mp4/ on purpose: /mp4/ queues the next
episode and starts an ingest when there is no copy. Watching here must never
fetch anything.
Run: .venv/bin/python3 -m pytest test_web_watch.py -q
"""
import json
import os
import pytest
import addon
TMDB = {
"tv/1": {"name": "Show", "first_air_date": "2020-01-01", "seasons": []},
"tv/1/season/2/episode/6": {"name": "Six", "runtime": 30},
"tv/1/season/2/episode/7": {"name": "Seven", "runtime": 30},
"movie/2": {"title": "Film", "release_date": "2021-05-05", "runtime": 30},
"tv/1/season/2": {"episodes": [
{"episode_number": 6, "name": "Six", "air_date": "2020-01-01"}]},
}
PROBE_AAC = {
"format": {"duration": "1800.0", "size": "1432636396", "bit_rate": "6370000"},
"streams": [
{"index": 0, "codec_type": "video", "codec_name": "h264", "profile": "High",
"width": 1920, "height": 800, "avg_frame_rate": "24000/1001",
"bit_rate": "6100000", "pix_fmt": "yuv420p"},
{"index": 1, "codec_type": "audio", "codec_name": "aac", "channels": 2,
"bit_rate": "128000", "tags": {"language": "eng"}}]}
PROBE_EAC3_FIRST = {
"format": {"duration": "1800.0", "size": "1", "bit_rate": "10800000"},
"streams": [
{"index": 0, "codec_type": "video", "codec_name": "h264", "profile": "High",
"width": 1920, "height": 802, "avg_frame_rate": "24/1", "bit_rate": "10000000"},
{"index": 1, "codec_type": "audio", "codec_name": "eac3", "channels": 6,
"tags": {"language": "eng"}},
{"index": 2, "codec_type": "audio", "codec_name": "aac", "channels": 2,
"tags": {"language": "hin"}}]}
@pytest.fixture(autouse=True)
def isolated(tmp_path, monkeypatch):
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path))
monkeypatch.setattr(addon, "INGEST_QUEUE_FILE", str(tmp_path / "queue.json"))
monkeypatch.setattr(addon, "INGEST_HISTORY_FILE", str(tmp_path / "history.jsonl"))
monkeypatch.setattr(addon, "LIBRARY_VERIFIED_FILE", str(tmp_path / "verified.json"))
monkeypatch.setattr(addon, "MP4_DIR", str(tmp_path / "mp4-cache"))
monkeypatch.setattr(addon, "_duration_cache", {})
monkeypatch.setattr(addon, "_quality_cache", {})
monkeypatch.setattr(addon, "_library_meta_cache", {})
monkeypatch.setattr(addon, "_mp4_jobs", {})
monkeypatch.setattr(addon, "_enough_space", lambda: True)
monkeypatch.setattr(addon, "_tmdb_get", lambda path, params=None: TMDB[path])
monkeypatch.setattr(addon, "_media_duration", lambda p: 1800.0)
probe = {"value": PROBE_AAC}
monkeypatch.setattr(addon, "_ffprobe_file", lambda p: probe["value"])
def refuse(*a, **k):
raise AssertionError("watching started a download")
monkeypatch.setattr(addon, "_start_mp4_job", refuse)
monkeypatch.setattr(addon, "_queue_next_episode", refuse)
monkeypatch.setattr(addon, "_ensure_ingest", refuse)
yield probe
@pytest.fixture
def client():
return addon.app.test_client()
def _hold(kind, season=0, episode=0, tmdb="1", body=b"0123456789" * 100):
meta = addon._library_meta(kind, f"tmdb:{tmdb}", season, episode)
path = addon._library_path(kind, meta["title"], meta["year"], meta["tmdb_id"],
season, episode, meta["episode_title"])
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "wb") as f:
f.write(body)
return path
# --- the file route ---------------------------------------------------------
def test_the_file_route_serves_a_held_episode(client):
_hold("series", 2, 6)
r = client.get("/web/file/series/1/2/6.mp4")
assert r.status_code == 200
assert r.mimetype == "video/mp4"
assert r.get_data() == b"0123456789" * 100
def test_the_file_route_answers_range_requests(client):
"""Range is what makes the browser player seekable."""
_hold("series", 2, 6)
r = client.get("/web/file/series/1/2/6.mp4", headers={"Range": "bytes=10-19"})
assert r.status_code == 206
assert r.get_data() == b"0123456789"
def test_the_file_route_serves_a_movie(client):
_hold("movie", tmdb="2")
assert client.get("/web/file/movie/2.mp4").status_code == 200
def test_the_file_route_404s_without_fetching_when_not_held(client):
assert client.get("/web/file/series/1/2/7.mp4").status_code == 404
def test_the_file_route_rejects_bad_ids(client):
assert client.get("/web/file/series/x/2/6.mp4").status_code == 404
assert client.get("/web/file/anime/1.mp4").status_code == 404
# --- the watch page ---------------------------------------------------------
def test_watch_page_plays_the_file(client):
_hold("series", 2, 6)
body = client.get("/web/watch/series/1/2/6").get_data(as_text=True)
assert "<video" in body
assert "src='/web/file/series/1/2/6.mp4'" in body
def test_watch_page_states_the_real_numbers(client):
_hold("series", 2, 6)
body = client.get("/web/watch/series/1/2/6").get_data(as_text=True)
assert "1920×800" in body
assert "h264" in body and "High" in body
assert "23.98 fps" in body
assert "6.1 Mbps" in body
assert "aac" in body and "eng" in body
def test_watch_page_reports_length_against_tmdb(client):
_hold("series", 2, 6)
body = client.get("/web/watch/series/1/2/6").get_data(as_text=True)
assert "30m 00s" in body and "100%" in body
def test_an_eac3_first_track_is_flagged_silent(client, isolated):
isolated["value"] = PROBE_EAC3_FIRST
_hold("series", 2, 6)
body = client.get("/web/watch/series/1/2/6").get_data(as_text=True)
assert "without sound" in body
assert "eac3" in body
def test_an_aac_copy_is_not_flagged(client):
_hold("series", 2, 6)
body = client.get("/web/watch/series/1/2/6").get_data(as_text=True)
assert "without sound" not in body
def test_the_source_comes_from_history(client):
path = _hold("series", 2, 6)
with open(addon.INGEST_HISTORY_FILE, "w") as f:
f.write(json.dumps({"ts": 1, "outcome": "done", "source": "hdhub",
"dest": path}) + "\n")
body = client.get("/web/watch/series/1/2/6").get_data(as_text=True)
assert "hdhub" in body
def test_watching_a_title_not_held_is_404(client):
r = client.get("/web/watch/series/1/2/7")
assert r.status_code == 404
def test_a_held_title_links_to_watch(client):
_hold("movie", tmdb="2")
body = client.get("/web/title/movie/2").get_data(as_text=True)
assert "href='/web/watch/movie/2'" in body
def test_a_held_episode_links_to_watch(client):
_hold("series", 2, 6)
body = client.get("/web/title/series/1?s=2").get_data(as_text=True)
assert "href='/web/watch/series/1/2/6'" in body
def test_quality_probe_is_cached_by_file_identity(client, monkeypatch):
path = _hold("series", 2, 6)
calls = []
monkeypatch.setattr(addon, "_ffprobe_file", lambda p: calls.append(p) or PROBE_AAC)
addon._quality(path)
addon._quality(path)
assert len(calls) == 1
def test_a_partial_copy_is_neither_served_nor_offered(client, monkeypatch):
"""The completeness gate applies here too: 30 min of a 100 min film is the
2026-08-31 truncation shape, and must not be judged as 'the copy'."""
monkeypatch.setitem(TMDB, "movie/2", dict(TMDB["movie/2"], runtime=100))
_hold("movie", tmdb="2")
assert client.get("/web/file/movie/2.mp4").status_code == 404
assert client.get("/web/watch/movie/2").status_code == 404
# Measured 2026-10-04 in Chromium: the browser plays the DEFAULT-flagged audio
# track, not the first. aac first + not default = 0 audio bytes; eac3 first +
# aac default = audible. (The earlier "first track" wording was confounded:
# that test's eac3 track was both first and default.)
PROBE_EAC3_FIRST_AAC_DEFAULT = {
"format": {"duration": "1800.0"},
"streams": [
{"index": 0, "codec_type": "video", "codec_name": "h264", "width": 1920, "height": 1080},
{"index": 1, "codec_type": "audio", "codec_name": "eac3", "channels": 6,
"disposition": {"default": 0}, "tags": {"language": "eng"}},
{"index": 2, "codec_type": "audio", "codec_name": "aac", "channels": 2,
"disposition": {"default": 1}, "tags": {"language": "eng"}}]}
PROBE_AAC_FIRST_EAC3_DEFAULT = {
"format": {"duration": "1800.0"},
"streams": [
{"index": 0, "codec_type": "video", "codec_name": "h264", "width": 1920, "height": 1080},
{"index": 1, "codec_type": "audio", "codec_name": "aac", "channels": 2,
"disposition": {"default": 0}, "tags": {"language": "eng"}},
{"index": 2, "codec_type": "audio", "codec_name": "eac3", "channels": 6,
"disposition": {"default": 1}, "tags": {"language": "eng"}}]}
def test_silence_follows_the_default_track_not_the_first(client, isolated):
_hold("series", 2, 6)
isolated["value"] = PROBE_EAC3_FIRST_AAC_DEFAULT
body = client.get("/web/watch/series/1/2/6").get_data(as_text=True)
assert "without sound" not in body
addon._quality_cache.clear()
isolated["value"] = PROBE_AAC_FIRST_EAC3_DEFAULT
body = client.get("/web/watch/series/1/2/6").get_data(as_text=True)
assert "without sound" in body and "default audio track" in body