"""Fetching a whole season -- or as much of it as has actually aired.
This is not new machinery. _queue_next_episode has fetched the NEXT episode
after each play since the library existed, "so a series becomes a local
library"; a season is that lookahead raised, running through the same ingest
queue, the same semaphore and the same resume-on-boot path.
Run: .venv/bin/python3 -m pytest test_season_ingest.py -q
"""
import os
import time
import pytest
import addon
@pytest.fixture(autouse=True)
def isolated(tmp_path, monkeypatch):
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path))
monkeypatch.setattr(addon, "INGEST_QUEUE_FILE",
os.path.join(str(tmp_path), "queue.json"))
monkeypatch.setattr(addon, "LIBRARY_VERIFIED_FILE",
os.path.join(str(tmp_path), "verified.json"))
monkeypatch.setattr(addon, "_library_meta_cache", {})
monkeypatch.setattr(addon, "_season_jobs", {})
monkeypatch.setattr(addon, "_mp4_jobs", {})
yield
def _settled(timeout=5.0):
"""Block until the season worker has finished.
Polls the job state rather than joining by thread name: the worker may not
have been scheduled yet when the assertion runs, and a name lookup that
finds nothing would pass by accident.
"""
deadline = time.time() + timeout
while time.time() < deadline:
states = list(addon._season_jobs.values())
if states and all(not s.startswith("running") for s in states):
return
time.sleep(0.01)
raise AssertionError(f"season job did not settle: {addon._season_jobs}")
def _season(*episodes):
"""episodes: (number, air_date) pairs."""
return {"episodes": [{"episode_number": n, "air_date": d}
for n, d in episodes]}
# --- what counts as released ------------------------------------------------
def test_only_aired_episodes_are_listed(monkeypatch):
"""TMDB lists a running season's unaired episodes too. Asking vidcore for
one costs a full browser resolve to learn nothing."""
monkeypatch.setattr(addon, "_tmdb_get", lambda p, params=None: _season(
(1, "2026-01-01"), (2, "2026-01-08"), (3, "2099-01-01")))
assert addon._released_episodes("1", 4) == [1, 2]
def test_an_episode_with_no_air_date_is_not_released(monkeypatch):
"""The field is how "released" is expressed. Guessing the other way queues
work that cannot succeed."""
monkeypatch.setattr(addon, "_tmdb_get", lambda p, params=None: _season(
(1, "2026-01-01"), (2, ""), (3, None)))
assert addon._released_episodes("1", 4) == [1]
def test_episodes_come_back_in_order(monkeypatch):
monkeypatch.setattr(addon, "_tmdb_get", lambda p, params=None: _season(
(3, "2026-01-01"), (1, "2026-01-01"), (2, "2026-01-01")))
assert addon._released_episodes("1", 4) == [1, 2, 3]
def test_a_tmdb_failure_yields_no_episodes_rather_than_raising(monkeypatch):
def boom(path, params=None):
raise RuntimeError("tmdb down")
monkeypatch.setattr(addon, "_tmdb_get", boom)
assert addon._released_episodes("1", 4) == []
# --- the season job ---------------------------------------------------------
@pytest.fixture
def season(monkeypatch):
"""A four-episode season, three aired, with the ingest itself stubbed."""
monkeypatch.setattr(addon, "_tmdb_get", lambda path, params=None: {
"tv/1": {"name": "Show", "first_air_date": "2020-01-01"},
"tv/1/season/4/episode/1": {"name": "One", "runtime": 50},
"tv/1/season/4": _season((1, "2026-01-01"), (2, "2026-01-02"),
(3, "2026-01-03"), (4, "2099-01-01")),
}[path])
ran = []
monkeypatch.setattr(addon, "_run_job",
lambda *a, **k: (ran.append(a), "done")[1])
return ran
def test_a_season_fetches_every_released_episode(season):
addon._start_season_job("tmdb:1", 4, 1080)
_settled()
assert [a[3] for a in season] == [1, 2, 3]
def test_the_unaired_episode_is_not_fetched(season):
addon._start_season_job("tmdb:1", 4, 1080)
_settled()
assert 4 not in [a[3] for a in season]
def test_episodes_already_held_are_skipped(season, monkeypatch):
monkeypatch.setattr(addon, "_library_has",
lambda k, i, s, e: e == 2)
addon._start_season_job("tmdb:1", 4, 1080)
_settled()
assert [a[3] for a in season] == [1, 3]
def test_a_forced_season_refetches_even_what_is_held(season, monkeypatch):
monkeypatch.setattr(addon, "_library_has", lambda k, i, s, e: True)
addon._start_season_job("tmdb:1", 4, 1080, force=True)
_settled()
assert [a[3] for a in season] == [1, 2, 3]
def test_the_same_season_is_not_started_twice(season, monkeypatch):
"""Pressing the button twice must not double every remux."""
monkeypatch.setattr(addon, "_season_jobs", {"season:tmdb:1:4:1080": "running"})
assert addon._start_season_job("tmdb:1", 4, 1080) == "running"
assert season == []
def test_episodes_run_one_at_a_time(season):
"""Through _run_job, not _start_mp4_job: a job per episode would leave
everything past the second blocked on the semaphore for up to
MP4_REMUX_TIMEOUT and reporting "remux slots busy"."""
addon._start_season_job("tmdb:1", 4, 1080)
_settled()
assert len(season) == 3 # sequential calls, not spawned threads
# --- the route --------------------------------------------------------------
def test_season_is_not_reachable_by_a_get():
assert addon.app.test_client().get("/library/season").status_code == 405
def test_season_route_starts_the_job(monkeypatch):
calls = []
monkeypatch.setattr(addon, "_start_season_job",
lambda *a, **k: (calls.append(a), "running")[1])
r = addon.app.test_client().post(
"/library/season",
data={"ident": "tmdb:1", "season": "4", "height": "1080"})
assert r.status_code == 303 and calls[0] == ("tmdb:1", 4, 1080)
def test_season_route_rejects_a_missing_ident():
r = addon.app.test_client().post("/library/season", data={"season": "4"})
assert r.status_code == 400