diff --git a/tests/tests_devices/test_eiger.py b/tests/tests_devices/test_eiger.py index 1f2d954..1b81a13 100644 --- a/tests/tests_devices/test_eiger.py +++ b/tests/tests_devices/test_eiger.py @@ -20,7 +20,7 @@ from jfjoch_client.models.measurement_statistics import MeasurementStatistics from ophyd import Staged from ophyd_devices.utils.psi_device_base_utils import DeviceStatus -from csaxs_bec.devices.jungfraujoch.eiger import Eiger +from csaxs_bec.devices.jungfraujoch.eiger import EigerError from csaxs_bec.devices.jungfraujoch.eiger_1_5m import Eiger1_5M from csaxs_bec.devices.jungfraujoch.eiger_9m import Eiger9M @@ -284,7 +284,9 @@ def test_eiger_on_complete(eiger_1_5m, raise_timeout): with ( mock.patch.object( - eiger.jfj_client.api, "status_get", return_value=BrokerStatus(state="Idle") + eiger.jfj_client.api, + "status_get", + return_value=BrokerStatus(state="Idle", message_severity="info"), ), mock.patch.object(eiger.jfj_client, "wait_for_idle", side_effect=mock_wait_for_idle), mock.patch.object( @@ -312,6 +314,43 @@ def test_eiger_on_complete(eiger_1_5m, raise_timeout): assert status.success == False if raise_timeout else True +def test_eiger_on_complete_error_message(eiger_1_5m): + """Test that on_complete raises if the message severity is error.""" + # Test that on_complete raises if message_severity is error + eiger = eiger_1_5m + unblock_wait_for_idle = threading.Event() + raise_timeout = False + + def mock_wait_for_idle(timeout: float, raise_on_timeout: bool) -> bool: + if unblock_wait_for_idle.wait(timeout): + if raise_timeout: + return False + return True + return False + + with ( + mock.patch.object( + eiger.jfj_client.api, + "status_get", + return_value=BrokerStatus(state="Idle", message_severity="error", message="Test error"), + ), + mock.patch.object(eiger.jfj_client, "wait_for_idle", side_effect=mock_wait_for_idle), + mock.patch.object( + eiger.jfj_client.api, + "statistics_data_collection_get", + return_value=MeasurementStatistics( + run_number=1, + images_collected=eiger.scan_info.msg.num_points + * eiger.scan_info.msg.scan_parameters["frames_per_trigger"], + ), + ), + ): + with pytest.raises(EigerError): + status = eiger.complete() + unblock_wait_for_idle.set() + status.wait(timeout=2) + + def test_eiger_file_event_callback(eiger_1_5m, tmp_path): """Test the file_event callback of the Eiger detector. This is equivalent for 9M and 1_5M.""" eiger = eiger_1_5m