test(flomni): add e2e coverage for tomo-queue command jobs

Covers add-time validation (unknown action, disallowed device, out-of-range/wrong-type params, unknown kwarg - each rejected without landing in the queue), the execution-time allow-list re-check as the real safety boundary, the tomo_queue_actions global var matching the live registry, tomo_queue_show()/the webpage generator surviving a command job, legacy kind-less jobs still running as tomograms alongside a real command-job device move, and the non-idempotent resume prompt (distinct from the executor's own start-confirmation prompt). Also fixes flomni_sim: it only snapshotted/restored whatever was already in the tomo_queue/tomo_progress global vars, so a stale entry left behind by an earlier crashed run or manual debugging session against the same shared sim Redis would contaminate every test that assumed it started from an empty queue. Now resets both to empty at setup, in addition to snapshotting/restoring around the test.
2026-07-13 09:06:12 +02:00
parent 30b5119a2f
commit bfe60ecd36
2 changed files with 202 additions and 1 deletions
@@ -22,12 +22,17 @@ def flomni_sim(bec_redis_fixture, bec_services_config_file_path, bec_servers):
    involves and why. The ``tomo_queue`` / ``tomo_progress`` global vars are
    snapshotted before the test and restored after, so tests don't leak
    queue/progress state into each other or into a real operator's session
-    on the same sim.
+    on the same sim. The queue is also reset to empty at setup: it lives in
+    the shared sim Redis, so a prior crashed run or manual debugging session
+    can leave stale entries behind that would otherwise contaminate every
+    test that assumes it starts from an empty queue.
    """
    bec, flomni = build_flomni(bec_services_config_file_path)

    queue_snapshot = bec.get_global_var(_QUEUE_VAR)
    progress_snapshot = bec.get_global_var(_PROGRESS_VAR)
+    bec.set_global_var(_QUEUE_VAR, [])
+    bec.set_global_var(_PROGRESS_VAR, None)

    try:
        yield flomni
@@ -0,0 +1,196 @@
+"""Step 2 -- command jobs. See AI_docs/TOMO_QUEUE_COMMAND_JOBS_PLAN.md.
+
+mokev/idgap (the real curated move targets) aren't in the flomni sim config
+(simulated_flomni.yaml) -- only on the real front end (bl_frontend.yaml), so
+the real device-moving path here is exercised against `ftray` (sample
+transfer tray), added to _TOMO_QUEUE_MOVE_DEVICES specifically for this
+sim-testability, per TOMO_QUEUE_TESTING.md.
+"""
+
+import datetime
+import uuid
+
+import pytest
+from _queue_helpers import FAST_STEPSIZE, add_short_job
+
+FTRAY_A = -50.0
+FTRAY_B = -10.0
+
+
+def test_add_command_validation_errors(flomni_sim):
+    flomni = flomni_sim
+
+    with pytest.raises(ValueError, match="Unknown queue action"):
+        flomni.tomo_queue_add_command({"action": "not_a_real_action", "kwargs": {}})
+
+    with pytest.raises(ValueError, match="not queue-movable"):
+        flomni.tomo_queue_add_command({"action": "move", "kwargs": {"positions": {"fsamx": 0.0}}})
+
+    with pytest.raises(ValueError, match="above the maximum"):
+        flomni.tomo_queue_add_command(
+            {"action": "optimize_idgap", "kwargs": {"search_range": 99.0}}
+        )
+
+    with pytest.raises(ValueError, match="must be a number"):
+        flomni.tomo_queue_add_command(
+            {"action": "optimize_idgap", "kwargs": {"search_range": "wide"}}
+        )
+
+    with pytest.raises(ValueError, match="has no parameter"):
+        flomni.tomo_queue_add_command({"action": "move", "kwargs": {"bogus_kwarg": 1}})
+
+    # None of the rejected adds should have landed in the queue.
+    assert flomni._tomo_queue_proxy.as_list() == []
+
+
+def test_move_execution_time_allowlist_recheck(flomni_sim):
+    """Validation is two-layer (doc §3.3): tomo_queue_add_command() already
+    refuses a disallowed device at add time (see
+    test_add_command_validation_errors), but _queue_action_move() must
+    independently refuse it again at execution time -- the actual safety
+    boundary, since a job could reach the queue by some other path than the
+    validated CLI call (bypassed here by appending the raw dict directly, the
+    same way test_legacy_queue_migration simulates a pre-existing queue
+    entry).
+    """
+    flomni = flomni_sim
+    job = {
+        "kind": "command",
+        "id": uuid.uuid4().hex,
+        "label": "sneaked past add-time validation",
+        "steps": [{"action": "move", "kwargs": {"positions": {"fsamx": 0.0}}}],
+        "idempotent": True,
+        "status": "pending",
+        "added_at": datetime.datetime.now().isoformat(),
+    }
+    flomni._tomo_queue_proxy.append(job)
+
+    with pytest.raises(ValueError, match="not queue-movable"):
+        flomni.tomo_queue_execute()
+
+    assert flomni._tomo_queue_proxy.as_list()[0]["status"] == "incomplete"
+
+
+def test_tomo_queue_actions_global_var_matches_registry(flomni_sim):
+    flomni = flomni_sim
+    published = flomni.client.get_global_var("tomo_queue_actions")
+    assert set(published["actions"]) == {"move", "optimize_idgap"}
+    assert published["move_devices"]["ftray"]["unit"] == "mm"
+    assert published["actions"]["move"]["idempotent"] is True
+    assert published["actions"]["optimize_idgap"]["params"]["search_range"]["default"] == 0.5
+
+
+def test_show_and_webpage_survive_command_job(flomni_sim, capsys):
+    flomni = flomni_sim
+    flomni.tomo_queue_add_command(
+        {"action": "move", "kwargs": {"positions": {"ftray": FTRAY_A}}},
+        label="move ftray",
+        idempotent=True,
+    )
+    capsys.readouterr()
+    jobs = flomni.tomo_queue_show()
+    out = capsys.readouterr().out
+    assert "CMD move{'positions': {'ftray': -50.0}}" in out
+    assert "[idem]" in out
+    assert jobs[0]["kind"] == "command"
+
+    from csaxs_bec.bec_ipython_client.plugins.flomni.flomni_webpage_generator import _render_html
+
+    html = _render_html(phone_numbers=[], has_tomo_queue=True, tomo_types={1: "x"})
+    assert "isCommand" in html  # the rendering branch exists; full JS isn't executed here
+
+
+def test_legacy_job_no_kind_defaults_to_tomo_and_command_moves_ftray(flomni_sim):
+    """B-style back-compat + the real command-job execution path, combined
+    into one mixed-queue run to keep the (only) real scan in this file to
+    one job.
+    """
+    flomni = flomni_sim
+    bec = flomni.client
+
+    # Start from a known position, not whatever a previous test run happened
+    # to leave ftray at -- ftray's position is real device-server state that
+    # persists across separate pytest invocations, unlike the tomo_queue/
+    # tomo_progress global vars the flomni_sim fixture snapshots/restores.
+    flomni._queue_action_move({"ftray": FTRAY_B})
+    ftray_before = bec.device_manager.devices.ftray.readback.get()
+    assert ftray_before == pytest.approx(FTRAY_B)
+
+    add_short_job(flomni, "legacy_tomo", FAST_STEPSIZE)
+    # Simulate a queue entry written before the "kind" field existed.
+    jobs = bec.get_global_var("tomo_queue")
+    del jobs[0]["kind"]
+    bec.set_global_var("tomo_queue", jobs)
+
+    flomni.tomo_queue_add_command(
+        {"action": "move", "kwargs": {"positions": {"ftray": FTRAY_A}}}, label="move ftray"
+    )
+    flomni.tomo_queue_add_command(
+        {"action": "optimize_idgap", "kwargs": {"search_range": 0.5}}, label="idgap stub"
+    )
+
+    flomni.tomo_queue_execute()
+
+    jobs = flomni._tomo_queue_proxy.as_list()
+    assert [j["status"] for j in jobs] == ["done", "done", "done"]
+    # Still no "kind" key -- proves it ran (and got marked done) via the
+    # back-compat default, not that execute() silently added one.
+    assert "kind" not in jobs[0]
+    assert "params" in jobs[0]
+    assert jobs[1]["kind"] == "command"
+
+    ftray_after = bec.device_manager.devices.ftray.readback.get()
+    assert ftray_after == pytest.approx(FTRAY_A)
+    assert ftray_after != pytest.approx(ftray_before)
+
+
+def test_non_idempotent_command_resume_prompts(flomni_sim, monkeypatch):
+    flomni = flomni_sim
+    bec = flomni.client
+    ftray = bec.device_manager.devices.ftray
+
+    # Use the same real move path the "move" action itself uses, rather than
+    # the device's underlying ophyd object directly (ftray's BEC wrapper is
+    # a Positioner here, unlike fsamx's DeviceBase -- no .obj to reach for).
+    flomni._queue_action_move({"ftray": FTRAY_B})
+    start = ftray.readback.get()
+    assert start == pytest.approx(FTRAY_B)
+
+    index = flomni.tomo_queue_add_command(
+        {"action": "move", "kwargs": {"positions": {"ftray": FTRAY_A}}},
+        label="non-idempotent move",
+        idempotent=False,
+    )
+    job_id = flomni._tomo_queue_proxy.as_list()[index]["id"]
+    flomni._tomo_queue_proxy.update_by_id(job_id, status="incomplete")
+
+    prompts = []
+    from csaxs_bec.bec_ipython_client.plugins.omny.omny_general_tools import OMNYTools
+
+    def _yesno(resume_answer):
+        # tomo_queue_execute()'s own top-level "OK to start?" confirmation
+        # goes through the same OMNYTools.yesno -- always let that one
+        # through, and only record/answer the *resume* prompt from
+        # _run_command_job, which is what this test is actually about.
+        def _fn(self, message, *a, **k):
+            if message.startswith("Starting automatic execution"):
+                return True
+            prompts.append(message)
+            return resume_answer
+
+        return _fn
+
+    # "no" -> must not re-run: ftray stays put, job still ends up "done".
+    monkeypatch.setattr(OMNYTools, "yesno", _yesno(False))
+    flomni.tomo_queue_execute()
+    assert len(prompts) == 1
+    assert ftray.readback.get() == pytest.approx(start)
+    assert flomni._tomo_queue_proxy.as_list()[0]["status"] == "done"
+
+    # Re-arm as incomplete and try again with "yes" -> must re-run.
+    flomni._tomo_queue_proxy.update_by_id(job_id, status="incomplete")
+    monkeypatch.setattr(OMNYTools, "yesno", _yesno(True))
+    flomni.tomo_queue_execute()
+    assert len(prompts) == 2
+    assert ftray.readback.get() == pytest.approx(FTRAY_A)
+    assert flomni._tomo_queue_proxy.as_list()[0]["status"] == "done"