"""The ingest queue survives a restart.
_mp4_jobs was in-memory only, so restarting the addon lost every job intent: a
copy that was mid-flight simply never resumed, and because a partial writes to
`<dest>.part` while the row keys off the finished file, nothing surfaced the
loss. The rolling library makes that worse -- it queues work the user never
explicitly asked for, so nobody is waiting to notice it vanished.
Nothing is deleted to recover. An orphaned .part is reused: ffmpeg runs with -y
and writes the same path, so a resumed ingest overwrites it in place.
Run: .venv/bin/python3 -m pytest test_ingest_queue.py -q
"""
import json
import os
import threading
import pytest
import addon
@pytest.fixture(autouse=True)
def queue_file(tmp_path, monkeypatch):
monkeypatch.setattr(addon, "INGEST_QUEUE_FILE",
os.path.join(str(tmp_path), "ingest_queue.json"))
monkeypatch.setattr(addon, "_mp4_jobs", {})
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path / "lib"))
yield
# _start_mp4_job spawns a real thread, and its final _set_job_state runs
# whenever that thread gets there. Left to escape, it lands AFTER monkeypatch
# has restored INGEST_QUEUE_FILE and writes to the REAL queue -- which is
# how a run of this suite once left `series:tmdb:1:1:1:720` in production.
# Joined here rather than in the tests: any test that starts a job has the
# same problem, including ones not yet written.
for t in threading.enumerate():
if t.name.startswith("mp4-") and t is not threading.current_thread():
t.join(10)
def _load():
with open(addon.INGEST_QUEUE_FILE) as f:
return json.load(f)
# --- persistence ------------------------------------------------------------
def test_a_job_state_is_written_to_disk():
addon._set_job_state("series:tmdb:1:1:1:720", "running")
assert _load()["series:tmdb:1:1:1:720"]["state"] == "running"
def test_a_state_change_is_persisted():
addon._set_job_state("k", "running")
addon._set_job_state("k", "done")
assert _load()["k"]["state"] == "done"
def test_the_job_arguments_are_persisted_so_it_can_be_restarted():
"""A key alone cannot be re-run -- the resume path needs the arguments."""
addon._set_job_state("k", "running", args=("series", "tmdb:1", 1, 1, 720))
assert _load()["k"]["args"] == ["series", "tmdb:1", 1, 1, 720]
def test_a_missing_queue_file_is_not_an_error():
assert addon._load_job_queue() == {}
def test_a_corrupt_queue_file_is_not_an_error():
"""State is a convenience, never a reason to refuse to boot."""
with open(addon.INGEST_QUEUE_FILE, "w") as f:
f.write("{not json")
assert addon._load_job_queue() == {}
def test_an_unwritable_queue_file_does_not_break_the_job(monkeypatch):
monkeypatch.setattr(addon, "INGEST_QUEUE_FILE", "/nonexistent/dir/q.json")
addon._set_job_state("k", "running") # must not raise
assert addon._mp4_jobs["k"] == "running"
# --- resume -----------------------------------------------------------------
@pytest.fixture
def resumed(monkeypatch):
started = []
monkeypatch.setattr(addon, "_start_mp4_job",
lambda *a, **k: started.append(tuple(a)) or "running")
return started
def test_a_job_interrupted_mid_flight_is_restarted(resumed):
addon._set_job_state("k", "running", args=("series", "tmdb:1", 1, 1, 720))
addon._resume_ingests()
assert resumed == [("series", "tmdb:1", 1, 1, 720)]
def test_a_finished_job_is_not_restarted(resumed):
addon._set_job_state("k", "done", args=("series", "tmdb:1", 1, 1, 720))
addon._resume_ingests()
assert resumed == []
def test_a_failed_job_is_not_restarted(resumed):
"""A failure that recurs every boot would launch a browser each time. The
user can still ask for it explicitly by playing the row."""
addon._set_job_state("k", "failed: could not resolve title",
args=("series", "tmdb:1", 1, 1, 720))
addon._resume_ingests()
assert resumed == []
def test_a_job_with_no_arguments_is_skipped_rather_than_guessed(resumed):
addon._set_job_state("k", "running")
addon._resume_ingests()
assert resumed == []
def test_resuming_survives_a_corrupt_file(resumed):
with open(addon.INGEST_QUEUE_FILE, "w") as f:
f.write("nonsense")
addon._resume_ingests() # must not raise
assert resumed == []
def test_a_resume_rehydrates_finished_state_and_restarts_live_ones(resumed):
"""A finished job must be remembered, or the next request for it starts a
duplicate ingest of something already on disk.
A running one is deliberately NOT rehydrated as "running" -- _start_mp4_job
refuses a key already marked running, so pre-setting it would silently
prevent the restart. The restarted job sets its own state.
"""
addon._set_job_state("done-one", "done", args=("series", "tmdb:1", 1, 1, 720))
addon._set_job_state("live-one", "running", args=("series", "tmdb:1", 1, 2, 720))
addon._mp4_jobs.clear() # simulate a fresh process
addon._resume_ingests()
assert addon._mp4_jobs["done-one"] == "done"
assert resumed == [("series", "tmdb:1", 1, 2, 720)]
assert "live-one" not in addon._mp4_jobs
# --- the real job path writes state ----------------------------------------
# --- progress ---------------------------------------------------------------
# Read off the .part file rather than parsed out of ffmpeg. It is exact, needs
# no -progress plumbing or Popen rewrite, and it cannot drift from reality --
# the bytes on disk ARE the progress. A percentage is deliberately not offered:
# it would need a total size we do not reliably know up front, and a wrong
# percentage reads worse than an honest byte count.
@pytest.fixture
def part_of(tmdb_stub, monkeypatch):
dest = addon._library_path("series", "Show", "2020", "1",
season=1, episode=1, episode_title="Pilot")
os.makedirs(os.path.dirname(dest), exist_ok=True)
return dest
@pytest.fixture
def tmdb_stub(monkeypatch):
monkeypatch.setattr(addon, "_library_meta", lambda *a, **k: {
"tmdb_id": "1", "title": "Show", "year": "2020",
"episode_title": "Pilot"})
def test_no_partial_means_no_progress(part_of):
assert addon._ingest_progress("series", "tmdb:1", 1, 1) is None
def test_a_partial_reports_its_size(part_of):
with open(addon._part_path(part_of), "wb") as f:
f.write(b"x" * (3 * 1024 * 1024))
assert addon._ingest_progress("series", "tmdb:1", 1, 1) == 3 * 1024 * 1024
def test_progress_is_not_reported_once_the_copy_is_finished(part_of):
with open(part_of, "wb") as f:
f.write(b"x" * 1024)
assert addon._ingest_progress("series", "tmdb:1", 1, 1) is None
def test_progress_survives_an_unidentifiable_title(monkeypatch):
monkeypatch.setattr(addon, "_library_meta", lambda *a, **k: None)
assert addon._ingest_progress("series", "tmdb:9", 1, 1) is None
def test_start_mp4_job_records_its_arguments(monkeypatch):
"""So that a restart mid-remux can pick it up again."""
monkeypatch.setattr(addon, "_run_ingest", lambda *a, **k: "done")
addon._start_mp4_job("series", "tmdb:1", 1, 1, 720)
entry = _load()[addon._mp4_key("series", "tmdb:1", 1, 1, 720)]
assert entry["args"] == ["series", "tmdb:1", 1, 1, 720]