"""Job control for the download manager (labs#13): phase, source, cancel, history.
The manager needs to say what a job is DOING (waiting / resolving /
downloading), where the bytes come FROM, to STOP one on request, and what
happened to every attempt AFTERWARDS. None of that existed: a job was one state
string, and the only history was the journal.
Cancel is the one with teeth. A cancelled job must:
- kill ffmpeg promptly (via _run_remux's existing poll loop),
- NOT fall through to the HdHub fallback, which would fetch the same title
from another source -- the opposite of what was asked,
- keep its .part (ffmpeg -y reuses it; nothing here deletes),
- record "cancelled", which _resume_ingests does not restart at boot.
Run: .venv/bin/python3 -m pytest test_job_control.py -q
"""
import json
import os
import threading
import time
import pytest
import addon
META = {"tmdb_id": "138502", "title": "X-Men '97", "year": "2024",
"episode_title": "Bright Eyes", "runtime": 30}
def _join_jobs():
for t in threading.enumerate():
if (t.name.startswith(("mp4-", "season-"))
and t is not threading.current_thread()):
t.join(15)
@pytest.fixture
def env(tmp_path, monkeypatch):
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_ROOT", str(tmp_path / "lib"))
monkeypatch.setattr(addon, "LIBRARY_VERIFIED_FILE", str(tmp_path / "verified.json"))
monkeypatch.setattr(addon, "_mp4_jobs", {})
monkeypatch.setattr(addon, "_job_info", {})
monkeypatch.setattr(addon, "_season_jobs", {})
monkeypatch.setattr(addon, "_season_info", {})
monkeypatch.setattr(addon, "_library_meta",
lambda kind, ident, s=0, e=0: dict(META, episode_title=f"Ep{e}"))
monkeypatch.setattr(addon, "_enough_space", lambda: True)
monkeypatch.setattr(addon, "_cap_variant", lambda url, h=None: url)
monkeypatch.setattr(addon, "_fetch_subtitles_for", lambda *a, **k: None)
monkeypatch.setattr(addon, "_copy_is_complete", lambda *a, **k: True)
monkeypatch.setattr(addon, "_notify_jellyfin", lambda *a, **k: None, raising=False)
calls = {"resolve": 0, "hdhub": 0}
def resolve(*a, **k):
calls["resolve"] += 1
return "http://cdn/m.m3u8"
monkeypatch.setattr(addon, "_resolve_vidcore", resolve)
def hdhub(*a, **k):
calls["hdhub"] += 1
return "no usable file"
monkeypatch.setattr(addon, "_ingest_from_hdhub", hdhub)
yield calls
_join_jobs()
def _write_dest(dest):
os.makedirs(os.path.dirname(dest), exist_ok=True)
with open(dest, "wb") as f:
f.write(b"x" * 2048)
def _history():
with open(addon.INGEST_HISTORY_FILE) as f:
return [json.loads(line) for line in f if line.strip()]
def _wait(pred, timeout=10):
end = time.time() + timeout
while time.time() < end:
if pred():
return True
time.sleep(0.05)
return False
KEY = "series:tmdb:138502:2:6:1080"
# --- history ----------------------------------------------------------------
def test_a_finished_ingest_appends_one_history_line(env, monkeypatch):
monkeypatch.setattr(addon, "_remux_to_mp4", lambda url, dest: _write_dest(dest))
addon._run_job("series", "tmdb:138502", 2, 6, 1080)
(h,) = _history()
assert h["key"] == KEY and h["outcome"] == "done"
assert h["source"] == "vidcore"
assert h["title"] == "X-Men '97" and h["label"] == "S02E06"
assert h["bytes"] == 2048
assert h["took"] >= 0 and h["ts"] > 0
def test_a_failure_is_recorded_with_its_reason(env, monkeypatch):
def boom(url, dest):
raise RuntimeError("cdn 403")
monkeypatch.setattr(addon, "_remux_to_mp4", boom)
addon._run_job("series", "tmdb:138502", 2, 6, 1080)
(h,) = _history()
assert h["outcome"].startswith("failed")
assert "cdn 403" in h["outcome"]
def test_an_hdhub_copy_records_hdhub_as_its_source(env, monkeypatch):
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: None)
def hdhub(kind, ident, s, e, h, dest, meta):
_write_dest(dest)
return None
monkeypatch.setattr(addon, "_ingest_from_hdhub", hdhub)
addon._run_job("series", "tmdb:138502", 2, 6, 1080)
assert _history()[0]["source"] == "hdhub"
def test_a_title_already_held_is_recorded_as_such(env, monkeypatch):
monkeypatch.setattr(addon, "_library_has", lambda *a, **k: True)
addon._run_job("series", "tmdb:138502", 2, 6, 1080)
(h,) = _history()
assert h["outcome"] == "done" and h["source"] is None
def test_history_reader_skips_a_torn_line_and_lists_newest_first(env):
with open(addon.INGEST_HISTORY_FILE, "w") as f:
f.write(json.dumps({"ts": 1, "key": "a"}) + "\n")
f.write('{"ts": 2, "key": "b"\n') # torn
f.write(json.dumps({"ts": 3, "key": "c"}) + "\n")
assert [h["key"] for h in addon._read_history()] == ["c", "a"]
def test_a_missing_history_file_reads_empty(env):
assert addon._read_history() == []
def test_history_appends_are_not_lost_under_concurrency(env):
threads = [threading.Thread(target=addon._append_history,
args=({"ts": i, "key": str(i)},))
for i in range(40)]
for t in threads:
t.start()
for t in threads:
t.join()
assert len(_history()) == 40
# --- phase + source ---------------------------------------------------------
def test_phase_and_source_are_visible_while_downloading(env, monkeypatch):
seen = {}
def remux(url, dest):
seen.update(addon._job_info.get(KEY) or {})
_write_dest(dest)
monkeypatch.setattr(addon, "_remux_to_mp4", remux)
addon._run_job("series", "tmdb:138502", 2, 6, 1080)
assert seen["phase"] == "downloading"
assert seen["source"] == "vidcore"
assert seen["dest"].endswith("S02E06 - Ep6.mp4")
def test_a_started_job_is_waiting_before_it_runs(env, monkeypatch):
gate = threading.Event()
monkeypatch.setattr(addon, "_remux_to_mp4", lambda url, dest: gate.wait(5))
sem = threading.Semaphore(0) # no free slots
monkeypatch.setattr(addon, "_mp4_sem", sem)
addon._start_mp4_job("series", "tmdb:138502", 2, 6, 1080)
assert _wait(lambda: (addon._job_info.get(KEY) or {}).get("phase") == "waiting")
addon._cancel_job(KEY)
sem.release()
gate.set()
# --- cancel -----------------------------------------------------------------
def _real_remux_that_runs_forever(tmp_path):
"""A stand-in for ffmpeg: a real process under the real _run_remux, so the
cancel path tested is the one production uses."""
def remux(url, dest):
part = addon._part_path(dest)
os.makedirs(os.path.dirname(part), exist_ok=True)
with open(part, "wb") as f:
f.write(b"partial")
addon._run_remux(["sleep", "60"], part, stall_timeout=120,
hard_timeout=120)
_write_dest(dest)
return remux
def test_cancel_kills_the_remux_promptly(env, monkeypatch, tmp_path):
monkeypatch.setattr(addon, "_remux_to_mp4", _real_remux_that_runs_forever(tmp_path))
addon._start_mp4_job("series", "tmdb:138502", 2, 6, 1080)
assert _wait(lambda: (addon._job_info.get(KEY) or {}).get("phase") == "downloading")
t0 = time.time()
assert addon._cancel_job(KEY) is True
assert _wait(lambda: addon._mp4_jobs.get(KEY) == "cancelled", timeout=10)
assert time.time() - t0 < addon.MP4_STALL_POLL * 3
def test_cancel_does_not_fall_through_to_hdhub(env, monkeypatch, tmp_path):
monkeypatch.setattr(addon, "_remux_to_mp4", _real_remux_that_runs_forever(tmp_path))
addon._start_mp4_job("series", "tmdb:138502", 2, 6, 1080)
assert _wait(lambda: (addon._job_info.get(KEY) or {}).get("phase") == "downloading")
addon._cancel_job(KEY)
_join_jobs()
assert env["hdhub"] == 0
def test_cancel_keeps_the_part_file(env, monkeypatch, tmp_path):
monkeypatch.setattr(addon, "_remux_to_mp4", _real_remux_that_runs_forever(tmp_path))
addon._start_mp4_job("series", "tmdb:138502", 2, 6, 1080)
assert _wait(lambda: (addon._job_info.get(KEY) or {}).get("phase") == "downloading")
dest = addon._job_info[KEY]["dest"]
addon._cancel_job(KEY)
_join_jobs()
assert os.path.exists(addon._part_path(dest))
assert not os.path.exists(dest)
def test_cancel_is_recorded_and_never_resumed_at_boot(env, monkeypatch, tmp_path):
monkeypatch.setattr(addon, "_remux_to_mp4", _real_remux_that_runs_forever(tmp_path))
addon._start_mp4_job("series", "tmdb:138502", 2, 6, 1080)
assert _wait(lambda: (addon._job_info.get(KEY) or {}).get("phase") == "downloading")
addon._cancel_job(KEY)
_join_jobs()
assert _history()[-1]["outcome"] == "cancelled"
with open(addon.INGEST_QUEUE_FILE) as f:
assert json.load(f)[KEY]["state"] == "cancelled"
started = []
monkeypatch.setattr(addon, "_start_mp4_job", lambda *a, **k: started.append(a))
addon._resume_ingests()
assert started == []
def test_cancel_while_waiting_never_resolves(env, monkeypatch):
sem = threading.Semaphore(0)
monkeypatch.setattr(addon, "_mp4_sem", sem)
monkeypatch.setattr(addon, "_remux_to_mp4", lambda url, dest: _write_dest(dest))
addon._start_mp4_job("series", "tmdb:138502", 2, 6, 1080)
assert _wait(lambda: (addon._job_info.get(KEY) or {}).get("phase") == "waiting")
addon._cancel_job(KEY)
sem.release()
_join_jobs()
assert addon._mp4_jobs[KEY] == "cancelled"
assert env["resolve"] == 0
def test_a_cancelled_job_can_be_started_again(env, monkeypatch):
addon._mp4_jobs[KEY] = "cancelled"
monkeypatch.setattr(addon, "_remux_to_mp4", lambda url, dest: _write_dest(dest))
assert addon._start_mp4_job("series", "tmdb:138502", 2, 6, 1080) == "running"
_join_jobs()
assert addon._mp4_jobs[KEY] == "done"
def test_cancelling_an_unknown_job_is_false(env):
assert addon._cancel_job("series:tmdb:1:1:1:1080") is False
def test_cancel_does_not_touch_other_jobs(env, monkeypatch, tmp_path):
monkeypatch.setattr(addon, "_remux_to_mp4", _real_remux_that_runs_forever(tmp_path))
other = "series:tmdb:138502:2:7:1080"
addon._start_mp4_job("series", "tmdb:138502", 2, 6, 1080)
addon._start_mp4_job("series", "tmdb:138502", 2, 7, 1080)
assert _wait(lambda: all((addon._job_info.get(k) or {}).get("phase") == "downloading"
for k in (KEY, other)))
addon._cancel_job(KEY)
assert _wait(lambda: addon._mp4_jobs.get(KEY) == "cancelled")
assert addon._mp4_jobs.get(other) == "running"
addon._cancel_job(other)
# --- season -----------------------------------------------------------------
def test_a_season_job_reports_its_position(env, monkeypatch):
monkeypatch.setattr(addon, "_released_episodes", lambda tmdb, s: [1, 2, 3])
gate = threading.Event()
seen = []
def remux(url, dest):
seen.append(dict(addon._season_info["season:tmdb:138502:2:1080"]))
_write_dest(dest)
monkeypatch.setattr(addon, "_remux_to_mp4", remux)
addon._start_season_job("tmdb:138502", 2, 1080)
_join_jobs()
assert [s["index"] for s in seen] == [1, 2, 3]
assert seen[0]["episodes"] == [1, 2, 3]
def test_cancelling_a_season_stops_the_remaining_episodes(env, monkeypatch, tmp_path):
monkeypatch.setattr(addon, "_released_episodes", lambda tmdb, s: [1, 2, 3])
monkeypatch.setattr(addon, "_remux_to_mp4", _real_remux_that_runs_forever(tmp_path))
skey = "season:tmdb:138502:2:1080"
addon._start_season_job("tmdb:138502", 2, 1080)
first = "series:tmdb:138502:2:1:1080"
assert _wait(lambda: (addon._job_info.get(first) or {}).get("phase") == "downloading")
assert addon._cancel_season(skey) is True
_join_jobs()
assert addon._mp4_jobs[first] == "cancelled"
assert "series:tmdb:138502:2:2:1080" not in addon._mp4_jobs
assert addon._season_jobs[skey].startswith("cancelled")
def test_cancelling_one_episode_of_a_season_skips_only_that_episode(env, monkeypatch, tmp_path):
monkeypatch.setattr(addon, "_released_episodes", lambda tmdb, s: [1, 2])
forever = _real_remux_that_runs_forever(tmp_path)
calls = []
def remux(url, dest):
calls.append(dest)
if len(calls) == 1:
forever(url, dest) # episode 1 hangs until cancelled
else:
_write_dest(dest)
monkeypatch.setattr(addon, "_remux_to_mp4", remux)
addon._start_season_job("tmdb:138502", 2, 1080)
first = "series:tmdb:138502:2:1:1080"
assert _wait(lambda: (addon._job_info.get(first) or {}).get("phase") == "downloading")
addon._cancel_job(first)
_join_jobs()
assert addon._mp4_jobs[first] == "cancelled"
assert addon._mp4_jobs["series:tmdb:138502:2:2:1080"] == "done"
assert addon._season_jobs["season:tmdb:138502:2:1080"] == "done (2)"