csaxs_bec/tests/tests_devices/test_smargon_device.py

"""Tests for the per-axis SmarGon devices (controller + motor + coordinator).

All run against the in-memory FakeTransport (no hardware/server).
"""

import pytest
from ophyd.utils import LimitError, ReadOnlyError

from csaxs_bec.devices.smargon import SmarGon, SmargopoloController, SmargopoloMotor
from csaxs_bec.devices.smargon.transport import FakeTransport, Mode
from csaxs_bec.devices.smargon.transport.errors import SmargopoloCommandError, SmargopoloModeError


@pytest.fixture(autouse=True)
def _reset_controller_singletons():
    """Each test gets fresh controllers (the singleton registry is class-level)."""
    SmargopoloController._controller_instances.clear()
    yield
    SmargopoloController._controller_instances.clear()


def _motor(name, scs_axis, **cfg):
    m = SmargopoloMotor(name=name, scs_axis=scs_axis, simulation=True, **cfg)
    m.wait_for_connection()
    return m


def _coordinator(**cfg):
    c = SmarGon(name="smargon", simulation=True, **cfg)
    c.wait_for_connection()
    return c


# -- controller singleton ---------------------------------------------------


def test_axes_and_coordinator_share_one_controller():
    coord = _coordinator()
    sgx = _motor("sgx", "SHX")
    sgy = _motor("sgy", "SHY")
    assert coord.controller is sgx.controller is sgy.controller
    assert coord.controller.transport is sgx.controller.transport


def test_real_and_sim_are_distinct_controllers():
    sim = SmargopoloMotor(name="a", scs_axis="SHX", simulation=True)
    real = SmargopoloMotor(name="b", scs_axis="SHX", simulation=False, host="h", port=1)
    assert sim.controller is not real.controller
    assert isinstance(sim.controller.transport, FakeTransport)


# -- motor as a BEC-native positioner ---------------------------------------


def test_motor_is_positioner_with_config_limits():
    sgx = _motor("sgx", "SHX", limits=[-2, 2])
    assert sgx.limits == (-2.0, 2.0)
    assert sgx.low_limit_travel.get() == -2.0 and sgx.high_limit_travel.get() == 2.0
    assert hasattr(sgx, "move") and hasattr(sgx, "stop")


def test_readback_is_aliased_to_device_name():
    """dev.wm relies on the readback key equalling the device name."""
    sgx = _motor("sgx", "SHX")
    assert "sgx" in sgx.read()  # not 'sgx_readback'
    assert "sgx_setpoint" in sgx.read()


def test_default_egu_from_axis():
    assert _motor("sgx", "SHX").egu == "mm"
    assert _motor("sgchi", "CHI").egu == "deg"


def test_refuses_move_until_referenced():
    sgx = _motor("sgx", "SHX")
    assert sgx.controller.get_mode() == "UNINITIALIZED"
    with pytest.raises(SmargopoloModeError):
        sgx.move(0.5)


def test_reference_then_move():
    coord = _coordinator()
    sgx = _motor("sgx", "SHX", limits=[-2, 2])
    coord.reference()  # shared controller -> sgx is now movable
    assert sgx.controller.is_ready()
    st = sgx.move(0.5, wait=True)
    assert st.done and st.success
    assert sgx.position == pytest.approx(0.5)


def test_set_is_nonblocking():
    coord = _coordinator()
    sgchi = _motor("sgchi", "CHI")
    coord.reference()
    st = sgchi.set(12.0)
    st.wait(timeout=5)
    assert st.success and sgchi.position == pytest.approx(12.0)


def test_soft_limits_enforced():
    coord = _coordinator()
    sgx = _motor("sgx", "SHX", limits=[-1, 1])
    coord.reference()
    with pytest.raises(LimitError):
        sgx.move(2.0)


def test_zero_zero_limits_mean_unbounded():
    coord = _coordinator()
    sgo = _motor("sgo", "OMEGA")  # no limits configured -> (0, 0)
    coord.reference()
    assert sgo.limits == (0.0, 0.0)
    sgo.move(123.0, wait=True)  # not rejected
    assert sgo.position == pytest.approx(123.0)


# -- coordinator ------------------------------------------------------------


def test_coordinator_reference_and_mode():
    coord = _coordinator()
    assert coord.get_mode() == "UNINITIALIZED"
    coord.reference()
    assert coord.is_ready() and coord.get_mode() == "READY"


def test_move_scs_coordinated():
    coord = _coordinator()
    sgx = _motor("sgx", "SHX", limits=[-2, 2])
    sgz = _motor("sgz", "SHZ", limits=[10, 22])
    coord.reference()
    coord.move_scs(SHX=0.3, SHZ=19.0, wait=True)
    assert sgx.position == pytest.approx(0.3)
    assert sgz.position == pytest.approx(19.0)


def test_move_scs_respects_limits():
    coord = _coordinator()
    _motor("sgz", "SHZ", limits=[10, 22])
    coord.reference()
    with pytest.raises(LimitError):
        coord.move_scs(SHZ=0.0)  # below the SHZ range


def test_motor_positions_and_transport_info():
    coord = _coordinator()
    assert set(coord.motor_positions()) == {"q1", "q2", "q3", "q4", "q5", "q6"}
    info = coord.transport_info()
    assert info["simulation"] is True and info["connected"] is True


def test_mode_signal_is_readonly():
    coord = _coordinator()
    assert coord.mode.get() == int(Mode.UNINITIALIZED)
    with pytest.raises(ReadOnlyError):
        coord.mode.put(2)


def test_controller_reachable_from_axis():
    coord = _coordinator()
    sgx = _motor("sgx", "SHX")
    # dev.<axis>.controller.reference() path
    sgx.controller.reference()
    assert coord.is_ready()


# -- fault handling ---------------------------------------------------------


def test_move_fails_when_target_rejected_up_front():
    coord = _coordinator()
    sgx = _motor("sgx", "SHX", limits=[-2, 2])
    coord.reference()
    coord.controller.transport.fail_next["move_absolute"] = SmargopoloCommandError("nope")
    st = sgx.set(0.5)
    with pytest.raises(SmargopoloCommandError):
        st.wait(timeout=5)
    assert not st.success


def test_move_aborts_when_controller_faults_mid_move():
    coord = _coordinator(gradual_motion=True)
    sgx = _motor("sgx", "SHX", gradual_motion=True)  # shares the gradual controller
    coord.reference()
    st = sgx.set(1e6)  # far target: gradual motion keeps it in-flight
    coord.controller.transport._mode = Mode.ERROR
    with pytest.raises(SmargopoloModeError):
        st.wait(timeout=5)
    assert not st.success and sgx.motor_is_moving.get() == 0


def test_move_scs_aborts_on_controller_fault():
    coord = _coordinator(gradual_motion=True)
    _motor("sgx", "SHX", gradual_motion=True)
    coord.reference()
    coord.controller.transport._mode = Mode.ERROR
    with pytest.raises(SmargopoloModeError):
        coord.move_scs(SHX=1e6, wait=True)