Merge pull request #113 from AllenNeuralDynamics/wip/pytest

Wip/pytest

Author: jsiegle

parallax/__init__.py

@@ -4,7 +4,7 @@
 
 import os
 
-__version__ = "1.1.3"
+__version__ = "1.1.4"
 
 # allow multiple OpenMP instances
 os.environ["KMP_DUPLICATE_LIB_OK"] = "True"

parallax/axis_filter.py

@@ -193,6 +193,7 @@ def __init__(self, model, camera_name):
         self.worker = None
         self.name = camera_name
         self.thread = None
+        self.threadDeleted = False
 
     def init_thread(self):
         """Initialize or reinitialize the worker and thread"""
@@ -207,7 +208,6 @@ def init_thread(self):
         self.thread.destroyed.connect(self.onThreadDestroyed)
         self.threadDeleted = False
 
-        #self.worker.frame_processed.connect(self.frame_processed)
         self.worker.frame_processed.connect(self.frame_processed.emit)
         self.worker.found_coords.connect(self.found_coords)
         self.worker.finished.connect(self.thread.quit)
@@ -266,16 +266,18 @@ def clean(self):
         if self.worker is not None:
             self.worker.stop_running()  # Signal the worker to stop
 
-        if not self.threadDeleted and self.thread.isRunning():
+        if self.thread and not self.threadDeleted and self.thread.isRunning():
             self.thread.quit()  # Ask the thread to quit
             self.thread.wait()  # Wait for the thread to finish
         self.thread = None  # Clear the reference to the thread
         self.worker = None  # Clear the reference to the worker
+        self.threadDeleted = True
         logger.debug(f"{self.name} Cleaned the thread")
 
     def onThreadDestroyed(self):
         """Flag if thread is deleted"""
         self.threadDeleted = True
+        self.thread = None
 
     def __del__(self):
         """Destructor for the filter object."""

parallax/probe_detect_manager.py

@@ -395,6 +395,7 @@ def __init__(self, model, camera_name):
         self.worker = None
         self.name = camera_name
         self.thread = None
+        self.threadDeleted = False
 
     def init_thread(self):
         """
@@ -537,11 +538,12 @@ def clean(self):
         if self.worker is not None:
             self.worker.stop_running()
 
-        if not self.threadDeleted and self.thread.isRunning():
+        if self.thread and not self.threadDeleted and self.thread.isRunning():
             self.thread.quit()  # Ask the thread to quit
             self.thread.wait()  # Wait for the thread to finish
         self.thread = None  # Clear the reference to the thread
         self.worker = None  # Clear the reference to the worker
+        self.threadDeleted = True
         logger.debug(f"{self.name} Cleaned the thread")
 
     def __del__(self):

parallax/recording_manager.py

@@ -21,11 +21,12 @@ def __init__(self, model):
         """Initialize recording manager"""
         self.model = model
         self.recording_camera_list = []
+        self.snapshot_camera_list = []
 
     def save_last_image(self, save_path, screen_widgets):
         """Saves the last captured image from all active camera feeds."""
         # Initialize the list to keep track of cameras from which an image has been saved
-        snapshot_camera_list = []
+        self.snapshot_camera_list = []
         # Get the directory path where the images will be saved
         if os.path.exists(save_path):
             print("\nSnapshot...")
@@ -34,7 +35,7 @@ def save_last_image(self, save_path, screen_widgets):
                 if screen.is_camera():
                     # Use custom name of the camera if it has one, otherwise use the camera's serial number
                     camera_name = screen.get_camera_name()
-                    if camera_name not in snapshot_camera_list:
+                    if camera_name not in self.snapshot_camera_list:
                         customName = screen.parent().title()
                         customName = customName if customName else camera_name
 
@@ -44,7 +45,7 @@ def save_last_image(self, save_path, screen_widgets):
                         )
 
                         # Add the camera to the list of cameras from which an image has been saved
-                        snapshot_camera_list.append(camera_name)
+                        self.snapshot_camera_list.append(camera_name)
                 else:
                     logger.debug("save_last_image) camera not found")
         else:

parallax/reticle_detect_manager.py

@@ -237,6 +237,7 @@ def __init__(self, camera_name):
         self.worker = None
         self.name = camera_name
         self.thread = None
+        self.threadDeleted = False
 
     def init_thread(self):
         """Initialize the worker thread."""
@@ -302,17 +303,19 @@ def clean(self):
         if self.worker is not None:
             self.worker.stop_running()  # Signal the worker to stop
 
-        if not self.threadDeleted and self.thread.isRunning():
+        if self.thread and not self.threadDeleted and self.thread.isRunning():
             logger.debug(f"{self.name} Stopping thread in {self.__class__.__name__}")
             self.thread.quit()  # Ask the thread to quit
             self.thread.wait()  # Wait for the thread to finish
         self.thread = None  # Clear the reference to the thread
         self.worker = None  # Clear the reference to the worker
+        self.threadDeleted = True
         logger.debug(f"{self.name} Cleaned the thread")
 
     def onThreadDestroyed(self):
         """Flag if thread is deleted"""
         self.threadDeleted = True
+        self.thread = None
 
     def __del__(self):
         """Destructor for the reticle detection manager."""

tests/test__setting.py

@@ -5,7 +5,7 @@
 from parallax.camera import MockCamera, PySpinCamera  
 from parallax.stage_listener import Stage, StageInfo 
 
-@pytest.fixture
+@pytest.fixture(scope='function')
 def model():
     """Fixture to initialize the Model object."""
     return Model(version="V2", bundle_adjustment=False)

tests/test_model.py

@@ -0,0 +1,112 @@
+# test_recording_manager.py
+
+import pytest
+import os
+from unittest.mock import Mock, MagicMock
+from parallax.recording_manager import RecordingManager
+
+@pytest.fixture
+def mock_model():
+    """Fixture to create a mock model object."""
+    return Mock()
+
+@pytest.fixture
+def mock_screen_widget():
+    """Fixture to create a mock screen widget."""
+    screen = Mock()
+    screen.is_camera = Mock(return_value=True)
+    screen.get_camera_name = Mock(return_value="MockCamera123")
+    screen.save_image = Mock()
+    screen.save_recording = Mock()
+    screen.stop_recording = Mock()
+    screen.parent = Mock()
+    screen.parent().title = Mock(return_value="MockCamera")
+    return screen
+
+@pytest.fixture
+def recording_manager(mock_model):
+    """Fixture to create a RecordingManager instance."""
+    return RecordingManager(mock_model)
+
+def test_save_last_image(recording_manager, mock_screen_widget, tmpdir):
+    """Test saving the last image for a camera."""
+    save_path = str(tmpdir)
+    screen_widgets = [mock_screen_widget]
+
+    # Call the method to save the last image.
+    recording_manager.save_last_image(save_path, screen_widgets)
+
+    # Assert that the save_image method was called.
+    mock_screen_widget.save_image.assert_called_once_with(
+        save_path, isTimestamp=True, name="MockCamera"
+    )
+    # Assert that the camera name was added to the snapshot_camera_list.
+    assert "MockCamera123" in recording_manager.snapshot_camera_list, (
+        f"{mock_screen_widget.get_camera_name()} was not tracked as a snapshot camera."
+    )
+
+def test_save_last_image_directory_not_exists(recording_manager, mock_screen_widget):
+    """Test saving the last image when the save path does not exist."""
+    save_path = "non_existent_directory"
+    screen_widgets = [mock_screen_widget]
+
+    # Call the method to save the last image.
+    recording_manager.save_last_image(save_path, screen_widgets)
+
+    # Assert that save_image was not called since the directory does not exist.
+    mock_screen_widget.save_image.assert_not_called()
+
+def test_save_recording(recording_manager, mock_screen_widget, tmpdir):
+    """Test starting a recording for a camera."""
+    save_path = str(tmpdir)
+    screen_widgets = [mock_screen_widget]
+
+    # Call the method to start recording.
+    recording_manager.save_recording(save_path, screen_widgets)
+
+    # Assert that the save_recording method was called.
+    mock_screen_widget.save_recording.assert_called_once_with(
+        save_path, isTimestamp=True, name="MockCamera"
+    )
+    # Assert that the camera name is in the list of currently recording cameras.
+    assert mock_screen_widget.get_camera_name() in recording_manager.recording_camera_list
+
+def test_save_recording_directory_not_exists(recording_manager, mock_screen_widget):
+    """Test starting a recording when the save path does not exist."""
+    save_path = "non_existent_directory"
+    screen_widgets = [mock_screen_widget]
+
+    # Call the method to start recording.
+    recording_manager.save_recording(save_path, screen_widgets)
+
+    # Assert that save_recording was not called since the directory does not exist.
+    mock_screen_widget.save_recording.assert_not_called()
+
+def test_stop_recording(recording_manager, mock_screen_widget):
+    """Test stopping a recording for a camera."""
+    screen_widgets = [mock_screen_widget]
+
+    # Add the camera to the recording list to simulate it being recorded.
+    recording_manager.recording_camera_list.append(mock_screen_widget.get_camera_name())
+
+    # Call the method to stop recording.
+    recording_manager.stop_recording(screen_widgets)
+
+    # Assert that the stop_recording method was called.
+    mock_screen_widget.stop_recording.assert_called_once()
+
+    # Assert that the camera was removed from the recording list.
+    assert mock_screen_widget.get_camera_name() not in recording_manager.recording_camera_list
+
+def test_stop_recording_not_recording_camera(recording_manager, mock_screen_widget):
+    """Test stopping a recording when the camera is not in the recording list."""
+    screen_widgets = [mock_screen_widget]
+
+    # Ensure the camera is not in the recording list.
+    recording_manager.recording_camera_list = []
+
+    # Call the method to stop recording.
+    recording_manager.stop_recording(screen_widgets)
+
+    # Assert that the stop_recording method was not called since the camera was not recording.
+    mock_screen_widget.stop_recording.assert_not_called()

tests/test_recording_manager.py

@@ -0,0 +1,102 @@
+# tests/test_screen_widget.py
+import pytest
+import numpy as np
+from unittest.mock import Mock
+from PyQt5.QtWidgets import QApplication
+from PyQt5.QtCore import Qt, QPoint
+from parallax.screen_widget import ScreenWidget
+
+@pytest.fixture(scope="function")
+def qapp():
+    """Fixture for creating a QApplication."""
+    app = QApplication([])
+    yield app
+    app.quit()
+
+@pytest.fixture(scope="function")
+def mock_camera():
+    """Fixture to create a mock camera object."""
+    camera = Mock()
+    camera.width = 4000
+    camera.height = 3000
+    camera.name.return_value = "MockCamera123"
+    camera.get_last_image_data.return_value = np.zeros((4000, 3000), dtype=np.uint8)
+    camera.get_last_image_data_singleFrame.return_value = np.zeros((4000, 3000), dtype=np.uint8)
+    return camera
+
+@pytest.fixture(scope="function")
+def mock_model():
+    model = Mock()
+    probe_detector = Mock()
+    model.add_probe_detector(probe_detector)
+    return model
+
+@pytest.fixture(scope="function")
+def screen_widget(qapp, mock_camera, mock_model):
+    """Fixture to create a ScreenWidget instance with a mock camera."""
+    return ScreenWidget(camera=mock_camera, model=mock_model)
+
+def test_simple(screen_widget):
+    assert True
+
+def test_screen_widget_initialization(screen_widget):
+    assert screen_widget.camera is not None
+    assert screen_widget.view_box is not None
+    assert screen_widget.image_item is not None
+
+def test_set_data(screen_widget, mock_camera):
+    # Prepare mock image data
+    data = np.random.randint(0, 256, (4000, 3000), dtype=np.uint8)
+
+    # Mock the process method in NoFilter and AxisFilter
+    screen_widget.filter.process = Mock()
+    screen_widget.axisFilter.process = Mock()
+    screen_widget.reticleDetector.process = Mock()
+    screen_widget.probeDetector.process = Mock()
+
+    # Call set_data to process the image
+    screen_widget.set_data(data)
+
+    # Assert that each filter's process method was called with the data
+    screen_widget.filter.process.assert_called_once_with(data)
+    screen_widget.axisFilter.process.assert_called_once_with(data)
+    screen_widget.reticleDetector.process.assert_called_once_with(data)
+    screen_widget.probeDetector.process.assert_called_once_with(
+        data, screen_widget.camera.get_last_capture_time.return_value
+    )
+
+def test_start_and_stop_acquisition_camera(screen_widget, mock_camera):
+    # Mock the methods for camera acquisition
+    mock_camera.begin_continuous_acquisition = Mock()
+    mock_camera.stop = Mock()
+
+    # Start acquisition and verify the method is called
+    screen_widget.start_acquisition_camera()
+    mock_camera.begin_continuous_acquisition.assert_called_once()
+
+    # Stop acquisition and verify the method is called
+    screen_widget.stop_acquisition_camera()
+    mock_camera.stop.assert_called_once_with(clean=False)
+
+def test_image_clicked_signal(screen_widget, qtbot):
+    with qtbot.waitSignal(screen_widget.selected, timeout=1000) as signal_waiter:
+        mock_event = Mock()
+        mock_event.pos = Mock(return_value=QPoint(100, 100))  # Use QPoint for pos()
+        mock_event.button = Mock(return_value=Qt.LeftButton)
+
+        # Simulate a mouse click on the image using the mock event.
+        screen_widget.image_clicked(mock_event)
+
+    emitted_args = signal_waiter.args
+    assert emitted_args[0] == screen_widget.camera_name, "The camera name is incorrect."
+    assert emitted_args[1] == (100, 100), "The clicked position is incorrect."
+
+def test_zoom_out(screen_widget):
+    # Mock the autoRange method of the view_box
+    screen_widget.view_box.autoRange = Mock()
+
+    # Call zoom_out
+    screen_widget.zoom_out()
+
+    # Verify that autoRange was called
+    screen_widget.view_box.autoRange.assert_called_once()