test(test-camers): Add unit tests for camera integration of prosilica and basler cameras

debye_bec/devices/cameras/debye_base_cam.py

@@ -6,7 +6,6 @@ import threading
 from typing import TYPE_CHECKING
 
 import numpy as np
-import time
 from bec_lib.logger import bec_logger
 from ophyd import DeviceStatus, StatusBase
 from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
@@ -27,7 +26,7 @@ class DebyeBaseCamera(PSIDeviceBase):
 
     def __init__(self, *, name: str, prefix: str = "", scan_info: ScanInfo | None = None, **kwargs):
         super().__init__(name=name, prefix=prefix, scan_info=scan_info, **kwargs)
-        self.image1 : "ImagePlugin_V35"
+        self.image1: "ImagePlugin_V35"
         self._update_frequency = 1  # Hz
         self._live_mode = False
         self._live_mode_event = None

tests/tests_devices/test_cameras.py

@@ -0,0 +1,68 @@
+"""Module to test prosilica and Basler cam integrations."""
+
+import threading
+from unittest import mock
+
+import ophyd
+import pytest
+from ophyd_devices.devices.areadetector.cam import AravisDetectorCam, ProsilicaDetectorCam
+from ophyd_devices.devices.areadetector.plugins import ImagePlugin_V35
+from ophyd_devices.tests.utils import MockPV, patch_dual_pvs
+
+from debye_bec.devices.cameras.basler_cam import BaslerCam
+from debye_bec.devices.cameras.prosilica_cam import ProsilicaCam
+
+# pylint: disable=protected-access
+# pylint: disable=redefined-outer-name
+
+
+@pytest.fixture(scope="function")
+def mock_basler():
+    """Fixture to mock the camera device."""
+    name = "cam"
+    prefix = "test:"
+    with mock.patch.object(ophyd, "cl") as mock_cl:
+        mock_cl.get_pv = MockPV
+        mock_cl.thread_class = threading.Thread
+        dev = BaslerCam(name=name, prefix=prefix)
+        patch_dual_pvs(dev)
+        yield dev
+
+
+def test_basler_init(mock_basler):
+    """Test the initialization of the Basler camera device."""
+    assert mock_basler.name == "cam"
+    assert mock_basler.prefix == "test:"
+    assert isinstance(mock_basler.cam1, AravisDetectorCam)
+    assert isinstance(mock_basler.image1, ImagePlugin_V35)
+    assert mock_basler._update_frequency == 1
+    assert mock_basler._live_mode is False
+    assert mock_basler._live_mode_event is None
+    assert mock_basler._task_status is None
+    assert mock_basler._n_rot90 == -1
+
+
+@pytest.fixture(scope="function")
+def mock_prosilica():
+    """Fixture to mock the camera device."""
+    name = "cam"
+    prefix = "test:"
+    with mock.patch.object(ophyd, "cl") as mock_cl:
+        mock_cl.get_pv = MockPV
+        mock_cl.thread_class = threading.Thread
+        dev = ProsilicaCam(name=name, prefix=prefix)
+        patch_dual_pvs(dev)
+        yield dev
+
+
+def test_prosilica_init(mock_prosilica):
+    """Test the initialization of the Prosilica camera device."""
+    assert mock_prosilica.name == "cam"
+    assert mock_prosilica.prefix == "test:"
+    assert isinstance(mock_prosilica.cam1, ProsilicaDetectorCam)
+    assert isinstance(mock_prosilica.image1, ImagePlugin_V35)
+    assert mock_prosilica._update_frequency == 1
+    assert mock_prosilica._live_mode is False
+    assert mock_prosilica._live_mode_event is None
+    assert mock_prosilica._task_status is None
+    assert mock_prosilica._n_rot90 == -1

tests/tests_devices/test_debye_base_cam.py

@@ -0,0 +1,85 @@
+"""Module to test camera base integration class for Debye."""
+
+import threading
+from unittest import mock
+
+import ophyd
+import pytest
+from ophyd_devices.tests.utils import MockPV, patch_dual_pvs
+
+from debye_bec.devices.cameras.debye_base_cam import DebyeBaseCamera
+
+# pylint: disable=protected-access
+# pylint: disable=redefined-outer-name
+
+
+@pytest.fixture(scope="function")
+def mock_cam():
+    """Fixture to mock the camera device."""
+    name = "cam"
+    prefix = "test:"
+    with mock.patch.object(ophyd, "cl") as mock_cl:
+        mock_cl.get_pv = MockPV
+        mock_cl.thread_class = threading.Thread
+        dev = DebyeBaseCamera(name=name, prefix=prefix)
+        patch_dual_pvs(dev)
+        yield dev
+
+
+def test_init(mock_cam):
+    """Test the initialization of the camera device."""
+    assert mock_cam.name == "cam"
+    assert mock_cam.prefix == "test:"
+    assert mock_cam._update_frequency == 1
+    assert mock_cam._live_mode is False
+    assert mock_cam._live_mode_event is None
+    assert mock_cam._task_status is None
+    assert mock_cam._n_rot90 == -1
+
+
+def test_start_live_mode(mock_cam):
+    """Test starting live mode."""
+
+    def mock_emit_to_bec(*args, **kwargs):
+        """Mock emit_to_bec method."""
+        while not mock_cam._live_mode_event.wait(1 / mock_cam._update_frequency):
+            pass
+
+    with mock.patch.object(mock_cam, "emit_to_bec", side_effect=mock_emit_to_bec):
+        mock_cam._start_live_mode()
+        assert mock_cam._live_mode_event is not None
+        assert mock_cam._task_status is not None
+        assert mock_cam._task_status.state == "running"
+        mock_cam._live_mode_event.set()
+        # Wait for the task to resolve
+        mock_cam._task_status.wait(timeout=5)
+        assert mock_cam._task_status.done is True
+
+
+def test_stop_live_mode(mock_cam):
+    """Test stopping live mode."""
+    with mock.patch.object(mock_cam, "_live_mode_event") as mock_live_mode_event:
+        mock_cam._stop_live_mode()
+        assert mock_live_mode_event.set.called
+        assert mock_cam._live_mode_event is None
+
+
+def test_live_mode_property(mock_cam):
+    """Test the live_mode property."""
+    assert mock_cam.live_mode is False
+    with mock.patch.object(mock_cam, "_start_live_mode") as mock_start_live_mode:
+        with mock.patch.object(mock_cam, "_stop_live_mode") as mock_stop_live_mode:
+            # Set to true
+            mock_cam.live_mode = True
+            assert mock_start_live_mode.called
+            assert mock_cam._live_mode is True
+            assert mock_start_live_mode.call_count == 1
+            # Second call should call _start_live_mode
+            mock_cam.live_mode = True
+            assert mock_start_live_mode.call_count == 1
+
+            # Set to false
+            mock_cam.live_mode = False
+            assert mock_stop_live_mode.called
+            assert mock_cam._live_mode is False
+            assert mock_stop_live_mode.call_count == 1