rick/LEDMatrix
1
0
Fork 0
mirror of https://github.com/ChuckBuilds/LEDMatrix.git synced 2026-10-04 06:15:09 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
2601cb4cbb370f0c1e0032af6af51a705f05c9ae
LEDMatrix/test/test_display_watchdog.py
T
ChuckandClaude Opus 5.5 64c7289593 feat(display): systemd watchdog and heartbeat for a frozen render loop (#687) 
If the render loop gets stuck inside a plugin's display(), ledmatrix.service
stays active and the panel stays frozen. This adds a way to detect that.

- src/display_watchdog.py (standard library only) sends sd_notify over
  $NOTIFY_SOCKET and writes /run/ledmatrix/display-heartbeat.json. Only the
  render thread counts: beats from other threads are ignored.
- ledmatrix.service: WatchdogSec=120, NotifyAccess=main,
  RuntimeDirectory=ledmatrix (0755), RestartSteps=4 and
  RestartMaxDelaySec=2min. It stays Type=simple. run.py widens the watchdog
  to 15 min for start-up, and load_plugin() does the same on the render
  thread. The loop arms after its first frame.
- /api/v3/health adds checks.display_loop: running, stalled (no heartbeat
  for over 60s, which makes the status degraded) or not_reported. With web
  login on, a caller who is not logged in still gets only healthy/degraded,
  and a stall degrades that answer.
- The update verifier requires a fresh heartbeat from the restarted display
  when the display it replaced was writing one. A frozen panel is rolled
  back.
- Existing installs get the systemd watchdog only after install_service.sh
  is re-run. The heartbeat works right away.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-30 11:15:31 -04:00

661 lines
26 KiB
Python
Raw Blame History

"""The render loop's liveness signals: systemd watchdog pings and the heartbeat.
A panel can freeze while ledmatrix.service stays "active" -- a render thread
stuck inside a plugin's display(). src/display_watchdog.py lets only the
render thread vouch for itself, to systemd (sd_notify WATCHDOG=1) and to the
web interface (a heartbeat file under /run/ledmatrix). These tests pin:
* the sd_notify wire format, including abstract-namespace sockets;
* that nothing is armed until the first frame, so start-up keeps its allowance;
* that beats from any other thread are ignored, so a stuck render thread
goes quiet even while the update worker and Vegas's tick thread carry on;
* the places the render loop checks in from (dwell sleeps, per-frame
display, Vegas's own loop, a plugin load's longer allowance).
"""
import json
import os
import socket
import sys
import threading
import time
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
os.environ.setdefault("EMULATOR", "true") # display_controller imports without hardware
from src import display_watchdog # noqa: E402
from src.display_watchdog import ( # noqa: E402
RenderWatchdog, heartbeat_age, notify, read_heartbeat, watchdog_usec)
WATCHDOG_120 = {'NOTIFY_SOCKET': '/run/systemd/notify', 'WATCHDOG_USEC': '120000000'}
class FakeSocket:
"""Records what notify() does with the socket it creates."""
def __init__(self, record, fail_connect=False):
self.record = record
self.fail_connect = fail_connect
self.closed = False
def connect(self, address):
self.record['address'] = address
if self.fail_connect:
raise ConnectionRefusedError('nobody listening')
def sendall(self, data):
self.record.setdefault('sent', []).append(data)
def close(self):
self.closed = True
self.record['closed'] = True
def fake_factory(record, fail_connect=False):
def factory(family, kind):
record['family'], record['type'] = family, kind
return FakeSocket(record, fail_connect)
return factory
@pytest.fixture
def af_unix(monkeypatch):
"""AF_UNIX for the fake-socket tests, even on a Python built without it."""
monkeypatch.setattr(socket, 'AF_UNIX', getattr(socket, 'AF_UNIX', 1), raising=False)
return socket.AF_UNIX
# -- sd_notify -------------------------------------------------------------
class TestNotify:
def test_sends_one_datagram_to_the_socket_path(self, af_unix):
record = {}
assert notify('WATCHDOG=1', {'NOTIFY_SOCKET': '/run/systemd/notify'},
socket_factory=fake_factory(record)) is True
assert record['family'] == af_unix
assert record['type'] & socket.SOCK_DGRAM == socket.SOCK_DGRAM
assert record['address'] == '/run/systemd/notify'
assert record['sent'] == [b'WATCHDOG=1']
assert record['closed']
def test_an_at_sign_means_the_abstract_namespace(self, af_unix):
record = {}
notify('READY=1\nSTATUS=Rendering', {'NOTIFY_SOCKET': '@/org/freedesktop/systemd1/notify'},
socket_factory=fake_factory(record))
assert record['address'] == '\0/org/freedesktop/systemd1/notify'
assert record['sent'] == [b'READY=1\nSTATUS=Rendering']
@pytest.mark.parametrize('address', [None, '', 'relative/path', 'vsock:2:1234'])
def test_no_usable_socket_sends_nothing(self, af_unix, address):
record = {}
env = {} if address is None else {'NOTIFY_SOCKET': address}
assert notify('WATCHDOG=1', env, socket_factory=fake_factory(record)) is False
assert record == {}
def test_a_failed_send_is_false_not_an_exception(self, af_unix):
record = {}
assert notify('WATCHDOG=1', {'NOTIFY_SOCKET': '/nope'},
socket_factory=fake_factory(record, fail_connect=True)) is False
assert record['closed']
@pytest.mark.skipif(not hasattr(socket, 'AF_UNIX') or os.name != 'posix',
reason='needs AF_UNIX datagram sockets')
def test_a_real_socket_receives_the_message(self, tmp_path):
path = str(tmp_path / 'notify')
server = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
try:
server.bind(path)
server.settimeout(2)
assert notify('WATCHDOG=1', {'NOTIFY_SOCKET': path}) is True
assert server.recv(4096) == b'WATCHDOG=1'
finally:
server.close()
@pytest.mark.skipif(not sys.platform.startswith('linux'),
reason='abstract sockets are Linux-only')
def test_a_real_abstract_socket_receives_the_message(self):
name = f'ledmatrix-test-{os.getpid()}-{time.monotonic_ns()}'
server = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
try:
server.bind('\0' + name)
server.settimeout(2)
assert notify('READY=1', {'NOTIFY_SOCKET': '@' + name}) is True
assert server.recv(4096) == b'READY=1'
finally:
server.close()
class TestWatchdogUsec:
def test_reads_the_units_value(self):
assert watchdog_usec({'WATCHDOG_USEC': '120000000'}) == 120_000_000
def test_meant_for_another_process(self):
assert watchdog_usec({'WATCHDOG_USEC': '120000000',
'WATCHDOG_PID': str(os.getpid() + 1)}) is None
def test_meant_for_this_process(self):
assert watchdog_usec({'WATCHDOG_USEC': '5000000',
'WATCHDOG_PID': str(os.getpid())}) == 5_000_000
@pytest.mark.parametrize('value', [None, '', 'abc', '0', '-5'])
def test_no_watchdog(self, value):
env = {} if value is None else {'WATCHDOG_USEC': value}
assert watchdog_usec(env) is None
# -- the render loop's side ----------------------------------------------------
class Clock:
def __init__(self, now=1000.0):
self.now = now
def __call__(self):
return self.now
def make(environ=WATCHDOG_120, heartbeat_dir=None, clock=None):
sent = []
wd = RenderWatchdog(environ=environ, send=lambda m: sent.append(m) or True,
clock=clock or Clock(), wall_clock=lambda: 1_700_000_000.0,
heartbeat_dir=heartbeat_dir, enable_faulthandler=False)
return wd, sent
def pings(sent):
return [m for m in sent if 'WATCHDOG=1' in m.split('\n')]
def on_other_thread(fn):
t = threading.Thread(target=fn)
t.start()
t.join()
class TestStartup:
def test_start_up_widens_the_limit(self):
wd, sent = make()
wd.begin_startup()
assert sent == [f'WATCHDOG_USEC={int(display_watchdog.STARTUP_ALLOWANCE_SECONDS * 1e6)}'
'\nSTATUS=Starting: loading plugins']
def test_start_up_never_shortens_a_longer_unit_limit(self):
wd, sent = make({'NOTIFY_SOCKET': '/x', 'WATCHDOG_USEC': str(3600 * 10**6)})
wd.begin_startup()
assert sent[0].startswith(f'WATCHDOG_USEC={3600 * 10**6}\n')
def test_without_a_watchdog_start_up_sends_nothing(self):
wd, sent = make({'NOTIFY_SOCKET': '/x'})
wd.begin_startup()
assert sent == []
class TestArming:
def test_nothing_before_the_render_loop_starts(self, tmp_path):
wd, sent = make(heartbeat_dir=str(tmp_path))
wd.note_frame() # a start-up screen
wd.beat()
wd.loop_pass()
assert sent == [] and not wd.armed
assert not (tmp_path / display_watchdog.HEARTBEAT_NAME).exists()
def test_nothing_before_the_first_frame(self, tmp_path):
wd, sent = make(heartbeat_dir=str(tmp_path))
wd.bind_render_thread()
wd.loop_pass() # the first pass has begun...
wd.beat() # ...and is, say, composing Vegas content
assert sent == [] and not wd.armed
assert not (tmp_path / display_watchdog.HEARTBEAT_NAME).exists()
def test_the_first_frame_arms_it(self, tmp_path):
wd, sent = make(heartbeat_dir=str(tmp_path))
wd.bind_render_thread()
wd.note_frame()
assert wd.armed
assert sent == ['READY=1\nWATCHDOG_USEC=120000000\nWATCHDOG=1\nSTATUS=Rendering']
heartbeat = json.loads((tmp_path / display_watchdog.HEARTBEAT_NAME).read_text())
assert heartbeat == {'pid': os.getpid(), 'mono': 1000.0, 'wall': 1_700_000_000.0}
def test_a_first_frame_pushed_by_another_thread_arms_on_the_next_render_beat(self):
"""A screen's first display() runs on PluginExecutor's thread."""
wd, sent = make()
wd.bind_render_thread()
on_other_thread(wd.note_frame)
assert not wd.armed and sent == []
wd.beat()
assert wd.armed and sent[0].startswith('READY=1\n')
def test_a_full_pass_with_nothing_drawn_arms_it(self):
"""No plugins enabled, or every screen empty: the loop is still alive."""
wd, sent = make()
wd.bind_render_thread()
wd.loop_pass()
assert not wd.armed
wd.loop_pass()
assert wd.armed and sent[0].startswith('READY=1\n')
def test_without_a_watchdog_it_still_says_ready_and_writes_the_heartbeat(self, tmp_path):
wd, sent = make({'NOTIFY_SOCKET': '/x'}, heartbeat_dir=str(tmp_path))
wd.bind_render_thread()
wd.note_frame()
assert sent == ['READY=1\nSTATUS=Rendering']
assert (tmp_path / display_watchdog.HEARTBEAT_NAME).exists()
sent_before = len(sent)
wd.beat()
assert len(sent) == sent_before # no WATCHDOG=1 without a watchdog
class TestBeats:
def test_pings_are_rate_limited(self, tmp_path):
clock = Clock()
wd, sent = make(heartbeat_dir=str(tmp_path), clock=clock)
wd.bind_render_thread()
wd.note_frame()
del sent[:]
for _ in range(100): # a burst of frames
wd.beat()
assert sent == []
clock.now += display_watchdog.BEAT_INTERVAL_SECONDS
wd.beat()
assert sent == ['WATCHDOG=1']
heartbeat = read_heartbeat(str(tmp_path / display_watchdog.HEARTBEAT_NAME))
assert heartbeat['mono'] == clock.now
def test_a_short_unit_limit_pings_more_often(self):
clock = Clock()
wd, sent = make({'NOTIFY_SOCKET': '/x', 'WATCHDOG_USEC': '3000000'}, clock=clock)
wd.bind_render_thread()
wd.note_frame()
del sent[:]
clock.now += 1.0 # a third of 3s
wd.beat()
assert sent == ['WATCHDOG=1']
def test_beats_from_other_threads_are_ignored(self, tmp_path):
"""The update worker, Vegas's tick thread and the prefetcher keep
running while the render thread is stuck; they must not keep the
watchdog fed on its behalf."""
clock = Clock()
wd, sent = make(heartbeat_dir=str(tmp_path), clock=clock)
wd.bind_render_thread()
wd.note_frame()
del sent[:]
before = (tmp_path / display_watchdog.HEARTBEAT_NAME).read_text()
clock.now += 60
on_other_thread(wd.beat)
on_other_thread(wd.note_frame)
on_other_thread(wd.loop_pass)
assert sent == []
assert (tmp_path / display_watchdog.HEARTBEAT_NAME).read_text() == before
def test_the_module_shortcuts_reach_the_process_instance(self, monkeypatch):
wd, sent = make()
monkeypatch.setattr(display_watchdog, 'watchdog', wd)
wd.bind_render_thread()
display_watchdog.note_frame()
assert wd.armed
with display_watchdog.extended(600, 'x'):
pass
display_watchdog.beat()
assert any('WATCHDOG_USEC=600000000' in m for m in sent)
class TestExtended:
def test_a_long_job_gets_the_longer_limit_then_the_units_back(self):
wd, sent = make()
wd.bind_render_thread()
wd.note_frame()
del sent[:]
with wd.extended(900, 'loading plugin weather'):
assert sent == ['WATCHDOG_USEC=900000000\nWATCHDOG=1\nSTATUS=Busy: loading plugin weather']
assert sent[-1] == 'WATCHDOG_USEC=120000000\nWATCHDOG=1\nSTATUS=Rendering'
def test_nested_jobs_restore_once(self):
wd, sent = make()
wd.bind_render_thread()
wd.note_frame()
del sent[:]
with wd.extended(900):
with wd.extended(900):
pass
assert len(sent) == 1 # the inner one neither re-extends nor restores
assert len(sent) == 2 and sent[-1].startswith('WATCHDOG_USEC=120000000\n')
def test_restored_even_when_the_job_fails(self):
wd, sent = make()
wd.bind_render_thread()
wd.note_frame()
with pytest.raises(RuntimeError):
with wd.extended(900):
raise RuntimeError('pip failed')
assert sent[-1].startswith('WATCHDOG_USEC=120000000\n')
def test_off_the_render_thread_or_before_arming_it_does_nothing(self):
"""Start-up loads run on a thread pool under the start-up allowance."""
wd, sent = make()
wd.bind_render_thread()
with wd.extended(900):
pass
wd.note_frame()
del sent[:]
on_other_thread(lambda: wd.extended(900).__enter__())
assert sent == []
class TestStopping:
def test_a_clean_stop_removes_the_heartbeat(self, tmp_path):
wd, sent = make(heartbeat_dir=str(tmp_path))
wd.bind_render_thread()
wd.note_frame()
wd.stopping()
assert sent[-1] == 'STOPPING=1'
assert not (tmp_path / display_watchdog.HEARTBEAT_NAME).exists()
def test_stopping_outside_systemd_is_harmless(self):
wd, sent = make({})
wd.stopping()
assert sent == []
class TestHeartbeatFile:
def test_the_directory_is_created_when_missing(self, tmp_path):
"""An install whose unit predates RuntimeDirectory=; the display is root."""
target = tmp_path / 'run' / 'ledmatrix'
wd, _ = make(heartbeat_dir=str(target))
wd.bind_render_thread()
wd.note_frame()
assert (target / display_watchdog.HEARTBEAT_NAME).is_file()
assert [p.name for p in target.iterdir()] == [display_watchdog.HEARTBEAT_NAME]
@pytest.mark.skipif(os.name != 'posix', reason='POSIX permissions')
def test_other_users_can_read_it(self, tmp_path):
wd, _ = make(heartbeat_dir=str(tmp_path))
wd.bind_render_thread()
wd.note_frame()
mode = (tmp_path / display_watchdog.HEARTBEAT_NAME).stat().st_mode & 0o777
assert mode == 0o644
def test_nowhere_to_write_is_not_an_error(self, tmp_path):
blocker = tmp_path / 'not-a-dir'
blocker.write_text('x')
clock = Clock()
wd, sent = make(heartbeat_dir=str(blocker / 'ledmatrix'), clock=clock)
wd.bind_render_thread()
wd.note_frame()
clock.now += 10
wd.beat()
assert wd.armed and pings(sent) # the watchdog works regardless
def test_windows_gets_no_heartbeat_by_default(self, monkeypatch):
monkeypatch.setattr(display_watchdog.os, 'name', 'nt')
wd = RenderWatchdog(environ={}, send=lambda m: True)
assert wd._heartbeat_path() is None
class TestHeartbeatAge:
def test_monotonic_is_preferred(self):
assert heartbeat_age({'mono': 100.0, 'wall': 0.0}, now_mono=112.5, now_wall=9e9) == 12.5
def test_the_wall_clock_is_the_fallback(self):
assert heartbeat_age({'wall': 50.0}, now_mono=1.0, now_wall=80.0) == 30.0
def test_a_monotonic_stamp_from_the_future_is_not_trusted(self):
"""Not the same clock -- fall back rather than report a fresh heartbeat."""
assert heartbeat_age({'mono': 500.0, 'wall': 50.0}, now_mono=100.0, now_wall=170.0) == 120.0
def test_no_time_at_all(self):
assert heartbeat_age({'pid': 1}) is None
assert heartbeat_age({'mono': True}) is None
def test_reading_a_missing_or_broken_file(self, tmp_path):
assert read_heartbeat(str(tmp_path / 'absent.json')) is None
(tmp_path / 'broken.json').write_text('{not json')
assert read_heartbeat(str(tmp_path / 'broken.json')) is None
(tmp_path / 'list.json').write_text('[1, 2]')
assert read_heartbeat(str(tmp_path / 'list.json')) is None
# -- where the render loop checks in -------------------------------------------
@pytest.fixture
def armed(monkeypatch):
"""A process watchdog bound to this thread and armed, with a clock to advance."""
clock = Clock()
wd, sent = make(clock=clock)
monkeypatch.setattr(display_watchdog, 'watchdog', wd)
wd.bind_render_thread()
wd.note_frame()
del sent[:]
return SimpleNamespace(wd=wd, sent=sent, clock=clock)
def _tick_clock(armed):
"""Advance the watchdog's clock past the rate limit on every beat check."""
original = armed.wd._clock
def advancing():
armed.clock.now += display_watchdog.BEAT_INTERVAL_SECONDS
return original()
armed.wd._clock = advancing
class TestCheckInPoints:
def test_the_dwell_sleep_checks_in(self, armed):
from src.display_controller import DisplayController
dc = object.__new__(DisplayController)
dc.current_display_mode = 'm'
dc.is_display_active = True
dc.on_demand_active = False
dc._tick_plugin_updates = lambda: None
dc._service_pending_changes = lambda: None
_tick_clock(armed)
dc._sleep_with_plugin_updates(0.05, tick_interval=0.01)
assert len(pings(armed.sent)) >= 3
def test_every_frame_of_a_screen_checks_in(self, armed):
from src.display_controller import DisplayController
dc = object.__new__(DisplayController)
dc.plugin_manager = None
plugin = MagicMock(plugin_id='p')
_tick_clock(armed)
for _ in range(3):
dc._display_once(plugin, 'm', accepts_display_mode=False)
assert len(pings(armed.sent)) == 3
def test_vegas_checks_in_every_frame_of_its_own_loop(self, armed):
"""An iteration runs for minutes without returning to run()."""
import threading as _threading
from src.vegas_mode.config import VegasModeConfig
from src.vegas_mode.coordinator import VegasModeCoordinator
coord = VegasModeCoordinator.__new__(VegasModeCoordinator)
coord.vegas_config = VegasModeConfig.from_config({'display': {'vegas_scroll': {
'enabled': True, 'max_cycle_duration': 60}}})
coord.render_pipeline = MagicMock(frame_interval=0.0, target_fps=90)
coord.display_manager = MagicMock()
coord._state_lock = _threading.Lock()
coord._is_active = True
coord._is_paused = False
coord._should_stop = False
coord._live_priority_active = False
coord._fps_last_health_log = 0.0
coord._fps_was_degraded = False
coord._interrupt_check = None
coord._interrupt_check_interval = 10
coord._update_callback = None
coord._update_tick_running = False
coord._check_static_plugin_trigger = lambda: None
frames = []
def run_frame():
frames.append(1)
if len(frames) == 5:
coord._should_stop = True
return False
return True
coord.run_frame = run_frame
_tick_clock(armed)
coord.run_iteration()
assert len(pings(armed.sent)) == 5
def test_each_plugin_fetched_for_a_vegas_cycle_checks_in(self, armed):
from src.vegas_mode.stream_manager import StreamManager
sm = StreamManager.__new__(StreamManager)
sm.plugin_manager = SimpleNamespace(plugins={})
_tick_clock(armed)
for plugin_id in ('a', 'b'):
sm._fetch_plugin_content(plugin_id)
assert len(pings(armed.sent)) == 2
def test_loading_a_plugin_gets_the_longer_limit(self, armed):
from src.plugin_system.plugin_manager import PluginManager
pm = PluginManager.__new__(PluginManager)
seen = []
pm._load_plugin = lambda plugin_id, force_enabled=False: seen.append(list(armed.sent)) or True
assert pm.load_plugin('weather') is True
allowance = int(display_watchdog.PLUGIN_LOAD_ALLOWANCE_SECONDS * 1e6)
assert seen[0] and seen[0][-1].startswith(f'WATCHDOG_USEC={allowance}\n')
assert armed.sent[-1].startswith('WATCHDOG_USEC=120000000\n')
class TestStuckRenderThread:
def test_a_render_thread_stuck_in_display_stops_the_pings(self, test_display_controller,
monkeypatch):
"""End to end through DisplayController.run(): pings flow while frames
do, stop while display() is stuck even though other threads keep
calling beat(), and nothing else in the process keeps them alive."""
sent = []
lock = threading.Lock()
def record(message):
with lock:
sent.append(message)
return True
# 0.3s limit -> a ping at most every 0.1s.
wd = RenderWatchdog(environ={'NOTIFY_SOCKET': '/x', 'WATCHDOG_USEC': '300000'},
send=record, heartbeat_dir=None, enable_faulthandler=False)
monkeypatch.setattr(display_watchdog, 'watchdog', wd)
stuck, release = threading.Event(), threading.Event()
class Plugin:
plugin_id = 'stuck-plugin'
needs_high_fps = True
enabled = True
calls = 0
def display(self, force_clear=False):
Plugin.calls += 1
display_watchdog.note_frame() # DisplayManager is mocked here
if Plugin.calls >= 60: # about half a second of frames
stuck.set()
release.wait(10)
raise KeyboardInterrupt # ends run() the way SIGTERM does
controller = test_display_controller
controller.available_modes = ['stuck-mode']
controller.plugin_modes = {'stuck-mode': Plugin()}
controller.mode_to_plugin_id = {'stuck-mode': 'stuck-plugin'}
controller.current_mode_index = 0
runner = threading.Thread(target=controller.run, daemon=True)
runner.start()
assert stuck.wait(10), 'the render loop never reached the plugin'
with lock:
before = len(pings(sent))
assert wd.armed and any(m.startswith('READY=1') for m in sent)
assert before >= 2, sent
# Other threads carry on while the render thread is stuck.
stop_others = threading.Event()
def busy_other_thread():
while not stop_others.is_set():
display_watchdog.beat()
display_watchdog.note_frame()
time.sleep(0.01)
other = threading.Thread(target=busy_other_thread, daemon=True)
other.start()
time.sleep(0.8) # well past the 0.3s limit
with lock:
after = len(pings(sent))
stop_others.set()
release.set()
other.join(5)
runner.join(10)
assert after == before, 'something other than the render thread fed the watchdog'
assert not runner.is_alive()
assert sent[-1] == 'STOPPING=1'
# -- the unit and the entry point ----------------------------------------------
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def _service_directives():
with open(os.path.join(ROOT, 'systemd', 'ledmatrix.service'), encoding='utf-8') as f:
lines = [line.strip() for line in f]
return dict(line.split('=', 1) for line in lines
if line and not line.startswith(('#', '[')) and '=' in line
and not line.startswith('Environment='))
def _seconds(value):
value = value.strip()
for suffix, factor in (('min', 60), ('s', 1)):
if value.endswith(suffix):
return float(value[:-len(suffix)]) * factor
return float(value)
class TestUnit:
def test_the_display_unit_has_a_watchdog_the_process_can_feed(self):
d = _service_directives()
# Type=notify would block "systemctl start/restart" until READY=1 --
# after plugins load -- and the web UI and the update check call those
# with short timeouts.
assert d['Type'] == 'simple'
assert d['NotifyAccess'] == 'main'
assert 'WatchdogSec' in d
def test_the_watchdog_outlasts_the_loops_longest_healthy_gap(self):
from src.plugin_system.plugin_executor import PluginExecutor
limit = _seconds(_service_directives()['WatchdogSec'])
executor_timeout = PluginExecutor().default_timeout
assert limit >= 3 * executor_timeout, (
"a screen's first display() may legitimately take the executor's "
f"{executor_timeout}s timeout; WatchdogSec={limit:.0f}s leaves too little margin")
assert limit < display_watchdog.STARTUP_ALLOWANCE_SECONDS
assert limit < display_watchdog.PLUGIN_LOAD_ALLOWANCE_SECONDS
assert display_watchdog.BEAT_INTERVAL_SECONDS * 4 <= limit
assert limit > display_watchdog.HEARTBEAT_STALE_SECONDS >= 2 * executor_timeout
def test_the_heartbeat_directory_is_created_readable_by_the_web_user(self):
d = _service_directives()
assert d['RuntimeDirectory'] == 'ledmatrix'
assert display_watchdog.HEARTBEAT_DIR == '/run/' + d['RuntimeDirectory']
assert d['RuntimeDirectoryMode'] == '0755'
def test_a_crash_loop_backs_off_instead_of_stopping_for_good(self):
"""A tripped start limit leaves the panel dark and refuses the web UI's
Start button and the update rollback's restart."""
d = _service_directives()
assert d['Restart'] == 'always'
assert 'StartLimitBurst' not in d
assert int(d['RestartSteps']) > 0
assert _seconds(d['RestartMaxDelaySec']) > _seconds(d['RestartSec'])
def test_run_py_widens_the_watchdog_before_importing_anything_heavy(self):
with open(os.path.join(ROOT, 'run.py'), encoding='utf-8') as f:
text = f.read()
call = text.index('display_watchdog.watchdog.begin_startup()')
assert call < text.index('from src.logging_config')
assert call < text.index('from src.display_controller')
def test_the_module_imports_nothing_else_from_src(self):
"""run.py loads it first thing; importing it must stay cheap."""
with open(os.path.join(ROOT, 'src', 'display_watchdog.py'), encoding='utf-8') as f:
imports = [line for line in f if line.startswith(('import ', 'from '))]
assert not [line for line in imports if 'src' in line], imports
Reference in New Issue View Git Blame Copy Permalink