csaxs_bec/omny_e2e_tests/test_tomo_queue_command_jobs.py

"""Step 2 -- command jobs. See AI_docs/TOMO_QUEUE_COMMAND_JOBS_PLAN.md.

ccm_energy/idgap (the plain "move" allow-list) aren't in the flomni sim
config (simulated_flomni.yaml) -- only on the real front end
(bl_frontend.yaml/bl_optics_hutch.yaml). Tests that only exercise add-time
validation/rendering (no tomo_queue_execute()) reference idgap for that
reason, while tests that actually execute the queue use the dedicated
"move_foptz_with_shift" action against foptz, which IS in the sim config
and is the only queue-movable path for foptz (it is deliberately excluded
from the plain "move" allow-list, see flomni.py's _TOMO_QUEUE_MOVE_DEVICES).
"""

import datetime
import uuid

import pytest
from _queue_helpers import FAST_STEPSIZE, add_short_job

FOPTZ_A = 15.0
FOPTZ_B = 5.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": "move_foptz_with_shift",
                "kwargs": {"foptz_target": 99.0, "manual_shift_x": 0.0, "manual_shift_y": 0.0},
            }
        )

    with pytest.raises(ValueError, match="must be a number"):
        flomni.tomo_queue_add_command(
            {
                "action": "move_foptz_with_shift",
                "kwargs": {"foptz_target": "far", "manual_shift_x": 0.0, "manual_shift_y": 0.0},
            }
        )

    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", "move_foptz_with_shift"}
    assert published["move_devices"]["idgap"]["unit"] == "mm"
    assert published["actions"]["move"]["idempotent"] is True
    assert (
        published["actions"]["move_foptz_with_shift"]["params"]["foptz_target"]["default"] == 17.0
    )


def test_show_and_webpage_survive_command_job(flomni_sim, capsys):
    flomni = flomni_sim
    flomni.tomo_queue_add_command(
        {"action": "move", "kwargs": {"positions": {"idgap": 6.0}}},
        label="move idgap",
        idempotent=True,
    )
    capsys.readouterr()
    jobs = flomni.tomo_queue_show()
    out = capsys.readouterr().out
    assert "CMD move{'positions': {'idgap': 6.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_foptz(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 foptz at -- foptz'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_foptz_with_shift(FOPTZ_B, 0.0, 0.0)
    foptz_before = bec.device_manager.devices.foptz.readback.get()
    assert foptz_before == pytest.approx(FOPTZ_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_foptz_with_shift",
            "kwargs": {"foptz_target": FOPTZ_A, "manual_shift_x": 1.0, "manual_shift_y": -2.0},
        },
        label="move foptz",
    )

    flomni.tomo_queue_execute()

    jobs = flomni._tomo_queue_proxy.as_list()
    assert [j["status"] for j in jobs] == ["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"

    foptz_after = bec.device_manager.devices.foptz.readback.get()
    assert foptz_after == pytest.approx(FOPTZ_A)
    assert foptz_after != pytest.approx(foptz_before)
    assert flomni.manual_shift_x == pytest.approx(1.0)
    assert flomni.manual_shift_y == pytest.approx(-2.0)


def test_non_idempotent_command_resume_prompts(flomni_sim, monkeypatch):
    flomni = flomni_sim
    bec = flomni.client
    foptz = bec.device_manager.devices.foptz

    # Use the same real move path the "move_foptz_with_shift" action itself
    # uses, rather than the device's underlying ophyd object directly
    # (foptz's BEC wrapper is a Positioner here, unlike fsamx's DeviceBase --
    # no .obj to reach for).
    flomni._queue_action_move_foptz_with_shift(FOPTZ_B, 0.0, 0.0)
    start = foptz.readback.get()
    assert start == pytest.approx(FOPTZ_B)

    index = flomni.tomo_queue_add_command(
        {
            "action": "move_foptz_with_shift",
            "kwargs": {"foptz_target": FOPTZ_A, "manual_shift_x": 0.0, "manual_shift_y": 0.0},
        },
        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_shared.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: foptz stays put, job still ends up "done".
    monkeypatch.setattr(OMNYTools, "yesno", _yesno(False))
    flomni.tomo_queue_execute()
    assert len(prompts) == 1
    assert foptz.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 foptz.readback.get() == pytest.approx(FOPTZ_A)
    assert flomni._tomo_queue_proxy.as_list()[0]["status"] == "done"