"""The ingest queue survives more than one job finishing at once.
Found by a flaky test, but it is not a test problem. _set_job_state truncated
the queue file and then wrote it, so a concurrent reader could catch it empty --
"Expecting value: line 1 column 1 (char 0)". _load_job_queue catches everything
and returns {} by design, so a torn read does not fail loudly: it reports an
empty queue, and the next write persists that. Every job on record, gone.
Two jobs finishing together was enough to hit it. A season finishing episode
after episode makes it routine, which is what made this worth fixing rather
than retrying.
Run: .venv/bin/python3 -m pytest test_queue_atomicity.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), "queue.json"))
monkeypatch.setattr(addon, "_mp4_jobs", {})
yield
def test_concurrent_writers_do_not_lose_each_others_entries():
"""Read-modify-write without a lock: two threads both load the queue, both
add their own key, and the second write drops the first."""
def write(n):
addon._set_job_state(f"key{n}", "done", args=("series", f"tmdb:{n}", 1, 1, 720))
threads = [threading.Thread(target=write, args=(n,)) for n in range(24)]
for t in threads:
t.start()
for t in threads:
t.join(10)
assert len(addon._load_job_queue()) == 24
def test_a_reader_never_sees_a_half_written_queue():
"""The failure the flaky test kept hitting, reproduced deliberately."""
addon._set_job_state("seed", "done", args=("series", "tmdb:1", 1, 1, 720))
errors = []
stop = threading.Event()
def reader():
while not stop.is_set():
try:
with open(addon.INGEST_QUEUE_FILE) as f:
json.load(f)
except FileNotFoundError:
pass # the rename window, not a torn read
except Exception as exc:
errors.append(exc)
def writer():
for n in range(150):
addon._set_job_state(f"k{n}", "running", args=("series", "x", 1, n, 720))
r = threading.Thread(target=reader, daemon=True)
r.start()
w = threading.Thread(target=writer)
w.start()
w.join(30)
stop.set()
r.join(5)
assert errors == [], f"reader saw a partial file: {errors[:3]}"
def test_the_queue_is_still_readable_afterwards():
addon._set_job_state("k", "done", args=("series", "tmdb:1", 1, 1, 720))
assert addon._load_job_queue()["k"]["args"] == ["series", "tmdb:1", 1, 1, 720]
def test_no_temp_file_is_left_behind():
addon._set_job_state("k", "done", args=("series", "tmdb:1", 1, 1, 720))
leftovers = [f for f in os.listdir(os.path.dirname(addon.INGEST_QUEUE_FILE))
if f.endswith(".tmp")]
assert leftovers == []
# --- a read failure must not become a write of nothing ----------------------
#
# The torn read above was only half of it. _load_job_queue answers {} for
# anything it cannot parse, by design, so that a bad file cannot stop the addon
# booting. That is right for a reader and catastrophic for a writer: it would
# rebuild the queue from nothing and persist it. One unreadable read, and every
# job on record is gone.
#
# This is not hypothetical. On 2026-08-31 the real queue went from twenty
# entries to a single one left behind by a test.
def test_an_unreadable_queue_is_not_overwritten_with_nothing():
with open(addon.INGEST_QUEUE_FILE, "w") as f:
f.write("{ this is not json")
addon._set_job_state("k", "done", args=("series", "tmdb:1", 1, 1, 720))
with open(addon.INGEST_QUEUE_FILE) as f:
assert f.read() == "{ this is not json"
def test_a_queue_that_is_not_an_object_is_not_overwritten():
with open(addon.INGEST_QUEUE_FILE, "w") as f:
f.write('["a list, somehow"]')
addon._set_job_state("k", "done", args=("series", "tmdb:1", 1, 1, 720))
with open(addon.INGEST_QUEUE_FILE) as f:
assert json.load(f) == ["a list, somehow"]
def test_a_missing_queue_is_still_created():
""""No file yet" and "file I cannot read" must not be treated alike."""
assert not os.path.exists(addon.INGEST_QUEUE_FILE)
addon._set_job_state("k", "done", args=("series", "tmdb:1", 1, 1, 720))
assert addon._load_job_queue()["k"]["state"] == "done"