from unittest.mock import MagicMock import pytest from pygui.backend.controllers.session_controller import SessionController @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_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