From a4e1ef87ee86d67b825345bed6ba34ca355f6ed6 Mon Sep 17 00:00:00 2001 From: wakonig_k Date: Tue, 6 Oct 2026 13:49:38 +0200 Subject: [PATCH] tests: add otf scan tests --- tests/tests_scans/conftest.py | 59 +++++++++++ tests/tests_scans/test_otf_scan.py | 164 +++++++++++++++++++++++++++++ xtreme_bec/scans/otf_scan.py | 2 +- 3 files changed, 224 insertions(+), 1 deletion(-) create mode 100644 tests/tests_scans/conftest.py create mode 100644 tests/tests_scans/test_otf_scan.py diff --git a/tests/tests_scans/conftest.py b/tests/tests_scans/conftest.py new file mode 100644 index 0000000..5d82562 --- /dev/null +++ b/tests/tests_scans/conftest.py @@ -0,0 +1,59 @@ +from unittest import mock + +import pytest +from bec_server.scan_server.tests.scan_fixtures import ( + MockCustomDevice, + device_manager, + nth_done_status_mock, + readout_priority, + v4_scan_assembler, +) + + +@pytest.fixture +def otf_scan_assembler(device_manager, readout_priority, v4_scan_assembler): + """Use BEC's assembler with the beamline-specific signals and RPC responses.""" + device_signals = { + "mono": ["sync_with_undulator"], + "mirror_theta": ["base_velocity", "velocity"], + "grating_beta": ["base_velocity", "velocity"], + "signals": ["store_updates"], + } + for name, signals in device_signals.items(): + device = MockCustomDevice( + name, + device_info={ + "signals": { + signal: { + "obj_name": f"{name}_{signal}", + "kind_str": "config", + "describe": {"precision": 3}, + } + for signal in signals + } + }, + limits=(250, 2000) if name == "mono" else (-10, 10), + ) + for signal in signals: + getattr(device, signal).put = mock.MagicMock() + device_manager.add_device(device) + + mono = device_manager.devices["mono"] + mono.set = mock.MagicMock() + mono.get_max_speeds = mock.MagicMock(return_value=(0.02, 0.03)) + mono.set_progress_range = mock.MagicMock() + readout_priority.monitored_devices = ["mono", "signals"] + + def assemble(e1=500.0, e2=800.0, time=2.0): + scan = v4_scan_assembler("otf_scan", e1=e1, e2=e2, time=time) + # The shared fixture's device dictionary needs aliases for dev. access. + for name in device_signals: + setattr(scan.dev, name, scan.dev[name]) + return scan + + return assemble + + +@pytest.fixture +def otf_scan(otf_scan_assembler): + return otf_scan_assembler() diff --git a/tests/tests_scans/test_otf_scan.py b/tests/tests_scans/test_otf_scan.py new file mode 100644 index 0000000..4bfe9b9 --- /dev/null +++ b/tests/tests_scans/test_otf_scan.py @@ -0,0 +1,164 @@ +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(), + ] diff --git a/xtreme_bec/scans/otf_scan.py b/xtreme_bec/scans/otf_scan.py index b26dbc7..79e5970 100644 --- a/xtreme_bec/scans/otf_scan.py +++ b/xtreme_bec/scans/otf_scan.py @@ -49,7 +49,7 @@ class OTFScan(ScanBase): #fmt: off e1: Annotated[float, ScanArgument(display_name="Energy 1", description="Start energy", units=Units.eV, ge=250, le=2000)], e2: Annotated[float, ScanArgument(display_name="Energy 2", description="End energy", units=Units.eV, ge=250, le=2000)], - time: Annotated[float, ScanArgument(display_name="Time", description="Travel time", units=Units.min, ge=0)], + time: Annotated[float, ScanArgument(display_name="Time", description="Travel time", units=Units.min, gt=0)], **kwargs, #fmt: on ):