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 <noreply@anthropic.com>

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"

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))

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