bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2025-06-23 11:27:57 +02:00
Code Issues Packages Projects Releases Wiki Activity

test: fix and improve tests camera on complete

Browse Source
This commit is contained in:
appel_c
2025-03-06 12:59:22 +01:00
parent 8cdcfe7a44
commit 0ed29209e4
3 changed files with 10 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
ophyd_devices/sim/sim_camera.py
View File

@ -136,6 +136,7 @@ class SimCamera(PSIDeviceBase, SimCameraControl):
def complete_cam():
"""Complete the camera acquisition."""
self.h5_writer.on_complete()
self._run_subs(
sub_type=self.SUB_FILE_EVENT,
file_path=self.file_path,

2
ophyd_devices/utils/psi_device_base_utils.py
View File

@ -59,7 +59,7 @@ class TaskStatus(DeviceStatus):
self._state = TaskState(value)
@property
def task_id(self) -> bool:
def task_id(self) -> str:
"""Get the task ID"""
return self._task_id

10
tests/test_simulation.py
View File

@ -549,7 +549,7 @@ def test_stage_camera_proxy_image_shape(
test_shape = (102, 77)
camera.image_shape.set(test_shape).wait()
proxy.stage()
image = camera.image.get()
image: np.ndarray = camera.image.get()
assert image.shape == (*reversed(test_shape), 3)
@ -595,6 +595,11 @@ def test_cam_stage_h5writer(camera):
def test_cam_complete(camera):
"""Test the complete method of SimCamera."""
finished_event = threading.Event()
def finished_cb():
finished_event.set()
with mock.patch.object(camera, "h5_writer") as mock_h5_writer:
status = camera.complete()
status_wait(status)
@ -603,7 +608,8 @@ def test_cam_complete(camera):
assert mock_h5_writer.on_complete.call_count == 0
camera.write_to_disk.put(True)
status = camera.complete()
status_wait(status)
status.add_callback(finished_cb)
finished_event.wait()
assert mock_h5_writer.on_complete.call_count == 1