fix: resolve CodeRabbit review and Codacy findings on #534

CodeRabbit raised six; all six were real.

The test double had drifted ahead of production. VisualTestDisplayManager
accepted set_scrolling_state(frame_hold=...) while DisplayManager did not,
so such a call passed every harness run and would raise TypeError on the
panel -- the one failure a safety harness exists to prevent. frame_hold
belongs to the change that adds it to DisplayManager (#523), so it moves
there and the double matches main again.

The harness swallowed exceptions from re-rendered frames. _settle_loop
re-renders a mode that came back blank, to give a scroll time to draw;
returning silently on a crash meant a mode that renders one good frame
and then explodes was reported as passing. Recorded on result.error now,
keeping the captured frame so the failure stays inspectable.

starlark-apps display() returned True after _display_frame() failed, so
the controller held a dead frame for the whole display_duration instead
of rotating on. _display_frame now returns bool on all three paths.

run_plugin_tests.py used env.setdefault for PYTHONPATH and
LEDMATRIX_CORE, so an inherited value won and the subprocess imported a
different core than the one under test -- ledmatrix-plugins#467 exactly.
Prepends PROJECT_ROOT and sets LEDMATRIX_CORE unconditionally.

The on-demand mailbox is polled after every frame, ~125x/second on a
scrolling mode, and the read is deliberately uncached, so it was that
many disk reads per second to find nothing. Floored at 250ms, which is
imperceptible for a web-UI click. Consuming it also deleted whatever was
present rather than what had just been processed, so a request posted
while the previous one was in flight was thrown away and never ran; the
delete is now keyed by request_id. That narrows the window rather than
closing it -- a true atomic claim needs a primitive the cache layer does
not offer, and the code says so rather than implying otherwise.

Codacy's 2 criticals were bandit B404/B603 on the subprocess call added
to run_plugin_tests.py. Fixed interpreter, argument list, no shell;
annotated with the repo's existing nosec convention. Bandit is clean on
the file.

Adds test/test_on_demand_mailbox.py (8), test_starlark_display_contract.py
(4) and two settle cases in test_harness_empty_claimed.py. 4, 4 and 2 of
those fail against the pre-fix code. Full suite: 3961 passed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9
This commit is contained in:
ChuckBuilds
2026-09-06 18:36:44 -04:00
co-authored by Claude Opus 5
parent d9f6b110bc
commit 80781dda46
8 changed files with 327 additions and 21 deletions
+58
View File
@@ -101,3 +101,61 @@ class TestSkippedResults:
r = _result(None, returned=None)
check_empty_claimed([r], strict=True)
assert r.empty_claimed is None
class TestSettleRecordsLaterFailures:
"""A mode that renders one good frame and then crashes is broken.
_settle_loop re-renders a mode that came back blank, to give a scroll or an
animation time to put something on the panel. Swallowing an exception from
those later frames meant the harness reported a passing result for a mode
that crashes as soon as it is asked for a second frame -- exactly the kind
of defect the harness exists to catch.
"""
class Boom:
"""Renders once, then raises."""
def __init__(self):
self.calls = 0
def display(self, force_clear=False):
self.calls += 1
raise RuntimeError("second frame exploded")
def _settle(self, inst, dm, result):
from src.plugin_system.testing import harness
harness._settle_loop(inst, "mode", dm, result, None)
def test_the_exception_is_recorded_on_the_result(self, monkeypatch):
from src.plugin_system.testing import harness
# Keep the probe short; this test is about the error, not the pacing.
monkeypatch.setattr(harness, "EMPTY_RECHECK_FRAMES", 1)
monkeypatch.setattr(harness, "EMPTY_RECHECK_STEP", 0)
result = _result(_blank())
assert result.error is None
self._settle(self.Boom(), _FakeDM(), result)
assert result.error is not None, "a crash on a later frame was swallowed"
assert "second frame exploded" in result.error
def test_the_already_captured_frame_is_kept(self, monkeypatch):
from src.plugin_system.testing import harness
monkeypatch.setattr(harness, "EMPTY_RECHECK_FRAMES", 1)
monkeypatch.setattr(harness, "EMPTY_RECHECK_STEP", 0)
image = _blank()
result = _result(image)
self._settle(self.Boom(), _FakeDM(), result)
assert result.image is image, "the good frame was discarded along with the error"
class _FakeDM:
"""Minimal display-manager double for _settle_loop."""
def get_image(self):
return _blank()
def check_overflow(self):
return None
+106
View File
@@ -0,0 +1,106 @@
"""The on-demand request mailbox: how often it is read, and how it is consumed.
The mailbox is a cache key the web process writes and the display process
reads. Two properties matter and neither is obvious from the call site:
* it is polled after every rendered frame, so an uncached read here is a
disk read at frame rate;
* consuming it must not throw away a request that arrived while the previous
one was being processed.
"""
from unittest.mock import MagicMock
import pytest
class TestPollingIsBounded:
"""_poll_on_demand_requests runs ~125x/second on a scrolling mode.
The read is deliberately uncached (memory_ttl=0) because a cached one
pinned the first request for an hour. That makes the call a real disk read,
so it needs a floor -- without one it was ~125 reads per second to find
nothing at all.
"""
def test_first_call_always_reads(self, test_display_controller):
c = test_display_controller
c.cache_manager.get = MagicMock(return_value=None)
c._poll_on_demand_requests()
assert c.cache_manager.get.call_count == 1
def test_immediate_second_call_does_not_read(self, test_display_controller):
c = test_display_controller
c.cache_manager.get = MagicMock(return_value=None)
c._poll_on_demand_requests()
for _ in range(50):
c._poll_on_demand_requests()
assert c.cache_manager.get.call_count == 1, "polling was not bounded"
def test_reads_again_once_the_interval_has_passed(self, test_display_controller, monkeypatch):
c = test_display_controller
c.cache_manager.get = MagicMock(return_value=None)
clock = {"t": 1000.0}
monkeypatch.setattr("src.display_controller.time.monotonic", lambda: clock["t"])
c._poll_on_demand_requests()
clock["t"] += c.ON_DEMAND_POLL_INTERVAL / 2
c._poll_on_demand_requests()
assert c.cache_manager.get.call_count == 1, "read before the interval elapsed"
clock["t"] += c.ON_DEMAND_POLL_INTERVAL
c._poll_on_demand_requests()
assert c.cache_manager.get.call_count == 2
def test_the_interval_is_short_enough_to_feel_instant(self, test_display_controller):
# A person clicking in the web UI must not notice the floor.
assert test_display_controller.ON_DEMAND_POLL_INTERVAL <= 0.5
class TestMailboxIsConsumedByIdentity:
"""Deleting whatever is in the mailbox loses a request that raced in."""
def _arrange(self, controller, first, later):
"""Mailbox returns `first`, then `later` on the pre-delete re-read."""
controller.on_demand_active = False
controller.on_demand_request_id = None
controller._last_on_demand_poll = None
reads = iter([first, later])
def fake_get(key, *a, **kw):
if key == 'display_on_demand_request':
return next(reads, later)
return None # processed-id lookup
controller.cache_manager.get = MagicMock(side_effect=fake_get)
controller.cache_manager.set = MagicMock()
controller.cache_manager.delete = MagicMock()
controller._activate_on_demand = MagicMock()
REQ_A = {'request_id': 'A', 'action': 'start', 'plugin_id': 'p', 'mode': 'm'}
REQ_B = {'request_id': 'B', 'action': 'start', 'plugin_id': 'p', 'mode': 'm'}
def test_own_request_is_deleted(self, test_display_controller):
c = test_display_controller
self._arrange(c, self.REQ_A, self.REQ_A)
c._poll_on_demand_requests()
c.cache_manager.delete.assert_called_once_with('display_on_demand_request')
def test_a_newer_request_is_left_for_the_next_poll(self, test_display_controller):
c = test_display_controller
self._arrange(c, self.REQ_A, self.REQ_B)
c._poll_on_demand_requests()
assert c.cache_manager.delete.call_count == 0, \
"request B was deleted without ever being processed"
def test_an_already_empty_mailbox_is_still_cleared(self, test_display_controller):
c = test_display_controller
self._arrange(c, self.REQ_A, None)
c._poll_on_demand_requests()
c.cache_manager.delete.assert_called_once_with('display_on_demand_request')
def test_the_request_is_still_processed(self, test_display_controller):
c = test_display_controller
self._arrange(c, self.REQ_A, self.REQ_B)
c._poll_on_demand_requests()
c._activate_on_demand.assert_called_once()
+82
View File
@@ -0,0 +1,82 @@
"""starlark-apps: display() must report what actually reached the panel.
The display controller skips a mode only on a boolean False. Returning True
after the frame update failed told it the mode had rendered, so it held a dead
frame for the whole display_duration instead of rotating on -- the same class
of defect as a display() that returns None.
"""
import sys
from pathlib import Path
from unittest.mock import MagicMock
import pytest
PLUGIN_DIR = Path(__file__).resolve().parent.parent / "plugin-repos" / "starlark-apps"
@pytest.fixture(scope="module")
def manager_module():
if not PLUGIN_DIR.exists():
pytest.skip("starlark-apps plugin is not checked out")
sys.path.insert(0, str(PLUGIN_DIR))
try:
import importlib
spec = importlib.util.spec_from_file_location(
"starlark_manager_under_test", PLUGIN_DIR / "manager.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
except Exception as e: # noqa: BLE001 - optional deps (pixlet, fcntl) may be absent
pytest.skip(f"starlark-apps manager is not importable here: {e}")
finally:
sys.path.remove(str(PLUGIN_DIR))
def _plugin(manager_module):
"""A manager with __init__ bypassed -- only display paths are under test."""
cls = manager_module.StarlarkAppsPlugin
inst = cls.__new__(cls)
inst.logger = MagicMock()
inst.display_manager = MagicMock()
inst.current_app = None
return inst
class _App:
def __init__(self, frames):
self.frames = frames
self.current_frame_index = 0
self.last_frame_time = 0.0
self.app_id = "app"
class TestDisplayFramePropagates:
def test_a_failed_update_returns_false(self, manager_module):
p = _plugin(manager_module)
p.current_app = _App([("frame", 100)])
p.display_manager.update_display.side_effect = RuntimeError("panel gone")
assert p._display_frame() is False
def test_a_good_update_returns_true(self, manager_module):
p = _plugin(manager_module)
p.current_app = _App([("frame", 100)])
assert p._display_frame() is True
def test_no_frames_returns_false(self, manager_module):
p = _plugin(manager_module)
p.current_app = _App([])
assert p._display_frame() is False
def test_display_reports_the_frame_failure(self, manager_module):
p = _plugin(manager_module)
p.current_app = _App([("frame", 100)])
p.display_manager.update_display.side_effect = RuntimeError("panel gone")
result = p.display()
assert result is False, "display() claimed success over a failed frame update"
assert isinstance(result, bool), "the controller only skips on a real bool"