feat: implement slot_error_boundary decorator and add unit tests for session and device controllers

2026-06-18 21:01:37 -07:00
parent 5514653b94
commit 7206334a27
6 changed files with 353 additions and 1 deletions
@@ -0,0 +1,142 @@
+import pytest
+import logging
+from unittest.mock import MagicMock, patch
+from pygui.backend.controllers.device_controller import DeviceController
+from pygui.backend.data.local_settings import LocalSettings
+from pygui.serialcomm.serial_port import WaferInfo
+
+@pytest.fixture
+def mock_settings():
+    s = LocalSettings()
+    s.save_data_dir = ""
+    s.connection_status = "Disconnected"
+    s.activity_log = []
+    s.last_wafer_info = {}
+    return s
+
+@pytest.fixture
+def controller(mock_settings):
+    import tempfile
+    import shutil
+    temp_dir = tempfile.mkdtemp()
+    yield DeviceController(mock_settings, temp_dir)
+    shutil.rmtree(temp_dir)
+
+def test_refresh_ports(controller):
+    controller._service.enumerate_ports = MagicMock(return_value=["COM1", "COM2"])
+    controller.refreshPorts()
+    assert "COM1" in controller.availablePorts
+
+def test_detect_wafer_spawns_thread(controller):
+    with patch("threading.Thread") as mock_thread:
+        info = WaferInfo(
+            family_code="P",
+            serial_number="P00001",
+            sensor_count=244,
+            mfg_date_hex="12345678",
+            sensor_assigned_hex="01",
+            locked_hex="00",
+            runtime=100,
+            cycle_count=10
+        )
+        controller._service.detect_all_ports = MagicMock(
+            return_value=("COM1", info)
+        )
+        controller.detectWafer()
+        mock_thread.assert_called_once()
+        assert controller.operationInProgress
+
+def test_read_memory_without_detect_return_error(controller):
+    emitted_result = None
+    def on_read_result(result):
+        nonlocal emitted_result
+        emitted_result = result
+
+    controller.readResult.connect(on_read_result)
+    controller.readMemoryAsync()
+    assert emitted_result is not None
+    assert "error" in emitted_result
+
+def test_setters_and_properties(controller):
+    controller.setSelectedPort("COM3")
+    assert controller.selectedPort == "COM3"
+
+    controller.setSaveDataDir("some/custom/dir")
+    assert controller.saveDataDir == "some/custom/dir"
+
+    assert controller.connectionStatus == "Disconnected"
+    assert controller.dataRowCount == 0
+    assert controller.dataColCount == 0
+    assert isinstance(controller.lastWaferInfo, list)
+    assert controller.dataModel is not None
+    assert controller.graphView is not None
+
+def test_clear_activity_log(controller):
+    controller._activity_log.append("[12:00:00] Some message")
+    assert len(controller._activity_log) > 0
+    controller.clearActivityLog()
+    assert len(controller._activity_log) == 0
+
+def test_clear_session(controller):
+    controller._connection_status = "Connected"
+    controller._selected_port = "COM1"
+    controller._last_wafer_info = {"serialNumber": "P00001"}
+    
+    controller.clearSession()
+    assert controller.connectionStatus == "Disconnected"
+    assert controller.selectedPort == ""
+    assert controller.lastWaferInfo == ["", "", 0, "", 0, 0]
+
+def test_erase_memory(controller):
+    controller._service.erase_wafer = MagicMock(return_value=True)
+    emitted_result = None
+    def on_erase_result(result):
+        nonlocal emitted_result
+        emitted_result = result
+
+    controller.eraseResult.connect(on_erase_result)
+    controller.eraseMemory("COM1")
+    assert emitted_result == {"success": True}
+    assert controller.connectionStatus == "Connected"
+
+def test_read_debug(controller):
+    controller._service.read_wafer_data = MagicMock(return_value=bytes([12] * 512))
+    emitted_result = None
+    def on_debug_result(result):
+        nonlocal emitted_result
+        emitted_result = result
+
+    controller.debugResult.connect(on_debug_result)
+    controller.readDebug("COM1")
+    assert emitted_result is not None
+    assert emitted_result.get("success") is True
+    assert emitted_result.get("sensor_bytes") == 512
+    assert emitted_result.get("debug_bytes") == 512
+
+def test_parse_and_save_data_and_get_chart_data(controller):
+    res = controller.getChartData()
+    assert res == {"success": False}
+
+    controller._raw_bytes = bytes([12] * 512)
+    emitted_result = None
+    def on_parsed(result):
+        nonlocal emitted_result
+        emitted_result = result
+
+    controller.parsedDataReady.connect(on_parsed)
+    controller.parseAndSaveData("P", "COM1")
+    
+    assert emitted_result is not None
+    assert emitted_result.get("success") is True
+    assert controller.dataRowCount == 1
+    assert controller.dataColCount == 244
+
+    chart_res = controller.getChartData()
+    assert chart_res.get("success") is True
+    assert "Sensor1" in chart_res.get("sensor_names")
+    assert len(chart_res.get("series")) == 244
+
+def test_logging_handler(controller):
+    logger = logging.getLogger("test_logger")
+    logger.info("Hello QML Activity Log")
+    assert any("Hello QML Activity Log" in line for line in controller._activity_log)
@@ -0,0 +1,48 @@
+import pytest
+from unittest.mock import MagicMock, patch
+from pygui.backend.controllers.session_controller import SessionController
+
+@pytest.fixture
+def controller():
+    return SessionController()
+
+def test_initial_default(controller):
+    assert controller.mode == "review"
+    assert controller.state == "idle"
+    assert controller.recording == False
+
+def test_playback_flow(controller):
+    controller.loadFile("tests/fixtures/sample_stream.csv")
+    controller.play()
+    initial_index = controller.frameIndex
+    controller._advance()
+    assert controller.frameIndex == initial_index + 1
+
+def test_speed_control(controller):
+    controller.setSpeed(2.0)
+    interval = controller._next_interval_ms()
+    controller.setSpeed(0.5)
+    assert controller._next_interval_ms() > interval
+
+def test_seek(controller):
+    controller.loadFile("tests/fixtures/sample_stream.csv")
+    controller.seek(1)
+    assert controller.frameIndex == 1
+
+def test_step_forward(controller):
+    controller.loadFile("tests/fixtures/sample_stream.csv")
+    initial = controller.frameIndex
+    controller.step(1)
+    assert controller.frameIndex == initial + 1
+
+def test_step_backward(controller):
+    controller.loadFile("tests/fixtures/sample_stream.csv")
+    controller.seek(2)
+    controller.step(-1)
+    assert controller.frameIndex == 1
+
+def test_pause_stop_timer(controller):
+    controller.loadFile("tests/fixtures/sample_stream.csv")
+    controller.seek(2)
+    controller.stop()
+    assert controller.frameIndex == 0
@@ -0,0 +1,99 @@
+"""Tests for slot_error_boundary decorator."""
+from __future__ import annotations
+
+import logging
+from unittest.mock import MagicMock
+
+from pygui.backend.utils import slot_error_boundary
+
+
+class _FakeSlot:
+    """Minimal QObject-like object to test the decorator against."""
+
+    def __init__(self, has_log_message=True):
+        self.logMessage = MagicMock() if has_log_message else None
+        self.__class__.__module__ = "test_module"
+
+
+class TestSlotErrorBoundary:
+
+    def test_passes_through_on_success(self):
+        @slot_error_boundary
+        def good(self) -> str:
+            return "ok"
+
+        result = good(_FakeSlot())
+        assert result == "ok"
+
+    def test_catches_and_logs_exception(self, caplog):
+        @slot_error_boundary
+        def bad(self) -> None:
+            raise ValueError("boom")
+
+        with caplog.at_level(logging.ERROR):
+            bad(_FakeSlot())
+
+        assert "ValueError" in caplog.text
+        assert "bad" in caplog.text
+
+    def test_emits_log_message_on_error(self):
+        obj = _FakeSlot(has_log_message=True)
+
+        @slot_error_boundary
+        def bad(self) -> None:
+            raise RuntimeError("fail")
+
+        bad(obj)
+
+        obj.logMessage.emit.assert_called_once()
+        emitted_msg = obj.logMessage.emit.call_args[0][0]
+        assert "bad" in emitted_msg
+        assert "fail" in emitted_msg
+
+    def test_returns_none_on_error(self):
+        @slot_error_boundary
+        def bad(self) -> None:
+            raise KeyError("missing")
+
+        result = bad(_FakeSlot())
+        assert result is None
+
+    def test_no_log_message_signal_is_ok(self):
+        """Controller without logMessage signal should not crash."""
+
+        class _NoLogMsg:
+            __module__ = "test_module"
+
+        @slot_error_boundary
+        def bad(self) -> None:
+            raise ValueError("boom")
+
+        # Should not raise — hasattr check guards the emit
+        result = bad(_NoLogMsg())
+        assert result is None
+
+    def test_preserves_function_name(self):
+        @slot_error_boundary
+        def named_method(self) -> None:
+            pass
+
+        assert named_method.__name__ == "named_method"
+
+    def test_passes_args_through(self):
+        @slot_error_boundary
+        def with_args(self, a: int, b: str) -> tuple[int, str]:
+            return (a, b)
+
+        result = with_args(_FakeSlot(), 42, "hello")
+        assert result == (42, "hello")
+
+    def test_exception_in_wrapper_does_not_leak(self):
+        """The decorator must never re-raise — QML would crash."""
+
+        @slot_error_boundary
+        def crashing(self) -> None:
+            raise TypeError("uncaught type error")
+
+        # Should return None, not raise
+        result = crashing(_FakeSlot())
+        assert result is None