tests: add otf scan tests

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(),
+    ]

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
     ):