From c3e7b9a2cc2670e6a4cf83e5e00f6aabc0a5918b Mon Sep 17 00:00:00 2001 From: Dawn Date: Tue, 8 Sep 2026 14:56:16 +0200 Subject: [PATCH] test: cover the sample camera pacing, the newest-frame drain and the frame slot CI's diff-coverage gate sat at 29% because the ZMQ fixes live in code no test reached: the subscriber's socket setup, _emit_image/notify_frame_ displayed, _recv_latest and the run() frame path, plus the main window's frame slot. The subscriber now runs against a real inproc socket (the drain and the in-flight cap are the fix, so a mock would prove nothing); run() is driven with a stubbed _recv_latest that hands over one JPEG and then stops. The busy overlay gets a check that cooling is the only blue busy state. The subscriber fixture takes qapp so no QImage is built before the QApplication exists. Co-Authored-By: Claude Fable 5.1 --- tests/unit/gui/test_busy_overlay.py | 19 ++++ tests/unit/gui/test_main_window.py | 31 +++++++ tests/unit/gui/test_prediction_subscriber.py | 93 ++++++++++++++++++++ 3 files changed, 143 insertions(+) diff --git a/tests/unit/gui/test_busy_overlay.py b/tests/unit/gui/test_busy_overlay.py index 6a71154b..e37f8234 100644 --- a/tests/unit/gui/test_busy_overlay.py +++ b/tests/unit/gui/test_busy_overlay.py @@ -50,3 +50,22 @@ def test_auto_centering_flag_overrides_alignment_gate(): auto_centering=True, ) assert viewing is not None and viewing.text.startswith("Viewing mode") + + +def test_robot_cooling_is_the_only_blue_busy_state(): + from types import SimpleNamespace + + from aare.gui.styles import BUSY_BLUE, BUSY_PSI_RED + + def _style(activity: str): + tell = SimpleNamespace(activity=SimpleNamespace(value=activity)) + style = build_busy_overlay_style(is_busy=True, tell_state=tell) # type: ignore[arg-type] + assert style is not None + return style + + assert _style("cooling").text == "ROBOT COOLING" + assert _style("cooling").badge_bg == BUSY_BLUE + # Every other busy state shares the one PSI red. + for activity in ("mounting", "unmounting", "drying", "unknown"): + assert _style(activity).badge_bg == BUSY_PSI_RED + assert _style("unknown").text == "BEAMLINE BUSY" diff --git a/tests/unit/gui/test_main_window.py b/tests/unit/gui/test_main_window.py index 4020f28a..5c07d5c0 100644 --- a/tests/unit/gui/test_main_window.py +++ b/tests/unit/gui/test_main_window.py @@ -606,3 +606,34 @@ def test_nonstaff_beamline_gate_popups(qtbot, mock_ui_state): with patch("aare.gui.main_window.QToolTip") as tip: qtbot.mousePress(aux_bar, Qt.MouseButton.LeftButton, pos=aux_bar.tabRect(1).center()) assert tip.showText.called, "aux-puck tab click must explain the lock" + + +def test_sample_camera_frame_paints_visible_views_and_acks(qtbot, mock_ui_state, monkeypatch): + from PySide6.QtGui import QImage + + with ( + patch("requests.get"), + patch("aare.gui.main_window.DAQWorker"), + patch("aare.gui.main_window.PredictionSubscriber"), + patch("aare.gui.main_window.VideoThread"), + patch("aare.gui.main_window.JFJochDBusClient"), + patch("aare.gui.main_window.jwt.decode") as mock_jwt, + ): + mock_jwt.return_value = {"sub": "testuser", "staff": True, "pgroups": ["p123"], "session": 15} + win = _make_window(qtbot) + + views = (win.sample_camera, win.compact_sample_camera, win.portrait_sample_camera) + # Nothing is shown in tests, so pretend every view is on screen to reach + # the paint branch. + monkeypatch.setattr(type(win.sample_camera), "isVisible", lambda self: True) + win.prediction_thread = MagicMock() + + win._on_sample_camera_frame(QImage(4, 6, QImage.Format.Format_RGB888)) + + assert all(v.pixmap_item.pixmap().width() == 4 for v in views) + # The ack is what paces the subscriber; it must fire after every frame. + win.prediction_thread.notify_frame_displayed.assert_called_once() + + # No subscriber (GUI started without a sample feed): the slot must not blow up. + win.prediction_thread = None + win._on_sample_camera_frame(QImage(4, 6, QImage.Format.Format_RGB888)) diff --git a/tests/unit/gui/test_prediction_subscriber.py b/tests/unit/gui/test_prediction_subscriber.py index c57552fb..092dfed4 100644 --- a/tests/unit/gui/test_prediction_subscriber.py +++ b/tests/unit/gui/test_prediction_subscriber.py @@ -25,3 +25,96 @@ def test_json_dict_is_parsed(): def test_json_non_dict_is_rejected(): assert _parse(b"[1, 2]") is None assert _parse(b"") is None + + +# --- Frame pacing and the multipart-safe conflate ------------------------- +# Real socket on an inproc endpoint: the drain and the in-flight cap are the +# fix, so they run against libzmq rather than a mock. + +import time + +import cv2 +import numpy as np +import pytest +import zmq +from PySide6.QtGui import QImage + + +@pytest.fixture +def sub(qapp): + s = PredictionSubscriber("inproc://samcam-test") + yield s + # run() nulls both on its own cleanup; only tear down what is left. + if s._sock is not None: + s._sock.close(0) + if s._ctx is not None: + s._ctx.term() + + +def test_emit_image_caps_frames_in_flight(sub): + image = QImage(4, 4, QImage.Format.Format_RGB888) + for _ in range(sub.MAX_FRAMES_IN_FLIGHT + 1): + sub._emit_image(image) + assert sub._frames_in_flight == sub.MAX_FRAMES_IN_FLIGHT + assert sub._dropped == 1 + + for _ in range(sub.MAX_FRAMES_IN_FLIGHT + 1): + sub.notify_frame_displayed() + assert sub._frames_in_flight == 0 # floors at zero, never negative + + +def test_recv_latest_keeps_only_the_newest_message(sub): + pub = sub._ctx.socket(zmq.PUB) + pub.bind("inproc://samcam-test") + try: + # Slow joiner: publish probes until the subscription has propagated. + deadline = time.monotonic() + 5 + while True: + pub.send_multipart([b"probe"]) + try: + sub._sock.recv_multipart(zmq.NOBLOCK) + break + except zmq.Again: + assert time.monotonic() < deadline, "subscription never propagated" + time.sleep(0.01) + while True: # flush any extra probes + try: + sub._sock.recv_multipart(zmq.NOBLOCK) + except zmq.Again: + break + sub._dropped = 0 + + for payload in (b"old", b"stale", b"newest"): + pub.send_multipart([b"hdr", payload]) + assert sub._recv_latest() == [b"hdr", b"newest"] + assert sub._dropped == 2 + finally: + pub.close(0) + + +def test_run_decodes_one_frame_and_reports_the_drops(sub, qtbot): + ok, jpeg = cv2.imencode(".jpg", np.zeros((4, 6, 3), dtype=np.uint8)) + assert ok + frame = [b'{"encoding": "jpeg"}', jpeg.tobytes()] + calls = [] + + def one_frame_then_stop(): + calls.append(1) + if len(calls) == 1: + return frame + sub.running = False + raise zmq.Again + + sub._recv_latest = one_frame_then_stop # type: ignore[method-assign] + sub._fps_window_start = time.perf_counter() - 10 # force the fps report + sub._dropped = 3 + images = [] + sub.image.connect(images.append) + + sub.run() + + assert len(images) == 1 and images[0].width() == 6 and images[0].height() == 4 + assert sub._frames_in_flight == 1 + assert sub._dropped == 0 # reported alongside the fps, then reset + assert sub._camera_available + assert sub._sock is None and sub._ctx is None # run() cleaned up