jack
255 lines
9.4 KiB
Python
255 lines
9.4 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_load_file_emits_error_for_empty_recording(controller, tmp_path):
|
|
"""A recording with a valid header but zero data rows (e.g. Record was
|
|
started/stopped without any live frames arriving) must surface a clear
|
|
loadFileError instead of silently doing nothing."""
|
|
empty_recording = tmp_path / "live_A00001_20260706_120427.csv"
|
|
empty_recording.write_text(
|
|
"Wafer ID=A00001\n"
|
|
"Acquisition Date=07/06/2026\n"
|
|
"Label,1,2\n"
|
|
"X (mm),0.0,10.0\n"
|
|
"Y (mm),10.0,-10.0\n"
|
|
"data\n"
|
|
)
|
|
|
|
mock_slot = MagicMock()
|
|
controller.loadFileError.connect(mock_slot)
|
|
controller.loadFile(str(empty_recording))
|
|
|
|
assert mock_slot.call_count == 1
|
|
assert "no recorded frames" in mock_slot.call_args[0][0]
|
|
assert controller.loadedFile == ""
|
|
|
|
def test_load_file_switches_to_review_mode(controller):
|
|
"""Loading a file for playback must always land in review mode, even if
|
|
the controller was left in live mode from a previous session — otherwise
|
|
a successful load can be invisible behind a stale Live toggle."""
|
|
controller._mode = "live"
|
|
controller.loadFile("tests/fixtures/sample_stream.csv")
|
|
assert controller.mode == "review"
|
|
|
|
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_compare_files_uses_full_run_not_first_100_frames(controller, tmp_path):
|
|
"""compareFiles must score the whole run, not just an initial 100-frame window.
|
|
|
|
Regression test: max_frames used to be hard-capped at 100, so a >100-frame
|
|
run's soak/cool phases were silently excluded from the DTW comparison.
|
|
"""
|
|
num_frames = 150
|
|
csv_path = tmp_path / "long_run.csv"
|
|
lines = ["Wafer ID=A12", "Acquisition Date=03/26/2025", "Label,1,2,3",
|
|
"X (mm),0,10,-10", "Y (mm),10,-10,-10", "data"]
|
|
for i in range(num_frames):
|
|
t = i * 0.5
|
|
lines.append(f"{t},{149.0 + i * 0.01},{148.5 + i * 0.01},{150.6 + i * 0.01}")
|
|
csv_path.write_text("\n".join(lines) + "\n")
|
|
|
|
mock_slot = MagicMock()
|
|
controller.comparisonResult.connect(mock_slot)
|
|
controller.compareFiles(str(csv_path), str(csv_path))
|
|
|
|
assert mock_slot.call_count == 1
|
|
result = mock_slot.call_args[0][0]
|
|
assert result["success"] is True
|
|
assert len(result["series_a"]) == num_frames
|
|
assert len(result["series_b"]) == num_frames
|
|
|
|
def test_compare_files_rejects_mismatched_wafer_types(controller, tmp_path):
|
|
"""compareFiles must refuse to compare two different wafer families.
|
|
|
|
Otherwise the overlap view silently truncates to min(sensor_count) and
|
|
pairs up unrelated physical sensors between the two wafer types.
|
|
"""
|
|
header = ["Wafer ID=A12", "Acquisition Date=03/26/2025", "Label,1,2,3",
|
|
"X (mm),0,10,-10", "Y (mm),10,-10,-10", "data",
|
|
"0.0,149.0,148.5,150.6"]
|
|
file_a = tmp_path / "A00055-20250326.csv"
|
|
file_b = tmp_path / "X00108-20250326.csv"
|
|
file_a.write_text("\n".join(header) + "\n")
|
|
file_b.write_text("\n".join(header) + "\n")
|
|
|
|
mock_slot = MagicMock()
|
|
controller.comparisonResult.connect(mock_slot)
|
|
controller.compareFiles(str(file_a), str(file_b))
|
|
|
|
assert mock_slot.call_count == 1
|
|
result = mock_slot.call_args[0][0]
|
|
assert result["success"] is False
|
|
assert "A" in result["error"] and "X" in result["error"]
|
|
|
|
def test_compare_files_rejects_recording_vs_read_mem(controller, tmp_path):
|
|
"""A live_-prefixed recording must not be compared against a read-memory
|
|
dump. Regression: stem[0] on "live_P0001_..." is "L", which used to be
|
|
misreported as a wafer-type mismatch ("L vs P") instead of this clearer
|
|
recording-vs-read-mem error.
|
|
"""
|
|
header = ["Wafer ID=P0001", "Acquisition Date=03/26/2025", "Label,1,2,3",
|
|
"X (mm),0,10,-10", "Y (mm),10,-10,-10", "data",
|
|
"0.0,149.0,148.5,150.6"]
|
|
recording = tmp_path / "live_P0001_20260706_143022.csv"
|
|
read_mem = tmp_path / "P0001-20260706_143500.csv"
|
|
recording.write_text("\n".join(header) + "\n")
|
|
read_mem.write_text("\n".join(header) + "\n")
|
|
|
|
mock_slot = MagicMock()
|
|
controller.comparisonResult.connect(mock_slot)
|
|
controller.compareFiles(str(recording), str(read_mem))
|
|
|
|
assert mock_slot.call_count == 1
|
|
result = mock_slot.call_args[0][0]
|
|
assert result["success"] is False
|
|
assert result["error"] == "Cannot compare a recording file with a read-memory file"
|
|
|
|
def test_compare_files_allows_two_recordings_of_same_wafer_type(controller, tmp_path):
|
|
"""Two live recordings of the same wafer family should compare fine —
|
|
the "live_" prefix must not be mistaken for the wafer type letter."""
|
|
header = ["Wafer ID=P0001", "Acquisition Date=03/26/2025", "Label,1,2,3",
|
|
"X (mm),0,10,-10", "Y (mm),10,-10,-10", "data",
|
|
"0.0,149.0,148.5,150.6"]
|
|
file_a = tmp_path / "live_P0001_20260706_143022.csv"
|
|
file_b = tmp_path / "live_P0002_20260706_150000.csv"
|
|
file_a.write_text("\n".join(header) + "\n")
|
|
file_b.write_text("\n".join(header) + "\n")
|
|
|
|
mock_slot = MagicMock()
|
|
controller.comparisonResult.connect(mock_slot)
|
|
controller.compareFiles(str(file_a), str(file_b))
|
|
|
|
assert mock_slot.call_count == 1
|
|
result = mock_slot.call_args[0][0]
|
|
assert result["success"] is True
|
|
|
|
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 |