import json from unittest.mock import MagicMock import pytest import pygui.backend.controllers.session_controller as session_controller_module from pygui.backend.controllers.session_controller import SessionController from pygui.backend.models.frame import Frame @pytest.fixture def controller(): return SessionController() def test_initial_default(controller): assert controller.mode == "review" assert controller.state == "idle" assert not controller.recording 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 def test_leaving_live_mode_blanks_wafer_map(controller): """Stopping a live session (tab switch → setMode) must not leave stale sensor values on the wafer map — it should read as blank.""" frame_signal = MagicMock() controller.frameUpdated.connect(frame_signal) controller.setMode("live") controller._on_live_frame(Frame(seq=1, time=0.0, values=[100.0] * 22)) assert controller.sensorValues, "sanity: live frame should populate the map" controller.setMode("review") assert controller.sensorValues == [] assert controller.stats == {} assert frame_signal.called, "QML needs frameUpdated to repaint blank" def test_leaving_live_mode_stops_stream(controller): controller.setMode("live") controller.setMode("review") assert controller._reader is None assert not controller._repaint_timer.isActive() def test_compare_files_integration(controller): # Mock the comparisonResult signal mock_slot = MagicMock() controller.comparisonResult.connect(mock_slot) # Two runs with the same first 3 frames but run B has 2 extra trailing frames. controller.compareFiles("tests/fixtures/sample_stream.csv", "tests/fixtures/sample_stream_b.csv") # Verify that the comparison signal is emitted with a success dict assert mock_slot.call_count == 1 args = mock_slot.call_args[0][0] assert args["success"] is True assert "distance" in args assert "series_a" in args assert "series_b" in args # time_a/time_b expose per-frame timestamps in seconds for the mismatched run lengths. assert args["time_a"] == [0.0, 0.5, 1.0] assert args["time_b"] == [0.0, 0.5, 1.0, 1.5, 2.0] assert args["frame_count_a"] == 3 assert args["frame_count_b"] == 5 def test_compare_files_same_file_is_gated(controller): """Comparing a run against itself should be rejected, not silently DTW'd.""" mock_slot = MagicMock() controller.comparisonResult.connect(mock_slot) csv_path = "tests/fixtures/sample_stream.csv" controller.compareFiles(csv_path, csv_path) assert mock_slot.call_count == 1 args = mock_slot.call_args[0][0] assert args["success"] is False assert "error" in args def test_comparison_result_reaches_qml(qapp, controller): """The result dict must arrive in QML as a real JS object with fields. Signal(object) delivers Python dicts to QML as an opaque empty wrapper; the signal must be declared with a dict/QVariantMap signature. """ from PySide6.QtQml import QQmlApplicationEngine engine = QQmlApplicationEngine() engine.rootContext().setContextProperty("streamController", controller) engine.loadData(b""" import QtQuick Item { id: probe property bool gotSuccess: false property real gotDistance: -1 Connections { target: streamController function onComparisonResult(result) { probe.gotSuccess = result.success === true probe.gotDistance = result.distance !== undefined ? result.distance : -1 } } } """) assert engine.rootObjects(), "probe QML failed to load" root = engine.rootObjects()[0] controller.compareFiles("tests/fixtures/sample_stream.csv", "tests/fixtures/sample_stream_b.csv") qapp.processEvents() assert root.property("gotSuccess") is True assert root.property("gotDistance") >= 0 def test_recording_toggle(controller, tmp_path): csv_path = str(tmp_path / "rec" / "live_test.csv") controller.startRecording(csv_path, "SN123") assert controller.recording controller.stopRecording() assert not controller.recording # Header written with wafer serial content = open(csv_path).read() assert "Wafer ID=SN123" in content def test_resync_count_default(controller): # No active reader -> zero assert controller.resyncCount == 0 controller._reader = MagicMock(resync_count=7) assert controller.resyncCount == 7 controller._reader = None def test_split_data_integration(controller): # Mock the splitResult signal mock_slot = MagicMock() controller.splitResult.connect(mock_slot) csv_path = "tests/fixtures/sample_stream.csv" controller.splitData(csv_path, 149.0) # Verify that the split signal is emitted assert mock_slot.call_count == 1 args = mock_slot.call_args[0][0] assert args["success"] is True assert "segments" in args def test_export_segment_after_split(qapp, controller, tmp_path): import shutil src = tmp_path / "sample.csv" shutil.copyfile("tests/fixtures/sample_stream.csv", src) controller.splitData(str(src), 149.0) exported = MagicMock() controller.segmentExported.connect(exported) controller.exportSegment(0) assert exported.call_count == 1 result = exported.call_args[0][0] assert result["success"] is True out = result["path"] assert out.startswith(str(tmp_path)) from pygui.backend.data.data_records import read_data_records assert len(read_data_records(out)) > 0 def test_export_segment_without_split(qapp, controller): exported = MagicMock() controller.segmentExported.connect(exported) controller.exportSegment(0) result = exported.call_args[0][0] assert result["success"] is False def test_export_segment_bad_index(qapp, controller, tmp_path): import shutil src = tmp_path / "sample.csv" shutil.copyfile("tests/fixtures/sample_stream.csv", src) controller.splitData(str(src), 149.0) exported = MagicMock() controller.segmentExported.connect(exported) controller.exportSegment(99) assert exported.call_args[0][0]["success"] is False def test_reset_live_trend_sets_start_time_and_emits_reset(controller, monkeypatch): monkeypatch.setattr(session_controller_module.time, "monotonic", lambda: 100.0) reset_seen = MagicMock() controller.trendReset.connect(reset_seen) controller._reset_live_trend() assert controller._stream_start_time == 100.0 assert reset_seen.called def test_on_live_frame_emits_trend_delta_with_elapsed_seconds(controller, monkeypatch): controller._stream_start_time = 100.0 monkeypatch.setattr(session_controller_module.time, "monotonic", lambda: 102.5) delta_seen = MagicMock() controller.trendDelta.connect(delta_seen) frame = Frame(seq=1, time=102.5, values=[10.0, 20.0, 30.0]) controller._on_live_frame(frame) payload = json.loads(delta_seen.call_args[0][0]) assert len(payload) == 1 elapsed, avg = payload[0] assert elapsed == pytest.approx(2.5) assert avg == pytest.approx(20.0) # ---- edge-center stats exposure (ADR-0002) ---- def _write_f_family_csv(path): """22-sensor F-family stream: pair (0,18) has delta 1.0, all others 0.""" n = 22 values = [149.0] * n values[0] = 150.0 lines = [ "Wafer ID=F12", "Acquisition Date=03/26/2025", "Label," + ",".join(str(i + 1) for i in range(n)), "X (mm)," + ",".join(str(i) for i in range(n)), "Y (mm)," + ",".join(str(i) for i in range(n)), "data", "0.0," + ",".join(f"{v:.1f}" for v in values), ] path.write_text("\n".join(lines) + "\n", encoding="utf-8") def test_stats_include_edge_center_for_known_family(controller, tmp_path): csv = tmp_path / "f_run.csv" _write_f_family_csv(csv) controller.loadFile(str(csv)) s = controller.stats assert s["ecMaxDelta"] == pytest.approx(1.0) assert s["ecMaxEdgeIndex"] == 1 and s["ecMaxCenterIndex"] == 19 # 1-origin assert s["ecMaxEdgeValue"] == pytest.approx(150.0) assert s["ecMaxCenterValue"] == pytest.approx(149.0) assert s["ecMinDelta"] == pytest.approx(0.0) def test_stats_edge_center_skips_replaced_sensor(controller, tmp_path): csv = tmp_path / "f_run.csv" _write_f_family_csv(csv) controller.loadFile(str(csv)) controller.replaceSensor(0, 149.0) # replaced: pair (0,18) must be skipped assert controller.stats["ecMaxDelta"] == pytest.approx(0.0) def test_stats_edge_center_absent_for_unknown_or_short_family(controller): controller.loadFile("tests/fixtures/sample_stream.csv") # A12, 3 sensors assert "ecMaxDelta" not in controller.stats