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_graph_data_empty_before_load(controller): assert controller.graphSensorNames == "" assert controller.graphSeriesJson == "[]" def test_graph_data_after_load(controller): controller.loadFile("tests/fixtures/sample_stream.csv") assert controller.graphSensorNames == "Sensor 1,Sensor 2,Sensor 3" series = json.loads(controller.graphSeriesJson) assert series == [ [149.0, 149.2, 149.1], [148.5, 148.4, 148.6], [150.6, 150.7, 150.5], ] def test_graph_data_cleared_after_unload(controller): controller.loadFile("tests/fixtures/sample_stream.csv") controller.unloadFile() assert controller.graphSensorNames == "" assert controller.graphSeriesJson == "[]" 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_binary_frame_decode_bounded_to_loaded_layout(controller, monkeypatch): """Regression for phantom sensor #239 polluting MIN/MAX/AVG: the wire payload carries a fixed-width channel array (up to 244 slots) wider than any single wafer family's real sensor count. Family "X" (xwafer.yaml, the only family live streaming is supported on) has 80 real sensors — everything past that in the payload is an unpopulated hardware slot and must never reach Frame.values.""" fake_transport = MagicMock() fake_transport.read.return_value = b"" monkeypatch.setattr( "pygui.serialcomm.serial_port.SerialPort.open_port", lambda *a, **k: fake_transport, ) controller.startStream("COM_FAKE", "X") try: assert len(controller._sensors) == 80 # sanity: xwafer.yaml loaded parse = controller._reader._parse payload = bytes(244 * 2) # full-width wire payload, well beyond the real 80 frame = parse(payload, seq=1) assert len(frame.values) == 80 finally: controller.stopStream() def test_live_error_surfaces_message_and_stops_stream(controller): """A fatal StreamReader error must reach QML with its message (for the Activity Log) and must tear the stream down cleanly.""" controller.setMode("live") controller._reader = MagicMock() error_seen = MagicMock() controller.liveError.connect(error_seen) controller._on_live_error("Binary stream resync failed: no sync marker") error_seen.assert_called_once_with("Binary stream resync failed: no sync marker") assert controller._reader is None def test_start_stream_refuses_non_x_family(controller, monkeypatch): """C# parity: only X-family wafer firmware supports live streaming. Family A produced ~10s of unrelated ASCII output, not real telemetry — startStream must refuse before ever opening the port.""" open_port = MagicMock() monkeypatch.setattr( "pygui.serialcomm.serial_port.SerialPort.open_port", open_port ) error_seen = MagicMock() controller.liveError.connect(error_seen) controller.startStream("COM_FAKE", "A") open_port.assert_not_called() error_seen.assert_called_once() assert "X-family" in error_seen.call_args[0][0] assert controller._reader is None def test_start_stream_allows_x_family(controller, monkeypatch): fake_transport = MagicMock() fake_transport.read.return_value = b"" monkeypatch.setattr( "pygui.serialcomm.serial_port.SerialPort.open_port", lambda *a, **k: fake_transport, ) error_seen = MagicMock() controller.liveError.connect(error_seen) controller.startStream("COM_FAKE", "X") try: error_seen.assert_not_called() assert controller._reader is not None finally: controller.stopStream() def test_stop_stream_skips_d2s_write_when_reader_already_exited(controller): """Reader's own worker thread can exit and close the transport (e.g. after a resync give-up) before stopStream() runs on the main thread. Writing D2S to that already-closed transport is a guaranteed EBADF — skip it instead of racing the close.""" controller.setMode("live") fake_reader = MagicMock() fake_reader._thread = MagicMock() fake_reader._thread.is_alive.return_value = False fake_transport = MagicMock() fake_transport.is_open = True fake_reader._transport = fake_transport controller._reader = fake_reader controller.stopStream() fake_transport.write.assert_not_called() 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