1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
"""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
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()
snapshot = {"upload_id": jobs.register_upload(source, 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"]["name"] = "renamed.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()
snapshot = {"upload_id": jobs.register_upload(source, 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([item for item in jobs.work.iterdir() if item.name != "uploads"]) == 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 [item for item in jobs.work.iterdir() if item.name != "uploads"]
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()
snapshot = {"upload_id": jobs.register_upload(source, 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()
|
