bec/csaxs_bec
0
0
Fork 1
Code Issues 17 Pull Requests 3 Actions Packages Projects Releases Wiki Activity

test: fix ddg and mcs tests

Browse Source
This commit is contained in:
appel_c
2026-06-23 13:03:55 +02:00
parent b77d8d3f94
commit 70ccbfd608
2 changed files with 17 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/devices/epics/mcs_card/mcs_card_csaxs.py
+10 -11
View File
@@ -487,13 +487,12 @@ class MCSCardCSAXS(PSIDeviceBase, MCSCard):
self._start_monitor_async_data_emission.clear() # Stop monitoring
# NOTE Important check as set_finished or set_exception should not be called
# if the status is already done (e.g. cancelled externally)
with self._rlock:
if status.done:
return # Already done and cancelled externally.
if exception is not None:
status.set_exception(exception)
else:
status.set_finished()
if status.done:
return # Already done and cancelled externally.
if exception is not None:
status.set_exception(exception)
else:
status.set_finished()
def _status_failed_callback(self, status: StatusBase) -> None:
"""Callback for status failure, the monitoring thread should be stopped."""
@@ -548,7 +547,7 @@ class MCSCardCSAXS(PSIDeviceBase, MCSCard):
self.software_channel_advance.put(1)
# Prepare and register status callback for the async monitoring loop
status_async_data = StatusBase(obj=self)
status_async_data = StatusBase(obj=self, timeout=5)
self._scan_done_callbacks.append(partial(self._status_callback, status_async_data))
# Set the event to start monitoring async data emission
@@ -556,14 +555,14 @@ class MCSCardCSAXS(PSIDeviceBase, MCSCard):
self._start_monitor_async_data_emission.set()
# Add CompareStatus for Acquiring DONE
status = CompareStatus(self.acquiring, ACQUIRING.DONE)
status.wait(timeout=3)
status = CompareStatus(self.acquiring, ACQUIRING.DONE, timeout=5)
# status.wait(timeout=3) # timeout is passed to individual status objects, so don't wait here.
# Combine both statuses
ret_status = status & status_async_data
# NOTE: Handle external stop/cancel, and stop monitoring
ret_status.add_callback(self._status_failed_callback)
ret_status.wait(timeout=3)
# ret_status.wait(timeout=3) # timeout is passed to individual status objects, so don't wait here.
self.cancel_on_stop(ret_status)
return ret_status
tests/tests_devices/test_delay_generator_csaxs.py
+7 -7
View File
@@ -260,7 +260,7 @@ def test_ddg1_prepare_mcs(mock_ddg1: DDG1, mock_mcs_csaxs: MCSCardCSAXS):
mcs.erase_start.put(0) # reset erase start
# Prepare MCS on trigger
st = ddg._prepare_mcs_on_trigger(mcs)
st = ddg.prepare_mcs_on_trigger()
assert st.done is False
assert st.success is False
assert mcs.erase_start.get() == 1 # erase started
@@ -356,16 +356,16 @@ def test_ddg1_on_trigger(mock_ddg1: DDG1):
#################################
# Scenario I - normal operation #
#################################
with mock.patch.object(ddg, "_prepare_mcs_on_trigger") as mock_prepare_mcs:
with mock.patch.object(ddg, "prepare_mcs_on_trigger") as mock_prepare_mcs:
# Set acquioring PV to acquiring
mcs = ddg.device_manager.devices.get("mcs", None)
assert mcs is not None
mcs.acquiring._read_pv.mock_data = 1 # Simulate acquiring started
mock_prepare_mcs.return_value = ophyd.StatusBase(done=True, success=True)
# MCS card is present and enabled, should call prepare_mcs_on_trigger
# and the status should resolve once acuiring goes from 1 to 0.
status = ddg.trigger()
assert status.done is False
mcs = ddg.device_manager.devices.get("mcs", None)
assert mcs is not None
mcs.acquiring._read_pv.mock_data = 1 # Simulate acquiring started
assert status.done is False
mcs.acquiring._read_pv.mock_data = 0 # Simulate acquiring stopped
status.wait(timeout=1) # Wait for the status to be done
assert status.done is True
@@ -407,7 +407,7 @@ def test_ddg1_on_trigger(mock_ddg1: DDG1):
ddg.state.event_status._read_pv.mock_data = STATUSBITS.ABORT_DELAY.value
ddg._start_polling()
assert ddg._poll_thread_run_event.is_set()
with mock.patch.object(ddg, "_prepare_mcs_on_trigger") as mock_prepare_mcs:
with mock.patch.object(ddg, "prepare_mcs_on_trigger") as mock_prepare_mcs:
status = ddg.trigger()
mock_prepare_mcs.assert_not_called() # MCS is disabled, should not be called
assert status.done is False