mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-05 06:45:09 +00:00
Merge branch 'main' of https://github.com/ChuckBuilds/LEDMatrix into fix/plugin-update-keeps-local-files
# Conflicts: # test/test_on_demand_live_and_restore.py
This commit is contained in:
Chuck
@@ -0,0 +1,326 @@
|
||||
"""Four web answers that disagreed with the rig they describe (found on ledpi).
|
||||
|
||||
1. POST /config/schedule refused the schedule GET returns on a fresh install
|
||||
(config.template.json: per-day, every day off, schedule disabled) with
|
||||
"At least one day must be enabled", as did /config/dim-schedule. A
|
||||
disabled schedule needs no enabled day.
|
||||
2. A brightness-only POST /config/main answered ``restart_required: true``,
|
||||
though the display applies brightness live (brightness.set over the
|
||||
socket, and the config watcher). The flag now says whether anything
|
||||
changed that the running display does not pick up by itself.
|
||||
3. /health stayed "healthy" with the display service stopped: only the
|
||||
sub-checks changed. Service inactive, no socket and no live heartbeat
|
||||
is now ``display_loop: stopped`` and "degraded".
|
||||
4. /display/current-status kept answering ``is_display_active: true`` from
|
||||
the cache for up to 120 s after the display stopped. With no socket and
|
||||
no live heartbeat it is now unknown.
|
||||
"""
|
||||
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from test._api_v3_test_helpers import api_v3_client, api_v3_module # noqa: F401,E402
|
||||
|
||||
from src import display_watchdog # noqa: E402
|
||||
from src.ipc import client as control_client # noqa: E402
|
||||
from web_interface import display_state # noqa: E402
|
||||
|
||||
REPO = Path(__file__).resolve().parent.parent
|
||||
TEMPLATE = json.loads((REPO / 'config' / 'config.template.json').read_text(encoding='utf-8'))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def store(api_v3_module, monkeypatch):
|
||||
state = {'config': {}, 'saves': 0}
|
||||
api_v3_module.api_v3.config_manager.load_config.side_effect = \
|
||||
lambda *a, **k: copy.deepcopy(state['config'])
|
||||
|
||||
def fake_save(_manager, config, **_kwargs):
|
||||
state['config'] = copy.deepcopy(config)
|
||||
state['saves'] += 1
|
||||
return True, ''
|
||||
|
||||
monkeypatch.setattr(api_v3_module, '_save_config_atomic', fake_save)
|
||||
return state
|
||||
|
||||
|
||||
# --- 1. schedules ---------------------------------------------------------------
|
||||
|
||||
SCHEDULE_ROUTES = [('/api/v3/config/schedule', 'schedule'),
|
||||
('/api/v3/config/dim-schedule', 'dim_schedule')]
|
||||
|
||||
|
||||
@pytest.mark.parametrize('route,section', SCHEDULE_ROUTES)
|
||||
def test_the_templates_disabled_per_day_schedule_saves_back(api_v3_client, store,
|
||||
route, section):
|
||||
stored = copy.deepcopy(TEMPLATE[section])
|
||||
stored['mode'] = 'per-day'
|
||||
assert stored['enabled'] is False
|
||||
assert not any(day['enabled'] for day in stored['days'].values())
|
||||
store['config'] = {section: copy.deepcopy(stored)}
|
||||
|
||||
read = api_v3_client.get(route).get_json()['data']
|
||||
resp = api_v3_client.post(route, json=read)
|
||||
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
saved = store['config'][section]
|
||||
assert saved['enabled'] is False and saved['mode'] == 'per-day'
|
||||
# The disabled days keep their times: switching one on finds them.
|
||||
assert saved['days'] == stored['days']
|
||||
|
||||
|
||||
@pytest.mark.parametrize('route,section', SCHEDULE_ROUTES)
|
||||
def test_an_enabled_per_day_schedule_still_needs_a_day(api_v3_client, store, route, section):
|
||||
body = copy.deepcopy(TEMPLATE[section])
|
||||
body.update(enabled=True, mode='per-day')
|
||||
resp = api_v3_client.post(route, json=body)
|
||||
assert resp.status_code == 400
|
||||
assert 'At least one day must be enabled' in resp.get_json()['message']
|
||||
assert store['saves'] == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize('route', [r for r, _ in SCHEDULE_ROUTES])
|
||||
def test_the_pickers_form_post_with_every_day_off_saves(api_v3_client, store, route):
|
||||
"""What schedule-picker.js posts: flat hidden inputs, booleans as strings,
|
||||
times for every day."""
|
||||
body = {'enabled': 'false', 'mode': 'per_day', 'start_time': '07:00', 'end_time': '23:00'}
|
||||
for day in ('monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday'):
|
||||
body.update({f'{day}_enabled': 'false', f'{day}_start': '06:30', f'{day}_end': '22:15'})
|
||||
resp = api_v3_client.post(route, json=body)
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
|
||||
|
||||
def test_an_invalid_time_on_a_disabled_day_is_dropped_not_refused(api_v3_client, store):
|
||||
body = {'enabled': False, 'mode': 'per-day',
|
||||
'days': {'monday': {'enabled': False, 'start_time': 'soon', 'end_time': '22:00'}}}
|
||||
resp = api_v3_client.post('/api/v3/config/schedule', json=body)
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert store['config']['schedule']['days']['monday'] == {'enabled': False,
|
||||
'end_time': '22:00'}
|
||||
|
||||
|
||||
# --- 2. restart_required on /config/main ------------------------------------------
|
||||
|
||||
STORED_MAIN = {
|
||||
'timezone': 'America/Chicago',
|
||||
'display': {
|
||||
'hardware': {'rows': 32, 'cols': 64, 'chain_length': 2, 'brightness': 90,
|
||||
'disable_hardware_pulsing': False, 'inverse_colors': False,
|
||||
'show_refresh_rate': False},
|
||||
'runtime': {'gpio_slowdown': 4},
|
||||
'display_durations': {'clock': 15},
|
||||
'use_short_date_format': False,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def main_store(store):
|
||||
store['config'] = copy.deepcopy(STORED_MAIN)
|
||||
return store
|
||||
|
||||
|
||||
def _save_main(client, body):
|
||||
with patch('web_interface.blueprints.api_v3.control_client.brightness_set',
|
||||
side_effect=control_client.ControlError('no_socket', 'x')):
|
||||
resp = client.post('/api/v3/config/main', data=json.dumps(body),
|
||||
content_type='application/json')
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
return resp.get_json()
|
||||
|
||||
|
||||
def test_a_brightness_only_save_needs_no_restart(api_v3_client, main_store):
|
||||
body = _save_main(api_v3_client, {'brightness': 40})
|
||||
assert main_store['config']['display']['hardware']['brightness'] == 40
|
||||
assert body['restart_required'] is False
|
||||
|
||||
|
||||
def test_a_brightness_save_on_a_config_without_a_display_section(api_v3_client, store):
|
||||
"""The route creates display.hardware and display.runtime on the way;
|
||||
empty sections are not a change."""
|
||||
store['config'] = {}
|
||||
assert _save_main(api_v3_client, {'brightness': 40})['restart_required'] is False
|
||||
|
||||
|
||||
def test_the_display_form_with_only_brightness_changed_needs_no_restart(api_v3_client,
|
||||
main_store):
|
||||
hw = STORED_MAIN['display']['hardware']
|
||||
body = {'__form_section': 'display', 'rows': 32, 'cols': 64, 'chain_length': 2,
|
||||
'brightness': 55, 'gpio_slowdown': 4}
|
||||
body.update({k: 'on' for k in ('disable_hardware_pulsing', 'inverse_colors',
|
||||
'show_refresh_rate') if hw[k]})
|
||||
assert _save_main(api_v3_client, body)['restart_required'] is False
|
||||
|
||||
|
||||
def test_a_mode_duration_needs_no_restart(api_v3_client, main_store):
|
||||
body = _save_main(api_v3_client, {'duration__clock': 40})
|
||||
assert main_store['config']['display']['display_durations']['clock'] == 40
|
||||
assert body['restart_required'] is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize('change', [{'rows': 64}, {'brightness': 40, 'chain_length': 3},
|
||||
{'gpio_slowdown': 2}, {'timezone': 'UTC'}])
|
||||
def test_a_setting_the_display_reads_at_startup_still_needs_one(api_v3_client, main_store,
|
||||
change):
|
||||
assert _save_main(api_v3_client, change)['restart_required'] is True
|
||||
|
||||
|
||||
def test_restart_needed_compares_leaves():
|
||||
from web_interface.blueprints.api_v3.config import restart_needed
|
||||
before = {'display': {'hardware': {'brightness': 90, 'rows': 32}}}
|
||||
assert not restart_needed(before, copy.deepcopy(before))
|
||||
assert not restart_needed(before, {'display': {'hardware': {'brightness': 10, 'rows': 32},
|
||||
'runtime': {}}})
|
||||
assert restart_needed(before, {'display': {'hardware': {'brightness': 90}}}) # removed
|
||||
assert not restart_needed({}, {'clock': {'enabled': True}}, live_paths=[('clock',)])
|
||||
assert restart_needed({}, {'clockwork': {'enabled': True}}, live_paths=[('clock',)])
|
||||
|
||||
|
||||
# --- 3 and 4. a stopped display -----------------------------------------------------
|
||||
|
||||
@pytest.fixture
|
||||
def no_display(monkeypatch, tmp_path):
|
||||
"""A Pi whose display service has stopped: the socket is expected here
|
||||
but does not answer, and systemd took the heartbeat's directory away."""
|
||||
monkeypatch.setattr(display_state, 'socket_supported', lambda: True)
|
||||
monkeypatch.setattr(display_state, 'client_socket_paths', lambda: [str(tmp_path / 'gone')])
|
||||
monkeypatch.setattr(display_state, 'read_state', lambda: None)
|
||||
path = tmp_path / 'display-heartbeat.json'
|
||||
monkeypatch.setattr(display_watchdog, 'HEARTBEAT_PATH', str(path))
|
||||
|
||||
def beat(age, pid=None):
|
||||
path.write_text(json.dumps({'pid': os.getpid() if pid is None else pid,
|
||||
'mono': time.monotonic() - age,
|
||||
'wall': time.time() - age}))
|
||||
return beat
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def service(monkeypatch):
|
||||
status = {'active': False, 'returncode': 3, 'stdout': 'inactive', 'stderr': ''}
|
||||
monkeypatch.setattr('web_interface.blueprints.api_v3.misc._get_display_service_status',
|
||||
lambda: dict(status))
|
||||
return status
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fresh_preview(tmp_path, monkeypatch):
|
||||
"""The preview frame the display left behind, under 60 s old: on its own
|
||||
it kept the hardware check "connected"."""
|
||||
from web_interface import display_preview
|
||||
snapshot = tmp_path / 'preview.png'
|
||||
snapshot.write_bytes(b'png')
|
||||
monkeypatch.setattr(display_preview, 'SNAPSHOT_PATH', str(snapshot))
|
||||
|
||||
|
||||
def _health(client):
|
||||
resp = client.get('/api/v3/health')
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
return resp.get_json()['data']
|
||||
|
||||
|
||||
class TestHealth:
|
||||
def test_a_stopped_display_service_is_degraded(self, api_v3_client, no_display, service,
|
||||
fresh_preview):
|
||||
data = _health(api_v3_client)
|
||||
assert data['services']['display_service']['status'] == 'inactive'
|
||||
assert data['checks']['display_loop']['status'] == 'stopped'
|
||||
assert data['status'] == 'degraded'
|
||||
|
||||
def test_a_service_still_starting_is_not(self, api_v3_client, no_display, service,
|
||||
fresh_preview):
|
||||
"""Active, before its socket and first heartbeat: not stopped."""
|
||||
service.update(active=True, stdout='active', returncode=0)
|
||||
data = _health(api_v3_client)
|
||||
assert data['checks']['display_loop']['status'] == 'not_reported'
|
||||
assert data['status'] == 'healthy'
|
||||
|
||||
def test_a_display_run_by_hand_is_not_stopped(self, api_v3_client, no_display, service,
|
||||
fresh_preview):
|
||||
"""The service is off but a display process beats (sudo python3 run.py)."""
|
||||
no_display(age=2)
|
||||
data = _health(api_v3_client)
|
||||
assert data['checks']['display_loop']['status'] == 'running'
|
||||
assert data['status'] == 'healthy'
|
||||
|
||||
@pytest.mark.parametrize('platform', ['no_unix_sockets', 'socket_off'])
|
||||
def test_without_a_socket_to_expect_nothing_changes(self, api_v3_client, no_display,
|
||||
service, fresh_preview, monkeypatch,
|
||||
platform):
|
||||
"""Windows and the dev server (no systemd unit), or the socket
|
||||
deliberately off: no heartbeat is no signal, as before."""
|
||||
if platform == 'no_unix_sockets':
|
||||
monkeypatch.setattr(display_state, 'socket_supported', lambda: False)
|
||||
else:
|
||||
monkeypatch.setattr(display_state, 'client_socket_paths', lambda: [])
|
||||
service.update(returncode=-1, stdout='', stderr='systemctl not found')
|
||||
data = _health(api_v3_client)
|
||||
assert data['checks']['display_loop']['status'] == 'not_reported'
|
||||
assert data['status'] == 'healthy'
|
||||
|
||||
def test_the_status_only_answer_says_degraded(self, api_v3_client, no_display, service,
|
||||
fresh_preview, monkeypatch):
|
||||
monkeypatch.setattr('web_interface.blueprints.api_v3.misc.request_is_authenticated',
|
||||
lambda: False)
|
||||
resp = api_v3_client.get('/api/v3/health')
|
||||
assert resp.get_json()['data'] == {'status': 'degraded'}
|
||||
|
||||
|
||||
class TestCurrentStatus:
|
||||
CACHED = {'mode': 'clock', 'plugin_id': 'clock', 'is_display_active': True,
|
||||
'on_demand_active': False, 'last_updated': None}
|
||||
|
||||
@pytest.fixture
|
||||
def cached(self, api_v3_module):
|
||||
entry = dict(self.CACHED, last_updated=time.time() - 30)
|
||||
cache = api_v3_module.api_v3.cache_manager
|
||||
cache.get.side_effect = lambda key, *a, **kw: (
|
||||
dict(entry) if key == 'display_current_state' else None)
|
||||
return entry
|
||||
|
||||
def _status(self, client):
|
||||
resp = client.get('/api/v3/display/current-status')
|
||||
assert resp.status_code == 200
|
||||
return resp.get_json()['data']
|
||||
|
||||
def test_a_stopped_display_is_not_reported_active(self, api_v3_client, no_display, cached):
|
||||
data = self._status(api_v3_client)
|
||||
assert not data.get('is_display_active')
|
||||
assert data['mode'] is None and data['last_updated'] is None
|
||||
assert data['source'] == 'cache'
|
||||
|
||||
def test_a_stale_heartbeat_is_not_active_either(self, api_v3_client, no_display, cached):
|
||||
no_display(age=display_watchdog.HEARTBEAT_STALE_SECONDS + 5)
|
||||
assert self._status(api_v3_client)['mode'] is None
|
||||
|
||||
@pytest.mark.skipif(os.name != 'posix', reason='process_exists answers only on POSIX')
|
||||
def test_a_heartbeat_from_a_dead_process_is_not_active(self, api_v3_client, no_display,
|
||||
cached):
|
||||
no_display(age=1, pid=2 ** 22 + 12345)
|
||||
assert self._status(api_v3_client)['mode'] is None
|
||||
|
||||
def test_a_live_heartbeat_without_a_socket_reads_the_cache(self, api_v3_client,
|
||||
no_display, cached):
|
||||
"""An older display with no socket, still running."""
|
||||
no_display(age=2)
|
||||
data = self._status(api_v3_client)
|
||||
assert data['mode'] == 'clock' and data['is_display_active'] is True
|
||||
|
||||
@pytest.mark.parametrize('platform', ['no_unix_sockets', 'socket_off'])
|
||||
def test_without_a_socket_to_expect_the_cache_answers(self, api_v3_client, no_display,
|
||||
cached, monkeypatch, platform):
|
||||
if platform == 'no_unix_sockets':
|
||||
monkeypatch.setattr(display_state, 'socket_supported', lambda: False)
|
||||
else:
|
||||
monkeypatch.setattr(display_state, 'client_socket_paths', lambda: [])
|
||||
data = self._status(api_v3_client)
|
||||
assert data['mode'] == 'clock' and data['is_display_active'] is True
|
||||
@@ -489,3 +489,171 @@ class TestConcurrency:
|
||||
|
||||
assert live["peak"] <= espn_dates.ESPN_CHUNK_WORKERS
|
||||
assert live["peak"] > 1, "chunks should actually overlap"
|
||||
|
||||
|
||||
class TestEdgeMonths:
|
||||
"""A window's partial edge months are asked whole and trimmed.
|
||||
|
||||
The default scoreboard window -- a fortnight either side of today -- spans
|
||||
two partial months, so it used to cost 29 day requests per league. ESPN's
|
||||
``dates=YYYYMMDD`` means a US Eastern day (verified against the live API
|
||||
on 2026-10-03, 417 of 417 soccer events), so a month answer trimmed to
|
||||
the window's Eastern days is what the day requests returned.
|
||||
"""
|
||||
|
||||
def test_a_fortnight_either_side_is_two_requests(self):
|
||||
planned = espn_dates.espn_request_chunks(date(2026, 9, 20), date(2026, 10, 18))
|
||||
assert planned == [
|
||||
("202609", (date(2026, 9, 20), date(2026, 9, 30))),
|
||||
("202610", (date(2026, 10, 1), date(2026, 10, 18))),
|
||||
]
|
||||
|
||||
def test_a_live_polls_two_days_stay_two_days(self):
|
||||
planned = espn_dates.espn_request_chunks(date(2026, 10, 2), date(2026, 10, 3))
|
||||
assert planned == [("20261002", None), ("20261003", None)]
|
||||
|
||||
def test_the_threshold_is_inclusive(self):
|
||||
n = espn_dates.ESPN_MONTH_COVER_MIN_DAYS
|
||||
short = espn_dates.espn_request_chunks(date(2026, 10, 1), date(2026, 10, n - 1))
|
||||
assert [chunk for chunk, _ in short] == [
|
||||
"202610%02d" % day for day in range(1, n)]
|
||||
enough = espn_dates.espn_request_chunks(date(2026, 10, 1), date(2026, 10, n))
|
||||
assert enough == [("202610", (date(2026, 10, 1), date(2026, 10, n)))]
|
||||
|
||||
def test_whole_months_and_short_edges_are_unchanged(self):
|
||||
planned = espn_dates.espn_request_chunks(date(2026, 8, 30), date(2026, 10, 2))
|
||||
assert planned == [
|
||||
("20260830", None), ("20260831", None), ("202609", None),
|
||||
("20261001", None), ("20261002", None),
|
||||
]
|
||||
|
||||
def test_without_time_zone_data_edges_stay_days(self, monkeypatch):
|
||||
monkeypatch.setattr(espn_dates, "_EASTERN", None)
|
||||
planned = espn_dates.espn_request_chunks(date(2026, 9, 20), date(2026, 10, 18))
|
||||
assert len(planned) == 29
|
||||
assert all(trim is None for _, trim in planned)
|
||||
|
||||
@pytest.mark.parametrize("start,end", [
|
||||
(date(2026, 9, 20), date(2026, 10, 18)),
|
||||
(date(2026, 1, 25), date(2026, 3, 3)),
|
||||
(date(2026, 12, 20), date(2027, 1, 9)),
|
||||
(date(2026, 10, 5), date(2026, 10, 9)),
|
||||
])
|
||||
def test_the_planned_requests_still_cover_every_day_exactly_once(self, start, end):
|
||||
covered = []
|
||||
for chunk, trim in espn_dates.espn_request_chunks(start, end):
|
||||
if trim is None:
|
||||
covered.extend(days_covered_by([chunk]))
|
||||
else:
|
||||
assert chunk == trim[0].strftime("%Y%m") == trim[1].strftime("%Y%m")
|
||||
covered.extend(trim[0] + timedelta(days=offset)
|
||||
for offset in range((trim[1] - trim[0]).days + 1))
|
||||
expected = [start + timedelta(days=offset) for offset in range((end - start).days + 1)]
|
||||
assert covered == expected
|
||||
|
||||
def test_a_trimmed_month_keeps_only_the_windows_eastern_days(self):
|
||||
september = [
|
||||
# 03:30Z on the 20th is still the 19th in New York: outside.
|
||||
{"id": "before", "date": "2026-09-20T03:30Z"},
|
||||
{"id": "first", "date": "2026-09-20T14:00Z"},
|
||||
{"id": "late", "date": "2026-09-30T23:30Z"},
|
||||
]
|
||||
october = [
|
||||
{"id": "oct1", "date": "2026-10-01T19:00Z"},
|
||||
# 03:30Z on the 19th is the evening of the 18th in New York: inside.
|
||||
{"id": "last", "date": "2026-10-19T03:30Z"},
|
||||
{"id": "after", "date": "2026-10-19T14:00Z"},
|
||||
{"id": "undated"},
|
||||
]
|
||||
session = FakeSession({"202609": september, "202610": october})
|
||||
|
||||
data = fetch_espn_date_chunks(session, URL, params={"dates": "20260920-20261018"})
|
||||
|
||||
assert sorted(call["dates"] for call in session.calls) == ["202609", "202610"]
|
||||
# An event with no readable date is kept, never dropped on a guess.
|
||||
assert [e["id"] for e in data["events"]] == ["first", "late", "oct1", "last", "undated"]
|
||||
|
||||
def test_eastern_standard_time_is_honoured_after_the_clocks_change(self):
|
||||
# 2026-11-01 ends daylight saving: Eastern is UTC-5 from then on.
|
||||
november = [
|
||||
{"id": "out", "date": "2026-11-15T04:30Z"}, # Nov 14, 23:30 EST
|
||||
{"id": "in", "date": "2026-11-15T05:30Z"}, # Nov 15, 00:30 EST
|
||||
]
|
||||
session = FakeSession({"202611": november})
|
||||
data = fetch_espn_date_chunks(session, URL, params={"dates": "20261115-20261121"})
|
||||
assert [e["id"] for e in data["events"]] == ["in"]
|
||||
|
||||
def test_a_capped_edge_month_re_asks_only_the_windows_days(self):
|
||||
full = [{"id": "cap%d" % i, "date": "2026-10-05T18:00Z"} for i in range(ESPN_MAX_LIMIT)]
|
||||
by_chunk = {"202610": full}
|
||||
by_chunk.update({"202610%02d" % day: [{"id": "o%02d" % day}] for day in range(1, 32)})
|
||||
session = FakeSession(by_chunk)
|
||||
|
||||
data = fetch_espn_date_chunks(session, URL, params={"dates": "20261001-20261010"})
|
||||
|
||||
sent = [call["dates"] for call in session.calls]
|
||||
assert sent[0] == "202610"
|
||||
assert sorted(sent[1:]) == ["202610%02d" % day for day in range(1, 11)]
|
||||
assert [e["id"] for e in data["events"]] == ["o%02d" % day for day in range(1, 11)]
|
||||
|
||||
|
||||
class TestProcessWideChunkCap:
|
||||
"""The chunk cap holds across windows, not per window.
|
||||
|
||||
A soccer board starting eight leagues fetches sixteen windows at once.
|
||||
With a pool of ``ESPN_CHUNK_WORKERS`` each, ~40 requests were in flight
|
||||
and every one past a session's pool opened a connection -- and a DNS
|
||||
lookup. On ledpi that was ~90 NameResolutionErrors per start.
|
||||
"""
|
||||
|
||||
def test_concurrent_windows_share_one_budget(self):
|
||||
live = {"now": 0, "peak": 0}
|
||||
guard = threading.Lock()
|
||||
|
||||
class CountingSession(FakeSession):
|
||||
def get(self, url, params=None, headers=None, timeout=None):
|
||||
with guard:
|
||||
live["now"] += 1
|
||||
live["peak"] = max(live["peak"], live["now"])
|
||||
try:
|
||||
time.sleep(0.01)
|
||||
return super().get(url, params=params, headers=headers, timeout=timeout)
|
||||
finally:
|
||||
with guard:
|
||||
live["now"] -= 1
|
||||
|
||||
sessions = [CountingSession() for _ in range(6)]
|
||||
# Six leagues, so the fetch service cannot merge them into one, on a
|
||||
# host with no token bucket: earlier tests may have spent ESPN's
|
||||
# burst, and a bucket paced at 20/s would serialise these by itself.
|
||||
threads = [
|
||||
threading.Thread(target=fetch_espn_date_chunks,
|
||||
args=(session, "https://scores.example.test/league%d" % index),
|
||||
kwargs={"params": {"dates": "20260101-20261231"}})
|
||||
for index, session in enumerate(sessions)
|
||||
]
|
||||
for thread in threads:
|
||||
thread.start()
|
||||
for thread in threads:
|
||||
thread.join(timeout=30)
|
||||
|
||||
assert all(len(session.calls) == 12 for session in sessions)
|
||||
assert live["peak"] <= espn_dates.ESPN_CHUNK_WORKERS
|
||||
assert live["peak"] > 1, "chunks should still overlap"
|
||||
|
||||
|
||||
def test_a_fresh_process_skips_the_doomed_range_request():
|
||||
"""Every start used to spend one 400 per window learning that ranges are
|
||||
still rejected -- eleven at once from a soccer board. A new process now
|
||||
starts inside the retry period instead."""
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
out = subprocess.run(
|
||||
[sys.executable, "-c",
|
||||
"import src.common.espn_dates as e; print(e._ranges_known_rejected())"],
|
||||
cwd=str(Path(__file__).resolve().parents[1]),
|
||||
capture_output=True, text=True, timeout=60,
|
||||
)
|
||||
assert out.stdout.strip() == "True", out.stderr
|
||||
|
||||
@@ -670,14 +670,14 @@ class TestCallerIdentity:
|
||||
assert _counters(global_service, plugin="football-scoreboard")["requests"] == 1
|
||||
|
||||
def test_espn_chunks_on_worker_threads_count_against_the_caller(self, global_service):
|
||||
from src.common.espn_dates import espn_date_chunks, fetch_espn_date_chunks, parse_espn_date_range
|
||||
from src.common.espn_dates import espn_request_chunks, fetch_espn_date_chunks, parse_espn_date_range
|
||||
|
||||
session = FakeSession(lambda url, kw: make_response(body=b'{"events": []}', url=url))
|
||||
dates = "20260801-20261015"
|
||||
with plugin_scope("baseball-scoreboard"):
|
||||
fetch_espn_date_chunks(session, "https://site.api.espn.com/s/scoreboard",
|
||||
params={"dates": dates})
|
||||
chunks = len(espn_date_chunks(*parse_espn_date_range(dates)))
|
||||
chunks = len(espn_request_chunks(*parse_espn_date_range(dates)))
|
||||
assert chunks > 1
|
||||
assert len(session.calls) == chunks
|
||||
assert _counters(global_service, plugin="baseball-scoreboard")["requests"] == chunks
|
||||
|
||||
@@ -0,0 +1,183 @@
|
||||
"""A frame the preview throttle skipped still reaches the snapshot.
|
||||
|
||||
The preview snapshot (/api/v3/display/current, the web UI's live preview) is
|
||||
only written from update_display(), at most once per write interval. A screen
|
||||
that draws its card once and then holds it -- soccer's recent/upcoming cards
|
||||
skip redundant redraws -- pushes exactly one frame. When that push lands inside
|
||||
the interval, e.g. a few milliseconds after the on-demand start's clear wrote a
|
||||
black frame, the throttle skips it and nothing ever writes it: on ledpi the
|
||||
preview stayed black for soccer's whole 15 s screen while the panel showed the
|
||||
card, and the next screen "rendered immediately".
|
||||
|
||||
Runs the real DisplayManager on the emulator, like test_display_dirty_tracking.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import types
|
||||
|
||||
os.environ["EMULATOR"] = "true"
|
||||
|
||||
import pytest
|
||||
from PIL import Image
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def dm(tmp_path_factory):
|
||||
from src.display_manager import DisplayManager
|
||||
DisplayManager._instance = None
|
||||
manager = DisplayManager({
|
||||
"display": {
|
||||
"hardware": {"rows": 32, "cols": 64, "chain_length": 2,
|
||||
"parallel": 1, "brightness": 90},
|
||||
"runtime": {"gpio_slowdown": 0},
|
||||
},
|
||||
}, suppress_test_pattern=True)
|
||||
manager._snapshot_path = str(
|
||||
tmp_path_factory.mktemp("owed_snapshot") / "led_matrix_preview.png")
|
||||
yield manager
|
||||
DisplayManager._instance = None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def viewer(dm, monkeypatch, tmp_path):
|
||||
"""A preview is open (1 s write interval); fresh snapshot bookkeeping."""
|
||||
monkeypatch.setattr(dm, "_viewer_is_fresh", lambda now: True)
|
||||
dm._viewer_was_fresh = True
|
||||
dm._snapshot_path = str(tmp_path / "snap.png")
|
||||
dm._last_snapshot_ts = 0.0
|
||||
dm._last_snapshot_touch_ts = 0.0
|
||||
dm._last_snapshot_digest = None
|
||||
dm._saved_snapshot_digest = None
|
||||
dm._snapshot_owed = False
|
||||
dm.set_scrolling_state(False)
|
||||
return dm
|
||||
|
||||
|
||||
def _lit(path):
|
||||
with Image.open(path) as img:
|
||||
return sum(1 for p in img.convert("RGB").getdata() if max(p) > 20)
|
||||
|
||||
|
||||
def _age_last_write(dm, seconds=2.0):
|
||||
"""As if `seconds` had passed since the last snapshot write."""
|
||||
dm._last_snapshot_ts -= seconds
|
||||
dm._last_snapshot_touch_ts -= seconds
|
||||
|
||||
|
||||
def _clear_then_draw_card(dm):
|
||||
"""The on-demand start's clear, then the card a few ms later."""
|
||||
dm.clear()
|
||||
dm.update_display() # black frame: written
|
||||
assert _lit(dm._snapshot_path) == 0
|
||||
dm.draw.rectangle([4, 4, 40, 20], fill=(255, 255, 0))
|
||||
dm.update_display() # the card: inside the interval
|
||||
|
||||
|
||||
def _controller(dm):
|
||||
from src import display_controller as dc_module
|
||||
controller = dc_module.DisplayController.__new__(dc_module.DisplayController)
|
||||
controller.plugin_manager = None
|
||||
controller.display_manager = dm
|
||||
return controller
|
||||
|
||||
|
||||
class _HoldingPlugin:
|
||||
"""Already showing its card: display() returns True and draws nothing."""
|
||||
|
||||
plugin_id = "holding"
|
||||
|
||||
def __init__(self):
|
||||
self.calls = 0
|
||||
|
||||
def display(self, display_mode=None, force_clear=False):
|
||||
self.calls += 1
|
||||
return True
|
||||
|
||||
|
||||
def test_a_held_card_reaches_the_preview_on_the_next_frame(viewer):
|
||||
dm = viewer
|
||||
_clear_then_draw_card(dm)
|
||||
assert _lit(dm._snapshot_path) == 0 # the throttle skipped the card
|
||||
|
||||
controller = _controller(dm)
|
||||
plugin = _HoldingPlugin()
|
||||
_age_last_write(dm)
|
||||
# The render loop's next frame: the plugin draws nothing and makes no
|
||||
# update_display() call, as soccer's switch cards do.
|
||||
assert controller._display_once(plugin, "soccer_eng.1_recent", True) is True
|
||||
assert plugin.calls == 1
|
||||
assert _lit(dm._snapshot_path) > 0
|
||||
|
||||
|
||||
def test_the_owed_write_still_waits_out_the_interval(viewer, monkeypatch):
|
||||
dm = viewer
|
||||
_clear_then_draw_card(dm)
|
||||
saves = []
|
||||
monkeypatch.setattr(dm, "_save_snapshot", lambda image: saves.append(image))
|
||||
dm.write_owed_snapshot() # still inside the interval
|
||||
assert saves == []
|
||||
_age_last_write(dm)
|
||||
dm.write_owed_snapshot()
|
||||
assert len(saves) == 1
|
||||
# Written: nothing is owed, so later frames do no work and the unchanged
|
||||
# frame is not encoded again.
|
||||
assert dm._snapshot_owed is False
|
||||
_age_last_write(dm)
|
||||
dm.write_owed_snapshot()
|
||||
assert len(saves) == 1
|
||||
|
||||
|
||||
def test_a_failed_owed_write_stays_owed_and_is_retried(viewer, monkeypatch):
|
||||
dm = viewer
|
||||
_clear_then_draw_card(dm)
|
||||
_age_last_write(dm)
|
||||
attempts = []
|
||||
|
||||
def failing_save(image):
|
||||
attempts.append(image)
|
||||
raise OSError("disk full")
|
||||
|
||||
monkeypatch.setattr(dm, "_save_snapshot", failing_save)
|
||||
dm.write_owed_snapshot() # the write fails
|
||||
assert len(attempts) == 1
|
||||
assert dm._snapshot_owed is True # still owed: a held screen
|
||||
saves = [] # makes no update_display()
|
||||
monkeypatch.setattr(dm, "_save_snapshot", lambda image: saves.append(image))
|
||||
dm.write_owed_snapshot() # retried on the next frame
|
||||
assert len(saves) == 1
|
||||
assert dm._snapshot_owed is False
|
||||
|
||||
|
||||
def test_nothing_owed_after_a_frame_that_was_written(viewer, monkeypatch):
|
||||
dm = viewer
|
||||
dm.draw.rectangle([0, 0, 8, 8], fill=(0, 255, 0))
|
||||
dm.update_display() # due: written at once
|
||||
assert dm._snapshot_owed is False
|
||||
calls = []
|
||||
monkeypatch.setattr(dm, "_write_snapshot_if_due",
|
||||
lambda *a, **k: calls.append(a))
|
||||
dm.write_owed_snapshot()
|
||||
assert calls == []
|
||||
|
||||
|
||||
def test_an_unchanged_frame_inside_the_interval_is_not_owed(viewer):
|
||||
dm = viewer
|
||||
dm.draw.rectangle([0, 0, 8, 8], fill=(0, 0, 255))
|
||||
dm.update_display()
|
||||
dm.update_display() # same frame, inside the interval
|
||||
assert dm._snapshot_owed is False
|
||||
|
||||
|
||||
def test_a_controller_without_the_hook_still_draws():
|
||||
"""Controllers built without a display manager (tests) are unaffected."""
|
||||
from src import display_controller as dc_module
|
||||
controller = dc_module.DisplayController.__new__(dc_module.DisplayController)
|
||||
controller.plugin_manager = None
|
||||
plugin = _HoldingPlugin()
|
||||
assert controller._display_once(plugin, "x", True) is True
|
||||
controller.display_manager = types.SimpleNamespace()
|
||||
assert controller._display_once(plugin, "x", True) is True
|
||||
assert plugin.calls == 2
|
||||
@@ -612,6 +612,12 @@ class TestEndToEnd:
|
||||
cached['display_current_state'] = {'mode': 'from-cache', 'last_updated': 1}
|
||||
path = str(tmp_path / 'control.sock')
|
||||
monkeypatch.setenv(c.SOCKET_PATH_ENV, path)
|
||||
# The display is this process, and its render loop is beating: the
|
||||
# cache is then still its answer once the socket goes.
|
||||
heartbeat = tmp_path / 'display-heartbeat.json'
|
||||
heartbeat.write_text(json.dumps({'pid': os.getpid(), 'mono': time.monotonic(),
|
||||
'wall': time.time()}))
|
||||
monkeypatch.setattr(display_watchdog, 'HEARTBEAT_PATH', str(heartbeat))
|
||||
hub = _hub_with_everything()
|
||||
server = ControlServer(path, state_hub=hub, keepalive=0.2)
|
||||
assert server.start()
|
||||
@@ -638,3 +644,8 @@ class TestEndToEnd:
|
||||
break
|
||||
time.sleep(0.05)
|
||||
assert (data['mode'], data['source']) == ('from-cache', 'cache')
|
||||
# Stopped: systemd takes the heartbeat's directory with it, and the
|
||||
# cache's last answer is no longer anyone's.
|
||||
heartbeat.unlink()
|
||||
data = _data(client, '/api/v3/display/current-status')
|
||||
assert (data['mode'], data['source']) == (None, 'cache')
|
||||
|
||||
@@ -221,7 +221,9 @@ class TestTheWebProcessNeverRunsAPlugin:
|
||||
def test_saving_its_section_through_the_main_config(self, web):
|
||||
body = web.post("/api/v3/config/main", {PLUGIN_ID: {"message": "via main"}})
|
||||
assert web.stored()["message"] == "via main"
|
||||
assert body["restart_required"] is True
|
||||
# The display's config watcher hands the section to the running
|
||||
# plugin (on_config_change), as for /plugins/config: no restart.
|
||||
assert body["restart_required"] is False
|
||||
assert web.ran() == []
|
||||
|
||||
def test_resetting_its_config(self, web):
|
||||
|
||||
Reference in New Issue
Block a user