From 7206334a276540db78b9e36919f68dbd3b94648d Mon Sep 17 00:00:00 2001 From: jack Date: Thu, 18 Jun 2026 21:01:37 -0700 Subject: [PATCH] feat: implement slot_error_boundary decorator and add unit tests for session and device controllers --- .../backend/controllers/device_controller.py | 14 ++ .../backend/controllers/session_controller.py | 22 ++- src/pygui/backend/utils.py | 29 ++++ tests/test_device_controller.py | 142 ++++++++++++++++++ tests/test_session_controller.py | 48 ++++++ tests/test_slot_error_boundary.py | 99 ++++++++++++ 6 files changed, 353 insertions(+), 1 deletion(-) create mode 100644 src/pygui/backend/utils.py create mode 100644 tests/test_device_controller.py create mode 100644 tests/test_session_controller.py create mode 100644 tests/test_slot_error_boundary.py diff --git a/src/pygui/backend/controllers/device_controller.py b/src/pygui/backend/controllers/device_controller.py index 73bdd00..dc8c1b0 100644 --- a/src/pygui/backend/controllers/device_controller.py +++ b/src/pygui/backend/controllers/device_controller.py @@ -12,6 +12,7 @@ from PySide6.QtCore import Property, QObject, Qt, Signal, Slot from pygui.backend.controllers.session_controller import SessionController from pygui.backend.data.local_settings import LocalSettings from pygui.backend.models.data_model import TemperatureTableModel +from pygui.backend.utils import slot_error_boundary from pygui.backend.visualization.graph_view import GraphView from pygui.serialcomm.data_parser import ( convert_to_temperatures, @@ -116,6 +117,7 @@ class DeviceController(QObject): return self._save_data_dir @Slot(str) + @slot_error_boundary def setSaveDataDir(self, path: str) -> None: """Set the directory for saving parsed CSV data files.""" self._save_data_dir = path @@ -128,6 +130,7 @@ class DeviceController(QObject): return self._selected_port @Slot(str) + @slot_error_boundary def setSelectedPort(self, port: str) -> None: """Set the active serial port from the port selector.""" self._selected_port = port @@ -170,6 +173,7 @@ class DeviceController(QObject): return self._graph_view @Slot(result=object) + @slot_error_boundary def getChartData(self) -> dict[str, Any]: """Extract chart-ready data from the current model. @@ -208,6 +212,7 @@ class DeviceController(QObject): self.activityLogUpdated.emit("\n".join(self._activity_log)) @Slot() + @slot_error_boundary def clearActivityLog(self) -> None: """Clear the activity log.""" self._activity_log.clear() @@ -215,6 +220,7 @@ class DeviceController(QObject): self._save_status() @Slot() + @slot_error_boundary def clearSession(self) -> None: """Clear the current session state, allowing a clean detect.""" self._connection_status = "Disconnected" @@ -242,6 +248,7 @@ class DeviceController(QObject): # ---- Public Slots ---- @Slot() + @slot_error_boundary def refreshPorts(self) -> None: """Scan for available serial ports and emit updated list.""" ports = self._service.enumerate_ports() @@ -251,6 +258,7 @@ class DeviceController(QObject): self._save_status() @Slot() + @slot_error_boundary def detectWafer(self) -> None: """Scan all serial ports for a wafer (runs in background thread). @@ -280,6 +288,7 @@ class DeviceController(QObject): self._detectFinished.emit(port, info) @Slot(object, object) + @slot_error_boundary def _handle_detect_finished( self, port: Optional[str], info: Optional[WaferInfo] ) -> None: @@ -304,6 +313,7 @@ class DeviceController(QObject): self._set_operation_progress(False) @Slot() + @slot_error_boundary def readMemoryAsync(self) -> None: """Read wafer memory in a background thread. @@ -346,6 +356,7 @@ class DeviceController(QObject): self._readFinished.emit(port, family_code, data) @Slot(str, str, object) + @slot_error_boundary def _handle_read_finished( self, port: str, family_code: str, data: Optional[bytes] ) -> None: @@ -367,6 +378,7 @@ class DeviceController(QObject): self._save_status() @Slot(str) + @slot_error_boundary def eraseMemory(self, port: str) -> None: """Send p1 erase command.""" self._set_operation_progress(True) @@ -385,6 +397,7 @@ class DeviceController(QObject): self._set_operation_progress(False) @Slot(str) + @slot_error_boundary def readDebug(self, port: str) -> None: """Read both D1 (sensor) and F1 (debug/cold-junction) data. @@ -428,6 +441,7 @@ class DeviceController(QObject): self._set_operation_progress(False) @Slot(str, str) + @slot_error_boundary def parseAndSaveData(self, family_code: str = "", port: str = "") -> None: """Parse raw bytes into temperatures and save to CSV. diff --git a/src/pygui/backend/controllers/session_controller.py b/src/pygui/backend/controllers/session_controller.py index ebc6907..0288f6a 100644 --- a/src/pygui/backend/controllers/session_controller.py +++ b/src/pygui/backend/controllers/session_controller.py @@ -14,6 +14,7 @@ from pygui.backend.models.frame_player import FramePlayer, frames_from_wafer_dat from pygui.backend.models.sensor_editor import SensorEditor from pygui.backend.models.session_model import SessionModel from pygui.backend.models.threshold_classifier import ThresholdConfig +from pygui.backend.utils import slot_error_boundary from pygui.backend.wafer.zwafer_models import Sensor from pygui.backend.wafer.zwafer_parser import ZWaferParser from pygui.serialcomm.stream_reader import StreamReader @@ -182,6 +183,7 @@ class SessionController(QObject): # ---- mode + thresholds ---- @Slot(str) + @slot_error_boundary def setMode(self, mode: str) -> None: if mode == self._mode: return @@ -191,6 +193,7 @@ class SessionController(QObject): self.modeChanged.emit() @Slot(float, float, bool) + @slot_error_boundary def setThresholds(self, set_point: float, margin: float, auto: bool) -> None: self._model.set_thresholds(ThresholdConfig(set_point, margin, auto)) if self._last: # re-band current frame in place @@ -198,6 +201,7 @@ class SessionController(QObject): # ---- review: file load + playback ---- @Slot(str) + @slot_error_boundary def loadFile(self, file_path: str) -> None: from pathlib import Path @@ -250,30 +254,37 @@ class SessionController(QObject): self._emit_current() @Slot() + @slot_error_boundary def play(self) -> None: self._play_timer.start(self._next_interval_ms()) @Slot() - def pause(self) -> None: self._play_timer.stop() + @slot_error_boundary + def pause(self) -> None: + self._play_timer.stop() @Slot() + @slot_error_boundary def stop(self) -> None: self._play_timer.stop() self._player.seek(0) self._emit_current() @Slot(int) + @slot_error_boundary def step(self, delta: int) -> None: self._play_timer.stop() self._player.step(delta) self._emit_current() @Slot(int) + @slot_error_boundary def seek(self, i: int) -> None: self._player.seek(i) self._emit_current() @Slot(float) + @slot_error_boundary def setSpeed(self, x: float) -> None: self._speed = max(0.1, x) if self._play_timer.isActive(): @@ -325,6 +336,7 @@ class SessionController(QObject): # ---- live: stream start/stop ---- @Slot(str, str) + @slot_error_boundary def startStream(self, port: str, family_code: str = "") -> None: from pygui.backend.wafer.wafer_layouts import load_layout_for_wafer_id @@ -389,6 +401,7 @@ class SessionController(QObject): self.stateChanged.emit() @Slot() + @slot_error_boundary def stopStream(self) -> None: self._repaint_timer.stop() if self._reader: @@ -409,6 +422,7 @@ class SessionController(QObject): self.stopRecording() @Slot(object) + @slot_error_boundary def _on_live_frame(self, frame: Frame) -> None: # Main thread (queued). Record raw, process edited; mark dirty for repaint. self._last_raw_frame = frame @@ -428,35 +442,41 @@ class SessionController(QObject): # ---- recording ---- @Slot(str, str) + @slot_error_boundary def startRecording(self, path: str, serial: str = "") -> None: self._recorder.start(path, self._sensors, serial) self.recordingChanged.emit() @Slot() + @slot_error_boundary def stopRecording(self) -> None: if self._recorder.is_recording: self._recorder.stop() self.recordingChanged.emit() @Slot(int, float) + @slot_error_boundary def replaceSensor(self, index: int, value: float) -> None: """Override sensor `index` to display `value` every frame.""" self._sensor_editor.set_replacement(index, value) self._reprocess_current() @Slot(int, float) + @slot_error_boundary def offsetSensor(self, index: int, delta: float) -> None: """Shift sensor `index` by `delta` every frame.""" self._sensor_editor.set_offset(index, delta) self._reprocess_current() @Slot(int) + @slot_error_boundary def clearSensorEdit(self, index: int) -> None: """Remove all overrides for sensor `index`.""" self._sensor_editor.clear(index) self._reprocess_current() @Slot() + @slot_error_boundary def clearSensorEdits(self) -> None: """Remove all sensor overrides.""" self._sensor_editor.clear() diff --git a/src/pygui/backend/utils.py b/src/pygui/backend/utils.py new file mode 100644 index 0000000..66adc5d --- /dev/null +++ b/src/pygui/backend/utils.py @@ -0,0 +1,29 @@ +import logging +from functools import wraps + + +def slot_error_boundary(func): + """Wrap a @Slot method so exceptions are logged + signalled to QML. + + Usage (stack under @Slot): + @Slot() + @slot_error_boundary + def myMethod(self) -> None: + ... + + On exception: + - Logs full traceback via logger.exception() + - Emits logMessage signal with human-readable error (if available) + - Returns None so QML never sees a crash + """ + @wraps(func) + def wrapper(self, *args, **kwargs): + try: + return func(self, *args, **kwargs) + except Exception as e: + logger = logging.getLogger(self.__class__.__module__) + logger.exception("Slot '%s' raised: %s", func.__name__, e) + if hasattr(self, "logMessage"): + self.logMessage.emit(f"Error in {func.__name__}: {str(e)}") + return None + return wrapper diff --git a/tests/test_device_controller.py b/tests/test_device_controller.py new file mode 100644 index 0000000..67f00df --- /dev/null +++ b/tests/test_device_controller.py @@ -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) diff --git a/tests/test_session_controller.py b/tests/test_session_controller.py new file mode 100644 index 0000000..ba02ea3 --- /dev/null +++ b/tests/test_session_controller.py @@ -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 diff --git a/tests/test_slot_error_boundary.py b/tests/test_slot_error_boundary.py new file mode 100644 index 0000000..3667fc4 --- /dev/null +++ b/tests/test_slot_error_boundary.py @@ -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