jack
135 lines
4.3 KiB
Python
135 lines
4.3 KiB
Python
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
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 controller.recording == False
|
|
|
|
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)
|
|
|
|
# Use sample_stream.csv for both arguments to test identical files compare
|
|
csv_path = "tests/fixtures/sample_stream.csv"
|
|
controller.compareFiles(csv_path, csv_path)
|
|
|
|
# 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
|
|
assert args["frame_offset"] == 0
|
|
|
|
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]
|
|
|
|
csv_path = "tests/fixtures/sample_stream.csv"
|
|
controller.compareFiles(csv_path, csv_path)
|
|
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 == True
|
|
|
|
controller.stopRecording()
|
|
assert controller.recording == False
|
|
|
|
# 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 |