diff --git a/omny_e2e_tests/conftest.py b/omny_e2e_tests/conftest.py index ca1f236..de9918f 100644 --- a/omny_e2e_tests/conftest.py +++ b/omny_e2e_tests/conftest.py @@ -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 diff --git a/omny_e2e_tests/test_tomo_queue_command_jobs.py b/omny_e2e_tests/test_tomo_queue_command_jobs.py new file mode 100644 index 0000000..dfb3d89 --- /dev/null +++ b/omny_e2e_tests/test_tomo_queue_command_jobs.py @@ -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"