chore: apply linting fixes and expand ruff coverage to test suite
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
"""Tests for src/pygui/backend/cluster_average.py."""
|
||||
import math
|
||||
import pytest
|
||||
|
||||
from pygui.backend.cluster_average import average_clusters, group_sensors_by_radius
|
||||
from pygui.backend.wafer.zwafer_models import Sensor
|
||||
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
"""Tests for src/pygui/backend/comparison.py."""
|
||||
|
||||
import numpy as np
|
||||
from pygui.backend.comparison import compare_runs
|
||||
|
||||
|
||||
|
||||
@@ -37,4 +37,4 @@ def test_recorded_data_rows_are_readable(tmp_path):
|
||||
assert records[0].time == 0.0
|
||||
assert records[0].values == [149.0, 148.0]
|
||||
assert records[1].time == 0.5
|
||||
assert records[1].values == [149.5, 148.5]
|
||||
assert records[1].values == [149.5, 148.5]
|
||||
|
||||
@@ -1,13 +1,14 @@
|
||||
"""Tests for serialcomm/data_parser.py binary parsing pipeline."""
|
||||
|
||||
import pytest
|
||||
|
||||
from pygui.serialcomm.data_parser import (
|
||||
_convert_hex_to_temp,
|
||||
convert_to_temperatures,
|
||||
csv_column_count,
|
||||
parse_binary_data,
|
||||
convert_to_temperatures,
|
||||
remove_trailing_zeros,
|
||||
save_to_csv,
|
||||
_convert_hex_to_temp,
|
||||
)
|
||||
|
||||
# ── csv_column_count ──────────────────────────────────────────────────────────
|
||||
@@ -271,7 +272,7 @@ class TestSaveToCsv:
|
||||
result = save_to_csv(data, "P", "P00001", str(tmp_path))
|
||||
assert result is not None
|
||||
headers = open(result).readline().strip().split(",")
|
||||
assert len(headers) == 60, f"data width 60 should win over display 48"
|
||||
assert len(headers) == 60, "data width 60 should win over display 48"
|
||||
assert headers[0] == "Sensor1"
|
||||
assert headers[-1] == "Sensor60"
|
||||
|
||||
|
||||
@@ -16,4 +16,4 @@ def test_reads_rows_after_data_marker(tmp_path):
|
||||
assert len(recs) == 2
|
||||
assert recs[0].time == 0.0
|
||||
assert recs[0].values == [149.0, 148.5]
|
||||
assert recs[1].values == [149.2, 148.4]
|
||||
assert recs[1].values == [149.2, 148.4]
|
||||
|
||||
@@ -1,10 +1,13 @@
|
||||
import pytest
|
||||
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()
|
||||
@@ -16,8 +19,8 @@ def mock_settings():
|
||||
|
||||
@pytest.fixture
|
||||
def controller(mock_settings):
|
||||
import tempfile
|
||||
import shutil
|
||||
import tempfile
|
||||
temp_dir = tempfile.mkdtemp()
|
||||
yield DeviceController(mock_settings, temp_dir)
|
||||
shutil.rmtree(temp_dir)
|
||||
@@ -81,7 +84,7 @@ 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 == ""
|
||||
@@ -142,7 +145,7 @@ def test_parse_and_save_data_and_get_chart_data(controller):
|
||||
|
||||
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
|
||||
@@ -159,16 +162,16 @@ def test_logging_handler(controller):
|
||||
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
|
||||
|
||||
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)
|
||||
@@ -187,19 +190,19 @@ def test_detect_wafer_clears_old_session(controller):
|
||||
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:
|
||||
|
||||
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]
|
||||
@@ -227,6 +230,6 @@ def test_device_service_port_caching(controller):
|
||||
|
||||
# Clear cache and scan again should invoke subprocess
|
||||
service.clear_port_scan_cache()
|
||||
ports3 = service.enumerate_ports()
|
||||
service.enumerate_ports()
|
||||
assert mock_run.call_count == 2
|
||||
|
||||
|
||||
@@ -1,30 +1,31 @@
|
||||
from pygui.backend.models.frame import Frame
|
||||
from pygui.backend.models.frame_player import FramePlayer
|
||||
|
||||
|
||||
def frames(n):
|
||||
return [Frame(seq=i, time=float(i), values=[149.0 + i])for i in range(n)]
|
||||
|
||||
def test_loads_and_reports_total():
|
||||
p = FramePlayer(); p.load(frames(3))
|
||||
assert p.total == 3 and p.index == 0
|
||||
assert p.total == 3 and p.index == 0
|
||||
current = p.current()
|
||||
assert current is not None
|
||||
assert current.values == [149.0]
|
||||
|
||||
|
||||
def test_step_forward_and_back_clamps():
|
||||
p = FramePlayer(); p.load(frames(3))
|
||||
assert p.step(1).index == 1
|
||||
assert p.step(1).index == 2
|
||||
assert p.step(1).index == 2
|
||||
assert p.step(-5).index == 0
|
||||
|
||||
|
||||
def test_seek():
|
||||
p = FramePlayer(); p.load(frames(5))
|
||||
assert p.seek(3).index == 3
|
||||
assert p.seek(99).index == 4
|
||||
|
||||
|
||||
def test_at_end():
|
||||
p = FramePlayer(); p.load(frames(2))
|
||||
assert not p.at_end
|
||||
p.seek(1)
|
||||
assert p.at_end
|
||||
assert p.at_end
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
import math
|
||||
|
||||
import pytest
|
||||
from pygui.backend.models.frame_stats import compute_stats, Stats
|
||||
|
||||
from pygui.backend.models.frame_stats import Stats, compute_stats
|
||||
|
||||
|
||||
def test_basic_stat():
|
||||
@@ -20,3 +22,9 @@ def test_empty_values_returns_zeros():
|
||||
|
||||
def test_ignores_nan():
|
||||
s = compute_stats([149.0, float("nan"), 151.0])
|
||||
assert s.min == 149.0 and s.min_index == 0
|
||||
assert s.max == 151.0 and s.max_index == 2
|
||||
assert s.diff == pytest.approx(2.0)
|
||||
assert s.avg == pytest.approx(150.0)
|
||||
assert s.sigma == pytest.approx(1.0)
|
||||
assert s.three_sigma == pytest.approx(3.0)
|
||||
|
||||
@@ -1,8 +1,6 @@
|
||||
"""Tests for src/pygui/backend/visualization/graph_view.py."""
|
||||
import json
|
||||
from unittest.mock import MagicMock, PropertyMock, patch
|
||||
|
||||
import pytest
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
|
||||
def _make_plot_widget():
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
"""Tests for src/pygui/backend/rbf_heatmap.py."""
|
||||
import numpy as np
|
||||
from pygui.backend.visualization.rbf_heatmap import interpolate_field, BACKEND
|
||||
|
||||
from pygui.backend.visualization.rbf_heatmap import BACKEND, interpolate_field
|
||||
|
||||
|
||||
def test_field_shape_and_interpolates_at_sites():
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
"""Tests for src/pygui/backend/sensor_editor.py."""
|
||||
import pytest
|
||||
from pygui.backend.models.sensor_editor import SensorEditor
|
||||
|
||||
|
||||
|
||||
@@ -1,7 +1,10 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from pygui.backend.controllers.session_controller import SessionController
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def controller():
|
||||
return SessionController()
|
||||
@@ -9,7 +12,7 @@ def controller():
|
||||
def test_initial_default(controller):
|
||||
assert controller.mode == "review"
|
||||
assert controller.state == "idle"
|
||||
assert controller.recording == False
|
||||
assert not controller.recording
|
||||
|
||||
def test_playback_flow(controller):
|
||||
controller.loadFile("tests/fixtures/sample_stream.csv")
|
||||
@@ -223,10 +226,10 @@ Item {
|
||||
def test_recording_toggle(controller, tmp_path):
|
||||
csv_path = str(tmp_path / "rec" / "live_test.csv")
|
||||
controller.startRecording(csv_path, "SN123")
|
||||
assert controller.recording == True
|
||||
assert controller.recording
|
||||
|
||||
controller.stopRecording()
|
||||
assert controller.recording == False
|
||||
assert not controller.recording
|
||||
|
||||
# Header written with wafer serial
|
||||
content = open(csv_path).read()
|
||||
@@ -252,4 +255,4 @@ def test_split_data_integration(controller):
|
||||
assert mock_slot.call_count == 1
|
||||
args = mock_slot.call_args[0][0]
|
||||
assert args["success"] is True
|
||||
assert "segments" in args
|
||||
assert "segments" in args
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
from pygui.backend.models.frame import Frame
|
||||
from pygui.backend.models.threshold_classifier import ThresholdConfig, BAND_HIGH, BAND_IN, BAND_LOW
|
||||
from pygui.backend.models.session_model import SessionModel, SessionUpdate
|
||||
|
||||
from pygui.backend.models.session_model import SessionModel
|
||||
from pygui.backend.models.threshold_classifier import BAND_HIGH, BAND_IN, BAND_LOW, ThresholdConfig
|
||||
|
||||
|
||||
def test_manual_banding_and_stats():
|
||||
@@ -12,17 +11,16 @@ def test_manual_banding_and_stats():
|
||||
assert update.stats.avg == 149.0
|
||||
assert (update.target, update.margin) == (149.0, 1.0)
|
||||
assert update.state in ("idle", "ramp", "set")
|
||||
|
||||
|
||||
def test_auto_banding_uses_mean_and_sigma():
|
||||
model = SessionModel(ThresholdConfig(auto=True))
|
||||
# Tight cluster around 149 -> small σ -> the 200.0 outlier reads HIGH
|
||||
update = model.process(Frame(seq=0, time=0.0, values=[149.0, 149.0, 149.0, 200.0]))
|
||||
assert update.bands[3] == BAND_HIGH
|
||||
assert update.target == update.stats.avg
|
||||
|
||||
|
||||
def test_set_thresholds_reband_on_next_frame():
|
||||
m = SessionModel(ThresholdConfig(set_point=149.0, margin=1.0, auto=False))
|
||||
m.set_thresholds(ThresholdConfig(set_point=149.0, margin=5.0, auto=False))
|
||||
upd = m.process(Frame(seq=0, time=0.0, values=[151.0]))
|
||||
assert upd.bands == [BAND_IN]
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
from pygui.backend.models.stability_detector import (
|
||||
StabilityDetector, STATE_IDLE, STATE_RAMP, STATE_SET,
|
||||
STATE_IDLE,
|
||||
STATE_RAMP,
|
||||
STATE_SET,
|
||||
StabilityDetector,
|
||||
)
|
||||
|
||||
|
||||
SET_POINT = 149.0
|
||||
|
||||
def make():
|
||||
@@ -11,16 +13,16 @@ def make():
|
||||
def test_idle_when_cold():
|
||||
d = make()
|
||||
assert d.update(avg=25.0, time=0.0, set_point=SET_POINT) == STATE_IDLE
|
||||
|
||||
|
||||
def test_ramp_while_far_from_setpoint():
|
||||
d = make()
|
||||
d.update(avg=100.0, time=0.0, set_point=SET_POINT)
|
||||
|
||||
|
||||
def test_ramp_until_settle_time_elapses():
|
||||
d = make()
|
||||
assert d.update(avg=149.2, time=0.0, set_point=SET_POINT) == STATE_RAMP
|
||||
assert d.update(avg=148.9, time=0.0, set_point=SET_POINT) == STATE_RAMP
|
||||
|
||||
|
||||
def test_set_after_holding_near_setpoint():
|
||||
d = make()
|
||||
d.update(avg=149.2, time=0.0, set_point=SET_POINT)
|
||||
@@ -32,4 +34,4 @@ def test_back_to_ramp_on_disturbance():
|
||||
d = make()
|
||||
for t in (0.0, 1.0, 2.5):
|
||||
d.update(149.0, t, set_point=SET_POINT)
|
||||
assert d.update(avg=160.0, time=3.0, set_point=SET_POINT) == STATE_RAMP
|
||||
assert d.update(avg=160.0, time=3.0, set_point=SET_POINT) == STATE_RAMP
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
import time
|
||||
import time
|
||||
|
||||
from pygui.backend.models.frame import Frame
|
||||
from pygui.serialcomm.stream_reader import StreamReader
|
||||
|
||||
|
||||
class FakeTransport:
|
||||
"""Yield canned data then blocks (returns b'')."""
|
||||
def __init__(self, lines):
|
||||
@@ -28,7 +30,7 @@ class FakeTransport:
|
||||
self._pos = idx + 1
|
||||
return res
|
||||
def close(self): self.closed = True
|
||||
|
||||
|
||||
def parse_line(raw: str, seq: int) -> Frame:
|
||||
parts = [float(x) for x in raw.split(",")]
|
||||
return Frame(seq = seq, time = parts[0], values = parts[1:])
|
||||
@@ -38,10 +40,10 @@ def test_reads_frames_and_counts_errors():
|
||||
transport = FakeTransport([b"0.0,149,148\n", b"garbage\n", b"0.5,150,149\n"])
|
||||
r = StreamReader(transport, parse_line,
|
||||
on_frame = got.append, on_error = lambda e: errors.append(e))
|
||||
|
||||
|
||||
r.start()
|
||||
time.sleep(0.1)
|
||||
r.stop()
|
||||
assert [f.values for f in got] == [[149.0, 148.0], [150.0, 149.0]]
|
||||
assert r.error_count == 1
|
||||
assert transport.closed
|
||||
assert transport.closed
|
||||
|
||||
@@ -1,8 +1,13 @@
|
||||
import math
|
||||
|
||||
from pygui.backend.models.threshold_classifier import (
|
||||
ThresholdConfig, classify, classify_all, resolve_bounds,
|
||||
BAND_IN, BAND_HIGH, BAND_LOW
|
||||
BAND_HIGH,
|
||||
BAND_IN,
|
||||
BAND_LOW,
|
||||
ThresholdConfig,
|
||||
classify,
|
||||
classify_all,
|
||||
resolve_bounds,
|
||||
)
|
||||
|
||||
|
||||
@@ -11,19 +16,18 @@ def test_classify_about_target_margin():
|
||||
assert classify(149.9, target=149.0, margin=1.0) ==BAND_IN # within
|
||||
assert classify(150.5, target=149.0, margin=1.0) ==BAND_HIGH # 1.5 above
|
||||
assert classify(147.5, target=149.0, margin=1.0) ==BAND_LOW # 1.5 below
|
||||
|
||||
|
||||
|
||||
|
||||
def test_manual_bounds_use_set_point_and_margin():
|
||||
cfg = ThresholdConfig(set_point=149.0, margin=1.0, auto=False)
|
||||
assert resolve_bounds([200.0, 0.0], cfg) == (149.0, 1.0)
|
||||
|
||||
|
||||
def test_auto_bounds_use_mean_and_sigma():
|
||||
cfg = ThresholdConfig(auto=True)
|
||||
target, margin = resolve_bounds([148.0, 150.0, 149.0], cfg)
|
||||
assert target == 149.0
|
||||
assert margin == math.sqrt(2 / 3)
|
||||
|
||||
assert margin == math.sqrt(2 / 3)
|
||||
|
||||
def test_classify_all_manual():
|
||||
cfg = ThresholdConfig(set_point=149.0, margin=1.0, auto=False)
|
||||
assert classify_all([149.0, 151.0, 147.0], cfg) == [BAND_IN, BAND_HIGH, BAND_LOW]
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
"""Tests for src/pygui/backend/wafer_layouts.py."""
|
||||
import pytest
|
||||
from pygui.backend.wafer.wafer_layouts import load_layout, available_families
|
||||
|
||||
from pygui.backend.wafer.wafer_layouts import available_families, load_layout
|
||||
from pygui.backend.wafer.zwafer_models import Sensor
|
||||
|
||||
|
||||
|
||||
@@ -1,9 +1,8 @@
|
||||
"""Tests for src/pygui/backend/visualization/wafer_map_item.py."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# Use the session-scoped qapp fixture from conftest.py so Qt Property/Signal
|
||||
# descriptors work properly.
|
||||
pytestmark = pytest.mark.usefixtures("qapp")
|
||||
|
||||
Reference in New Issue
Block a user