import logging from unittest.mock import MagicMock, patch import pytest 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 shutil import tempfile 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_spawns_thread(controller): with patch("threading.Thread") as mock_thread: controller.eraseMemory("COM1") mock_thread.assert_called_once() assert controller.operationInProgress def test_erase_memory_handler(controller): emitted_result = None def on_erase_result(result): nonlocal emitted_result emitted_result = result controller.eraseResult.connect(on_erase_result) controller._handle_erase_finished(True) assert emitted_result == {"success": True} assert controller.connectionStatus == "Connected" assert not controller.operationInProgress def test_read_debug_spawns_thread(controller): with patch("threading.Thread") as mock_thread: controller.readDebug("COM1") mock_thread.assert_called_once() assert controller.operationInProgress def test_read_debug_handler(controller): emitted_result = None def on_debug_result(result): nonlocal emitted_result emitted_result = result controller.debugResult.connect(on_debug_result) controller._handle_debug_finished({ "success": True, "sensor_bytes": 512, "debug_bytes": 512 }) 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 assert controller.connectionStatus == "Connected" assert not controller.operationInProgress 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) def test_fresh_session_on_startup(mock_settings): import shutil import tempfile # Configure mock settings with some data to simulate previous session mock_settings.activity_log = ["[12:00:00] Old message"] mock_settings.selected_port = "COM9" mock_settings.last_wafer_info = {"familyCode": "P", "serialNumber": "P00002"} mock_settings.data_row_count = 50 mock_settings.data_col_count = 244 temp_dir = tempfile.mkdtemp() try: new_controller = DeviceController(mock_settings, temp_dir) # Should start completely fresh (empty/default) ignoring the persisted status assert len(new_controller._activity_log) == 0 assert new_controller.selectedPort == "" assert new_controller.lastWaferInfo == ["", "", 0, "", 0, 0] assert new_controller.dataRowCount == 0 assert new_controller.dataColCount == 0 finally: shutil.rmtree(temp_dir) def test_detect_wafer_clears_old_session(controller): controller._selected_port = "COM1" controller._last_wafer_info = {"familyCode": "P", "serialNumber": "P00001"} controller._data_row_count = 10 controller._data_col_count = 244 controller._activity_log.append("[12:00:00] Previous message") # We mock detect_all_ports to not run long or fail controller._service.detect_all_ports = MagicMock(return_value=(None, None)) with patch("threading.Thread"): controller.detectWafer() # Verify it cleared all session variables and logs immediately at start of detect assert controller.selectedPort == "" assert controller.lastWaferInfo == ["", "", 0, "", 0, 0] assert controller.dataRowCount == 0 assert controller.dataColCount == 0 # Log should only contain "Scanning all ports for wafer ..." assert len(controller._activity_log) == 1 assert "Scanning all ports" in controller._activity_log[0] def test_device_service_port_caching(controller): service = controller._service # Mock comports list with patch("serial.tools.list_ports.comports", return_value=[]), \ patch("subprocess.run") as mock_run: # Mock subprocess to return empty mock_run.return_value.returncode = 0 mock_run.return_value.stdout = "" # Clear cache first service.clear_port_scan_cache() # First scan should invoke list_ports and subprocess ports1 = service.enumerate_ports() assert mock_run.call_count == 1 # Second scan within 2 seconds should return cached list without invoking subprocess ports2 = service.enumerate_ports() assert mock_run.call_count == 1 assert ports1 == ports2 # Clear cache and scan again should invoke subprocess service.clear_port_scan_cache() service.enumerate_ports() assert mock_run.call_count == 2