jack
190 lines
6.7 KiB
Python
190 lines
6.7 KiB
Python
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)
|
|
|
|
def test_fresh_session_on_startup(mock_settings):
|
|
import tempfile
|
|
import shutil
|
|
|
|
# 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") as mock_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]
|
|
|