76 lines
3.1 KiB
Python
76 lines
3.1 KiB
Python
# pylint: skip-file
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
from bec_server.scan_server.tests.scan_fixtures import *
|
|
from bec_server.scan_server.tests.scan_hook_tests import *
|
|
|
|
NIDAQ_CONTINUOUS_SCAN_DEFAULT_HOOK_TESTS = [
|
|
("prepare_scan", [assert_prepare_scan_reads_baseline_devices]),
|
|
("open_scan", [assert_scan_open_called]),
|
|
("stage", [assert_stage_all_devices_called]),
|
|
("pre_scan", [assert_pre_scan_called]),
|
|
("unstage", [assert_unstage_all_devices_called]),
|
|
("close_scan", [assert_close_scan_waits_for_baseline_and_closes]),
|
|
]
|
|
|
|
|
|
def _assemble_nidaq_continuous_scan(v4_scan_assembler, **overrides):
|
|
params = {"scan_duration": 10.0, "daq": "nidaq", "compression": False}
|
|
params.update(overrides)
|
|
return v4_scan_assembler("nidaq_continuous_scan", **params)
|
|
|
|
|
|
@pytest.mark.parametrize(("hook_name", "hook_tests"), NIDAQ_CONTINUOUS_SCAN_DEFAULT_HOOK_TESTS)
|
|
def test_nidaq_continuous_scan_v4_default_hooks(
|
|
v4_scan_assembler, nth_done_status_mock, hook_name, hook_tests
|
|
):
|
|
scan = _assemble_nidaq_continuous_scan(v4_scan_assembler)
|
|
|
|
run_scan_tests(scan, [(hook_name, hook_tests)], nth_done_status_mock=nth_done_status_mock)
|
|
|
|
|
|
def test_nidaq_continuous_scan_v4_prepare_scan_updates_metadata(v4_scan_assembler):
|
|
scan = _assemble_nidaq_continuous_scan(v4_scan_assembler)
|
|
scan.actions.add_scan_report_instruction_device_progress = mock.MagicMock()
|
|
baseline_status = mock.MagicMock()
|
|
scan.actions.read_baseline_devices = mock.MagicMock(return_value=baseline_status)
|
|
|
|
scan.prepare_scan()
|
|
scan.actions._build_scan_status_message("open")
|
|
|
|
assert scan.scan_info.additional_scan_parameters["scan_duration"] == 10.0
|
|
assert scan.scan_info.additional_scan_parameters["compression"] is False
|
|
assert scan.scan_info.readout_priority_modification["async"] == ["nidaq"]
|
|
scan.actions.add_scan_report_instruction_device_progress.assert_called_once_with(scan.daq)
|
|
scan.actions.read_baseline_devices.assert_called_once_with(wait=False)
|
|
assert scan._baseline_readout_status is baseline_status
|
|
|
|
|
|
def test_nidaq_continuous_scan_v4_scan_core_reads_until_complete(
|
|
v4_scan_assembler, nth_done_status_mock
|
|
):
|
|
scan = _assemble_nidaq_continuous_scan(v4_scan_assembler)
|
|
kickoff_status = mock.MagicMock()
|
|
completion_status = nth_done_status_mock(resolve_after=3)
|
|
scan.actions.kickoff = mock.MagicMock(return_value=kickoff_status)
|
|
scan.actions.complete = mock.MagicMock(return_value=completion_status)
|
|
scan.actions.read_monitored_devices = mock.MagicMock()
|
|
|
|
with mock.patch("debye_bec.scans.nidaq_continuous_scan.time.sleep"):
|
|
scan.scan_core()
|
|
|
|
scan.actions.kickoff.assert_called_once_with(device=scan.daq, wait=False)
|
|
kickoff_status.wait.assert_called_once_with(timeout=5)
|
|
scan.actions.complete.assert_called_once_with(device=scan.daq, wait=False)
|
|
assert scan.actions.read_monitored_devices.call_count == 2
|
|
|
|
|
|
def test_nidaq_continuous_scan_v4_post_scan_completes_all_devices(v4_scan_assembler):
|
|
scan = _assemble_nidaq_continuous_scan(v4_scan_assembler)
|
|
scan.actions.complete_all_devices = mock.MagicMock()
|
|
|
|
scan.post_scan()
|
|
|
|
scan.actions.complete_all_devices.assert_called_once_with()
|