"""Library ingest: fetched copies land in the Jellyfin library, not a cache.
A finished copy is not cache -- it is library content, and it is permanent. It
goes to /mnt/usb/{Shows,Movies} in the layout Jellyfin identifies on, where it is
indexed and served like any other media and Kodi plays it through the same
plugin:// path as the rest of the library. Seek, resume and pause all work there
natively, which is what makes the muxed-fMP4 problem stop mattering for anything
we have fetched.
Naming carries the TMDB id explicitly ([tmdbid-138502]). The 2026-07-28 library
work found Jellyfin's own identification is ~92% right and not safe to rename
blindly -- but that was retrofitting unknown files. Here the title was resolved
FROM its TMDB id, so identity is asserted rather than inferred.
The volume is exFAT (verified 2026-08-11: findmnt -> exfat /dev/sdb1). That
forbids " * / : < > ? \\ | in names, which is why the existing library holds
`Ghost in the Shell：STAND ALONE COMPLEX` with a FULLWIDTH colon -- an existing
convention this follows rather than invents.
Run: .venv/bin/python3 -m pytest test_library_ingest.py -q
"""
import os
import pytest
import addon
@pytest.fixture(autouse=True)
def tmp_library(tmp_path, monkeypatch):
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path))
# Redirected for every test, not just the ones that care: the verified store
# lives beside addon.py in the deployed checkout, and a test that wrote to
# the real one would quietly mark production files as trustworthy.
monkeypatch.setattr(addon, "LIBRARY_VERIFIED_FILE",
os.path.join(str(tmp_path), "library_verified.json"))
monkeypatch.setattr(addon, "_duration_cache", {})
yield
# --- character safety (exFAT) -----------------------------------------------
def test_forbidden_exfat_characters_are_replaced_not_dropped():
"""exFAT rejects " * / : < > ? \\ | outright. Replacing preserves
readability; dropping would turn "Ep: One" into "Ep One" and lose the
distinction from a real title."""
out = addon._safe_component('A:B/C\\D*E?F"G<H>I|J')
for bad in '"*/:<>?\\|':
assert bad not in out
assert "A" in out and "J" in out
def test_colon_becomes_the_fullwidth_colon_already_used_in_the_library():
"""Matches `Ghost in the Shell：STAND ALONE COMPLEX` on the volume today."""
assert addon._safe_component("Ghost in the Shell: SAC") == \
"Ghost in the Shell： SAC"
def test_control_characters_and_edge_whitespace_are_stripped():
assert "\n" not in addon._safe_component("Two\nLines")
assert addon._safe_component(" padded ") == "padded"
def test_trailing_dots_are_stripped():
"""A component ending in '.' is rejected or silently altered on
exFAT/Windows, which would break the later os.replace."""
assert not addon._safe_component("Episode.").endswith(".")
def test_a_component_never_becomes_empty():
"""An empty component would collapse the path and put the file in the
parent directory."""
assert addon._safe_component("///") != ""
assert addon._safe_component("") != ""
# --- series naming ----------------------------------------------------------
def test_series_path_is_the_jellyfin_layout():
p = addon._library_path("series", "X-Men '97", "2024", "138502",
season=2, episode=7, episode_title="Bright Eyes")
rel = os.path.relpath(p, addon.LIBRARY_ROOT)
assert rel == os.path.join(
"Shows",
"X-Men '97 (2024) [tmdbid-138502]",
"Season 02",
"X-Men '97 (2024) - S02E07 - Bright Eyes.mp4",
)
def test_season_and_episode_are_zero_padded():
p = addon._library_path("series", "Show", "2020", "1",
season=1, episode=3, episode_title="T")
assert "Season 01" in p and "S01E03" in p
def test_double_digit_seasons_are_not_truncated():
p = addon._library_path("series", "Show", "2020", "1",
season=12, episode=134, episode_title="T")
assert "Season 12" in p and "S12E134" in p
def test_a_missing_episode_title_still_produces_a_valid_name():
"""TMDB does not always carry an episode title. The file must still be
named unambiguously -- Jellyfin keys on the SxxExx token."""
p = addon._library_path("series", "Show", "2020", "1",
season=1, episode=3, episode_title="")
assert p.endswith("Show (2020) - S01E03.mp4")
def test_a_slash_in_an_episode_title_cannot_create_a_directory():
p = addon._library_path("series", "Show", "2020", "1", season=1, episode=1,
episode_title="Part 1/2")
assert os.path.basename(p) == "Show (2020) - S01E01 - Part 1-2.mp4"
# --- movie naming -----------------------------------------------------------
def test_movie_path_is_the_jellyfin_layout():
p = addon._library_path("movie", "Blade Runner", "1982", "78")
rel = os.path.relpath(p, addon.LIBRARY_ROOT)
assert rel == os.path.join(
"Movies",
"Blade Runner (1982) [tmdbid-78]",
"Blade Runner (1982).mp4",
)
def test_a_missing_year_does_not_leave_empty_parentheses():
"""TMDB occasionally has no release date. '( )' would both look broken and
change how Jellyfin parses the name."""
p = addon._library_path("movie", "Untitled", "", "99")
assert "()" not in p and "( )" not in p
assert "[tmdbid-99]" in p
# --- containment ------------------------------------------------------------
def test_hostile_values_cannot_escape_the_library_root():
for evil in ("../../etc/passwd", "..", "a/../../b", "/etc/shadow"):
for kind in ("series", "movie"):
p = os.path.realpath(addon._library_path(
kind, evil, "2020", evil, season=1, episode=1,
episode_title=evil))
assert p.startswith(os.path.realpath(addon.LIBRARY_ROOT) + os.sep)
def test_the_tmdb_id_is_the_identifying_token():
"""Two shows sharing a name and year must not share a folder."""
a = addon._library_path("series", "Ghosts", "2019", "95396",
season=1, episode=1, episode_title="P")
b = addon._library_path("series", "Ghosts", "2019", "87739",
season=1, episode=1, episode_title="P")
assert os.path.dirname(a) != os.path.dirname(b)
# --- the partial file must be on the same filesystem ------------------------
def test_the_part_file_is_a_sibling_of_its_destination():
"""os.replace() across filesystems fails with EXDEV. Writing the temp file
into the repo's cache dir and renaming onto /mnt/usb would fail on every
ingest, so the partial must sit beside its destination."""
dest = addon._library_path("series", "Show", "2020", "1",
season=1, episode=1, episode_title="T")
assert os.path.dirname(addon._part_path(dest)) == os.path.dirname(dest)
# --- metadata lookup --------------------------------------------------------
# Naming is only safe if the identity behind it is. A title we cannot identify
# must NOT be ingested under a guessed name: the 2026-07-28 cleanup existed
# precisely because wrongly-named media had accumulated, and a bad file in the
# library costs more than a missing one.
@pytest.fixture
def tmdb(monkeypatch):
"""Stand in for TMDB. Tests must never reach the network."""
calls = []
responses = {}
def fake(path, params=None):
calls.append(path)
if path not in responses:
raise RuntimeError(f"unexpected TMDB call: {path}")
return responses[path]
monkeypatch.setattr(addon, "_tmdb_get", fake)
monkeypatch.setattr(addon, "_library_meta_cache", {})
return type("T", (), {"calls": calls, "responses": responses})()
def test_series_metadata_comes_from_the_tmdb_id(tmdb):
tmdb.responses["tv/138502"] = {"name": "X-Men '97", "first_air_date": "2024-03-20"}
tmdb.responses["tv/138502/season/2/episode/7"] = {"name": "Bright Eyes"}
m = addon._library_meta("series", "tmdb:138502", 2, 7)
assert m == {"tmdb_id": "138502", "title": "X-Men '97", "year": "2024",
"episode_title": "Bright Eyes", "runtime": None}
def test_movie_metadata_comes_from_the_tmdb_id(tmdb):
tmdb.responses["movie/78"] = {"title": "Blade Runner", "release_date": "1982-06-25"}
m = addon._library_meta("movie", "tmdb:78")
assert m["title"] == "Blade Runner" and m["year"] == "1982"
assert m["episode_title"] == ""
def test_an_imdb_ident_is_translated_to_a_tmdb_id(tmdb):
"""The cinemeta path supplies tt-ids, but the library names by TMDB id."""
tmdb.responses["find/tt10310354"] = {
"tv_results": [{"id": 138502}], "movie_results": []}
tmdb.responses["tv/138502"] = {"name": "X-Men '97", "first_air_date": "2024-03-20"}
tmdb.responses["tv/138502/season/2/episode/7"] = {"name": "Bright Eyes"}
m = addon._library_meta("series", "tt10310354", 2, 7)
assert m["tmdb_id"] == "138502"
def test_an_unresolvable_ident_yields_no_metadata(tmdb):
"""No id, no ingest -- rather than inventing a folder name."""
tmdb.responses["find/tt00000000"] = {"tv_results": [], "movie_results": []}
assert addon._library_meta("series", "tt00000000", 1, 1) is None
def test_a_tmdb_failure_yields_no_metadata_rather_than_a_guess(tmdb):
"""_tmdb_get raises for anything unstubbed, so this exercises the real
failure path: the lookup must swallow it and decline, not propagate."""
assert addon._library_meta("series", "tmdb:999999", 1, 1) is None
def test_a_missing_episode_title_is_not_fatal(tmdb):
"""TMDB carries the show but not always the episode name. The episode is
still ingestable -- Jellyfin keys on SxxExx."""
tmdb.responses["tv/1"] = {"name": "Show", "first_air_date": "2020-01-01"}
tmdb.responses["tv/1/season/1/episode/1"] = {}
m = addon._library_meta("series", "tmdb:1", 1, 1)
assert m["episode_title"] == "" and m["title"] == "Show"
def test_a_show_with_no_air_date_still_resolves(tmdb):
tmdb.responses["tv/1"] = {"name": "Show"}
tmdb.responses["tv/1/season/1/episode/1"] = {"name": "Pilot"}
m = addon._library_meta("series", "tmdb:1", 1, 1)
assert m["year"] == "" and m["title"] == "Show"
def test_metadata_is_cached_across_calls(tmdb):
"""Ingest asks per episode; the show lookup must not be repeated for every
episode of a season."""
tmdb.responses["tv/1"] = {"name": "Show", "first_air_date": "2020-01-01"}
tmdb.responses["tv/1/season/1/episode/1"] = {"name": "Pilot"}
addon._library_meta("series", "tmdb:1", 1, 1)
before = len(tmdb.calls)
addon._library_meta("series", "tmdb:1", 1, 1)
assert len(tmdb.calls) == before
# --- one copy per episode, regardless of height -----------------------------
def test_an_existing_copy_at_any_height_counts_as_present(tmdb):
"""The library holds one copy per episode. A 1080p re-request must not
produce a second file beside an existing 720p one -- and nothing is ever
deleted to make room for an upgrade."""
tmdb.responses["tv/1"] = {"name": "Show", "first_air_date": "2020-01-01"}
tmdb.responses["tv/1/season/1/episode/1"] = {"name": "Pilot"}
dest = addon._library_path("series", "Show", "2020", "1",
season=1, episode=1, episode_title="Pilot")
assert addon._library_has("series", "tmdb:1", 1, 1) is False
os.makedirs(os.path.dirname(dest), exist_ok=True)
open(dest, "wb").write(b"x")
assert addon._library_has("series", "tmdb:1", 1, 1) is True
def test_an_unidentifiable_title_is_reported_absent_not_crashing(tmdb):
assert addon._library_has("series", "tmdb:999999", 1, 1) is False
# --- the remux writes where it is told --------------------------------------
@pytest.fixture
def fake_ffmpeg(monkeypatch):
"""Stand in for ffmpeg: record the argv and create whatever output it was
asked for, so the surrounding file handling is what gets exercised."""
class Fake:
runs = []
fail = False
# What the source says it is, and what ffmpeg actually produced. Equal
# by default: the interesting cases set them apart.
playlist_seconds = 1800.0
output_seconds = 1800.0
class Result:
returncode = 0
stderr = ""
# The seam is _run_remux, NOT subprocess.run. A stalled copy is now ended by
# watching the output file grow, which needs Popen and a poll loop -- so a
# fixture that patches subprocess.run stops standing in for anything and
# every test using it silently shells out to real ffmpeg against
# `http://cdn/x.m3u8`, where the reconnect flags make it retry for minutes.
def fake_run(argv, out_path, stall_timeout=None, hard_timeout=None):
Fake.runs.append(argv)
Result.returncode = 1 if Fake.fail else 0
if not Fake.fail:
with open(out_path, "wb") as f:
f.write(b"video" * 100)
return Result
monkeypatch.setattr(addon, "_run_remux", fake_run)
monkeypatch.setattr(addon, "_playlist_duration",
lambda url: Fake.playlist_seconds)
monkeypatch.setattr(addon, "_media_duration",
lambda path: Fake.output_seconds)
return Fake
def test_remux_creates_the_destination_directory(fake_ffmpeg, tmp_path):
"""The season folder will not exist on the first episode of a show. The old
code made MP4_DIR, which is the wrong volume entirely."""
dest = os.path.join(str(tmp_path), "Shows", "S (2020) [tmdbid-1]",
"Season 01", "S (2020) - S01E01.mp4")
addon._remux_to_mp4("http://cdn/x.m3u8", dest)
assert os.path.exists(dest)
def test_remux_writes_to_part_then_renames(fake_ffmpeg, tmp_path):
"""ffmpeg must be handed the .part path, never the final one: the moov atom
is written last, so a crash mid-remux would otherwise leave a truncated file
that _library_has reports as a complete copy."""
dest = os.path.join(str(tmp_path), "out.mp4")
addon._remux_to_mp4("http://cdn/x.m3u8", dest)
assert fake_ffmpeg.runs[0][-1] == addon._part_path(dest)
assert os.path.exists(dest)
assert not os.path.exists(addon._part_path(dest))
def test_a_failed_remux_leaves_no_final_file(fake_ffmpeg, tmp_path):
dest = os.path.join(str(tmp_path), "out.mp4")
fake_ffmpeg.fail = True
with pytest.raises(RuntimeError):
addon._remux_to_mp4("http://cdn/x.m3u8", dest)
assert not os.path.exists(dest)
# --- the remux is only done when it reached the end -------------------------
#
# 2026-08-26: Strange New Worlds S04E04 and S04E05 were ingested as 10.6 and 9.8
# minutes of a 51- and 55-minute episode, and "Furious" S01E07 as 27.7 of 52.
# ffmpeg exited 0 on all three and the log said `ingest done`. An HLS input that
# stops early -- an expired token, a 403 on a segment part-way through the
# ladder -- is an end-of-stream to ffmpeg, not an error: it finalises the moov
# atom over what it got and reports success. os.replace then published a whole
# file containing a fifth of an episode, and because _library_has asks only
# "does a non-empty file exist", every later request answered "done" and no
# re-fetch was ever attempted. A partial copy is worse than no copy: no copy
# retries, a partial one is permanent.
#
# So an exit code is not evidence of completeness. The playlist states its own
# length in EXTINF and the output can be measured, so compare them.
def test_a_remux_that_stopped_early_is_not_published(fake_ffmpeg, tmp_path):
"""The SNW S04E04 shape: ffmpeg exit 0, real file, one fifth of the run."""
dest = os.path.join(str(tmp_path), "out.mp4")
fake_ffmpeg.playlist_seconds = 3009.6
fake_ffmpeg.output_seconds = 636.7
with pytest.raises(RuntimeError, match="incomplete"):
addon._remux_to_mp4("http://cdn/x.m3u8", dest)
assert not os.path.exists(dest)
def test_the_partial_is_kept_so_the_next_run_overwrites_it(fake_ffmpeg, tmp_path):
"""Nothing is deleted to recover. ffmpeg runs with -y and writes the same
path, so leaving the .part costs nothing and keeps the evidence."""
dest = os.path.join(str(tmp_path), "out.mp4")
fake_ffmpeg.playlist_seconds = 3009.6
fake_ffmpeg.output_seconds = 636.7
with pytest.raises(RuntimeError):
addon._remux_to_mp4("http://cdn/x.m3u8", dest)
assert os.path.exists(addon._part_path(dest))
def test_a_complete_remux_is_published(fake_ffmpeg, tmp_path):
dest = os.path.join(str(tmp_path), "out.mp4")
fake_ffmpeg.playlist_seconds = 3009.6
fake_ffmpeg.output_seconds = 3009.5
addon._remux_to_mp4("http://cdn/x.m3u8", dest)
assert os.path.exists(dest)
def test_a_dropped_final_segment_is_not_treated_as_truncation(fake_ffmpeg, tmp_path):
"""EXTINF sums and container duration do not have to agree to the frame.
The tolerance must absorb that drift while still catching a real loss --
the smallest real one seen was 53% of the episode."""
dest = os.path.join(str(tmp_path), "out.mp4")
fake_ffmpeg.playlist_seconds = 3009.6
fake_ffmpeg.output_seconds = 3009.6 * addon.REMUX_MIN_COMPLETENESS
addon._remux_to_mp4("http://cdn/x.m3u8", dest)
assert os.path.exists(dest)
def test_an_unmeasurable_playlist_does_not_block_a_good_ingest(fake_ffmpeg, tmp_path,
monkeypatch):
"""Same rule as _cap_variant: a check that cannot run must not become a new
way for a working stream to fail. Unverifiable falls back to the old
behaviour rather than refusing."""
dest = os.path.join(str(tmp_path), "out.mp4")
monkeypatch.setattr(addon, "_playlist_duration", lambda url: None)
addon._remux_to_mp4("http://cdn/x.m3u8", dest)
assert os.path.exists(dest)
def test_an_unmeasurable_output_is_refused(fake_ffmpeg, tmp_path, monkeypatch):
"""The opposite case is NOT symmetric. If the playlist told us how long the
episode is and the output cannot be probed at all, that is a broken file,
not a missing measurement."""
dest = os.path.join(str(tmp_path), "out.mp4")
monkeypatch.setattr(addon, "_media_duration", lambda path: None)
with pytest.raises(RuntimeError, match="incomplete"):
addon._remux_to_mp4("http://cdn/x.m3u8", dest)
assert not os.path.exists(dest)
def test_playlist_duration_sums_extinf():
body = ("#EXTM3U\n#EXT-X-PLAYLIST-TYPE:VOD\n"
"#EXTINF:6.006,\nseg0.m4s\n"
"#EXTINF:6.006,\nseg1.m4s\n"
"#EXTINF:3.500,\nseg2.m4s\n"
"#EXT-X-ENDLIST\n")
assert addon._parse_playlist_duration(body) == pytest.approx(15.512)
def test_a_playlist_with_no_segments_has_no_duration():
"""A master playlist carries no EXTINF. Reporting 0 would make every ingest
look complete; reporting nothing makes it unverifiable, which is honest."""
assert addon._parse_playlist_duration("#EXTM3U\n#EXT-X-STREAM-INF:x\na.m3u8\n") is None
# --- ingest orchestration ---------------------------------------------------
@pytest.fixture
def ingest(monkeypatch, tmdb):
"""A resolvable series with a stubbed remux, so only orchestration runs."""
tmdb.responses["tv/1"] = {"name": "Show", "first_air_date": "2020-01-01"}
tmdb.responses["tv/1/season/1/episode/1"] = {"name": "Pilot"}
remuxed = []
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: "http://cdn/master.m3u8")
monkeypatch.setattr(addon, "_cap_variant", lambda url, h: f"{url}?h={h}")
def fake_remux(url, dest):
remuxed.append((url, dest))
os.makedirs(os.path.dirname(dest), exist_ok=True)
open(dest, "wb").write(b"x" * 10)
monkeypatch.setattr(addon, "_remux_to_mp4", fake_remux)
return remuxed
def test_ingest_writes_into_the_library(ingest):
state = addon._run_ingest("series", "tmdb:1", 1, 1, 720)
assert state == "done"
dest = addon._library_path("series", "Show", "2020", "1",
season=1, episode=1, episode_title="Pilot")
assert os.path.exists(dest)
assert ingest and ingest[0][1] == dest
def test_ingest_declines_a_title_it_cannot_identify(ingest, tmdb):
"""No identity, no file -- not a guessed folder name."""
state = addon._run_ingest("series", "tmdb:999999", 1, 1, 720)
assert state.startswith("failed")
assert ingest == []
def test_ingest_is_a_noop_when_the_library_already_has_it(ingest):
addon._run_ingest("series", "tmdb:1", 1, 1, 720)
ingest.clear()
state = addon._run_ingest("series", "tmdb:1", 1, 1, 1080)
assert state == "done"
assert ingest == [], "a second height must not produce a second file"
def test_ingest_fails_cleanly_when_the_source_cannot_be_resolved(ingest, monkeypatch):
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: None)
state = addon._run_ingest("series", "tmdb:1", 1, 1, 720)
assert state.startswith("failed")
assert ingest == []
# --- serving prefers the library, without orphaning the legacy cache --------
@pytest.fixture
def served(monkeypatch, tmdb, tmp_path):
tmdb.responses["tv/1"] = {"name": "Show", "first_air_date": "2020-01-01"}
tmdb.responses["tv/1/season/1/episode/1"] = {"name": "Pilot"}
monkeypatch.setattr(addon, "MP4_DIR", os.path.join(str(tmp_path), "cache"))
os.makedirs(addon.MP4_DIR, exist_ok=True)
monkeypatch.setattr(addon, "_prefetch_vidcore", lambda *a, **k: None)
monkeypatch.setattr(addon, "_start_mp4_job", lambda *a, **k: "running")
return tmdb
def _write_library_copy(body=b"library"):
dest = addon._library_path("series", "Show", "2020", "1",
season=1, episode=1, episode_title="Pilot")
os.makedirs(os.path.dirname(dest), exist_ok=True)
open(dest, "wb").write(body)
return dest
def test_the_library_copy_is_what_gets_served(served):
dest = _write_library_copy(b"library-bytes")
assert addon._playable_path("series", "tmdb:1", 1, 1, 720) == dest
def test_a_legacy_cache_copy_is_still_served(served):
"""The one file made by hand before the library existed must not become
unreachable -- nothing is deleted, so nothing may be orphaned either."""
legacy = addon._mp4_path("series", "tmdb:1", 1, 1, 720)
open(legacy, "wb").write(b"legacy")
assert addon._playable_path("series", "tmdb:1", 1, 1, 720) == legacy
def test_the_library_wins_over_a_legacy_copy(served):
dest = _write_library_copy()
open(addon._mp4_path("series", "tmdb:1", 1, 1, 720), "wb").write(b"legacy")
assert addon._playable_path("series", "tmdb:1", 1, 1, 720) == dest
def test_nothing_playable_yields_none(served):
assert addon._playable_path("series", "tmdb:1", 1, 1, 720) is None
def test_the_route_serves_the_library_copy_with_range_support(served):
_write_library_copy(b"0123456789" * 10)
client = addon.app.test_client()
r = client.get("/mp4/series/tmdb:1:1:1.mp4?h=720",
headers={"Range": "bytes=0-9"})
assert r.status_code == 206 # seeking depends on this
assert r.get_data() == b"0123456789"
def test_a_row_is_offered_once_the_library_holds_it(served):
_write_library_copy()
with addon.app.test_request_context("/stream/series/fxl:1:1:1.json"):
out = []
addon._append_mp4_row(out, "series", "tmdb:1", 1, 1)
assert len(out) == 1 and "/mp4/" in out[0]["url"]
def test_no_copy_yet_still_offers_a_row_that_will_make_one(served):
"""Superseded the old 'no copy means no row'. That WAS the chicken-and-egg:
withholding the row until a copy existed meant no copy could ever be made
through the UI."""
with addon.app.test_request_context("/stream/series/fxl:1:1:1.json"):
out = []
addon._append_mp4_row(out, "series", "tmdb:1", 1, 1)
assert len(out) == 1
assert "in library" not in out[0]["title"]
# --- triggers ---------------------------------------------------------------
# Until now nothing could START an ingest: _append_mp4_row only offered a /mp4/
# row once the copy existed, and only a /mp4/ request created one. The two
# guarded each other, so the single file on disk was made by hand.
@pytest.fixture
def triggered(monkeypatch, tmdb, tmp_path):
tmdb.responses["tv/1"] = {"name": "Show", "first_air_date": "2020-01-01"}
tmdb.responses["tv/1/season/1/episode/1"] = {"name": "Pilot"}
monkeypatch.setattr(addon, "MP4_DIR", os.path.join(str(tmp_path), "cache"))
os.makedirs(addon.MP4_DIR, exist_ok=True)
monkeypatch.setattr(addon, "_prefetch_vidcore", lambda *a, **k: None)
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: "http://cdn/m.m3u8")
monkeypatch.setattr(addon, "_cap_variant", lambda url, h: f"{url}?h={h}")
started = []
monkeypatch.setattr(addon, "_start_mp4_job",
lambda *a, **k: started.append(a) or "running")
return started
def test_playing_a_stream_starts_the_copy(triggered):
"""Pressing play is the honest 'I chose this' signal."""
addon.app.test_client().get("/vidcore/series/tmdb:1:1:1.m3u8?h=720")
assert triggered, "playing must enqueue the copy"
def test_playing_does_not_re_fetch_what_the_library_already_has(triggered):
_write_library_copy()
addon.app.test_client().get("/vidcore/series/tmdb:1:1:1.m3u8?h=720")
assert triggered == []
def test_an_unfinished_copy_redirects_to_the_live_stream(triggered):
"""The row must be playable the moment it is clicked. 503 would surface as a
playback failure while the copy silently proceeded."""
r = addon.app.test_client().get("/mp4/series/tmdb:1:1:1.mp4?h=720")
assert r.status_code == 302
assert "cdn" in r.headers["Location"]
assert triggered, "and it must start the copy"
def test_a_finished_copy_is_served_rather_than_redirected(triggered):
_write_library_copy(b"0123456789")
r = addon.app.test_client().get("/mp4/series/tmdb:1:1:1.mp4?h=720")
assert r.status_code == 200
assert triggered == []
def test_the_keep_row_is_offered_before_any_copy_exists(triggered):
"""This is the chicken-and-egg: the row had to exist for a copy to be made,
and a copy had to exist for the row to be offered."""
with addon.app.test_request_context("/stream/series/fxl:1:1:1.json"):
out = []
addon._append_mp4_row(out, "series", "tmdb:1", 1, 1)
assert len(out) == 1 and "/mp4/" in out[0]["url"]
def test_the_row_says_when_the_copy_is_already_held(triggered):
_write_library_copy()
with addon.app.test_request_context("/stream/series/fxl:1:1:1.json"):
out = []
addon._append_mp4_row(out, "series", "tmdb:1", 1, 1)
assert "in library" in out[0]["title"]
def test_an_unidentifiable_title_offers_no_keep_row(triggered):
"""Nothing can be ingested without an identity, so the row would be a lie."""
with addon.app.test_request_context("/stream/series/fxl:9:1:1.json"):
out = []
addon._append_mp4_row(out, "series", "tmdb:999999", 1, 1)
assert out == []
# --- next-episode queueing --------------------------------------------------
# The only trigger that fetches something the user did NOT directly choose, so
# it is the one that must know when to stop. Two independent bounds: TMDB says
# whether a next episode exists at all, and the reputation store says whether
# the source actually has it.
@pytest.fixture
def rolling(monkeypatch, tmdb, tmp_path):
tmdb.responses["tv/1"] = {"name": "Show", "first_air_date": "2020-01-01"}
for e in (1, 2, 3):
tmdb.responses[f"tv/1/season/1/episode/{e}"] = {"name": f"E{e}"}
tmdb.responses["tv/1/season/1"] = {
"episodes": [{"episode_number": 1}, {"episode_number": 2}]}
tmdb.responses["tv/1/season/2"] = {"episodes": [{"episode_number": 1}]}
tmdb.responses["tv/1/season/3"] = {"episodes": []}
tmdb.responses["tv/1/season/2/episode/1"] = {"name": "S2E1"}
monkeypatch.setattr(addon, "MP4_DIR", os.path.join(str(tmp_path), "cache"))
os.makedirs(addon.MP4_DIR, exist_ok=True)
monkeypatch.setattr(addon, "_prefetch_vidcore", lambda *a, **k: None)
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: "http://cdn/m.m3u8")
monkeypatch.setattr(addon, "_cap_variant", lambda url, h: url)
monkeypatch.setattr(addon, "_vidcore_rep", {})
queued = []
monkeypatch.setattr(addon, "_start_mp4_job",
lambda *a, **k: queued.append(a) or "running")
return queued
def test_the_next_episode_in_a_season_is_found(rolling):
assert addon._next_episode("1", 1, 1) == (1, 2)
def test_the_end_of_a_season_rolls_into_the_next(rolling):
assert addon._next_episode("1", 1, 2) == (2, 1)
def test_the_end_of_the_series_stops(rolling):
"""Season 3 exists in TMDB but is empty, and there is no season 4."""
assert addon._next_episode("1", 2, 1) is None
def test_playing_an_episode_queues_the_next_one(rolling):
addon.app.test_client().get("/vidcore/series/tmdb:1:1:1.m3u8?h=720")
assert (("series", "tmdb:1", 1, 2, 720)) in [tuple(q) for q in rolling]
def test_playing_a_library_copy_also_rolls_forward(rolling):
"""Otherwise the rolling library stalls the moment you start watching from
it -- the local file is served without ever touching the vidcore route."""
dest = addon._library_path("series", "Show", "2020", "1",
season=1, episode=1, episode_title="E1")
os.makedirs(os.path.dirname(dest), exist_ok=True)
open(dest, "wb").write(b"x" * 10)
addon.app.test_client().get("/mp4/series/tmdb:1:1:1.mp4?h=720")
assert (("series", "tmdb:1", 1, 2, 720)) in [tuple(q) for q in rolling]
def test_a_next_episode_the_source_lacks_is_not_queued(rolling):
"""A known miss is a positive finding that vidcore does not have it. Queueing
it would spend a browser launch to rediscover that on every play."""
addon._vidcore_rep[addon._rep_key("series", "tmdb:1", 1, 2)] = {
"status": "miss", "ts": addon.time.time(), "strikes": 2}
addon.app.test_client().get("/vidcore/series/tmdb:1:1:1.m3u8?h=720")
assert [q for q in rolling if q[3] == 2] == []
def test_an_inconclusive_next_episode_is_not_queued(rolling):
"""Inconclusive must NOT harden into 'go and fetch it' either -- one strike
is not a finding. It stays queueable only once it is genuinely unknown, not
while a previous attempt was ambiguous."""
addon._vidcore_rep[addon._rep_key("series", "tmdb:1", 1, 2)] = {
"status": "miss", "ts": addon.time.time(), "strikes": 1}
addon.app.test_client().get("/vidcore/series/tmdb:1:1:1.m3u8?h=720")
nxt = [q for q in rolling if q[3] == 2]
assert nxt == [], "a pending strike should not trigger a speculative fetch"
def test_queueing_does_not_chain_beyond_one(rolling):
"""Watching one episode must not enqueue a whole season."""
addon.app.test_client().get("/vidcore/series/tmdb:1:1:1.m3u8?h=720")
assert len([q for q in rolling if q[1] == "tmdb:1"]) <= 2
def test_a_movie_never_queues_anything(rolling, tmdb):
tmdb.responses["movie/5"] = {"title": "Film", "release_date": "2020-01-01"}
addon.app.test_client().get("/vidcore/movie/tmdb:5.m3u8?h=720")
assert all(q[0] != "series" for q in rolling)
def test_a_tmdb_failure_stops_rolling_rather_than_guessing(rolling, monkeypatch):
monkeypatch.setattr(addon, "_tmdb_get",
lambda *a, **k: (_ for _ in ()).throw(RuntimeError("down")))
assert addon._next_episode("1", 1, 1) is None
# --- the vidcore rows also prefer a local copy ------------------------------
def test_a_vidcore_row_redirects_to_the_local_copy(rolling):
"""Playing "720p vidcore" on a title already in the library streamed from
the CDN, ignoring the seekable file sitting on disk."""
dest = addon._library_path("series", "Show", "2020", "1",
season=1, episode=1, episode_title="E1")
os.makedirs(os.path.dirname(dest), exist_ok=True)
open(dest, "wb").write(b"x" * 10)
r = addon.app.test_client().get("/vidcore/series/tmdb:1:1:1.m3u8?h=720")
assert r.status_code == 302
assert "/mp4/series/tmdb:1:1:1.mp4" in r.headers["Location"]
def test_the_local_check_happens_before_the_browser_resolve(rolling, monkeypatch):
"""A resolve costs ~11s and launches a browser. Paying that for a title we
already hold on disk is the whole point of checking first."""
resolved = []
monkeypatch.setattr(addon, "_resolve_vidcore",
lambda *a, **k: resolved.append(a) or "http://cdn/m.m3u8")
dest = addon._library_path("series", "Show", "2020", "1",
season=1, episode=1, episode_title="E1")
os.makedirs(os.path.dirname(dest), exist_ok=True)
open(dest, "wb").write(b"x" * 10)
addon.app.test_client().get("/vidcore/series/tmdb:1:1:1.m3u8?h=720")
assert resolved == [], "no browser should be launched for a local title"
def test_without_a_local_copy_the_vidcore_row_still_streams(rolling):
r = addon.app.test_client().get("/vidcore/series/tmdb:1:1:1.m3u8?h=720")
assert r.status_code == 302
assert "/mp4/" not in r.headers["Location"]
def test_a_movie_vidcore_row_also_prefers_the_local_copy(rolling, tmdb):
tmdb.responses["movie/5"] = {"title": "Film", "release_date": "2020-01-01"}
dest = addon._library_path("movie", "Film", "2020", "5")
os.makedirs(os.path.dirname(dest), exist_ok=True)
open(dest, "wb").write(b"x" * 10)
r = addon.app.test_client().get("/vidcore/movie/tmdb:5.m3u8?h=720")
assert r.status_code == 302
assert "/mp4/movie/tmdb:5.mp4" in r.headers["Location"]
def test_ingest_resolves_at_run_time_not_at_queue_time(ingest, monkeypatch):
"""Variant URLs are signed and expire in ~4h. A job that resolved when it
was queued would fetch a dead link once the queue backs up."""
calls = []
monkeypatch.setattr(addon, "_resolve_vidcore",
lambda *a, **k: calls.append(a) or "http://cdn/m.m3u8")
addon._run_ingest("series", "tmdb:1", 1, 1, 720)
assert calls, "the ingest must resolve for itself"
# --- a partial copy already on disk is not a copy ---------------------------
#
# 0.13.2 stopped NEW partials being published. It could do nothing about the
# three already in the library, and nothing about the way they were being
# served: _local_redirect answers before the resolve, so Strange New Worlds
# S04E04 was handed to the player from disk at 10.6 minutes while a complete
# 50-minute upstream sat there unused.
#
# The gate goes in _playable_path and _library_has rather than in the row that
# offers the copy, because Jacktook caches its source list for 24h -- a row can
# be a day stale, but these are on the path of every single play.
#
# Both, and not just the serving side: if only _playable_path refused, the file
# would stop being served and _library_has would still report "already held", so
# the ingest that should replace it would decline to run. Refusing in both is
# what makes the repair automatic.
@pytest.fixture
def on_disk(tmdb, monkeypatch):
"""A library copy of a 50-minute episode whose measured length is settable.
Returns (path, seconds) where seconds["value"] is what ffprobe would say.
"""
tmdb.responses["tv/1"] = {"name": "Show", "first_air_date": "2020-01-01"}
tmdb.responses["tv/1/season/1/episode/1"] = {"name": "Pilot", "runtime": 50}
meta = addon._library_meta("series", "tmdb:1", 1, 1)
path = addon._library_path("series", meta["title"], meta["year"],
meta["tmdb_id"], 1, 1, meta["episode_title"])
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "wb") as f:
f.write(b"x" * 1024)
seconds = {"value": 3000.0}
monkeypatch.setattr(addon, "_media_duration", lambda p: seconds["value"])
return path, seconds
def test_metadata_carries_the_tmdb_runtime(tmdb):
"""Free: it rides on the episode lookup the title already needed."""
tmdb.responses["tv/1"] = {"name": "Show", "first_air_date": "2020-01-01"}
tmdb.responses["tv/1/season/1/episode/1"] = {"name": "Pilot", "runtime": 51}
assert addon._library_meta("series", "tmdb:1", 1, 1)["runtime"] == 51
def test_a_partial_copy_is_not_served(on_disk):
"""The SNW S04E04 shape: 10.6 minutes of a 51-minute episode on disk."""
path, seconds = on_disk
seconds["value"] = 636.7
assert addon._playable_path("series", "tmdb:1", 1, 1) is None
def test_a_partial_copy_does_not_block_its_own_replacement(on_disk):
"""The half that makes the repair automatic rather than manual."""
path, seconds = on_disk
seconds["value"] = 636.7
assert addon._library_has("series", "tmdb:1", 1, 1) is False
def test_a_complete_copy_is_still_served(on_disk):
path, seconds = on_disk
assert addon._playable_path("series", "tmdb:1", 1, 1) == path
assert addon._library_has("series", "tmdb:1", 1, 1) is True
def test_editorial_drift_does_not_condemn_a_good_copy(on_disk):
"""TMDB's runtime is not a measurement. The fourteen known-good copies in
the library measured 92-101% of it, so the floor has to sit below that --
which is why this is 0.85 and the remux check, comparing against a real
EXTINF sum, is 0.99."""
path, seconds = on_disk
seconds["value"] = 3000.0 * 0.92
assert addon._playable_path("series", "tmdb:1", 1, 1) == path
def test_a_copy_verified_against_its_source_outranks_tmdb(on_disk):
"""The loop-breaker. If upstream genuinely holds only a short version, the
remux check passes it (it matched its own playlist) and a runtime gate would
then reject it on every play, re-fetching the same short file for ever."""
path, seconds = on_disk
seconds["value"] = 636.7
addon._record_verified(path, 636.7, 636.7)
assert addon._playable_path("series", "tmdb:1", 1, 1) == path
assert addon._library_has("series", "tmdb:1", 1, 1) is True
def test_an_unknown_runtime_leaves_the_copy_alone(on_disk, tmdb, monkeypatch):
"""TMDB does not always carry a runtime. Unknown must mean "leave it",
never "condemn it" -- the opposite asymmetry to the remux check, because
there we had just written the file and knew what it should be."""
path, seconds = on_disk
monkeypatch.setattr(addon, "_library_meta_cache", {})
tmdb.responses["tv/1/season/1/episode/1"] = {"name": "Pilot"}
seconds["value"] = 1.0
assert addon._playable_path("series", "tmdb:1", 1, 1) == path
def test_an_unprobeable_copy_is_left_alone(on_disk, monkeypatch):
path, _ = on_disk
monkeypatch.setattr(addon, "_media_duration", lambda p: None)
assert addon._playable_path("series", "tmdb:1", 1, 1) == path
def test_the_duration_is_probed_once_per_file(on_disk, monkeypatch):
"""Every play of a local copy asks. An ffprobe per request would be a new
cost on the one path that was fast."""
path, _ = on_disk
calls = []
monkeypatch.setattr(addon, "_media_duration",
lambda p: (calls.append(p), 3000.0)[1])
addon._playable_path("series", "tmdb:1", 1, 1)
addon._playable_path("series", "tmdb:1", 1, 1)
assert len(calls) == 1
def test_a_rewritten_file_is_probed_again(on_disk, monkeypatch):
"""The cache key carries size and mtime, so a re-ingest invalidates it
without anyone having to remember to."""
path, _ = on_disk
calls = []
monkeypatch.setattr(addon, "_media_duration",
lambda p: (calls.append(p), 3000.0)[1])
addon._playable_path("series", "tmdb:1", 1, 1)
with open(path, "wb") as f:
f.write(b"y" * 4096)
addon._playable_path("series", "tmdb:1", 1, 1)
assert len(calls) == 2
def test_a_verified_remux_records_what_it_measured(fake_ffmpeg, tmp_path):
dest = os.path.join(str(tmp_path), "out.mp4")
fake_ffmpeg.playlist_seconds = 3009.6
fake_ffmpeg.output_seconds = 3009.5
addon._remux_to_mp4("http://cdn/x.m3u8", dest)
assert os.path.abspath(dest) in addon._load_verified()
def test_a_refused_remux_records_nothing(fake_ffmpeg, tmp_path):
dest = os.path.join(str(tmp_path), "out.mp4")
fake_ffmpeg.playlist_seconds = 3009.6
fake_ffmpeg.output_seconds = 636.7
with pytest.raises(RuntimeError):
addon._remux_to_mp4("http://cdn/x.m3u8", dest)
assert addon._load_verified() == {}