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