Mirko Holler
30b5119a2f
Type-tags queue jobs with a "kind" field (back-compat: missing kind defaults to "tomo"), and adds "command" jobs that reconfigure the beamline (currently: absolute moves on an allow-listed device, plus an optimize_idgap stub) between tomograms instead of running a scan. Command jobs are added via tomo_queue_add_command() against a curated named-action registry (_TOMO_QUEUE_ACTIONS) - not arbitrary code - so every queued action is auditable and safe to persist across a kernel restart. Validation is two-layer: schema checks at add time, and each action re-checks its own arguments against live hardware at execution time, which is the actual safety boundary. Crash-resume for a command job re-runs the whole step sequence from the top if every step is idempotent, or prompts the operator otherwise. flomni_webpage_generator.py is guarded against command jobs, which have no "params" key. See AI_docs/TOMO_QUEUE_COMMAND_JOBS_PLAN.md for the full design.
121 lines
4.8 KiB
Python
121 lines
4.8 KiB
Python
"""Checklist section C -- concurrency, the path the GUI will use. See TOMO_QUEUE_TESTING.md.
|
|
|
|
Session A (executing the queue) is always a subprocess -- tomo_queue_execute()
|
|
must run on a process's real main thread (see _queue_helpers.ProgressSampler).
|
|
Session B (editing concurrently) is the pytest process's own flomni_sim
|
|
connection: a second, independent client hitting the same redis-backed
|
|
tomo_queue global var, exactly like a second operator's kernel or the future
|
|
GUI dialog would.
|
|
"""
|
|
|
|
import datetime
|
|
import uuid
|
|
|
|
from _queue_helpers import add_short_job, spawn_queue_subprocess, wait_until
|
|
|
|
|
|
def test_concurrent_queue_edits_from_second_client(
|
|
flomni_sim, bec_services_config_file_path, tmp_path
|
|
):
|
|
"""C8: while session A executes the queue, session B reorders the
|
|
pending tail, deletes a pending job, and appends a new one, all mid-job1.
|
|
Session A must pick up every change at the next job boundary and write
|
|
statuses to the right jobs -- the real-redis version of what the offline
|
|
mock harness proved in memory (section 2 of the doc).
|
|
"""
|
|
flomni = flomni_sim
|
|
bec = flomni.client
|
|
|
|
add_short_job(flomni, "job1")
|
|
add_short_job(flomni, "job2")
|
|
add_short_job(flomni, "job3")
|
|
add_short_job(flomni, "job4")
|
|
jobs = flomni._tomo_queue_proxy.as_list()
|
|
ids = {j["label"]: j["id"] for j in jobs}
|
|
|
|
proc = spawn_queue_subprocess(bec_services_config_file_path, tmp_path / "sessionA.log")
|
|
try:
|
|
wait_until(lambda: flomni._tomo_queue_proxy.as_list()[0]["status"] == "running", timeout=30)
|
|
|
|
# -- session B's edits, all mid-job1: reorder the pending tail
|
|
# (job4 ahead of job2), delete job3, append job5. Deliberately does
|
|
# NOT touch any live tomo_* param global var here (only the
|
|
# tomo_queue list itself) -- job1's own live scan in session A is
|
|
# reading those same params every projection; touching them mid-run
|
|
# would perturb the *running* job, which is a different hazard than
|
|
# the one this test is about (see TOMO_QUEUE_COMMAND_JOBS_PLAN's
|
|
# "Done must be blocked while running" rule for the GUI).
|
|
current = {j["id"]: j for j in bec.get_global_var("tomo_queue")}
|
|
reordered = [
|
|
current[ids["job1"]], # the running job, left in place
|
|
current[ids["job4"]],
|
|
current[ids["job2"]],
|
|
]
|
|
bec.set_global_var("tomo_queue", reordered)
|
|
|
|
job5 = dict(current[ids["job1"]]) # clone an existing job's params snapshot
|
|
job5.update(
|
|
id=uuid.uuid4().hex,
|
|
label="job5",
|
|
status="pending",
|
|
added_at=datetime.datetime.now().isoformat(),
|
|
)
|
|
flomni._tomo_queue_proxy.append(job5)
|
|
|
|
returncode = proc.wait(timeout=420) # 4 real jobs x ~8 projections each
|
|
finally:
|
|
if proc.poll() is None:
|
|
proc.kill()
|
|
proc.wait(timeout=10)
|
|
|
|
log_text = (tmp_path / "sessionA.log").read_text()
|
|
assert returncode == 0, f"session A failed (see log):\n{log_text[-4000:]}"
|
|
|
|
jobs = flomni._tomo_queue_proxy.as_list()
|
|
labels_in_order = [j["label"] for j in jobs]
|
|
statuses = {j["label"]: j["status"] for j in jobs}
|
|
|
|
assert "job3" not in labels_in_order, "job3 was deleted mid-run and must not reappear"
|
|
assert labels_in_order == [
|
|
"job1",
|
|
"job4",
|
|
"job2",
|
|
"job5",
|
|
], f"queue should reflect the reorder+append made mid-run, got {labels_in_order}"
|
|
assert statuses == {"job1": "done", "job4": "done", "job2": "done", "job5": "done"}
|
|
|
|
|
|
def test_delete_running_job_mid_run(flomni_sim, bec_services_config_file_path, tmp_path):
|
|
"""C9: deleting the *running* job mid-run (unprotected at CLI level --
|
|
the GUI will block it, per section 5 of the doc) must degrade
|
|
gracefully: update_by_id() returns False, job1's scan finishes on its
|
|
own terms, nothing crashes, and no status gets written to the wrong job.
|
|
"""
|
|
flomni = flomni_sim
|
|
|
|
add_short_job(flomni, "job1")
|
|
add_short_job(flomni, "job2")
|
|
|
|
proc = spawn_queue_subprocess(bec_services_config_file_path, tmp_path / "sessionA.log")
|
|
try:
|
|
wait_until(lambda: flomni._tomo_queue_proxy.as_list()[0]["status"] == "running", timeout=30)
|
|
|
|
# Delete the running job (index 0) out from under session A.
|
|
flomni.tomo_queue_delete(0)
|
|
|
|
returncode = proc.wait(timeout=180)
|
|
finally:
|
|
if proc.poll() is None:
|
|
proc.kill()
|
|
proc.wait(timeout=10)
|
|
|
|
log_text = (tmp_path / "sessionA.log").read_text()
|
|
assert (
|
|
returncode == 0
|
|
), f"session A should degrade gracefully, not crash (see log):\n{log_text[-4000:]}"
|
|
|
|
jobs = flomni._tomo_queue_proxy.as_list()
|
|
labels = [j["label"] for j in jobs]
|
|
assert labels == ["job2"], f"deleted job1 must not reappear, got {labels}"
|
|
assert jobs[0]["status"] == "done", "job2 should still have run after job1's scan finished"
|