133 lines
5.2 KiB
Python
133 lines
5.2 KiB
Python
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
from bec_server.scan_server.tests.fixtures import * # noqa: F401, F403
|
|
|
|
from xtreme_bec.scans.hyst_scan import HystScan
|
|
|
|
# pylint: disable=missing-function-docstring
|
|
# pylint: disable=protected-access
|
|
|
|
|
|
@pytest.fixture
|
|
def hyst_scan(scan_assembler, device_manager_mock):
|
|
"""HystScan instance with samx as field motor (flyer) and samy as mono."""
|
|
return scan_assembler(
|
|
HystScan,
|
|
device_manager_mock.devices["samx"], # field_motor / flyer
|
|
0.0, # start_field
|
|
0.5, # end_field
|
|
device_manager_mock.devices["samy"], # mono / energy_motor
|
|
600.0, # energy1
|
|
640.0, # energy2
|
|
ramp_rate=3.0, # distinct from default_ramp_rate=2 for clearer assertions
|
|
parameter={"args": {}, "kwargs": {}},
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"in_args, num_steps, reference_positions",
|
|
[
|
|
([600], 5, [600, 600, 600, 600, 600]),
|
|
([600, 640], 6, [600, 640, 600, 640, 600, 640]),
|
|
([600, 620, 640], 9, [600, 620, 640, 620, 600, 620, 640, 620, 600]),
|
|
([1, 2, 3, 4], 10, [1, 2, 3, 4, 3, 2, 1, 2, 3, 4]),
|
|
],
|
|
)
|
|
def test_get_next_scan_motor_position(in_args, num_steps, reference_positions, hyst_scan):
|
|
hyst_scan.energy_positions = in_args
|
|
gen = hyst_scan._get_next_scan_motor_position()
|
|
assert [next(gen) for _ in range(num_steps)] == reference_positions
|
|
|
|
|
|
def test_hyst_scan_init(scan_assembler, device_manager_mock):
|
|
field_motor = device_manager_mock.devices["samx"]
|
|
mono = device_manager_mock.devices["samy"]
|
|
scan = scan_assembler(
|
|
HystScan,
|
|
field_motor, 0.0, 0.5,
|
|
mono, 600.0, 640.0,
|
|
parameter={"args": {}, "kwargs": {}},
|
|
)
|
|
assert scan.flyer is field_motor
|
|
assert scan.energy_motor is mono
|
|
assert scan.flyer_positions == [0.0, 0.5]
|
|
assert scan.energy_positions == [600.0, 640.0]
|
|
assert scan.ramp_rate == HystScan.default_ramp_rate
|
|
assert scan.scan_motors == [mono, field_motor]
|
|
|
|
|
|
def test_hyst_scan_prepare_positions(hyst_scan):
|
|
with mock.patch.object(hyst_scan, "_check_limits"):
|
|
list(hyst_scan.prepare_positions())
|
|
assert hyst_scan.positions == [[600.0], [640.0]]
|
|
assert hyst_scan.num_pos == 0
|
|
|
|
|
|
def test_hyst_scan_scan_core(hyst_scan, device_manager_mock, ScanStubStatusMock):
|
|
"""scan_core ramps the field while stepping the energy motor once per loop iteration."""
|
|
field_motor = device_manager_mock.devices["samx"]
|
|
mono = device_manager_mock.devices["samy"]
|
|
|
|
def fake_done():
|
|
yield False # first while-check: enter loop
|
|
yield True # second while-check: exit loop
|
|
|
|
def fake_set(*args, **kwargs):
|
|
yield "fake_set"
|
|
return ScanStubStatusMock(done_func=fake_done)
|
|
|
|
def fake_rpc(*args, **kwargs):
|
|
yield "fake_rpc"
|
|
|
|
def fake_read(*args, **kwargs):
|
|
yield "fake_read"
|
|
|
|
# Replace connector with a MagicMock so send_client_info is freely callable
|
|
hyst_scan.connector = mock.MagicMock()
|
|
|
|
devices_cls = type(device_manager_mock.devices)
|
|
with mock.patch.object(hyst_scan.stubs, "set", side_effect=fake_set) as mock_set:
|
|
with mock.patch.object(
|
|
hyst_scan.stubs, "send_rpc_and_wait", side_effect=fake_rpc
|
|
) as mock_rpc:
|
|
with mock.patch.object(hyst_scan.stubs, "read", side_effect=fake_read) as mock_read:
|
|
with mock.patch.object(devices_cls, "monitored_devices", return_value=[]):
|
|
output = list(hyst_scan.scan_core())
|
|
|
|
assert output == [
|
|
"fake_rpc", # send_rpc_and_wait(flyer, "ramprate.set", default_ramp_rate)
|
|
"fake_set", # set(flyer, start_field=0.0)
|
|
"fake_rpc", # send_rpc_and_wait(flyer, "ramprate.set", ramp_rate=3.0)
|
|
"fake_set", # set(flyer, end_field=0.5) ← status captured here
|
|
"fake_set", # set(energy_motor, energy1=600.0) — inside loop
|
|
"fake_read", # read(flyer, energy_motor) — inside loop
|
|
"fake_rpc", # send_rpc_and_wait(flyer, "ramprate.set", default_ramp_rate) — after loop
|
|
]
|
|
assert hyst_scan.num_pos == 1
|
|
assert hyst_scan.point_id == 1
|
|
|
|
# Ramp-rate sequence: brake to default → set scan rate → restore default
|
|
assert mock_rpc.call_args_list[0] == mock.call(
|
|
field_motor, "ramprate.set", HystScan.default_ramp_rate
|
|
)
|
|
assert mock_rpc.call_args_list[1] == mock.call(field_motor, "ramprate.set", 3.0)
|
|
assert mock_rpc.call_args_list[2] == mock.call(
|
|
field_motor, "ramprate.set", HystScan.default_ramp_rate
|
|
)
|
|
|
|
# Field motor: move to start position, then begin flying to end (non-blocking)
|
|
assert mock_set.call_args_list[0] == mock.call(device=field_motor, value=0.0)
|
|
assert mock_set.call_args_list[1] == mock.call(device=field_motor, value=0.5, wait=False)
|
|
# Energy motor steps to first energy position inside the loop
|
|
assert mock_set.call_args_list[2] == mock.call(device=mono, value=600.0)
|
|
|
|
# Read issued once (one loop iteration), with flyer and energy motor
|
|
mock_read.assert_called_once_with(device=[field_motor, mono], point_id=0)
|
|
|
|
# Progress message sent for the energy step
|
|
hyst_scan.connector.send_client_info.assert_called_once_with(
|
|
"Moving mono to 600.0.", source="scan_server", expire=10
|
|
)
|