jack
9e3bec9031
- Parse F1 debug data via family-independent temp formula (bits 4-14 scaled by 2**(i-8), sign at bit 0) - Save sensor+debug temps side-by-side to debug_info.csv - Wire READ DEBUG button in StatusActionsPanel - Rename ReplayChart → RunChart; move into GraphTab - Extract PanelBox/SectionTitle/PanelSlider in ReadoutPanel to cut ~300 lines of duplicated styling - Add rightRailWidth token to Theme for consistent rail sizing - Wrap ReadoutPanel in ScrollView so Thresholds card stays reachable when E-C delta rows appear
303 lines
10 KiB
Python
303 lines
10 KiB
Python
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 _make_p_family_bytes(value: int = 0x0100) -> bytes:
|
|
"""Synthetic single-block P-family binary: 244 valid words + 12 overhead."""
|
|
data = bytearray()
|
|
for i in range(256):
|
|
word = value if i < 244 else 0
|
|
data += word.to_bytes(2, byteorder="little")
|
|
return bytes(data)
|
|
|
|
|
|
def test_debug_worker_decodes_and_saves_csv(controller, tmp_path):
|
|
controller._last_wafer_info = {"familyCode": "P"}
|
|
controller._save_data_dir = str(tmp_path)
|
|
controller._service.read_wafer_data = MagicMock(
|
|
side_effect=[_make_p_family_bytes(), _make_p_family_bytes()]
|
|
)
|
|
|
|
captured = {}
|
|
controller._debugFinished.connect(captured.update)
|
|
|
|
controller._debug_worker("COM1")
|
|
|
|
assert captured.get("success") is True
|
|
assert captured.get("sensor_bytes") == 512
|
|
assert captured.get("debug_bytes") == 512
|
|
assert "csv_path" in captured
|
|
from pathlib import Path
|
|
assert Path(captured["csv_path"]).exists()
|
|
assert Path(captured["csv_path"]).name == "debug_info.csv"
|
|
|
|
|
|
def test_debug_worker_reports_error_when_csv_save_fails(controller, tmp_path):
|
|
controller._last_wafer_info = {"familyCode": "P"}
|
|
# Point save dir somewhere save_debug_csv cannot write to, so it returns None.
|
|
controller._save_data_dir = str(tmp_path / "debug_info.csv") # a file, not a dir
|
|
(tmp_path / "debug_info.csv").write_text("occupied")
|
|
controller._service.read_wafer_data = MagicMock(
|
|
side_effect=[_make_p_family_bytes(), _make_p_family_bytes()]
|
|
)
|
|
|
|
captured = {}
|
|
controller._debugFinished.connect(captured.update)
|
|
|
|
controller._debug_worker("COM1")
|
|
|
|
assert captured.get("success") is not True
|
|
assert "error" in captured
|
|
|
|
|
|
def test_debug_handler_logs_csv_path(controller):
|
|
controller._handle_debug_finished({
|
|
"success": True,
|
|
"sensor_bytes": 512,
|
|
"debug_bytes": 512,
|
|
"csv_path": "/tmp/debug_info.csv",
|
|
})
|
|
assert "/tmp/debug_info.csv" in controller.activityLog
|
|
|
|
|
|
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
|
|
|
|
|
|
|
|
def test_export_dir_created_under_save_data_dir(controller):
|
|
from pathlib import Path
|
|
|
|
result = controller.exportDir()
|
|
|
|
assert result == str(Path(controller.saveDataDir) / "Export")
|
|
assert Path(result).is_dir()
|