diff --git a/tests/tests_devices/test_phoenix_trigger.py b/tests/tests_devices/test_phoenix_trigger.py new file mode 100644 index 0000000..891ea0c --- /dev/null +++ b/tests/tests_devices/test_phoenix_trigger.py @@ -0,0 +1,97 @@ +import threading +import time +from unittest import mock + +import numpy as np +import ophyd +import pytest +from bec_server.device_server.tests.utils import DMMock +from ophyd_devices.interfaces.base_classes.psi_detector_base import DeviceTimeoutError +from ophyd_devices.tests.utils import MockPV, patch_dual_pvs + +from phoenix_bec.devices.phoenix_trigger import SAMPLING, PhoenixTrigger + + +@pytest.fixture(scope="function") +def mock_trigger(): + name = "phoenix_trigger" + prefix = "X07MB-OP2:" + dm = DMMock() + with mock.patch.object(dm, "connector"): + with ( + mock.patch( + "ophyd_devices.interfaces.base_classes.psi_detector_base.FileWriter" + ) as filemixin, + mock.patch( + "ophyd_devices.interfaces.base_classes.psi_detector_base.PSIDetectorBase._update_service_config" + ) as mock_service_config, + ): + with mock.patch.object(ophyd, "cl") as mock_cl: + mock_cl.get_pv = MockPV + mock_cl.thread_class = threading.Thread + with mock.patch.object(PhoenixTrigger, "_init"): + det = PhoenixTrigger(name=name, prefix=prefix, device_manager=dm) + patch_dual_pvs(det) + det.TIMEOUT_FOR_SIGNALS = 0.1 + yield det + + +def test_phoenix_trigger_init(mock_trigger): + """Test PhoenixTrigger init""" + assert mock_trigger.name == "phoenix_trigger" + assert mock_trigger.prefix == "X07MB-OP2:" + + +def test_phoenix_trigger_stage(mock_trigger): + """Test PhoenixTrigger on_stage""" + with mock.patch.object(mock_trigger.scaninfo, "load_scan_metadata") as mock_load_scan_metadata: + mock_trigger.scaninfo.scan_type = "step" + mock_trigger.scaninfo.exp_time = exp_time = 1 + mock_trigger.stage() + assert mock_load_scan_metadata.call_count == 1 + assert mock_trigger.start_csmpl.get() == 0 + assert mock_trigger.total_cycles.get() == np.ceil(exp_time * 5) + assert mock_trigger.smpl.get() == 1 + + +def test_phoenix_trigger_unstage(mock_trigger): + """Test PhoenixTrigger on_unstage""" + with mock.patch.object(mock_trigger.custom_prepare, "on_stop") as mock_on_stop: + mock_trigger.unstage() + assert mock_on_stop.call_count == 1 + + +def test_phoenix_trigger_stop(mock_trigger): + """Test PhoenixTrigger on_stop""" + with mock.patch.object(mock_trigger.smpl, "put") as mock_smpl_put: + mock_trigger.smpl_done._read_pv.mock_data = SAMPLING.RUNNING + mock_trigger.stop() + assert mock_trigger.stopped is True + assert mock_trigger.total_cycles.get() == 5 + assert mock_trigger.start_csmpl.get() == 1 + assert mock_smpl_put.call_args_list == [mock.call(1), mock.call(1)] + + +def test_phoenix_trigger_trigger(mock_trigger): + """Test PhoenixTrigger on_trigger + + First test that the trigger timeouts due to readback from smpl_done not being done. + Afterwards, check that status object resolved correctly if smpl_done is done. + """ + exp_time = 0.05 + mock_trigger.device_manager.add_device("falcon_nohdf5") + falcon_state = mock_trigger.device_manager.devices.falcon_nohdf5.state = mock.MagicMock() + falcon_state.get = mock.MagicMock(return_value=1) + mock_trigger.scaninfo.scan_type = "step" + mock_trigger.scaninfo.exp_time = exp_time + + with mock.patch.object( + mock_trigger.custom_prepare, "wait_with_status", return_value=mock.MagicMock() + ) as mock_wait_with_status: + status = mock_trigger.trigger() + assert mock_wait_with_status.call_count == 1 + assert mock_wait_with_status.call_args[1]["signal_conditions"] == [ + (mock_trigger.smpl_done.get, SAMPLING.DONE) + ] + assert mock_wait_with_status.call_args[1]["timeout"] == 5 * exp_time + assert mock_wait_with_status.call_args[1]["check_stopped"] is True