"""One library file, one ingest -- whatever ident asked for it.
The in-flight guard in _start_mp4_job keys on `kind:ident:season:episode:height`.
But the same episode reaches this service under two idents: Kodi's cinemeta rows
send `tt…`, the flixlab catalog and the web UI send `tmdb:…`. Both resolve
through _library_meta to the SAME _library_path, and therefore the same .part --
yet they are different keys, so neither job sees the other. Two ffmpeg processes
then write one .part. (labs#12.)
The web UI makes this likelier: the point of it is to start a download from
somewhere other than the Kodi row that may already be fetching it.
Run: .venv/bin/python3 -m pytest test_one_file_one_ingest.py -q
"""
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}
@pytest.fixture
def ingest_env(tmp_path, monkeypatch):
monkeypatch.setattr(addon, "INGEST_QUEUE_FILE",
os.path.join(str(tmp_path), "ingest_queue.json"))
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path / "lib"))
monkeypatch.setattr(addon, "LIBRARY_VERIFIED_FILE",
os.path.join(str(tmp_path), "library_verified.json"))
monkeypatch.setattr(addon, "_mp4_jobs", {})
monkeypatch.setattr(addon, "_library_meta", lambda *a, **k: dict(META))
monkeypatch.setattr(addon, "_enough_space", lambda: True)
monkeypatch.setattr(addon, "_resolve_vidcore", lambda *a, **k: "http://cdn/m.m3u8")
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)
calls = {"n": 0, "live": 0, "peak": 0}
lock = threading.Lock()
def fake_remux(variant_url, dest):
with lock:
calls["n"] += 1
calls["live"] += 1
calls["peak"] = max(calls["peak"], calls["live"])
time.sleep(0.3) # long enough for the other to start
os.makedirs(os.path.dirname(dest), exist_ok=True)
with open(dest, "wb") as f:
f.write(b"x" * 1024)
with lock:
calls["live"] -= 1
monkeypatch.setattr(addon, "_remux_to_mp4", fake_remux)
yield calls
# Same reason as test_ingest_queue: a job thread that outlives the patches
# writes its final state to the REAL queue.
for t in threading.enumerate():
if t.name.startswith("mp4-") and t is not threading.current_thread():
t.join(10)
def _join_jobs():
for t in threading.enumerate():
if t.name.startswith("mp4-") and t is not threading.current_thread():
t.join(10)
def test_two_idents_for_one_episode_fetch_it_once(ingest_env):
addon._start_mp4_job("series", "tt16026746", 2, 7, 1080)
addon._start_mp4_job("series", "tmdb:138502", 2, 7, 1080)
_join_jobs()
assert ingest_env["peak"] == 1, "two remuxes wrote the same .part at once"
assert ingest_env["n"] == 1, "the second ident fetched a copy already held"
def test_the_second_job_reports_done_not_failed(ingest_env):
"""It got what it asked for. Reporting a failure here is the #11 shape --
an outcome misreported -- and would put a false red row on /library."""
addon._start_mp4_job("series", "tt16026746", 2, 7, 1080)
addon._start_mp4_job("series", "tmdb:138502", 2, 7, 1080)
_join_jobs()
assert addon._mp4_jobs["series:tt16026746:2:7:1080"] == "done"
assert addon._mp4_jobs["series:tmdb:138502:2:7:1080"] == "done"
def test_different_episodes_still_run_concurrently(ingest_env, monkeypatch):
"""The guard is per FILE. It must not serialise unrelated ingests behind
one another -- that would turn the remux semaphore's two slots into one."""
metas = {1: dict(META, episode_title="One"), 2: dict(META, episode_title="Two")}
monkeypatch.setattr(addon, "_library_meta",
lambda kind, ident, s=0, e=0: metas[e])
addon._start_mp4_job("series", "tmdb:138502", 2, 1, 1080)
addon._start_mp4_job("series", "tmdb:138502", 2, 2, 1080)
_join_jobs()
assert ingest_env["n"] == 2
assert ingest_env["peak"] == 2