289 lines
9.4 KiB
Python
289 lines
9.4 KiB
Python
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
import yaml
|
|
from ophyd import EpicsSignalRO, PVPositioner
|
|
from ophyd.utils.errors import StatusTimeoutError
|
|
from ophyd_devices.devices.undulator import UndulatorBase
|
|
from ophyd_devices.tests.utils import patched_device
|
|
|
|
from xtreme_bec.devices.undulator import (
|
|
UndulatorAlpha,
|
|
UndulatorGap,
|
|
UndulatorHarmonic,
|
|
UndulatorMode,
|
|
UndulatorModeEnum,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mode():
|
|
with patched_device(UndulatorMode, name="id_mode") as device:
|
|
device.select_control._read_pv.mock_data = 1
|
|
device.cmu_online._read_pv.mock_data = 4
|
|
device.done._read_pv.mock_data = 1
|
|
yield device
|
|
device.destroy()
|
|
|
|
|
|
def finish_move(device, status):
|
|
device.done._read_pv.mock_data = 0
|
|
assert not status.done
|
|
device.done._read_pv.mock_data = 1
|
|
status.wait(timeout=1)
|
|
assert status.success
|
|
|
|
|
|
def test_mode_uses_undulator_signals_and_standard_set(mode):
|
|
assert isinstance(mode, UndulatorBase)
|
|
assert mode.setpoint.pvname == "X07MA-UIND:MODE"
|
|
assert mode.value_string.pvname == "X07MA-UIND:MODE"
|
|
assert isinstance(mode.value_string, EpicsSignalRO)
|
|
assert mode.done.pvname == "X07MA-UIND:DONE"
|
|
assert mode.stop_signal.pvname == "X07MA-UIND:STOP"
|
|
assert mode.select_control.pvname == "X07MA-UIND:SCTRL"
|
|
assert mode.hints == {"fields": ["id_mode"]}
|
|
assert not hasattr(mode, "set_mode")
|
|
|
|
status = mode.set("Linear H")
|
|
|
|
assert mode.setpoint.get() == 1
|
|
assert mode.read()["id_mode"]["value"] == 1
|
|
assert mode.position == 1
|
|
assert not status.done
|
|
finish_move(mode, status)
|
|
|
|
|
|
@pytest.mark.parametrize("selected", list(UndulatorModeEnum))
|
|
@pytest.mark.parametrize("input_type", ["label", "enum", "index"])
|
|
def test_set_accepts_every_mode(mode, selected, input_type):
|
|
index = list(UndulatorModeEnum).index(selected)
|
|
target = {"label": selected.value, "enum": selected, "index": index}[input_type]
|
|
|
|
mode.setpoint._read_pv.mock_data = (index + 1) % len(UndulatorModeEnum)
|
|
status = mode.set(target)
|
|
|
|
assert mode.setpoint.get() == index
|
|
assert not status.done
|
|
finish_move(mode, status)
|
|
|
|
|
|
def test_same_mode_while_busy_still_writes_and_waits_for_done(mode):
|
|
mode.done._read_pv.mock_data = 0
|
|
with patch.object(mode.setpoint, "put", wraps=mode.setpoint.put) as put:
|
|
status = mode.set("Off")
|
|
|
|
put.assert_called_once_with(0, wait=True)
|
|
assert not status.done
|
|
finish_move(mode, status)
|
|
|
|
|
|
@pytest.mark.parametrize("target", ["Circular+", UndulatorModeEnum.CIRCULAR_PLUS, 4])
|
|
@pytest.mark.parametrize("method", ["set", "move"])
|
|
def test_idle_same_mode_completes_without_write(mode, target, method):
|
|
mode.setpoint._read_pv.mock_data = 4
|
|
with patch.object(mode.setpoint, "put") as put:
|
|
status = getattr(mode, method)(target, timeout=1)
|
|
status.wait(timeout=1)
|
|
|
|
put.assert_not_called()
|
|
assert status.done
|
|
assert status.success
|
|
assert status.target == 4
|
|
|
|
|
|
def test_move_normalizes_labels_and_preserves_callback_and_timeout(mode):
|
|
moved_cb = Mock()
|
|
status = mode.move("Circular+", wait=False, timeout=1, moved_cb=moved_cb)
|
|
|
|
assert mode.setpoint.get() == 4
|
|
assert status.timeout == 1
|
|
moved_cb.assert_not_called()
|
|
finish_move(mode, status)
|
|
moved_cb.assert_called_once_with(status, obj=mode)
|
|
|
|
|
|
@pytest.mark.parametrize("method", ["set", "move"])
|
|
@pytest.mark.parametrize(
|
|
["target", "error_type", "message"],
|
|
[
|
|
(-1, ValueError, "Invalid mode index"),
|
|
(-8, ValueError, "Invalid mode index"),
|
|
(8, ValueError, "Invalid mode index"),
|
|
("unknown", ValueError, "Invalid mode string"),
|
|
(1.0, TypeError, "Mode must be"),
|
|
(None, TypeError, "Mode must be"),
|
|
(True, TypeError, "Mode must be"),
|
|
(False, TypeError, "Mode must be"),
|
|
],
|
|
)
|
|
def test_invalid_modes_do_not_write(mode, method, target, error_type, message):
|
|
with patch.object(mode.setpoint, "put") as put:
|
|
with pytest.raises(error_type, match=message):
|
|
getattr(mode, method)(target, wait=False)
|
|
put.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("target", ["Off", "Circular+", 4])
|
|
@pytest.mark.parametrize("method", ["set", "move"])
|
|
def test_operator_control_blocks_moves_before_status_creation(mode, method, target):
|
|
mode.select_control._read_pv.mock_data = 0
|
|
|
|
with patch.object(PVPositioner, "move") as parent_move:
|
|
with pytest.raises(PermissionError, match="operator controlled"):
|
|
getattr(mode, method)(target, wait=False)
|
|
|
|
parent_move.assert_not_called()
|
|
assert mode.setpoint.get() == 0
|
|
|
|
|
|
@pytest.mark.parametrize("force", [False, True])
|
|
def test_direct_setpoint_writes_obey_control_validation(mode, force):
|
|
mode.select_control._read_pv.mock_data = 0
|
|
|
|
with pytest.raises(PermissionError, match="operator controlled"):
|
|
mode.setpoint.put(4, force=force)
|
|
|
|
assert mode.setpoint.get() == 0
|
|
|
|
|
|
def test_stop_writes_stop_and_waits_for_done(mode):
|
|
status = mode.set("Circular-")
|
|
mode.done._read_pv.mock_data = 0
|
|
|
|
mode.stop()
|
|
|
|
assert mode.stop_signal.get() == 1
|
|
assert not status.done
|
|
finish_move(mode, status)
|
|
|
|
|
|
def test_operator_control_skips_stop_write(mode):
|
|
mode.select_control._read_pv.mock_data = 0
|
|
|
|
mode.stop()
|
|
mode.stop()
|
|
|
|
assert mode.stop_signal.get() == 0
|
|
|
|
|
|
def test_move_times_out_without_done_transition(mode):
|
|
status = mode.set("Linear H", timeout=0.01)
|
|
|
|
with pytest.raises(StatusTimeoutError):
|
|
status.wait(timeout=1)
|
|
|
|
assert status.done
|
|
assert not status.success
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
["device_name", "device_class", "suffix"],
|
|
[
|
|
("id_mode", UndulatorMode, "MODE"),
|
|
("id_harmonic", UndulatorHarmonic, "HARMONIC"),
|
|
("id_gap", UndulatorGap, "GAP-SP"),
|
|
],
|
|
)
|
|
def test_frontend_undulator_configuration(device_name, device_class, suffix):
|
|
config_path = Path(__file__).parents[2] / "xtreme_bec/device_configs/frontend.yaml"
|
|
device_config = yaml.safe_load(config_path.read_text())[device_name]
|
|
assert device_config["deviceClass"] == f"xtreme_bec.devices.undulator.{device_class.__name__}"
|
|
config = device_config["deviceConfig"]
|
|
|
|
assert config == {"prefix": "X07MA-UIND:"}
|
|
with patched_device(device_class, name=device_name, **config) as device:
|
|
assert device.setpoint.pvname == f"X07MA-UIND:{suffix}"
|
|
assert device.done.pvname == "X07MA-UIND:DONE"
|
|
assert device.cmu_online.pvname == "X07MA-OP-CMU:ONLINE"
|
|
device.destroy()
|
|
|
|
|
|
@pytest.fixture(params=[UndulatorGap, UndulatorHarmonic, UndulatorAlpha, UndulatorMode])
|
|
def undulator(request):
|
|
with patched_device(request.param, name="undulator", prefix="TEST:UIND:") as device:
|
|
device.select_control._read_pv.mock_data = 1
|
|
device.cmu_online._read_pv.mock_data = 4
|
|
device.done._read_pv.mock_data = 1
|
|
yield device
|
|
device.destroy()
|
|
|
|
|
|
def test_cmu_state_uses_absolute_read_only_pv(undulator):
|
|
assert undulator.cmu_online.pvname == "X07MA-OP-CMU:ONLINE"
|
|
assert isinstance(undulator.cmu_online, EpicsSignalRO)
|
|
assert "undulator_cmu_online" not in undulator.read()
|
|
|
|
|
|
@pytest.mark.parametrize("state", [0, 1, 2, 3, 5, 6])
|
|
@pytest.mark.parametrize("target", [0, 1])
|
|
@pytest.mark.parametrize("method", ["move", "set"])
|
|
def test_cmu_must_be_online_before_moves(undulator, state, target, method):
|
|
undulator.cmu_online._read_pv.mock_data = state
|
|
|
|
with patch.object(PVPositioner, "move") as parent_move:
|
|
with pytest.raises(PermissionError, match="CMU must be On-line"):
|
|
getattr(undulator, method)(target, wait=False)
|
|
|
|
parent_move.assert_not_called()
|
|
assert undulator.setpoint.get() == 0
|
|
|
|
|
|
@pytest.mark.parametrize("state", [0, 1, 2, 3, 5, 6])
|
|
@pytest.mark.parametrize("force", [False, True])
|
|
def test_cmu_validation_applies_to_direct_setpoint_writes(undulator, state, force):
|
|
undulator.cmu_online._read_pv.mock_data = state
|
|
|
|
with pytest.raises(PermissionError, match="CMU must be On-line"):
|
|
undulator.setpoint.put(1, force=force)
|
|
|
|
assert undulator.setpoint.get() == 0
|
|
|
|
|
|
@pytest.mark.parametrize("state", [0, 1, 2, 3, 5, 6])
|
|
def test_cmu_validation_skips_stop_writes(undulator, state):
|
|
undulator.cmu_online._read_pv.mock_data = state
|
|
|
|
assert undulator.stop_signal.put(1) is None
|
|
undulator.stop()
|
|
undulator.stop()
|
|
|
|
assert undulator.stop_signal.get() == 0
|
|
|
|
|
|
def test_online_cmu_allows_moves_and_done_completion(undulator):
|
|
status = undulator.set(1)
|
|
|
|
assert undulator.setpoint.get() == 1
|
|
assert not status.done
|
|
if undulator.readback is not None:
|
|
undulator.readback._read_pv.mock_data = 1
|
|
finish_move(undulator, status)
|
|
assert undulator.position == 1
|
|
|
|
|
|
def test_online_cmu_preserves_pending_stop_behavior(undulator):
|
|
status = undulator.set(1)
|
|
undulator.done._read_pv.mock_data = 0
|
|
|
|
undulator.stop()
|
|
|
|
assert undulator.stop_signal.get() == 1
|
|
assert not status.done
|
|
finish_move(undulator, status)
|
|
|
|
|
|
def test_online_cmu_still_requires_beamline_control(undulator):
|
|
undulator.select_control._read_pv.mock_data = 0
|
|
|
|
with patch.object(PVPositioner, "move") as parent_move:
|
|
with pytest.raises(PermissionError, match="operator controlled"):
|
|
undulator.set(1)
|
|
parent_move.assert_not_called()
|
|
with pytest.raises(PermissionError, match="operator controlled"):
|
|
undulator.setpoint.put(1)
|
|
undulator.stop()
|
|
assert undulator.setpoint.get() == 0
|
|
assert undulator.stop_signal.get() == 0
|