165 lines
6.5 KiB
Python
165 lines
6.5 KiB
Python
from unittest import mock
|
|
|
|
import pytest
|
|
from bec_lib.bec_errors import ScanInputValidationError
|
|
from bec_server.scan_server.scans.scan_base import ScanType
|
|
from bec_server.scan_server.tests.scan_hook_tests import DEFAULT_HOOK_TESTS, run_scan_tests
|
|
|
|
from xtreme_bec.scans.otf_scan import OTFScan
|
|
|
|
|
|
@pytest.mark.parametrize(("hook_name", "hook_tests"), DEFAULT_HOOK_TESTS)
|
|
def test_default_hooks(otf_scan, nth_done_status_mock, hook_name, hook_tests):
|
|
run_scan_tests(otf_scan, [(hook_name, hook_tests)], nth_done_status_mock)
|
|
|
|
|
|
def test_assembly(otf_scan):
|
|
assert isinstance(otf_scan, OTFScan)
|
|
assert otf_scan.scan_info.scan_type == ScanType.HARDWARE_TRIGGERED
|
|
assert (otf_scan.e1, otf_scan.e2, otf_scan.time) == (500.0, 800.0, 2.0)
|
|
assert otf_scan.mono is otf_scan.dev["mono"]
|
|
assert otf_scan._baseline_readout_status is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"parameters",
|
|
[{"e1": 249.0}, {"e1": 2001.0}, {"e2": 249.0}, {"e2": 2001.0}, {"time": -1.0}, {"time": 0.0}],
|
|
)
|
|
def test_assembly_rejects_invalid_scan_parameters(otf_scan_assembler, parameters):
|
|
with pytest.raises(ScanInputValidationError):
|
|
otf_scan_assembler(**parameters)
|
|
|
|
|
|
@pytest.mark.parametrize(("e1", "e2"), [(250.0, 2000.0), (2000.0, 250.0)])
|
|
def test_assembly_accepts_energy_boundaries(otf_scan_assembler, e1, e2):
|
|
scan = otf_scan_assembler(e1=e1, e2=e2, time=0.1)
|
|
|
|
assert (scan.e1, scan.e2, scan.time) == (e1, e2, 0.1)
|
|
|
|
|
|
def test_prepare_routes_adc_to_async_and_reports_mono_progress(otf_scan, nth_done_status_mock):
|
|
baseline_status = nth_done_status_mock()
|
|
otf_scan.actions.read_baseline_devices = mock.MagicMock(return_value=baseline_status)
|
|
|
|
otf_scan.prepare_scan()
|
|
|
|
assert otf_scan.scan_info.readout_priority_modification["async"] == ["signals"]
|
|
assert otf_scan.scan_info.scan_report_instructions == [{"device_progress": ["mono"]}]
|
|
priority = otf_scan.scan_info.readout_priority_modification
|
|
assert [device.name for device in otf_scan.dev.monitored_devices(priority)] == ["mono"]
|
|
assert [device.name for device in otf_scan.dev.async_devices(priority)] == ["signals"]
|
|
assert otf_scan._baseline_readout_status is baseline_status
|
|
assert baseline_status.wait_calls == 0
|
|
|
|
|
|
@pytest.mark.parametrize(("e1", "e2"), [(500.0, 800.0), (800.0, 500.0)])
|
|
def test_pre_scan_positions_before_configuring_scan_speed(otf_scan_assembler, e1, e2):
|
|
scan = otf_scan_assembler(e1=e1, e2=e2, time=1.5)
|
|
calls = mock.MagicMock()
|
|
calls.attach_mock(scan.dev.mirror_theta.base_velocity.put, "theta_base")
|
|
calls.attach_mock(scan.dev.grating_beta.base_velocity.put, "beta_base")
|
|
calls.attach_mock(scan.dev.mirror_theta.velocity.put, "theta_speed")
|
|
calls.attach_mock(scan.dev.grating_beta.velocity.put, "beta_speed")
|
|
calls.attach_mock(scan.mono.set, "move")
|
|
calls.attach_mock(scan.mono.get_max_speeds, "max_speeds")
|
|
calls.attach_mock(scan.mono.set_progress_range, "progress")
|
|
calls.attach_mock(scan.mono.sync_with_undulator.put, "sync")
|
|
scan.actions.pre_scan_all_devices = mock.MagicMock()
|
|
calls.attach_mock(scan.actions.pre_scan_all_devices, "pre_scan")
|
|
|
|
scan.pre_scan()
|
|
|
|
assert calls.mock_calls == [
|
|
mock.call.theta_base(0.008),
|
|
mock.call.beta_base(0.008),
|
|
mock.call.theta_speed(0.08),
|
|
mock.call.beta_speed(0.08),
|
|
mock.call.move(e1),
|
|
mock.call.move().wait(),
|
|
mock.call.max_speeds(e1=e1, e2=e2, travel_time=90.0),
|
|
mock.call.progress(e1=e1, e2=e2),
|
|
mock.call.theta_base(0.002),
|
|
mock.call.beta_base(0.003),
|
|
mock.call.theta_speed(0.02),
|
|
mock.call.beta_speed(0.03),
|
|
mock.call.sync(1),
|
|
mock.call.pre_scan(),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("resolve_after", [1, 2, 4])
|
|
def test_scan_core_stores_adc_and_reads_while_moving(otf_scan, nth_done_status_mock, resolve_after):
|
|
status = nth_done_status_mock(resolve_after=resolve_after)
|
|
calls = mock.MagicMock()
|
|
calls.attach_mock(otf_scan.dev.signals.store_updates.put, "store")
|
|
otf_scan.actions.set = mock.MagicMock(return_value=status)
|
|
otf_scan.actions.read_monitored_devices = mock.MagicMock()
|
|
calls.attach_mock(otf_scan.actions.set, "move")
|
|
calls.attach_mock(otf_scan.actions.read_monitored_devices, "read")
|
|
|
|
with mock.patch("xtreme_bec.scans.otf_scan.time.sleep") as sleep:
|
|
calls.attach_mock(sleep, "sleep")
|
|
otf_scan.scan_core()
|
|
|
|
assert calls.mock_calls == [
|
|
mock.call.store(1),
|
|
mock.call.move(otf_scan.mono, 800.0, wait=False),
|
|
*[
|
|
call
|
|
for _ in range(resolve_after - 1)
|
|
for call in (mock.call.read(), mock.call.sleep(1))
|
|
],
|
|
]
|
|
|
|
|
|
def test_at_each_point_reads_monitored_devices(otf_scan):
|
|
otf_scan.actions.read_monitored_devices = mock.MagicMock()
|
|
|
|
otf_scan.at_each_point()
|
|
|
|
otf_scan.actions.read_monitored_devices.assert_called_once_with()
|
|
|
|
|
|
@pytest.mark.parametrize("hook", ["post_scan", "on_exception"])
|
|
def test_cleanup_disables_acquisition_and_restores_velocity(otf_scan, hook):
|
|
calls = mock.MagicMock()
|
|
calls.attach_mock(otf_scan.dev.signals.store_updates.put, "store")
|
|
otf_scan.actions.complete_all_devices = mock.MagicMock()
|
|
calls.attach_mock(otf_scan.actions.complete_all_devices, "complete")
|
|
calls.attach_mock(otf_scan.mono.sync_with_undulator.put, "sync")
|
|
calls.attach_mock(otf_scan.dev.mirror_theta.base_velocity.put, "theta_base")
|
|
calls.attach_mock(otf_scan.dev.grating_beta.base_velocity.put, "beta_base")
|
|
calls.attach_mock(otf_scan.dev.mirror_theta.velocity.put, "theta_speed")
|
|
calls.attach_mock(otf_scan.dev.grating_beta.velocity.put, "beta_speed")
|
|
|
|
if hook == "on_exception":
|
|
otf_scan.on_exception(RuntimeError("scan interrupted"))
|
|
else:
|
|
otf_scan.post_scan()
|
|
|
|
assert calls.mock_calls == [
|
|
mock.call.store(0),
|
|
*([mock.call.complete()] if hook == "post_scan" else []),
|
|
mock.call.sync(0),
|
|
mock.call.theta_base(0.008),
|
|
mock.call.beta_base(0.008),
|
|
mock.call.theta_speed(0.08),
|
|
mock.call.beta_speed(0.08),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("has_baseline", [False, True])
|
|
def test_close_waits_for_baseline_before_closing_and_checking_statuses(otf_scan, has_baseline):
|
|
calls = mock.MagicMock()
|
|
otf_scan.actions.close_scan = calls.close
|
|
otf_scan.actions.check_for_unchecked_statuses = calls.check
|
|
otf_scan._baseline_readout_status = calls.baseline if has_baseline else None
|
|
|
|
otf_scan.close_scan()
|
|
|
|
assert calls.mock_calls == [
|
|
*([mock.call.baseline.wait()] if has_baseline else []),
|
|
mock.call.close(),
|
|
mock.call.check(),
|
|
]
|