From 67b55f71caccfe4cd0aaf46ef519d034a5464105 Mon Sep 17 00:00:00 2001 From: appel_c Date: Tue, 6 Jan 2026 10:46:48 +0100 Subject: [PATCH] wp deprecate --- tests/tests_devices/test_pilatus_csaxs.py | 449 ---------------------- 1 file changed, 449 deletions(-) delete mode 100644 tests/tests_devices/test_pilatus_csaxs.py diff --git a/tests/tests_devices/test_pilatus_csaxs.py b/tests/tests_devices/test_pilatus_csaxs.py deleted file mode 100644 index 8b04372..0000000 --- a/tests/tests_devices/test_pilatus_csaxs.py +++ /dev/null @@ -1,449 +0,0 @@ -# pylint: skip-file -import os -import threading -from unittest import mock - -import ophyd -import pytest -from bec_lib import messages -from bec_lib.endpoints import MessageEndpoints -from bec_server.device_server.tests.utils import DMMock -from ophyd_devices.tests.utils import MockPV - -from csaxs_bec.devices.epics.pilatus_csaxs import PilatuscSAXS -from csaxs_bec.devices.tests_utils.utils import patch_dual_pvs - - -@pytest.fixture(scope="function") -def mock_det(): - name = "pilatus" - prefix = "X12SA-ES-PILATUS300K:" - dm = DMMock() - with mock.patch.object(dm, "connector"): - with ( - mock.patch("ophyd_devices.interfaces.base_classes.bec_device_base.FileWriter"), - mock.patch( - "ophyd_devices.interfaces.base_classes.psi_detector_base.PSIDetectorBase._update_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(PilatuscSAXS, "_init"): - det = PilatuscSAXS(name=name, prefix=prefix, device_manager=dm) - patch_dual_pvs(det) - yield det - - -@pytest.mark.parametrize("trigger_source, detector_state", [(1, 0)]) -# TODO rewrite this one, write test for init_detector, init_filewriter is tested -def test_init_detector(mock_det, trigger_source, detector_state): - """Test the _init function: - - This includes testing the functions: - - _init_detector - - _stop_det - - _set_trigger - --> Testing the filewriter is done in test_init_filewriter - - Validation upon setting the correct PVs - - """ - mock_det.custom_prepare.on_init() # call the method you want to test - assert mock_det.cam.acquire.get() == detector_state - assert mock_det.cam.trigger_mode.get() == trigger_source - - -@pytest.mark.parametrize( - "scaninfo, stopped, expected_exception", - [ - ( - { - "eacc": "e12345", - "num_points": 500, - "frames_per_trigger": 1, - "filepath": "test.h5", - "scan_id": "123", - "mokev": 12.4, - }, - False, - False, - ), - ( - { - "eacc": "e12345", - "num_points": 500, - "frames_per_trigger": 1, - "filepath": "test.h5", - "scan_id": "123", - "mokev": 12.4, - }, - True, - False, - ), - ], -) -def test_stage(mock_det, scaninfo, stopped, expected_exception): - path = "tmp" - mock_det.filepath_raw = path - with mock.patch.object( - mock_det.custom_prepare, "publish_file_location" - ) as mock_publish_file_location: - mock_det.scaninfo.num_points = scaninfo["num_points"] - mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"] - mock_det.filewriter.compile_full_filename.return_value = scaninfo["filepath"] - mock_det.device_manager.add_device("mokev", value=12.4) - mock_det.stopped = stopped - with ( - mock.patch.object(mock_det.custom_prepare, "prepare_data_backend") as mock_data_backend, - mock.patch.object( - mock_det.custom_prepare, "update_readout_time" - ) as mock_update_readout_time, - ): - mock_det.filepath.set(scaninfo["filepath"]).wait() - if expected_exception: - with pytest.raises(Exception): - mock_det.timeout = 0.1 - mock_det.stage() - else: - mock_det.stage() - mock_data_backend.assert_called_once() - mock_update_readout_time.assert_called() - # Check _prep_det - assert mock_det.cam.num_images.get() == int( - scaninfo["num_points"] * scaninfo["frames_per_trigger"] - ) - assert mock_det.cam.num_frames.get() == 1 - - mock_publish_file_location.assert_called_once_with( - done=False, successful=False, metadata={"input_path": path} - ) - - -def test_pre_scan(mock_det): - mock_det.custom_prepare.on_pre_scan() - assert mock_det.cam.acquire.get() == 1 - - -@pytest.mark.parametrize( - "readout_time, expected_value", [(1e-3, 3e-3), (3e-3, 3e-3), (5e-3, 5e-3), (None, 3e-3)] -) -def test_update_readout_time(mock_det, readout_time, expected_value): - if readout_time is None: - mock_det.custom_prepare.update_readout_time() - assert mock_det.readout_time == expected_value - else: - mock_det.scaninfo.readout_time = readout_time - mock_det.custom_prepare.update_readout_time() - assert mock_det.readout_time == expected_value - - -@pytest.mark.parametrize( - "scaninfo", - [ - ( - { - "filepath": "test.h5", - "filepath_raw": "test5_raw.h5", - "successful": True, - "done": False, - "scan_id": "123", - } - ), - ( - { - "filepath": "test.h5", - "filepath_raw": "test5_raw.h5", - "successful": False, - "done": True, - "scan_id": "123", - } - ), - ], -) -def test_publish_file_location(mock_det, scaninfo): - mock_det.scaninfo.scan_id = scaninfo["scan_id"] - mock_det.filepath.set(scaninfo["filepath"]).wait() - mock_det.filepath_raw = scaninfo["filepath_raw"] - mock_det.custom_prepare.publish_file_location( - done=scaninfo["done"], - successful=scaninfo["successful"], - metadata={"input_path": scaninfo["filepath_raw"]}, - ) - if scaninfo["successful"] is None: - msg = messages.FileMessage( - file_path=scaninfo["filepath"], - done=scaninfo["done"], - metadata={"input_path": scaninfo["filepath_raw"]}, - ) - else: - msg = messages.FileMessage( - file_path=scaninfo["filepath"], - done=scaninfo["done"], - metadata={"input_path": scaninfo["filepath_raw"]}, - successful=scaninfo["successful"], - ) - expected_calls = [ - mock.call( - MessageEndpoints.public_file(scaninfo["scan_id"], mock_det.name), - msg, - pipe=mock_det.connector.pipeline.return_value, - ), - mock.call( - MessageEndpoints.file_event(mock_det.name), - msg, - pipe=mock_det.connector.pipeline.return_value, - ), - ] - assert mock_det.connector.set_and_publish.call_args_list == expected_calls - - -@pytest.mark.parametrize( - "requests_state, expected_exception, url_delete, url_put", - [ - ( - True, - False, - "http://x12sa-pd-2:8080/stream/pilatus_2", - "http://xbl-daq-34:8091/pilatus_2/stop", - ), - ( - False, - False, - "http://x12sa-pd-2:8080/stream/pilatus_2", - "http://xbl-daq-34:8091/pilatus_2/stop", - ), - ], -) -def test_stop_detector_backend(mock_det, requests_state, expected_exception, url_delete, url_put): - with ( - mock.patch.object( - mock_det.custom_prepare, "send_requests_delete" - ) as mock_send_requests_delete, - mock.patch.object(mock_det.custom_prepare, "send_requests_put") as mock_send_requests_put, - ): - instance_delete = mock_send_requests_delete.return_value - instance_delete.ok = requests_state - instance_put = mock_send_requests_put.return_value - instance_put.ok = requests_state - if expected_exception: - mock_det.custom_prepare.stop_detector_backend() - mock_send_requests_delete.assert_called_once_with(url=url_delete) - mock_send_requests_put.assert_called_once_with(url=url_put) - instance_delete.raise_for_status.called_once() - instance_put.raise_for_status.called_once() - else: - mock_det.custom_prepare.stop_detector_backend() - mock_send_requests_delete.assert_called_once_with(url=url_delete) - mock_send_requests_put.assert_called_once_with(url=url_put) - - -@pytest.mark.parametrize( - "scaninfo, data_msgs, urls, requests_state, expected_exception", - [ - ( - { - "filepath_raw": "pilatus_2.h5", - "eacc": "e12345", - "scan_number": 1000, - "scan_directory": "S00000_00999", - "num_points": 500, - "frames_per_trigger": 1, - "headers": {"Content-Type": "application/json", "Accept": "application/json"}, - }, - [ - { - "source": [ - { - "searchPath": "/", - "searchPattern": "glob:*.cbf", - "destinationPath": ( - "/sls/X12SA/data/e12345/Data10/pilatus_2/S00000_00999" - ), - } - ] - }, - [ - "zmqWriter", - "e12345", - { - "addr": "tcp://x12sa-pd-2:8888", - "dst": ["file"], - "numFrm": 500, - "timeout": 2000, - "ifType": "PULL", - "user": "e12345", - }, - ], - ["zmqWriter", "e12345", {"frmCnt": 500, "timeout": 2000}], - ], - [ - "http://x12sa-pd-2:8080/stream/pilatus_2", - "http://xbl-daq-34:8091/pilatus_2/run", - "http://xbl-daq-34:8091/pilatus_2/wait", - ], - True, - False, - ), - ( - { - "filepath_raw": "pilatus_2.h5", - "eacc": "e12345", - "scan_number": 1000, - "scan_directory": "S00000_00999", - "num_points": 500, - "frames_per_trigger": 1, - "headers": {"Content-Type": "application/json", "Accept": "application/json"}, - }, - [ - { - "source": [ - { - "searchPath": "/", - "searchPattern": "glob:*.cbf", - "destinationPath": ( - "/sls/X12SA/data/e12345/Data10/pilatus_2/S00000_00999" - ), - } - ] - }, - [ - "zmqWriter", - "e12345", - { - "addr": "tcp://x12sa-pd-2:8888", - "dst": ["file"], - "numFrm": 500, - "timeout": 2000, - "ifType": "PULL", - "user": "e12345", - }, - ], - ["zmqWriter", "e12345", {"frmCnt": 500, "timeout": 2000}], - ], - [ - "http://x12sa-pd-2:8080/stream/pilatus_2", - "http://xbl-daq-34:8091/pilatus_2/run", - "http://xbl-daq-34:8091/pilatus_2/wait", - ], - False, # return of res.ok is False! - True, - ), - ], -) -def test_prep_file_writer(mock_det, scaninfo, data_msgs, urls, requests_state, expected_exception): - with ( - mock.patch.object(mock_det.custom_prepare, "close_file_writer") as mock_close_file_writer, - mock.patch.object(mock_det.custom_prepare, "stop_file_writer") as mock_stop_file_writer, - mock.patch.object(mock_det, "filewriter") as mock_filewriter, - mock.patch.object(mock_det.custom_prepare, "create_directory") as mock_create_directory, - mock.patch.object(mock_det.custom_prepare, "send_requests_put") as mock_send_requests_put, - ): - mock_det.scaninfo.scan_number = scaninfo["scan_number"] - mock_det.scaninfo.num_points = scaninfo["num_points"] - mock_det.scaninfo.frames_per_trigger = scaninfo["frames_per_trigger"] - mock_det.scaninfo.username = scaninfo["eacc"] - mock_filewriter.compile_full_filename.return_value = scaninfo["filepath_raw"] - mock_filewriter.get_scan_directory.return_value = scaninfo["scan_directory"] - instance = mock_send_requests_put.return_value - instance.ok = requests_state - instance.raise_for_status.side_effect = Exception - - if expected_exception: - with pytest.raises(Exception): - mock_det.timeout = 0.1 - mock_det.custom_prepare.prepare_data_backend() - mock_close_file_writer.assert_called_once() - mock_stop_file_writer.assert_called_once() - instance.raise_for_status.assert_called_once() - else: - mock_det.custom_prepare.prepare_data_backend() - - mock_close_file_writer.assert_called_once() - mock_stop_file_writer.assert_called_once() - - # Assert values set on detector - assert mock_det.cam.file_path.get() == "/dev/shm/zmq/" - assert ( - mock_det.cam.file_name.get() - == f"{scaninfo['eacc']}_2_{scaninfo['scan_number']:05d}" - ) - assert mock_det.cam.auto_increment.get() == 1 - assert mock_det.cam.file_number.get() == 0 - assert mock_det.cam.file_format.get() == 0 - assert mock_det.cam.file_template.get() == "%s%s_%5.5d.cbf" - # Remove last / from destinationPath - mock_create_directory.assert_called_once_with( - os.path.join(data_msgs[0]["source"][0]["destinationPath"]) - ) - assert mock_send_requests_put.call_count == 3 - - calls = [ - mock.call(url=url, data=data_msg, headers=scaninfo["headers"]) - for url, data_msg in zip(urls, data_msgs) - ] - for call, mock_call in zip(calls, mock_send_requests_put.call_args_list): - assert call == mock_call - - -def test_complete(mock_det): - path = "tmp" - mock_det.filepath_raw = path - with ( - mock.patch.object(mock_det.custom_prepare, "finished") as mock_finished, - mock.patch.object( - mock_det.custom_prepare, "publish_file_location" - ) as mock_publish_file_location, - ): - mock_det.complete() - assert mock_finished.call_count == 1 - call = mock.call(done=True, successful=True, metadata={"input_path": path}) - assert mock_publish_file_location.call_args == call - - -def test_stop(mock_det): - with ( - mock.patch.object(mock_det.custom_prepare, "stop_detector") as mock_stop_det, - mock.patch.object(mock_det.custom_prepare, "stop_file_writer") as mock_stop_file_writer, - mock.patch.object(mock_det.custom_prepare, "close_file_writer") as mock_close_file_writer, - ): - mock_det.stop() - mock_stop_det.assert_called_once() - mock_stop_file_writer.assert_called_once() - mock_close_file_writer.assert_called_once() - assert mock_det.stopped is True - - -@pytest.mark.parametrize( - "stopped, mcs_stage_state, expected_exception", - [ - (False, ophyd.Staged.no, False), - (True, ophyd.Staged.no, True), - (False, ophyd.Staged.yes, True), - ], -) -def test_finished(mock_det, stopped, mcs_stage_state, expected_exception): - with ( - mock.patch.object(mock_det, "device_manager") as mock_dm, - mock.patch.object(mock_det.custom_prepare, "stop_file_writer") as mock_stop_file_friter, - mock.patch.object(mock_det.custom_prepare, "stop_detector") as mock_stop_det, - mock.patch.object(mock_det.custom_prepare, "close_file_writer") as mock_close_file_writer, - ): - mock_dm.devices.mcs.obj._staged = mcs_stage_state - mock_det.stopped = stopped - if expected_exception: - with pytest.raises(Exception): - mock_det.timeout = 0.1 - mock_det.custom_prepare.finished() - assert mock_det.stopped is stopped - mock_stop_file_friter.assert_called() - mock_stop_det.assert_called_once() - mock_close_file_writer.assert_called_once() - else: - mock_det.custom_prepare.finished() - if stopped: - assert mock_det.stopped is stopped - - mock_stop_file_friter.assert_called() - mock_stop_det.assert_called_once() - mock_close_file_writer.assert_called_once()