"""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"