aboutsummaryrefslogtreecommitdiff
path: root/lib/project/tests/test_gui_jobs.py
diff options
context:
space:
mode:
Diffstat (limited to 'lib/project/tests/test_gui_jobs.py')
-rw-r--r--lib/project/tests/test_gui_jobs.py124
1 files changed, 124 insertions, 0 deletions
diff --git a/lib/project/tests/test_gui_jobs.py b/lib/project/tests/test_gui_jobs.py
new file mode 100644
index 0000000..6116a72
--- /dev/null
+++ b/lib/project/tests/test_gui_jobs.py
@@ -0,0 +1,124 @@
+"""Bounded single-worker scheduling and cancellation through real process cleanup."""
+import os
+import sys
+import threading
+import time
+
+import pytest
+
+from producer import gui_jobs, setup
+from producer.gui import rooted
+
+
+def terminal(jobs, job_id):
+ deadline = time.monotonic() + 10
+ while time.monotonic() < deadline:
+ job = jobs.get(job_id)
+ if job["status"] in ("succeeded", "failed", "cancelled"):
+ return job
+ time.sleep(0.01)
+ pytest.fail("Job failed to finish")
+
+
+def test_single_worker_bounded_queue_and_running_process_cancellation(wav, tmp_path, monkeypatch):
+ source = wav()
+ started = threading.Event()
+ pid_file = tmp_path / "child.pid"
+ calls = []
+ def slow_scan(path, progress):
+ calls.append(path)
+ if len(calls) == 1:
+ started.set()
+ setup.run_process([sys.executable, "-c",
+ f"import os,time;open({str(pid_file)!r},'w').write(str(os.getpid()));time.sleep(60)"],
+ "Waiting for worker", progress)
+ return {"bins": 2048}
+ monkeypatch.setattr(gui_jobs, "waveform", slow_scan)
+ jobs = gui_jobs.Jobs(tmp_path, lambda value: rooted(tmp_path, value))
+ snapshot = {"path": source.name, "input_id": gui_jobs.identity(source)}
+ try:
+ first = jobs.submit("scan", snapshot)
+ assert started.wait(5)
+ # Public snapshots cannot mutate the immutable in-flight job.
+ first["snapshot"]["path"] = "missing.wav"
+ queued = [jobs.submit("scan", snapshot) for _ in range(3)]
+ with pytest.raises(gui_jobs.Busy):
+ jobs.submit("scan", snapshot)
+ assert len(calls) == 1
+ assert jobs.cancel(queued[0]["id"])["status"] == "cancelled"
+ deadline = time.monotonic() + 5
+ while not pid_file.exists() and time.monotonic() < deadline:
+ time.sleep(0.01)
+ pid = int(pid_file.read_text())
+ jobs.cancel(first["id"])
+ assert terminal(jobs, first["id"])["status"] == "cancelled"
+ with pytest.raises(ProcessLookupError):
+ os.kill(pid, 0)
+ assert terminal(jobs, queued[1]["id"])["status"] == "succeeded"
+ assert terminal(jobs, queued[2]["id"])["cached"]
+ assert len(calls) == 2
+ finally:
+ jobs.close()
+ assert not jobs.work.exists()
+
+
+def test_history_and_result_cache_are_bounded_and_evicted_media_expires(wav, tmp_path, monkeypatch):
+ source = wav()
+ monkeypatch.setattr(gui_jobs, "MAX_HISTORY", 5)
+ monkeypatch.setattr(gui_jobs, "CACHE_ENTRIES", 2)
+ def preview(path, work, snapshot, progress):
+ (work / "original.wav").write_bytes(b"preview")
+ (work / "processed.wav").write_bytes(b"preview")
+ return {}
+ monkeypatch.setattr(gui_jobs, "render_preview", preview)
+ jobs = gui_jobs.Jobs(tmp_path, lambda value: rooted(tmp_path, value))
+ snapshot = {"path": source.name, "input_id": gui_jobs.identity(source)}
+ try:
+ results = []
+ for index in range(10):
+ job = jobs.submit("preview", snapshot | {"start": index})
+ result = terminal(jobs, job["id"])
+ assert result["status"] == "succeeded", result
+ results.append(result["result"]["result_id"])
+ assert len(jobs.list()) == 5
+ assert len(jobs.cache) == 2
+ assert len(list(jobs.work.iterdir())) == 2
+ with pytest.raises(KeyError):
+ jobs.media(results[0], "original")
+ assert jobs.media(results[-1], "original") == b"preview"
+ with pytest.raises(KeyError):
+ jobs.media(results[-1], "../original")
+ # Byte budget independently evicts even below the entry-count limit.
+ monkeypatch.setattr(gui_jobs, "CACHE_BYTES", 1)
+ with jobs.condition:
+ jobs._evict()
+ assert not jobs.cache
+ assert not list(jobs.work.iterdir())
+ finally:
+ jobs.close()
+
+
+def test_queued_input_replacement_is_rejected_and_shutdown_cancels_scan(wav, tmp_path, monkeypatch):
+ source = wav()
+ started = threading.Event()
+ def scan(path, progress):
+ started.set()
+ while True:
+ progress("Scanning", 1, 10)
+ time.sleep(0.01)
+ monkeypatch.setattr(gui_jobs, "waveform", scan)
+ jobs = gui_jobs.Jobs(tmp_path, lambda value: rooted(tmp_path, value))
+ snapshot = {"path": source.name, "input_id": gui_jobs.identity(source)}
+ running = jobs.submit("scan", snapshot)
+ assert started.wait(5)
+ queued = jobs.submit("scan", snapshot)
+ with source.open("ab") as handle:
+ handle.write(b"replacement")
+ jobs.cancel(running["id"])
+ assert terminal(jobs, queued["id"])["status"] == "failed"
+ assert "changed" in jobs.get(queued["id"])["error"]
+ # Submit with updated identity and exercise app shutdown cleanup.
+ last = jobs.submit("scan", snapshot | {"input_id": gui_jobs.identity(source)})
+ jobs.close()
+ assert jobs.get(last["id"])["status"] == "cancelled"
+ assert not jobs.work.exists()