rick/LEDMatrix
1
0
Fork 0
mirror of https://github.com/ChuckBuilds/LEDMatrix.git synced 2026-10-10 09:06:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge main into claude/on-demand-error-clears-session

Browse Source
This commit is contained in:
Chuck
2026-10-09 11:55:22 -04:00
parent 4b9fc3c755 1bcb524fc3
commit 9234332c5a
161 changed files with 2731 additions and 4252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/_api_v3_test_helpers.py
+1 -2
View File
@@ -41,8 +41,7 @@ def mock_plugin_catalog():
"""
from src.plugin_system.plugin_catalog import PluginCatalog
catalog = MagicMock(spec=PluginCatalog)
for name in ('plugins_dir', 'config_manager', 'schema_manager',
'plugin_manifests', 'plugin_directories'):
for name in ('plugins_dir', 'plugin_manifests', 'plugin_directories'):
setattr(catalog, name, MagicMock())
return catalog
test/fixtures/api_v3_url_map.json
Vendored
+9
View File
@@ -175,6 +175,15 @@
"POST"
]
],
[
"/api/v3/config/refresh-rate",
"api_v3.get_refresh_rate",
[
"GET",
"HEAD",
"OPTIONS"
]
],
[
"/api/v3/config/schedule",
"api_v3.get_schedule_config",
test/js/README.md
+1 -1
View File
@@ -71,7 +71,7 @@ server has none.
| `dom/test_raw_json_page.js` | yes | The Config Editor tab (`js/pages/raw-json.js`): one POST per Save after repeated swaps, Format/Validate, invalid JSON never sent, a save survives a swap, the old global entry points |
| `dom/test_schedule_page.js` | yes | The Schedule tab (`js/pages/schedule.js`) with the real `schedule-picker` widget: both pickers drawn once per swap from the saved config, one notification per save answer after repeated swaps, the brightness label, a late widget waited for, the old global entry points |
| `dom/test_visibility_service.js` | yes (no server) | `js/core/visibility.js` with the real `LEDVisibility` from `app-shell.js` and the real registry: start/stop with the active tab and the browser tab's visibility, no interval while hidden or after a swap-out, registrations independent, the no-`LEDVisibility` fallback |
| `dom/test_display_page.js` | yes | The Display tab (`js/pages/display.js`) with the real `plugin-order-list` widget and `LEDVisibility`: one page, one sync interval and one action per control after repeated swaps, the sync poll only while on screen and never after a swap-out, sync states as text, the debounced scroll-speed hint, `updateSyncUI`'s entry point |
| `dom/test_display_page.js` | yes | The Display tab (`js/pages/display.js`) with the real `plugin-order-list` widget and `LEDVisibility`: one page, one sync interval and one action per control after repeated swaps, the sync poll only while on screen and never after a swap-out, sync states as text, the debounced scroll-speed hint, the refresh-cap hint and its "Use N Hz" button, `updateSyncUI`'s entry point |
| `dom/test_general_page.js` | yes | The General tab (`js/pages/general.js`) with the real `timezone-selector` widget: the picker drawn once per swap, one request per Security action after repeated swaps, hostile token names stay text, refused/network/login answers, a write survives a swap, `webLogin`'s entry points |
| `dom/test_backup_restore_page.js` | yes | The Backup & Restore tab (`js/pages/backup-restore.js`): one request per action after repeated swaps, the upload and restore options, reads cancelled and writes not on a swap, hostile names stay text, the old global entry points |
| `dom/test_tools_sections.js` | yes | The Tools tab's MQTT bridge and Pixlet editor sections: form prefill, the write-only password (blank means unchanged), the running-session banner and countdown, and that the editor link points at the host you loaded the page from |
test/js/dom/test_display_page.js
+11
View File
@@ -88,6 +88,8 @@ const ok = (l, c, x) => c ? (pass++, console.log(' ok ' + l))
let syncAnswer = { status: 'success', data: { role: 'leader', state: 'no_peer' } };
let syncMode = 'ok';
let advice = smooth;
const shortfall = { measured_hz: 110.4, planned_hz: 120, suggested_cap_hz: 100, slow_percent: 8 };
let refreshAnswer = { status: 'success', data: { planned_hz: 120, measured_hz: 110.4, shortfall } };
const requests = [];
function fakeFetch(url, init = {}) {
requests.push(url);
@@ -99,6 +101,7 @@ const ok = (l, c, x) => c ? (pass++, console.log(' ok ' + l))
});
if (url === '/api/v3/plugins/installed') return respond(200, { status: 'success', data: { plugins } });
if (url.startsWith('/api/v3/config/scroll-speed-advice?')) return respond(200, advice);
if (url === '/api/v3/config/refresh-rate') return respond(200, refreshAnswer);
if (url === '/api/v3/sync/status') {
if (syncMode === 'network') return Promise.reject(new TypeError('Failed to fetch'));
if (syncMode === 'login') return respond(401, { status: 'error' }, { 'X-LEDMatrix-Login': '/login' });
@@ -148,6 +151,14 @@ const ok = (l, c, x) => c ? (pass++, console.log(' ok ' + l))
// ── first load ──────────────────────────────────────────────────────────
ok('one plugin-list request on start', count('/api/v3/plugins/installed') === 1, requests);
ok('one scroll-speed hint request on start (after the debounce)', count('/api/v3/config/scroll-speed-advice') === 1, requests);
const refreshHint = $('limit_refresh_rate_hz_hint');
ok('a panel short of its cap says so, as text, with a button for a cap it can hold',
/about 110 Hz, below this 120 Hz cap.*8% slower/.test(refreshHint.textContent)
&& refreshHint.querySelector('button').textContent === 'Use 100 Hz', refreshHint.textContent);
refreshHint.querySelector('button').click();
ok('the button fills the field and says to save and restart',
$('limit_refresh_rate_hz').value === '100' && /Save, then restart/.test(refreshHint.textContent),
[$('limit_refresh_rate_hz').value, refreshHint.textContent]);
ok('the saved role is standalone: no sync request, no interval work',
$('sync_role').value === 'standalone' && syncPolls() === 0, [$('sync_role').value, syncPolls()]);
ok('the sync poll interval runs while the tab is on screen', intervals.size === 1
test/js/unit/test_html_escaping.js
+8 -35
View File
@@ -12,9 +12,10 @@
//
// CodeQL reported 83 js/incomplete-html-attribute-sanitization alerts for
// exactly this. The web UI now has one implementation, window.LEDEscape in
// app-early.js, and the old per-file escapers are one-line names for it. This
// suite runs LEDEscape and every one of those names as shipped, and fails if a
// hand-rolled escaper appears anywhere else in web_interface/.
// app-early.js, which every page and widget calls directly (BaseWidget keeps
// an escapeHtml method for plugin widgets). This suite runs LEDEscape and that
// method as shipped, and fails if a hand-rolled escaper appears anywhere else
// in web_interface/.
const fs = require('fs');
const path = require('path');
@@ -85,30 +86,6 @@ const ESCAPERS = [
['app-early.js (LEDEscape.attr)', null, null, 'attr'],
['base-widget.js (BaseWidget.escapeHtml)',
'static/v3/js/widgets/base-widget.js', 'escapeHtml(text) {', 'escapeHtml', true],
['plugins_manager.js (top-level escapeHtml)',
'static/v3/plugins_manager.js', 'function escapeHtml(text) {', 'escapeHtml', false],
['plugins_manager.js (starlark escapeHtml)',
'static/v3/plugins_manager.js', 'function escapeHtml(str) {', 'escapeHtml', false],
['json-file-manager.js (_esc)',
'static/v3/js/widgets/json-file-manager.js', '_esc(str) {', '_esc', true],
['plugin-file-manager.js (escHtml)',
'static/v3/js/widgets/plugin-file-manager.js', 'function escHtml(s) {', 'escHtml', false],
['plugins_manager.js (escapeAttribute)',
'static/v3/plugins_manager.js', 'function escapeAttribute(text) {', 'escapeAttribute', false],
['notification.js (escapeHtml)',
'static/v3/js/widgets/notification.js', 'function escapeHtml(text) {', 'escapeHtml', false],
['google-calendar-picker.js (escapeHtml)',
'static/v3/js/widgets/google-calendar-picker.js', 'function escapeHtml(str) {', 'escapeHtml', false],
['text-input.js (escapeHtml)',
'static/v3/js/widgets/text-input.js', 'function escapeHtml(text) {', 'escapeHtml', false],
['slider.js (escapeAttr)',
'static/v3/js/widgets/slider.js', 'function escapeAttr(text) {', 'escapeAttr', false],
['tools.html (escHtml)',
'templates/v3/partials/tools.html', 'function escHtml(s) {', 'escHtml', false],
['tools.html (phEscape)',
'templates/v3/partials/tools.html', 'function phEscape(s) {', 'phEscape', false],
['logs.html (escapeHtml)',
'templates/v3/partials/logs.html', 'function escapeHtml(text) {', 'escapeHtml', false],
// cache.html, backup_restore.html, operation_history.html and display.html
// have no script any more (display.html's two escapers were never called):
// their js/pages/ modules draw server data with textContent, and each
@@ -169,9 +146,7 @@ console.log('\n4b. LEDEscape.jsStringAttr: a JS string literal that survives an
console.log('\n4c. no hand-rolled escaper outside app-early.js');
{
const skip = new Set(['static/v3/js/app-early.js',
// documentation example, kept self-contained on purpose
'static/v3/js/widgets/example-color-picker.js']);
const skip = new Set(['static/v3/js/app-early.js']);
const found = [];
const walk = dir => fs.readdirSync(dir, { withFileTypes: true }).forEach(e => {
const p = path.join(dir, e.name);
@@ -308,7 +283,7 @@ console.log('\n6. url-input onInput: previewLink.href is guarded at the sink');
// ── plugin-file-manager: cell edits travel via data-*, not inline handlers ──
// A JSON key/day from an uploaded file used to be spliced, HTML-escaped,
// into an oninput="...('${escHtml(col)}'...)" attribute. escHtml neutralises
// into an oninput="...('${escHtml(col)}'...)" attribute. Escaping neutralises
// a quote for an ordinary attribute, but here the value also has to survive
// as a *JS string literal* -- the browser HTML-decodes the attribute before
// running it as script, which turns the escaped quote back into a real one
@@ -332,11 +307,10 @@ console.log("\n7. plugin-file-manager: cell edits never go through an inline han
process.exit(1);
}
const escHtmlFn = loadFn('static/v3/js/widgets/plugin-file-manager.js', 'function escHtml(s) {', 'escHtml', false);
const renderEntryTableSrc = extractFn('function renderEntryTable(fieldId, container, content) {');
const calls = [];
const fakeWindow = { _pfmCellEdit: (fieldId, day, col, value) => calls.push({ fieldId, day, col, value }) };
const fakeWindow = { LEDEscape, _pfmCellEdit: (fieldId, day, col, value) => calls.push({ fieldId, day, col, value }) };
class FakeContainer {
constructor() { this._html = ''; this._listeners = {}; }
@@ -354,12 +328,11 @@ console.log("\n7. plugin-file-manager: cell edits never go through an inline han
}
// eslint-disable-next-line no-eval
const renderEntryTable = eval(`(function(getState, escHtml, safeSetHTML, window){
const renderEntryTable = eval(`(function(getState, safeSetHTML, window){
${renderEntryTableSrc}
return renderEntryTable;
})`)(
() => ({ entriesPerPage: 20, _tablePage: 1 }),
escHtmlFn,
(target, html) => { target.innerHTML = html; },
fakeWindow
);
test/js/unit/test_inline_handler_escaping.js
+3 -3
View File
@@ -62,12 +62,12 @@ global.setGridHtmlIfChanged = (container, html) => { container.innerHTML = html;
// eslint-disable-next-line no-eval
eval([
'function escapeHtml(text) {', 'function escapeAttribute(text) {', 'function jsStringAttr(value) {',
'function jsStringAttr(value) {',
'function renderPluginStore(plugins) {', 'function renderSavedRepositories(repositories) {',
'function renderCustomRegistryPlugins(plugins, registryUrl) {',
].map(extract).join('\n') + '\nglobal.jsStringAttr = jsStringAttr; global.escapeHtml = escapeHtml;'
].map(extract).join('\n') + '\nglobal.jsStringAttr = jsStringAttr;'
+ '\nglobal.renderPluginStore = renderPluginStore; global.renderSavedRepositories = renderSavedRepositories;'
+ '\nglobal.renderCustomRegistryPlugins = renderCustomRegistryPlugins; global.escapeAttribute = escapeAttribute;');
+ '\nglobal.renderCustomRegistryPlugins = renderCustomRegistryPlugins;');
// ── minimal HTML start-tag tokenizer ───────────────────────────────────────
function decodeEntities(s) {
test/js/unit/test_render_cards.js
+2 -2
View File
@@ -16,7 +16,7 @@ const container = {
innerHTML: '',
querySelectorAll: () => [], // no skeletons in this harness
};
// escapeHtml() escapes via a detached element, so mirror what a browser does
// LEDEscape.html() escapes via a detached element, so mirror what a browser does
// when you read innerHTML back off textContent: & < > are escaped, quotes are not.
class FakeEl {
set textContent(v) { this._t = String(v == null ? '' : v); }
@@ -36,7 +36,7 @@ global.PLUGIN_DEBUG = false;
global.debugLog = () => {};
function setupInstalledEventDelegation() {} // stubbed; tested separately
eval(slice('function escapeHtml(text)', '\nfunction isNewPlugin'));
eval(slice('function jsStringAttr(value)', '\nfunction isNewPlugin'));
eval(slice('function renderInstalledCards(plugins, total)',
'// Set up event delegation for plugin action buttons'));
test/js/unit/test_store_registry_fields.js
+1 -1
View File
@@ -51,7 +51,7 @@ global.installedPlugins = [];
// eslint-disable-next-line no-eval
eval([
'function escapeHtml(text) {', 'function escapeAttribute(text) {', 'function jsStringAttr(value) {',
'function jsStringAttr(value) {',
'function isStorePluginInstalled(pluginIdOrPlugin) {',
'function findInstalledStorePlugin(pluginIdOrPlugin) {', 'function renderPluginStore(plugins) {',
].map(extract).join('\n') + '\nglobal.renderPluginStore = renderPluginStore;'
test/js/unit/test_update_all.js
+3 -2
View File
@@ -253,9 +253,10 @@ const noSleep = { sleep: async () => {} };
for (const endpoint of bad) codes.push(await refusal(endpoint));
ok('an endpoint that could leave the API path is refused before fetch()',
codes.every(c => c === 'INVALID_ENDPOINT') && urls.length === 0, { codes, urls });
await PluginAPI.resetPluginConfig('a/../b&x=1');
global.debugLog = () => {}; // GETs go through the throttler, which logs
await PluginAPI.getPluginHealth('a/../b&x=1');
ok('a plugin id is encoded into the URL, not spliced into it',
urls[0] === '/api/v3/plugins/config/reset?plugin_id=a%2F..%2Fb%26x%3D1', urls);
urls[0] === '/api/v3/plugins/health/a%2F..%2Fb%26x%3D1', urls);
delete global.fetch;
}
test/test_api_v3_display_modes.py
+12 -1
View File
@@ -7,7 +7,7 @@ manifest.json off disk and reimplemented PluginManager's own fallbacks.
"""
import json
from unittest.mock import MagicMock
from unittest.mock import MagicMock, patch
import pytest
@@ -155,3 +155,14 @@ class TestOneBadConfigSectionDoesNotBlankTheList:
side_effect=RuntimeError("GET https://x/y?api_key=SEC123 failed"))
body = api_v3_client.get('/api/v3/display/modes').get_json()
assert 'SEC123' not in json.dumps(body)
class TestOnDemandUsesTheRegisteredSpelling:
def test_a_mode_differing_in_case_is_sent_as_registered(self, client):
with patch('web_interface.blueprints.api_v3.display._deliver_on_demand',
return_value=('socket', None)) as deliver:
response = client.post('/api/v3/display/on-demand/start',
json={'plugin_id': 'football-scoreboard',
'mode': 'NFL_LIVE', 'start_service': False})
assert response.status_code == 200, response.get_json()
assert deliver.call_args.args[0]['mode'] == 'nfl_live'
test/test_api_v3_install_restart_installed_id.py
+123
View File
@@ -0,0 +1,123 @@
"""POST /plugins/install asks for a restart by the id the plugin installed as.
A store install needs a display restart when config.json already enables the
plugin (a reinstall, or a config carried over): the display loads a plugin
when its ``enabled`` flag changes, and this flag did not. The route read the
flag under the registry id it was given. An aliased entry installs under
another id -- ``weather`` installs a directory whose manifest declares
``ledmatrix-weather``, and its config section is ``ledmatrix-weather`` -- so
reinstalling an enabled Weather never reported that a restart was needed,
and the display kept running the old copy.
"""
import json
from unittest.mock import MagicMock
import pytest
from test._api_v3_test_helpers import api_v3_client, api_v3_module # noqa: F401
INSTALL = "/api/v3/plugins/install"
@pytest.fixture
def store(api_v3_module, tmp_path):
"""The store installs registry entry ``weather`` as ``installed_id``."""
manager = api_v3_module.api_v3.plugin_store_manager
manager.install_plugin.return_value = True
manager.get_registry_info.return_value = None
manager._find_plugin_path.return_value = None
def installs_as(installed_id):
path = tmp_path / installed_id
path.mkdir()
(path / "manifest.json").write_text(json.dumps({"id": installed_id}),
encoding="utf-8")
manager._find_plugin_path.side_effect = (
lambda pid: path if pid == "weather" else None)
manager.installs_as = installs_as
return manager
@pytest.fixture
def config(api_v3_module):
"""config.json with an ``enabled`` flag for each plugin id given."""
def sections(enabled):
api_v3_module.api_v3.config_manager.load_config.return_value = {
plugin_id: {"enabled": flag} for plugin_id, flag in enabled.items()}
return sections
@pytest.fixture
def queued(api_v3_module):
queue = MagicMock()
def enqueue(operation_type, plugin_id, operation_callback=None):
queue.callback_result = operation_callback(MagicMock())
return "op-1"
queue.enqueue_operation.side_effect = enqueue
api_v3_module.api_v3.operation_queue = queue
return queue
def _direct(client):
return client.post(INSTALL, json={"plugin_id": "weather"}).get_json()
def _queued(client, queue):
client.post(INSTALL, json={"plugin_id": "weather"})
return queue.callback_result
class TestDirectInstall:
def test_an_aliased_install_enabled_under_its_installed_id_asks_for_a_restart(
self, api_v3_client, store, config):
store.installs_as("ledmatrix-weather")
config({"ledmatrix-weather": True})
body = _direct(api_v3_client)
assert body["status"] == "success"
assert body["restart_required"] is True
assert body["restart_message"]
def test_an_enabled_section_under_the_registry_id_alone_does_not(
self, api_v3_client, store, config):
"""The display knows the plugin as ledmatrix-weather; nothing runs
under a section called weather."""
store.installs_as("ledmatrix-weather")
config({"weather": True})
assert _direct(api_v3_client)["restart_required"] is False
def test_an_aliased_install_that_is_not_enabled_needs_no_restart(
self, api_v3_client, store, config):
store.installs_as("ledmatrix-weather")
config({"ledmatrix-weather": False})
assert _direct(api_v3_client)["restart_required"] is False
def test_an_install_under_its_own_id_is_unchanged(self, api_v3_client, store, config):
store.installs_as("weather")
config({"weather": True})
assert _direct(api_v3_client)["restart_required"] is True
def test_an_install_that_cannot_be_found_uses_the_requested_id(
self, api_v3_client, store, config):
config({"weather": True})
assert _direct(api_v3_client)["restart_required"] is True
class TestQueuedInstall:
def test_an_aliased_install_enabled_under_its_installed_id_asks_for_a_restart(
self, api_v3_client, store, config, queued):
store.installs_as("ledmatrix-weather")
config({"ledmatrix-weather": True})
result = _queued(api_v3_client, queued)
assert result["success"] is True
assert result["restart_required"] is True
assert result["restart_message"]
def test_an_enabled_section_under_the_registry_id_alone_does_not(
self, api_v3_client, store, config, queued):
store.installs_as("ledmatrix-weather")
config({"weather": True})
assert _queued(api_v3_client, queued)["restart_required"] is False
test/test_cache_manager.py
+1 -77
View File
@@ -1,7 +1,7 @@
"""
Tests for CacheManager and cache components.
Tests cache functionality including memory cache, disk cache, strategy, and metrics.
Tests cache functionality including memory cache, disk cache, and strategy.
"""
import pytest
@@ -11,7 +11,6 @@ from src.cache_manager import CacheManager
from src.cache.memory_cache import MemoryCache
from src.cache.disk_cache import DiskCache
from src.cache.cache_strategy import CacheStrategy
from src.cache.cache_metrics import CacheMetrics
from datetime import datetime
@@ -26,7 +25,6 @@ class TestCacheManager:
assert hasattr(cm, '_memory_cache_component')
assert hasattr(cm, '_disk_cache_component')
assert hasattr(cm, '_strategy_component')
assert hasattr(cm, '_metrics_component')
def test_set_and_get(self, tmp_path):
"""Test basic set and get operations."""
@@ -196,50 +194,6 @@ class TestMemoryCache:
assert stats["max_size"] == 1000 # default
class TestCacheMetrics:
"""Test CacheMetrics functionality."""
def test_record_hit(self):
"""Test recording cache hit."""
metrics = CacheMetrics()
metrics.record_hit()
stats = metrics.get_metrics()
# get_metrics() returns calculated values, not raw hits/misses
assert stats['total_requests'] == 1
assert stats['cache_hit_rate'] == 1.0 # 1 hit out of 1 request
def test_record_miss(self):
"""Test recording cache miss."""
metrics = CacheMetrics()
metrics.record_miss()
stats = metrics.get_metrics()
# get_metrics() returns calculated values, not raw hits/misses
assert stats['total_requests'] == 1
assert stats['cache_hit_rate'] == 0.0 # 0 hits out of 1 request
def test_record_fetch_time(self):
"""Test recording fetch time."""
metrics = CacheMetrics()
metrics.record_fetch_time(0.5)
stats = metrics.get_metrics()
assert stats['fetch_count'] == 1
assert stats['total_fetch_time'] == 0.5
assert stats['average_fetch_time'] == 0.5
def test_cache_hit_rate(self):
"""Test cache hit rate calculation."""
metrics = CacheMetrics()
metrics.record_hit()
metrics.record_hit()
metrics.record_miss()
stats = metrics.get_metrics()
assert stats['cache_hit_rate'] == pytest.approx(0.666, abs=0.01)
class TestDiskCache:
"""Test DiskCache functionality."""
@@ -370,36 +324,6 @@ class TestDiskCache:
# Should handle gracefully
assert result is None or isinstance(result, dict)
def test_record_background_hit(self):
"""Test recording background cache hit."""
metrics = CacheMetrics()
metrics.record_hit(cache_type='background')
stats = metrics.get_metrics()
assert stats['total_requests'] == 1
assert stats['background_hit_rate'] == 1.0
def test_record_background_miss(self):
"""Test recording background cache miss."""
metrics = CacheMetrics()
metrics.record_miss(cache_type='background')
stats = metrics.get_metrics()
assert stats['total_requests'] == 1
assert stats['background_hit_rate'] == 0.0
def test_multiple_fetch_times(self):
"""Test recording multiple fetch times."""
metrics = CacheMetrics()
metrics.record_fetch_time(0.5)
metrics.record_fetch_time(1.0)
metrics.record_fetch_time(0.3)
stats = metrics.get_metrics()
assert stats['fetch_count'] == 3
assert stats['total_fetch_time'] == 1.8
assert stats['average_fetch_time'] == pytest.approx(0.6, abs=0.01)
class TestDiskCacheWriteEconomy:
test/test_cache_strategy_intervals.py
+6 -56
View File
@@ -1,12 +1,10 @@
"""CacheStrategy intervals, pinned across the whole input grid.
The strategy table used to carry a per-sport defaults dict whose every value
was 60, and a soccer branch identical to its else. These tests pin the
returned strategy for every data type x sport key x config shape, so
simplifying the lookup cannot change what any caller gets back. They were
written against the pre-cleanup code and pass on it unchanged, except for
the legacy `<sport>_scoreboard` config shape (see below), which that code
still read.
was 60, a soccer branch identical to its else, and a config lookup of
`<sport>_scoreboard` sections that only the replaced built-in scoreboards
had. These tests pin the returned strategy for every data type x sport key,
so simplifying the lookup cannot change what any caller gets back.
"""
import pytest
@@ -14,46 +12,6 @@ import pytest
from src.cache.cache_strategy import CacheStrategy
class _Cfg:
def __init__(self, config):
self.config = config
class _NoConfigAttr:
pass
# Plugin config sections are keyed by plugin id. Their intervals belong to the
# plugin, and the strategy table has never read them.
_PLUGIN_ID_CONFIG = {
pid: {"live_update_interval": 5, "recent_update_interval": 7,
"upcoming_update_interval": 9}
for pid in ("football-scoreboard", "basketball-scoreboard",
"baseball-scoreboard", "hockey-scoreboard", "soccer-scoreboard")
}
# `<sport>_scoreboard` sections come from the built-in scoreboards the plugin
# system replaced. An install upgraded from that era can still carry them in
# config.json (nothing deletes them). No current caller passes a sport key to
# the strategy, but a stale section must not steer cache TTLs if one does.
_LEGACY_SCOREBOARD_CONFIG = {
f"{sport}_scoreboard": {"live_update_interval": 5,
"recent_update_interval": 7,
"upcoming_update_interval": 9}
for sport in ("nfl", "nba", "mlb", "nhl", "soccer", "ncaa_fb",
"ncaa_baseball", "ncaam_basketball", "milb")
}
CONFIG_MANAGERS = {
"no_config_manager": None,
"empty_config": _Cfg({}),
"plugin_id_config": _Cfg(_PLUGIN_ID_CONFIG),
"legacy_scoreboard_config": _Cfg(_LEGACY_SCOREBOARD_CONFIG),
"config_is_none": _Cfg(None),
"config_is_not_a_dict": _Cfg("x"),
"config_manager_without_config": _NoConfigAttr(),
}
SPORT_KEYS = [None, "", "nfl", "nba", "mlb", "nhl", "soccer", "ncaa_fb",
"ncaa_baseball", "ncaam_basketball", "milb",
"football-scoreboard", "curling"]
@@ -93,16 +51,8 @@ def _expected(data_type, sport_key):
return FIXED.get(data_type, DEFAULT)
@pytest.mark.parametrize("cm_name", sorted(CONFIG_MANAGERS))
def test_live_interval_is_60_for_every_sport(cm_name):
strategy = CacheStrategy(config_manager=CONFIG_MANAGERS[cm_name])
for sport_key in SPORT_KEYS:
assert strategy.get_sport_live_interval(sport_key) == 60, sport_key
@pytest.mark.parametrize("cm_name", sorted(CONFIG_MANAGERS))
def test_strategy_table_for_every_data_type_and_sport(cm_name):
strategy = CacheStrategy(config_manager=CONFIG_MANAGERS[cm_name])
def test_strategy_table_for_every_data_type_and_sport():
strategy = CacheStrategy()
data_types = ["live_scores", "sports_live", *FIXED, "unknown", ""]
for data_type in data_types:
for sport_key in SPORT_KEYS:
test/test_deprecation.py
+38 -1
View File
@@ -32,9 +32,46 @@ DEPRECATED_3_9 = {
],
}
#: Deprecated after the October 2026 over-engineering audit, for removal in
#: 3.10.0: nothing in core, the monorepo or the registry's third-party plugins
#: calls them.
DEPRECATED_3_10 = {
"src.logo_downloader.LogoDownloader": [
"fetch_teams_data", "extract_teams_from_data", "download_missing_logos_for_league",
"download_all_ncaa_football_logos", "download_all_missing_logos",
"convert_image_to_rgba", "convert_all_logos_to_rgba",
],
"src.config_manager.ConfigManager": [
"rollback_config", "list_backups", "validate_config_file", "get_secret",
"cleanup_orphaned_plugin_configs", "validate_all_plugin_configs",
],
"src.common.api_helper.APIHelper": [
"fetch_espn_scoreboard", "fetch_espn_standings", "fetch_espn_rankings",
"set_cache", "get_cache", "set_rate_limit", "get_request_stats",
],
"src.plugin_system.testing.plugin_test_base.PluginTestCase": ["setUp"],
"src.background_data_service.BackgroundDataService": [
"get_result", "is_request_complete", "get_request_status",
],
"src.plugin_system.plugin_manager.PluginManager": [
"get_all_plugins", "get_plugin_info", "get_all_plugin_info",
"get_plugin_display_modes", "find_plugin_for_mode",
],
"src.plugin_system.plugin_state.PluginStateManager": [
"is_loaded", "is_running", "is_error", "get_last_update", "get_error_info",
"get_state_info",
],
"src.cache_manager.CacheManager": ["load_cache", "generate_sport_cache_key"],
"src.font_manager.FontManager": ["measure_text", "get_native_bdf_size"],
"src.base_odds_manager.BaseOddsManager": ["get_odds_for_games", "format_odds_summary"],
"src.dynamic_team_resolver.DynamicTeamResolver": [
"get_available_dynamic_teams", "is_dynamic_team",
],
}
#: Every pinned marker: (class path, method) -> the release that removes it.
PINNED = {(path, name): removal
for removal, table in (("3.9.0", DEPRECATED_3_9),)
for removal, table in (("3.9.0", DEPRECATED_3_9), ("3.10.0", DEPRECATED_3_10))
for path, names in table.items() for name in names}
test/test_display_pending_changes.py
-1
View File
@@ -145,7 +145,6 @@ def vegas_coordinator(controller):
coord.render_pipeline.target_fps = float(coord.vegas_config.target_fps)
coord.stream_manager = MagicMock()
coord.display_manager = controller.display_manager
coord.stats = {'cycles_completed': 0, 'interruptions': 0}
coord._state_lock = threading.Lock()
coord._is_active = True
coord._is_paused = False
test/test_error_aggregator.py
-49
View File
@@ -218,24 +218,6 @@ class TestPatternDetection:
assert pattern is not None
assert pattern.severity in ["error", "critical"]
def test_pattern_callback_called(self):
"""Pattern detection callback should be called."""
aggregator = ErrorAggregator(pattern_threshold=2)
callback_called = []
def callback(pattern):
callback_called.append(pattern)
aggregator.on_pattern_detected(callback)
# Trigger pattern
for _ in range(3):
aggregator.record_error(error=ValueError("Pattern trigger"))
assert len(callback_called) == 1
assert callback_called[0].error_type == "ValueError"
class TestErrorSummary:
"""Test error summary generation."""
@@ -318,37 +300,6 @@ class TestPluginHealth:
assert health["recent_error_count"] == 10
class TestRecordClearing:
"""Test clearing old records."""
def test_clear_old_records(self):
"""Old records should be cleared."""
aggregator = ErrorAggregator()
# Add a record
aggregator.record_error(error=ValueError("Old error"))
# Manually age the record
aggregator._records[0].timestamp = datetime.now() - timedelta(hours=48)
# Clear records older than 24 hours
cleared = aggregator.clear_old_records(max_age_hours=24)
assert cleared == 1
assert len(aggregator._records) == 0
def test_recent_records_not_cleared(self):
"""Recent records should not be cleared."""
aggregator = ErrorAggregator()
aggregator.record_error(error=ValueError("Recent error"))
cleared = aggregator.clear_old_records(max_age_hours=24)
assert cleared == 0
assert len(aggregator._records) == 1
class TestThreadSafety:
"""Test thread safety of error aggregator."""
test/test_espn_payload.py
+162
View File
@@ -0,0 +1,162 @@
"""Tests for src/common/espn_payload.py and its use by BackgroundDataService."""
import copy
import time
from unittest.mock import MagicMock, Mock, patch
import pytest
from src.background_data_service import BackgroundDataService, shutdown_background_service
from src.common.espn_payload import is_espn_scoreboard_url, slim_scoreboard_payload
SCOREBOARD = "https://site.api.espn.com/apis/site/v2/sports/baseball/mlb/scoreboard"
def _event():
"""One event carrying every key the slimming drops and a sample of the
keys scoreboards read, at the depth ESPN puts them."""
competitor = {
"id": "10",
"homeAway": "home",
"score": "5",
"team": {"abbreviation": "NYY", "logo": "https://a/l.png",
"links": [{"href": "https://espn.com/team"}]},
"records": [{"summary": "90-60"}],
"linescores": [{"value": 1}],
"statistics": [{"name": "hits", "displayValue": "9"}],
"leaders": [{"name": "avg", "leaders": [{"athlete": {"id": "1"}}]}],
"probables": [{"athlete": {"id": "2"}, "statistics": []}],
}
return {
"id": "401",
"date": "2026-10-01T23:05Z",
"links": [{"href": "https://espn.com/game"}],
"status": {"type": {"state": "post"}},
"competitions": [{
"status": {"type": {"state": "post", "shortDetail": "Final"},
"featuredAthletes": [{"athlete": {"id": "3"}}]},
"competitors": [competitor, dict(copy.deepcopy(competitor), homeAway="away")],
"odds": [{"details": "NYY -150", "overUnder": 8.5}],
"situation": {"outs": 2},
"notes": [{"headline": "Game 1"}],
"broadcasts": [{"names": ["FOX"]}],
"venue": {"fullName": "Yankee Stadium"},
"leaders": [{"name": "hits"}],
"headlines": [{"description": "recap"}],
"highlights": [{"links": {"source": {}}}],
"geoBroadcasts": [{"media": {"shortName": "FOX"}}],
}],
}
class TestSlimScoreboardPayload:
def test_drops_exactly_the_listed_keys(self):
payload = {"leagues": [{"id": "10"}], "events": [_event()]}
slim_scoreboard_payload(payload)
event = payload["events"][0]
competition = event["competitions"][0]
assert "links" not in event
for key in ("leaders", "headlines", "highlights", "geoBroadcasts"):
assert key not in competition
assert "featuredAthletes" not in competition["status"]
for competitor in competition["competitors"]:
assert "leaders" not in competitor
assert "probables" not in competitor
assert "links" not in competitor["team"]
def test_keeps_everything_else_unchanged(self):
"""Removing the dropped keys from the original by hand gives exactly
the slimmed payload: nothing else moved, changed or went missing."""
original = {"leagues": [{"id": "10"}], "events": [_event(), _event()]}
expected = copy.deepcopy(original)
for event in expected["events"]:
del event["links"]
competition = event["competitions"][0]
for key in ("leaders", "headlines", "highlights", "geoBroadcasts"):
del competition[key]
del competition["status"]["featuredAthletes"]
for competitor in competition["competitors"]:
del competitor["leaders"], competitor["probables"]
del competitor["team"]["links"]
assert slim_scoreboard_payload(original) == expected
def test_in_place_and_returns_payload(self):
payload = {"events": [_event()]}
assert slim_scoreboard_payload(payload) is payload
@pytest.mark.parametrize("payload", [
None, [], "x", {}, {"events": None}, {"events": "x"},
{"events": [None, 1, "x", {"competitions": None}]},
{"events": [{"competitions": [None, {"status": None, "competitors": None}]}]},
{"events": [{"competitions": [{"competitors": [None, {"team": None}]}]}]},
])
def test_odd_shapes_pass_through(self, payload):
before = copy.deepcopy(payload)
assert slim_scoreboard_payload(payload) == before
class TestIsEspnScoreboardUrl:
@pytest.mark.parametrize("url", [
SCOREBOARD,
SCOREBOARD + "/",
"http://site.api.espn.com/apis/site/v2/sports/football/college-football/scoreboard",
])
def test_scoreboards(self, url):
assert is_espn_scoreboard_url(url)
@pytest.mark.parametrize("url", [
None, "", 12,
"https://site.api.espn.com/apis/site/v2/sports/baseball/mlb/teams",
"https://site.api.espn.com/apis/site/v2/sports/football/nfl/summary",
"https://example.com/scoreboard",
"https://espn.com.evil.example/apis/x/scoreboard",
"https://notespn.com/apis/x/scoreboard",
])
def test_not_scoreboards(self, url):
assert not is_espn_scoreboard_url(url)
@pytest.fixture
def service():
shutdown_background_service()
cache = MagicMock()
cache.get.return_value = None
svc = BackgroundDataService(cache, max_workers=1, request_timeout=5)
yield svc
svc.shutdown(wait=False)
shutdown_background_service()
def _run(service, url, **kwargs):
response = Mock(status_code=200)
response.json.return_value = {"events": [_event()]}
response.raise_for_status.return_value = None
delivered = []
with patch.object(service.session, "get", return_value=response):
req_id = service.submit_fetch_request(
sport="mlb", year=2026, url=url, cache_key="mlb_schedule_window_14_7",
callback=lambda result: delivered.append(result.data), **kwargs)
deadline = time.time() + 5
while not service.is_request_complete(req_id) and time.time() < deadline:
time.sleep(0.02)
cached = service.cache_manager.set.call_args[0][1]
return cached, delivered
class TestBackgroundServiceSlims:
def test_espn_scoreboard_is_cached_and_delivered_slimmed(self, service):
cached, delivered = _run(service, SCOREBOARD)
competition = cached["events"][0]["competitions"][0]
assert "leaders" not in competition
assert "probables" not in competition["competitors"][0]
assert competition["odds"] and competition["situation"]
# The callback sees the very payload that was cached.
assert delivered and delivered[0] is cached
def test_opt_out_caches_whole_response(self, service):
cached, _ = _run(service, SCOREBOARD, slim_payload=False)
assert cached == {"events": [_event()]}
def test_other_urls_untouched(self, service):
cached, _ = _run(service, "https://example.com/feed")
assert cached == {"events": [_event()]}
test/test_frame_ops.py
-3
View File
@@ -330,9 +330,6 @@ class _Stream:
def get_grouped_content_for_composition(self):
return self.groups[0]
def get_active_plugin_ids(self):
return [pid for pid, _ in self.groups[0]]
def take_next_group(self, count=None, offscreen_only=False):
if self._i >= len(self.groups):
return []
test/test_frame_timing.py
+77
View File
@@ -807,3 +807,80 @@ def test_a_process_with_the_gc_monitor_exits_cleanly():
assert proc.returncode == 0, proc.stderr
assert "Exception ignored" not in proc.stderr
assert "installed at exit: False" in proc.stdout
SLOW = 1 / 110.0 # a panel that cannot reach a 120 Hz cap
def _windows(recorder, n, interval, start=0.0):
for i in range(n):
_feed(recorder, [interval] * 200, start=start + 50.0 * i)
_aggregate(recorder)
def _shortfall_warnings(caplog):
return [r for r in caplog.records
if r.name == "src.common.frame_timing" and "Limit Refresh Rate" in r.getMessage()]
def test_a_panel_slower_than_its_cap_is_reported_once(tmp_path, caplog):
r = _recorder(tmp_path)
r.plan_refresh(120.0)
caplog.set_level("WARNING")
_windows(r, 3, SLOW) # adopted on the 2nd window, checked on the 4th
assert _shortfall_warnings(caplog) == []
_windows(r, 3, SLOW, start=1000.0)
warnings = _shortfall_warnings(caplog)
assert len(warnings) == 1
assert "about 110 Hz" in warnings[0].getMessage()
assert "to 100 Hz" in warnings[0].getMessage()
def test_a_panel_that_reaches_its_cap_is_not_reported(tmp_path, caplog):
r = _recorder(tmp_path)
r.plan_refresh(100.0)
caplog.set_level("WARNING")
_windows(r, 6, PERIOD)
assert _shortfall_warnings(caplog) == []
def test_without_a_planned_rate_nothing_is_checked(tmp_path, caplog):
# The emulator and the fallback canvas: DisplayManager never calls
# plan_refresh(), since their frames are not paced by a panel.
r = _recorder(tmp_path)
caplog.set_level("WARNING")
_windows(r, 6, SLOW)
assert _shortfall_warnings(caplog) == []
def test_the_snapshot_records_the_planned_rate(tmp_path):
r = _recorder(tmp_path)
assert r.snapshot()["planned_refresh_hz"] is None
r.plan_refresh(120.0)
assert r.snapshot()["planned_refresh_hz"] == 120.0
def test_windows_the_period_rejected_do_not_count_toward_the_warning(tmp_path, caplog):
# A loaded start fixed 60 Hz (two windows agreed); the panel really runs at
# 100 Hz, but a window that much faster is ignored by the estimate, so the
# period stays 60 Hz. Warning "60 Hz is under your 100 Hz cap" would be wrong.
r = _recorder(tmp_path)
r.plan_refresh(100.0)
caplog.set_level("WARNING")
_windows(r, 2, 1 / 60.0)
assert abs(1.0 / r.refresh_period - 60.0) < 0.5
_windows(r, 6, PERIOD, start=1000.0)
assert abs(1.0 / r.refresh_period - 60.0) < 0.5 # still ignored
assert _shortfall_warnings(caplog) == []
def test_one_disagreeing_window_restarts_the_run(tmp_path, caplog):
r = _recorder(tmp_path)
r.plan_refresh(120.0)
caplog.set_level("WARNING")
_windows(r, 3, SLOW) # two windows toward three
_windows(r, 1, 1 / 250.0, start=1000.0) # far faster: rejected, resets
_windows(r, 1, SLOW, start=2000.0)
assert _shortfall_warnings(caplog) == []
_windows(r, 2, SLOW, start=3000.0) # three in a row now
assert len(_shortfall_warnings(caplog)) == 1
test/test_install_os_support.py
+59 -10
View File
@@ -50,9 +50,11 @@ def _stub(bin_dir: Path, name: str, body: str) -> None:
path.chmod(0o755)
def _stubs(tmp_path: Path, python_version="3.11", network="NetworkManager") -> Path:
def _stubs(tmp_path: Path, python_version="3.11", network="NetworkManager",
active=(), packages=()) -> Path:
"""python3 reports ``python_version`` (None: not installed); systemctl
reports ``network`` as the only active unit; dpkg lists no desktop."""
reports ``network`` and ``active`` as the only active units; dpkg-query
lists ``packages`` as installed (none by default, so no desktop)."""
bin_dir = tmp_path / "bin"
bin_dir.mkdir(exist_ok=True)
if python_version is None:
@@ -61,10 +63,14 @@ def _stubs(tmp_path: Path, python_version="3.11", network="NetworkManager") -> P
else:
_stub(bin_dir, "python3", f'case "$*" in *"%d.%d.%d"*) echo "{python_version}.1" ;; '
f'*) echo "{python_version}" ;; esac\n')
_stub(bin_dir, "systemctl",
f'case "$*" in *"is-active --quiet {network}") exit 0 ;; esac\nexit 3\n')
units = "|".join(f'*"is-active --quiet {unit}"' for unit in (network, *active))
_stub(bin_dir, "systemctl", f'case "$*" in {units}) exit 0 ;; esac\nexit 3\n')
_stub(bin_dir, "dpkg", "exit 0\n")
_stub(bin_dir, "dpkg-query", "exit 1\n")
if packages:
listing = "".join(f"ii {name}\\n" for name in packages)
_stub(bin_dir, "dpkg-query", f'printf "{listing}"\n')
else:
_stub(bin_dir, "dpkg-query", "exit 1\n")
_stub(bin_dir, "ping", "exit 0\n")
return bin_dir
@@ -149,20 +155,22 @@ class TestLibrary:
# --- first_time_install.sh's OS check ------------------------------------------
def _os_check_section() -> str:
def _os_check_section(marker_root: str = "/nonexistent") -> str:
"""first_time_install.sh from the OS check up to the next section, with
the desktop-marker directories pointed somewhere that cannot exist."""
the desktop-marker directories moved under ``marker_root`` (by default
somewhere that cannot exist)."""
text = FIRST_TIME.read_text(encoding="utf-8").replace("\r\n", "\n")
start = text.index("# Check OS version")
end = text.index("# The user who ran the installer")
section = text[start:end]
for marker in ("/usr/share/raspberrypi-ui-mods", "/usr/share/xsessions"):
assert marker in section
section = section.replace(marker, "/nonexistent" + marker)
section = section.replace(marker, marker_root + marker)
return section
def run_os_check(tmp_path: Path, release: str, **stub_args) -> subprocess.CompletedProcess:
def run_os_check(tmp_path: Path, release: str, marker_root: str = "/nonexistent",
**stub_args) -> subprocess.CompletedProcess:
"""Run the OS check as the installer would, from a copy of the project
layout so ``$(dirname "$0")/scripts/install/lib_os.sh`` resolves."""
project = tmp_path / "project"
@@ -171,7 +179,7 @@ def run_os_check(tmp_path: Path, release: str, **stub_args) -> subprocess.Comple
script = project / "first_time_install.sh"
script.write_text("set -Eeuo pipefail\n"
"trap 'echo ERR-TRAP line $LINENO >&2; exit 99' ERR\n"
+ _os_check_section() + '\necho "SECTION-DONE"\n',
+ _os_check_section(marker_root) + '\necho "SECTION-DONE"\n',
encoding="utf-8", newline="\n")
env = _env(tmp_path, release, _stubs(tmp_path, **stub_args))
return subprocess.run(["bash", str(script)], capture_output=True, text=True, env=env)
@@ -230,6 +238,47 @@ class TestInstallerOsCheck:
result = run_os_check(tmp_path, "trixie", python_version="3.13")
assert "✓ NetworkManager is managing the network" in result.stdout
# A running desktop stops the install; one that is only installed warns.
@pytest.mark.parametrize("unit", ["display-manager", "lightdm", "gdm", "sddm"])
def test_running_desktop_stops(self, tmp_path, unit):
result = run_os_check(tmp_path, "trixie", python_version="3.13", active=(unit,))
assert result.returncode == 1, result.stdout + result.stderr
assert "A desktop is running" in result.stdout
assert "multi-user.target" in result.stdout
assert "SECTION-DONE" not in result.stdout
@pytest.mark.parametrize("package", [
"raspberrypi-ui-mods", "rpd-wayland-core", "rpd-x-core", "xfce4",
"lxde-core", "gnome-shell", "kde-plasma-desktop", "plasma-workspace:arm64",
"task-desktop", "task-mate-desktop",
])
def test_installed_desktop_that_is_not_running_warns(self, tmp_path, package):
result = run_os_check(tmp_path, "trixie", python_version="3.13",
packages=("bash", package))
assert result.returncode == 0, result.stdout + result.stderr
assert "Desktop packages are installed, but no desktop is running" in result.stdout
assert "✓ OS requirements met" in result.stdout
def test_desktop_session_files_warn(self, tmp_path):
(tmp_path / "markers" / "usr" / "share" / "xsessions").mkdir(parents=True)
result = run_os_check(tmp_path, "trixie", python_version="3.13",
marker_root=str(tmp_path / "markers"))
assert result.returncode == 0, result.stdout + result.stderr
assert "Desktop packages are installed, but no desktop is running" in result.stdout
@pytest.mark.parametrize("packages", [
# libblockdev contains "kde" mid-word; the old check stopped on it.
("libblockdev-crypto3", "libblockdev3:arm64"),
("gnome-keyring", "xfce4-terminal", "xfconf", "lxde-icon-theme",
"kde-cli-tools", "gnome-session-common", "task-ssh-server", "rpd-plym-splash"),
])
def test_lite_with_desktop_named_parts_is_lite(self, tmp_path, packages):
result = run_os_check(tmp_path, "trixie", python_version="3.13", packages=packages)
assert result.returncode == 0, result.stdout + result.stderr
assert "✓ Lite version confirmed" in result.stdout
assert "WARNING: Desktop" not in result.stdout
# --- check_system_compatibility.sh ---------------------------------------------
test/test_ipc_contract.py
+3 -1
View File
@@ -60,7 +60,9 @@ class TestRoundTrip:
def test_start_args_round_trip(self):
args = OnDemandStartArgs(plugin_id='clock', mode='clock_main', duration=45.0,
pinned=True)
assert OnDemandStartArgs.from_dict(_wire(args.to_dict())) == args
wire = _wire({'plugin_id': 'clock', 'mode': 'clock_main', 'duration': 45.0,
'pinned': True})
assert OnDemandStartArgs.from_dict(wire) == args
def test_encoded_messages_are_ascii_single_lines(self):
data = c.encode_message({'v': 1, 'id': 'x', 'cmd': 'ping',
test/test_ipc_display_stage2.py
-1
View File
@@ -349,7 +349,6 @@ class TestVegasChecksEveryFrame:
coord.render_pipeline.frame_interval = 0.0
coord.render_pipeline.target_fps = 90
coord.display_manager = MagicMock()
coord.stats = {'cycles_completed': 0, 'interruptions': 0}
coord._state_lock = threading.Lock()
coord._is_active = True
coord._is_paused = False
test/test_ipc_server.py
+14 -14
View File
@@ -215,7 +215,7 @@ class TestWhereTheServerListens:
assert srv.server_socket_path({}) is None
assert ControlServer('x.sock').start() is False
with pytest.raises(client.ControlError) as e:
client.ping(paths=['x.sock'])
client.request(Command.PING, paths=['x.sock'])
assert e.value.reason == 'unsupported'
@@ -271,7 +271,7 @@ def _read_line(s):
class TestLiveSocket:
def test_client_round_trip(self, live, sock_path):
live()
assert client.ping(paths=[sock_path]) == {'pong': True}
assert client.request(Command.PING, paths=[sock_path]) == {'pong': True}
assert client.hello(paths=[sock_path])['version'] == 1
assert client.on_demand_status(paths=[sock_path])['current_mode'] == 'clock'
@@ -335,7 +335,7 @@ class TestLiveSocket:
assert s.recv(10) == b'' # and hung up
finally:
s.close()
assert client.ping(paths=[sock_path]) == {'pong': True}
assert client.request(Command.PING, paths=[sock_path]) == {'pong': True}
def test_a_client_that_hangs_up_mid_message(self, live, sock_path):
server = live()
@@ -343,7 +343,7 @@ class TestLiveSocket:
s.sendall(b'{"v":1,"id":"half","cmd":"on_demand.st')
s.close()
time.sleep(0.2)
assert client.ping(paths=[sock_path]) == {'pong': True}
assert client.request(Command.PING, paths=[sock_path]) == {'pong': True}
assert server.drain() == []
def test_a_slow_client_is_dropped_and_blocks_nobody(self, live, sock_path):
@@ -352,7 +352,7 @@ class TestLiveSocket:
try:
slow.sendall(b'{"v":1,') # ...and never finishes
t0 = time.monotonic()
assert client.ping(paths=[sock_path]) == {'pong': True}
assert client.request(Command.PING, paths=[sock_path]) == {'pong': True}
assert time.monotonic() - t0 < 0.5, 'a slow client held up another'
assert slow.recv(100) == b'' # hung up on, not answered
finally:
@@ -380,7 +380,7 @@ class TestLiveSocket:
for s in held:
s.close()
time.sleep(0.3)
assert client.ping(paths=[sock_path]) == {'pong': True}
assert client.request(Command.PING, paths=[sock_path]) == {'pong': True}
def test_many_concurrent_clients(self, live, sock_path):
server = live(queue_size=64)
@@ -408,7 +408,7 @@ class TestLiveSocket:
s = ControlServer(sock_path, status_provider=lambda: status)
try:
assert s.start()
assert client.ping(paths=[sock_path]) == {'pong': True}
assert client.request(Command.PING, paths=[sock_path]) == {'pong': True}
finally:
s.close()
@@ -416,7 +416,7 @@ class TestLiveSocket:
live()
second = ControlServer(sock_path, status_provider=lambda: status)
assert second.start() is False
assert client.ping(paths=[sock_path]) == {'pong': True}
assert client.request(Command.PING, paths=[sock_path]) == {'pong': True}
def test_a_regular_file_is_never_removed(self, sock_path):
with open(sock_path, 'w') as f:
@@ -434,24 +434,24 @@ class TestLiveSocket:
assert second.start()
try:
first.close() # must not unlink second's file
assert client.ping(paths=[sock_path]) == {'pong': True}
assert client.request(Command.PING, paths=[sock_path]) == {'pong': True}
finally:
second.close()
def test_client_reasons(self, sock_path, tmp_path):
with pytest.raises(client.ControlError) as e:
client.ping(paths=[sock_path])
client.request(Command.PING, paths=[sock_path])
assert e.value.reason == 'no_socket'
dead = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
dead.bind(sock_path)
try:
with pytest.raises(client.ControlError) as e:
client.ping(paths=[sock_path])
client.request(Command.PING, paths=[sock_path])
assert e.value.reason == 'refused'
finally:
dead.close()
with pytest.raises(client.ControlError) as e:
client.ping(paths=[])
client.request(Command.PING, paths=[])
assert e.value.reason == 'disabled'
with pytest.raises(client.ControlError) as e:
client.on_demand_start('x', None, None, paths=[sock_path])
@@ -464,7 +464,7 @@ class TestLiveSocket:
try:
t0 = time.monotonic()
with pytest.raises(client.ControlError) as e:
client.ping(paths=[sock_path], timeout=0.3)
client.request(Command.PING, paths=[sock_path], timeout=0.3)
assert e.value.reason == 'timeout'
assert time.monotonic() - t0 < 1.0
finally:
@@ -542,7 +542,7 @@ class TestPermissions:
os.setgroups([])
os.setgid(gid)
os.setuid(nobody.pw_uid)
result = json.dumps(client.ping(paths=[sock_path]))
result = json.dumps(client.request(Command.PING, paths=[sock_path]))
except client.ControlError as e:
result = 'error:' + e.reason
except Exception as e: # report anything else to the parent
test/test_ipc_state_stream.py
+1 -1
View File
@@ -522,7 +522,7 @@ class TestLiveStream:
# max_clients is 2 and three streams are open: subscribers gave
# their request slots back.
for _ in range(4):
assert client.ping(paths=[path]) == {'pong': True}
assert client.request(Command.PING, paths=[path]) == {'pong': True}
hub.publish('display', _display(mode='weather'), volatile=('last_updated',))
assert _wait_until(lambda: all(
s.latest()['state']['display']['mode'] == 'weather' for s in subs))
test/test_live_display_modes.py
+265
View File
@@ -0,0 +1,265 @@
"""The web interface sees the display modes the display actually registered (#668).
A plugin may compute its modes from its config: soccer-scoreboard registers
``soccer_<league>_live/recent/upcoming`` for every league the user adds under
``custom_leagues``, and no manifest can list those ahead of time. The display
always rotated them -- DisplayController._register_loaded_plugin prefers
``plugin.modes`` -- but the web process reads plugins as files, so its mode
listing (/display/modes, the on-demand dialog) and find_plugin_for_mode
(/display/on-demand/start with a mode and no plugin_id) saw only manifests.
The display now records each plugin's registered modes in its plugin state,
the runtime snapshot carries them, and PluginCatalog prefers them while the
snapshot is live, falling back to the manifest when it is not.
"""
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent))
from src.cache_manager import CacheManager # noqa: E402
from src.plugin_system import plugin_runtime as rt # noqa: E402
from src.plugin_system.plugin_catalog import PluginCatalog # noqa: E402
from src.plugin_system.plugin_runtime import ( # noqa: E402
PluginRuntimePublisher, build_runtime_snapshot, read_plugin_runtime,
view_from_snapshot,
)
from src.plugin_system.plugin_state import PluginState, PluginStateManager # noqa: E402
from test._api_v3_test_helpers import api_v3_client, api_v3_module # noqa: F401,E402
DECLARED = ["soccer_eng.1_live", "soccer_eng.1_recent", "soccer_eng.1_upcoming"]
CUSTOM = ["soccer_sco.1_live", "soccer_sco.1_recent", "soccer_sco.1_upcoming"]
REGISTERED = DECLARED + CUSTOM
def _loaded_states(modes=None):
states = PluginStateManager()
states.set_state("soccer-scoreboard", PluginState.ENABLED)
states.record_loaded("soccer-scoreboard", "2.24.1")
if modes is not None:
states.record_modes("soccer-scoreboard", modes)
return states
@pytest.fixture
def shared_cache(tmp_path, monkeypatch):
"""Two cache managers over one directory: the display's and the web's."""
monkeypatch.setattr(CacheManager, "_get_writable_cache_dir",
lambda self: str(tmp_path / "cache"))
(tmp_path / "cache").mkdir()
display_cache, web_cache = CacheManager(), CacheManager()
yield display_cache, web_cache
display_cache.stop_cleanup_thread()
web_cache.stop_cleanup_thread()
@pytest.fixture
def plugins_dir(tmp_path):
root = tmp_path / "plugins"
for plugin_id, modes in (("soccer-scoreboard", DECLARED), ("clock-simple", ["clock"])):
(root / plugin_id).mkdir(parents=True)
(root / plugin_id / "manifest.json").write_text(json.dumps({
"id": plugin_id, "name": plugin_id, "version": "1.0.0",
"class_name": "P", "display_modes": modes}), encoding="utf-8")
return root
# --- The display records what it registered ---------------------------------
class TestStateManagerRecordsModes:
def test_runtime_records_carry_them(self):
assert _loaded_states(REGISTERED).runtime_records()[
"soccer-scoreboard"]["modes"] == REGISTERED
def test_none_until_registered(self):
assert _loaded_states().runtime_records()["soccer-scoreboard"]["modes"] is None
def test_a_new_list_is_a_change_the_same_one_is_not(self):
"""change_count drives the publisher: re-registering an unchanged
plugin must not cost an SD-card write."""
states = _loaded_states(DECLARED)
before = states.change_count
states.record_modes("soccer-scoreboard", list(DECLARED))
assert states.change_count == before
states.record_modes("soccer-scoreboard", REGISTERED)
assert states.change_count == before + 1
def test_ignored_for_a_plugin_that_is_not_loaded(self):
states = PluginStateManager()
states.record_modes("ghost", ["ghost"])
assert "ghost" not in states.runtime_records()
def test_unload_forgets_them(self):
states = _loaded_states(REGISTERED)
states.clear_state("soccer-scoreboard")
assert "soccer-scoreboard" not in states.runtime_records()
def test_a_reload_starts_without_them_until_registered_again(self):
states = _loaded_states(REGISTERED)
states.record_loaded("soccer-scoreboard", "2.25.0")
assert states.runtime_records()["soccer-scoreboard"]["modes"] is None
class TestControllerRecordsOnRegistration:
def test_plugin_modes_reach_the_state_manager(self, test_display_controller):
"""_register_loaded_plugin is the one path every load, enable and
reload goes through."""
c = test_display_controller
states = _loaded_states()
plugin = MagicMock()
plugin.modes = list(REGISTERED)
c.plugin_manager.state_manager = states
c.plugin_manager.get_plugin = MagicMock(return_value=plugin)
c.plugin_manager.plugin_manifests = {"soccer-scoreboard": {"display_modes": DECLARED}}
c._register_loaded_plugin("soccer-scoreboard")
assert states.runtime_records()["soccer-scoreboard"]["modes"] == REGISTERED
def test_a_failing_state_manager_does_not_break_registration(self, test_display_controller):
c = test_display_controller
plugin = MagicMock()
plugin.modes = ["clock"]
c.plugin_manager.state_manager.record_modes = MagicMock(side_effect=RuntimeError("x"))
c.plugin_manager.get_plugin = MagicMock(return_value=plugin)
c.plugin_manager.plugin_manifests = {}
assert c._register_loaded_plugin("clock-simple") == ["clock"]
assert c.mode_to_plugin_id["clock"] == "clock-simple"
# --- The snapshot carries them; only a live view reports them ---------------
class TestSnapshotAndView:
NOW = 1_800_000_000.0
def _view(self, states, running=True, published_at=None):
snapshot = build_runtime_snapshot(states, started_at=1.0, now=self.NOW,
running=running)
if published_at is not None:
snapshot["published_at"] = published_at
return view_from_snapshot(snapshot, now=self.NOW)
def test_live_view_reports_the_registered_modes(self):
assert self._view(_loaded_states(REGISTERED)).display_modes(
"soccer-scoreboard") == REGISTERED
def test_stale_and_stopped_views_report_nothing(self):
states = _loaded_states(REGISTERED)
assert self._view(states, published_at=self.NOW - 10_000).display_modes(
"soccer-scoreboard") is None
assert self._view(states, running=False).display_modes("soccer-scoreboard") is None
def test_unregistered_or_unknown_plugins_report_nothing(self):
view = self._view(_loaded_states())
assert view.display_modes("soccer-scoreboard") is None
assert view.display_modes("not-loaded") is None
def test_a_runaway_list_is_bounded(self):
modes = [f"m{i}" for i in range(1000)] + ["x" * 500]
snapshot = build_runtime_snapshot(_loaded_states(modes), started_at=1.0, now=self.NOW)
published = snapshot["plugins"]["soccer-scoreboard"]["modes"]
assert len(published) == rt._MAX_MODES
def test_a_mode_name_is_kept_whole_or_dropped(self):
long_mode = "x" * (rt._ID_CHARS + 1)
snapshot = build_runtime_snapshot(_loaded_states(["ok", long_mode]),
started_at=1.0, now=self.NOW)
assert snapshot["plugins"]["soccer-scoreboard"]["modes"] == ["ok"]
def test_non_strings_from_a_hand_made_snapshot_are_dropped(self):
snapshot = {"schema": rt.SNAPSHOT_SCHEMA, "running": True,
"published_at": self.NOW, "plugins": {
"p": {"loaded": True, "modes": ["a", 3, None]}}}
assert view_from_snapshot(snapshot, now=self.NOW).display_modes("p") == ["a"]
# --- The web's catalog prefers them -------------------------------------------
class TestCatalog:
def _catalog(self, plugins_dir, web_cache):
catalog = PluginCatalog(plugins_dir,
runtime_source=lambda: read_plugin_runtime(web_cache))
catalog.discover_plugins()
return catalog
def test_live_display_modes_win_over_the_manifest(self, plugins_dir, shared_cache):
display_cache, web_cache = shared_cache
PluginRuntimePublisher(display_cache, _loaded_states(REGISTERED)).tick()
catalog = self._catalog(plugins_dir, web_cache)
assert catalog.get_plugin_display_modes("soccer-scoreboard") == REGISTERED
def test_a_custom_league_mode_resolves_to_its_plugin(self, plugins_dir, shared_cache):
"""What /display/on-demand/start does with a mode and no plugin_id."""
display_cache, web_cache = shared_cache
PluginRuntimePublisher(display_cache, _loaded_states(REGISTERED)).tick()
catalog = self._catalog(plugins_dir, web_cache)
assert catalog.find_plugin_for_mode("SOCCER_SCO.1_LIVE") == "soccer-scoreboard"
def test_a_plugin_the_display_has_not_loaded_falls_back_to_its_manifest(
self, plugins_dir, shared_cache):
display_cache, web_cache = shared_cache
PluginRuntimePublisher(display_cache, _loaded_states(REGISTERED)).tick()
catalog = self._catalog(plugins_dir, web_cache)
assert catalog.get_plugin_display_modes("clock-simple") == ["clock"]
assert catalog.find_plugin_for_mode("clock") == "clock-simple"
def test_a_mode_the_display_dropped_does_not_resolve_by_manifest(
self, plugins_dir, shared_cache):
display_cache, web_cache = shared_cache
PluginRuntimePublisher(display_cache, _loaded_states(CUSTOM)).tick()
catalog = self._catalog(plugins_dir, web_cache)
assert catalog.find_plugin_for_mode("soccer_eng.1_live") is None
def test_a_stopped_display_falls_back_to_manifests(self, plugins_dir, shared_cache):
display_cache, web_cache = shared_cache
publisher = PluginRuntimePublisher(display_cache, _loaded_states(REGISTERED))
publisher.tick()
publisher.stop()
catalog = self._catalog(plugins_dir, web_cache)
assert catalog.get_plugin_display_modes("soccer-scoreboard") == DECLARED
assert catalog.find_plugin_for_mode("soccer_sco.1_live") is None
def test_no_runtime_source_is_manifests_only(self, plugins_dir):
catalog = PluginCatalog(plugins_dir)
catalog.discover_plugins()
assert catalog.get_plugin_display_modes("soccer-scoreboard") == DECLARED
def test_a_failing_runtime_source_is_manifests_only(self, plugins_dir):
def broken():
raise OSError("cache gone")
catalog = PluginCatalog(plugins_dir, runtime_source=broken)
catalog.discover_plugins()
assert catalog.get_plugin_display_modes("soccer-scoreboard") == DECLARED
def test_one_listing_reads_the_view_once(self, plugins_dir):
source = MagicMock(return_value=None)
catalog = PluginCatalog(plugins_dir, runtime_source=source)
catalog.discover_plugins()
for _ in range(10):
catalog.get_plugin_display_modes("soccer-scoreboard")
catalog.find_plugin_for_mode("clock")
assert source.call_count == 1
class TestDisplayModesRoute:
def test_lists_the_custom_league_modes(self, api_v3_module, api_v3_client, # noqa: F811
plugins_dir, shared_cache):
display_cache, web_cache = shared_cache
PluginRuntimePublisher(display_cache, _loaded_states(REGISTERED)).tick()
api = api_v3_module.api_v3
api.plugin_catalog = PluginCatalog(
plugins_dir, runtime_source=lambda: read_plugin_runtime(web_cache))
api.config_manager.load_config = MagicMock(return_value={
"soccer-scoreboard": {"enabled": True}})
response = api_v3_client.get("/api/v3/display/modes")
assert response.status_code == 200, response.get_data(as_text=True)
modes = {m["mode"]: m for m in response.get_json()["data"]["modes"]}
assert set(modes) == set(REGISTERED)
assert modes["soccer_sco.1_live"]["plugin_id"] == "soccer-scoreboard"
test/test_operation_queue_pending_and_trim.py
+3 -3
View File
@@ -54,7 +54,7 @@ def test_second_pending_operation_for_a_plugin_is_refused(op_queue):
assert _wait_for(lambda: op_queue.get_operation_status(first).status
== OperationStatus.COMPLETED)
# Once it has finished, the plugin accepts a new operation again.
op_queue.enqueue_operation(OperationType.UPDATE, "demo")
op_queue.enqueue_operation(OperationType.UNINSTALL, "demo")
def test_operations_map_is_trimmed_with_history(op_queue):
@@ -64,8 +64,8 @@ def test_operations_map_is_trimmed_with_history(op_queue):
assert _wait_for(lambda: all(
(op_queue.get_operation_status(i) is None
or op_queue.get_operation_status(i).status == OperationStatus.COMPLETED)
for i in ids) and len(op_queue.get_operation_history()) == 3)
for i in ids) and len(op_queue._operation_history) == 3)
assert len(op_queue._operations) == 3
kept = {op.operation_id for op in op_queue.get_operation_history()}
kept = {op.operation_id for op in op_queue._operation_history}
assert set(op_queue._operations) == kept
test/test_plugin_runtime_snapshot.py
+1 -1
View File
@@ -162,7 +162,7 @@ class TestPublisher:
assert snapshot["stale_after"] == rt.STALE_AFTER
assert snapshot["plugins"] == {"clock": {
"loaded": True, "state": "enabled", "error": None,
"version": "1.0.0", "loaded_at": 10.0}}
"version": "1.0.0", "loaded_at": 10.0, "modes": None}}
def test_changes_are_throttled_and_quiet_displays_refresh(self):
cache = MagicMock()
test/test_render_gate.py
-1
View File
@@ -404,7 +404,6 @@ class TestVegasWiring:
off = VegasModeConfig.from_config(
{"display": {"vegas_scroll": {"prefetch_gate": False}}})
assert off.prefetch_gate is False
assert off.to_dict()["prefetch_gate"] is False
assert VegasModeConfig.from_config(
{"display": {"vegas_scroll": {}}}).prefetch_gate is True
test/test_scroll_config.py
+44
View File
@@ -21,6 +21,7 @@ from src.common.scroll_config import ( # noqa: E402
refresh_hz_from_config,
resolve,
)
from src.common import scroll_config # noqa: E402
class FakeHelper:
@@ -504,3 +505,46 @@ class TestSpeedAdvice:
got = solve_crisp(50, 125.74)
assert got.steppiness == "smooth"
assert got.pixels_per_frame == 1
class TestRefreshShortfall:
"""A panel that cannot reach its cap runs every scroll slow."""
def test_the_ledmatrix_rig_is_told_to_cap_at_100(self):
# Pi 4, 2x128x64 on adafruit-hat-pwm under a 120 Hz cap: measured
# 107.6-113.1 Hz, and frame_timing reports the fast end.
s = scroll_config.refresh_shortfall(113.1, 120)
assert s == {"measured_hz": 113.1, "planned_hz": 120.0,
"suggested_cap_hz": 100, "slow_percent": 6}
def test_a_panel_that_holds_its_cap_is_fine(self):
assert scroll_config.refresh_shortfall(99.95, 100) is None
assert scroll_config.refresh_shortfall(97.5, 100) is None
def test_a_panel_that_beats_its_cap_is_fine(self):
assert scroll_config.refresh_shortfall(125.7, 120) is None
def test_nothing_measured_says_nothing(self):
assert scroll_config.refresh_shortfall(None, 120) is None
assert scroll_config.refresh_shortfall(0, 120) is None
assert scroll_config.refresh_shortfall("fast", 120) is None
def test_the_suggestion_leaves_headroom_under_the_measurement(self):
assert scroll_config.holdable_cap(113.1) == 100
assert scroll_config.holdable_cap(95.0) == 90
# 5% under 105 is 99.75: 100 would sit inside the panel's drift.
assert scroll_config.holdable_cap(105.0) == 90
assert scroll_config.holdable_cap(9.0) is None
assert scroll_config.holdable_cap(None) is None
def test_the_log_line_names_the_cap_to_use(self):
text = scroll_config.describe_refresh_shortfall(
scroll_config.refresh_shortfall(113.1, 120))
assert "about 113 Hz" in text and "120 Hz" in text
assert "6% slower" in text
assert "Set Limit Refresh Rate to 100 Hz" in text
def test_no_suggestion_for_a_panel_too_slow_for_any_cap(self):
text = scroll_config.describe_refresh_shortfall(
scroll_config.refresh_shortfall(9.0, 100))
assert "Set Limit Refresh Rate" not in text
test/test_scroll_helper_lazy_image.py
-3
View File
@@ -187,9 +187,6 @@ def test_vegas_extends_without_building_the_image(no_fromarray):
def get_grouped_content_for_composition(self):
return groups[0]
def get_active_plugin_ids(self):
return ["a"]
def take_next_group(self, count=None, offscreen_only=False):
self.i += 1
return groups[self.i] if self.i < len(groups) else []
test/test_sports_helpers.py
+42 -12
View File
@@ -8,7 +8,9 @@ loses those tests with it.
The parity class is what keeps "byte-identical" true after this lands. Point
LEDMATRIX_PLUGINS at a ledmatrix-plugins checkout and every promoted body is
compared, as a docstring-stripped AST, against every plugin copy that carries
it. Without the variable it skips rather than fails, since core CI has no
it. A copy that is gone counts as adopted when the plugin imports
src.common.sports_helpers (plugins#563/#564 did that for every scoreboard).
Without the variable it skips rather than fails, since core CI has no
plugins checkout; ledmatrix-plugins CI runs the same comparison against core
(scripts/check_sports_helpers_parity.py, ledmatrix-plugins#495).
"""
@@ -572,6 +574,24 @@ def _core_definitions():
return out
def _sports_source(root, sport):
return (root / f"{sport}-scoreboard" / "sports.py").read_text(encoding="utf-8")
def _adopted(source):
"""Gone is fine once the plugin uses the module; otherwise the finder is
not seeing its copy."""
name = sports_helpers.__name__
for node in ast.walk(ast.parse(source)):
if isinstance(node, ast.ImportFrom):
if node.module == name or any(
f"{node.module}.{a.name}" == name for a in node.names):
return True
elif isinstance(node, ast.Import) and any(a.name == name for a in node.names):
return True
return False
class TestParityWithPlugins:
@pytest.mark.parametrize("name", sorted(PROMOTED))
def test_body_matches_every_plugin_copy(self, name):
@@ -580,11 +600,11 @@ class TestParityWithPlugins:
ours = _dump(_core_definitions()[name])
drifted, missing = [], []
for sport in carriers:
defs = _definitions(ast.parse(
(root / f"{sport}-scoreboard" / "sports.py").read_text(encoding="utf-8")))
theirs = defs[where].get(plugin_name)
source = _sports_source(root, sport)
theirs = _definitions(ast.parse(source))[where].get(plugin_name)
if theirs is None:
missing.append(sport)
if not _adopted(source):
missing.append(sport)
elif _dump(theirs) != ours:
drifted.append(sport)
assert missing == [], f"{plugin_name} no longer in: {missing}"
@@ -594,10 +614,20 @@ class TestParityWithPlugins:
@pytest.mark.parametrize("sport", SCOREBOARDS)
def test_constants_match(self, sport):
root = _plugins_root()
defs = _definitions(ast.parse(
(root / f"{sport}-scoreboard" / "sports.py").read_text(encoding="utf-8")))
assert ast.literal_eval(defs["module"]["_MIN_WINDOW_DAYS"].value) == MIN_WINDOW_DAYS
assert ast.literal_eval(defs["module"]["_MAX_WINDOW_DAYS"].value) == MAX_WINDOW_DAYS
gap = defs["SportsCore"]["_DWELL_REENTRY_GAP_SECONDS"].value
assert math.isclose(ast.literal_eval(gap), SportsHelpersMixin._DWELL_REENTRY_GAP_SECONDS)
source = _sports_source(_plugins_root(), sport)
defs = _definitions(ast.parse(source))
expected = {
("module", "_MIN_WINDOW_DAYS"): MIN_WINDOW_DAYS,
("module", "_MAX_WINDOW_DAYS"): MAX_WINDOW_DAYS,
("SportsCore", "_DWELL_REENTRY_GAP_SECONDS"):
SportsHelpersMixin._DWELL_REENTRY_GAP_SECONDS,
}
missing = []
for (where, name), value in expected.items():
node = defs[where].get(name)
if node is None:
if not _adopted(source):
missing.append(name)
else:
assert math.isclose(ast.literal_eval(node.value), value), name
assert missing == [], f"not found in {sport}: {missing}"
test/test_sync_manager.py
+17 -46
View File
@@ -78,7 +78,6 @@ def make_manager(role=SyncRole.STANDALONE, hw_config=None):
mgr._last_leader_frame_time = 0.0
mgr._frame_lock = threading.Lock()
mgr._leader_ip = None
mgr._on_new_cycle = None
mgr._on_scroll_image = None
mgr._pending_scroll_image = None
mgr._scroll_image_lock = threading.Lock()
@@ -495,31 +494,35 @@ class TestFollowerRecvLoop:
assert mgr._peer_compatible is True
assert mgr.logger.error.called is False
def test_scroll_x_switches_to_follower_and_builds_cycle(self):
def test_scroll_x_switches_to_follower(self):
mgr = make_manager(role=SyncRole.FOLLOWER)
calls = []
mgr._on_new_cycle = lambda: calls.append(1)
self._drive(mgr, json.dumps({"t": "sx", "x": 12.34}).encode())
assert mgr._follower_state is FollowerState.FOLLOWER
assert mgr.get_latest_scroll_x() == 12.34
assert calls == [1]
def test_scroll_x_while_already_following_does_not_rebuild(self):
def test_scroll_x_while_already_following_updates_the_position(self):
mgr = make_manager(role=SyncRole.FOLLOWER)
mgr._follower_state = FollowerState.FOLLOWER
calls = []
mgr._on_new_cycle = lambda: calls.append(1)
self._drive(mgr, json.dumps({"t": "sx", "x": 5.0}).encode())
assert mgr.get_latest_scroll_x() == 5.0
assert calls == []
def test_new_cycle_message_triggers_callback(self):
@pytest.mark.parametrize("payload", [{"t": "nc"}, {"t": "some-future-type"}])
def test_an_older_or_newer_leaders_message_is_ignored(self, payload):
# Older leaders send "nc" at each new cycle. Nothing uses it, and a
# follower must take it -- or any type it does not know -- quietly:
# not as a frame, not as a malformed packet, no error back-off.
mgr = make_manager(role=SyncRole.FOLLOWER)
mgr._follower_state = FollowerState.FOLLOWER
calls = []
mgr._on_new_cycle = lambda: calls.append(1)
self._drive(mgr, json.dumps({"t": "nc"}).encode())
assert calls == [1]
sleeps = MagicMock()
with patch.object(sync_manager, "time",
SimpleNamespace(time=time.time, monotonic=time.monotonic,
sleep=sleeps)):
self._drive(mgr, json.dumps(payload).encode())
assert mgr._follower_state is FollowerState.FOLLOWER
assert mgr.get_latest_frame() is None
assert mgr.get_latest_scroll_x() is None
assert not mgr.logger.debug.called
sleeps.assert_not_called()
def test_non_object_json_does_not_reach_the_outer_handler(self):
# A bare JSON scalar parses, then msg.get() raises AttributeError.
@@ -547,28 +550,6 @@ class TestFollowerRecvLoop:
assert mgr.get_latest_scroll_x() is None
sleeps.assert_not_called()
def test_callback_failure_is_not_mistaken_for_a_malformed_packet(self, monkeypatch):
# A payload that parses is a control message, full stop. If the
# callback it triggers raises one of the types the field guard
# catches, that fault belongs to the callback: it must not send
# the packet to the image decoder, which would report it as a
# decode error and bury the real cause. The loop still survives
# it — the outer handler catches it like any other fault.
mgr = make_manager(role=SyncRole.FOLLOWER)
mgr._follower_state = FollowerState.FOLLOWER
def boom():
raise ValueError("callback is broken")
mgr._on_new_cycle = boom
fake_clock(monkeypatch, sleep_fn=MagicMock())
self._drive(mgr, json.dumps({"t": "nc"}).encode())
logged = " | ".join(str(c) for c in mgr.logger.debug.call_args_list)
assert "callback is broken" in logged
assert "frame decode error" not in logged
assert "malformed control message" not in logged
def test_oversized_legacy_frame_is_rejected_before_decode(self, monkeypatch):
# The UDP path is reachable by any host on the LAN, so it caps
# dimensions before load() just as the TCP image server does.
@@ -603,12 +584,9 @@ class TestFollowerRecvLoop:
for payload in (b'{"t": "sx", "x": ' + literal.encode() + b'}',
json.dumps({"t": "sx", "x": literal}).encode()):
mgr = make_manager(role=SyncRole.FOLLOWER)
calls = []
mgr._on_new_cycle = lambda: calls.append(1)
self._drive(mgr, payload)
assert mgr.get_latest_scroll_x() is None
assert mgr._follower_state is FollowerState.STANDALONE
assert calls == []
def test_non_finite_scroll_x_leaves_a_good_value_in_place(self):
# The reject must not clear the last usable position either — a
@@ -697,17 +675,10 @@ class TestSendControlMessages:
msg = json.loads(mgr._send_sock.sendto.call_args[0][0].decode())
assert msg == {"t": "sx", "x": 3.14}
def test_send_new_cycle(self):
mgr = self._connected_leader()
mgr.send_new_cycle()
msg = json.loads(mgr._send_sock.sendto.call_args[0][0].decode())
assert msg == {"t": "nc"}
def test_control_messages_noop_when_disconnected(self):
mgr = self._connected_leader()
mgr._leader_state = LeaderState.NO_PEER
mgr.send_scroll_x(1.0)
mgr.send_new_cycle()
assert not mgr._send_sock.sendto.called
def test_set_leader_width(self):
test/test_vegas_config.py
+8 -41
View File
@@ -1,10 +1,12 @@
"""
Tests for src/vegas_mode/config.py
Covers VegasModeConfig: from_config, to_dict, get_frame_interval,
Covers VegasModeConfig: from_config, get_frame_interval,
get_ordered_plugins, validate.
"""
import dataclasses
import pytest
from src.vegas_mode.config import VegasModeConfig
@@ -49,7 +51,8 @@ class TestVegasModeConfigDefaults:
(Path(__file__).resolve().parent.parent / "config"
/ "config.template.json").read_text(encoding="utf-8"))
shipped = template["display"]["vegas_scroll"]
defaults = VegasModeConfig().to_dict()
defaults = dataclasses.asdict(VegasModeConfig())
defaults["excluded_plugins"] = sorted(defaults["excluded_plugins"])
mismatched = {k: (v, defaults[k]) for k, v in shipped.items()
if k in defaults and defaults[k] != v}
assert not mismatched, f"template vs code default: {mismatched}"
@@ -57,9 +60,9 @@ class TestVegasModeConfigDefaults:
def test_missing_keys_read_the_field_defaults(self):
# from_config used to repeat every default; with no keys set it must
# produce exactly the dataclass defaults.
assert VegasModeConfig.from_config({}).to_dict() == VegasModeConfig().to_dict()
assert (VegasModeConfig.from_config({"display": {"vegas_scroll": {}}}).to_dict()
== VegasModeConfig().to_dict())
assert VegasModeConfig.from_config({}) == VegasModeConfig()
assert (VegasModeConfig.from_config({"display": {"vegas_scroll": {}}})
== VegasModeConfig())
# ---------------------------------------------------------------------------
@@ -114,42 +117,6 @@ class TestFromConfig:
assert cfg.frame_based_scrolling is False
# ---------------------------------------------------------------------------
# to_dict
# ---------------------------------------------------------------------------
class TestToDict:
def test_roundtrip(self):
original = VegasModeConfig(
enabled=True,
scroll_speed=75.0,
separator_width=24,
plugin_order=["a", "b"],
excluded_plugins={"z"},
target_fps=100,
)
d = original.to_dict()
assert d["enabled"] is True
assert d["scroll_speed"] == 75.0
assert d["separator_width"] == 24
assert d["plugin_order"] == ["a", "b"]
assert "z" in d["excluded_plugins"]
assert d["target_fps"] == 100
def test_excluded_plugins_is_list(self):
cfg = VegasModeConfig(excluded_plugins={"x"})
d = cfg.to_dict()
assert isinstance(d["excluded_plugins"], list)
def test_all_keys_present(self):
d = VegasModeConfig().to_dict()
for key in ("enabled", "scroll_speed", "separator_width", "plugin_order",
"excluded_plugins", "target_fps", "buffer_ahead",
"frame_based_scrolling", "scroll_delay",
"dynamic_duration_enabled", "min_cycle_duration", "max_cycle_duration"):
assert key in d
# ---------------------------------------------------------------------------
# get_frame_interval
# ---------------------------------------------------------------------------
test/test_vegas_continuous_refresh.py
-1
View File
@@ -189,7 +189,6 @@ class TestCoordinatorWiring:
coordinator.render_pipeline.is_cycle_complete.return_value = False
coordinator.render_pipeline.should_recompose.return_value = False
coordinator.stream_manager = MagicMock()
coordinator.stats = {'cycles_completed': 0}
coordinator._state_lock = threading.Lock()
coordinator._is_active = True
coordinator._is_paused = False
test/test_vegas_coordinator_config.py
-1
View File
@@ -25,7 +25,6 @@ def _coordinator(active=True):
c.render_pipeline.is_cycle_complete.return_value = False
c.stream_manager = MagicMock()
c.plugin_adapter = MagicMock()
c.stats = {'cycles_completed': 0, 'config_updates': 0}
c._state_lock = threading.Lock()
c._is_active = active
c._is_paused = False
test/test_vegas_coordinator_iteration.py
-2
View File
@@ -27,7 +27,6 @@ def _coordinator(plugins):
coord.stream_manager = MagicMock()
coord.display_manager = MagicMock()
coord.plugin_manager = SimpleNamespace(plugins=plugins, get_plugin=plugins.get)
coord.stats = {'cycles_completed': 0, 'interruptions': 0}
coord._state_lock = threading.Lock()
coord._is_active = True
coord._is_paused = False
@@ -99,7 +98,6 @@ def test_vegas_resumes_after_a_live_priority_pause():
def test_stop_clears_a_live_priority_pause():
coord = _live_coordinator(['nfl_live'])
coord._start_time = None
coord._restore_switch_interval = lambda: None
coord._remove_render_gate = lambda: None
coord.run_iteration()
test/test_vegas_crisp_pacing.py
-3
View File
@@ -26,9 +26,6 @@ class FakeStream:
def get_grouped_content_for_composition(self):
return [('a', [Image.new('RGB', (4000, H), (255, 255, 255))])]
def get_active_plugin_ids(self):
return ['a']
class FakeDM:
width = W
test/test_vegas_density.py
+3 -20
View File
@@ -3,6 +3,7 @@ Tests for the Vegas mode density work: dead-space trimming in PluginAdapter
and the configurable lead-in gap in ScrollHelper.
"""
import dataclasses
from contextlib import contextmanager
import pytest
@@ -298,9 +299,6 @@ class TestPluginBoundaryGaps:
def get_grouped_content_for_composition(self):
return grouped
def get_active_plugin_ids(self):
return [pid for pid, _ in grouped]
class DM:
width = DISPLAY_W
height = DISPLAY_H
@@ -615,10 +613,10 @@ class TestConfigSurface:
assert cfg.lead_in_width == 0
assert cfg.content_padding == 8
def test_round_trips_through_to_dict(self):
def test_round_trips_through_asdict(self):
cfg = VegasModeConfig(trim_threshold=20, lead_in_width=64)
restored = VegasModeConfig.from_config(
{'display': {'vegas_scroll': cfg.to_dict()}})
{'display': {'vegas_scroll': dataclasses.asdict(cfg)}})
assert restored.trim_threshold == 20
assert restored.lead_in_width == 64
@@ -698,9 +696,6 @@ class TestMeasuredSeparation:
def get_grouped_content_for_composition(self):
return grouped
def get_active_plugin_ids(self):
return [pid for pid, _ in grouped]
class DM:
width = DISPLAY_W
height = DISPLAY_H
@@ -816,9 +811,6 @@ class TestCycleEndsBeforeWrap:
def get_grouped_content_for_composition(self):
return [('a', [Image.new('RGB', (strip_width, DISPLAY_H), (255, 255, 255))])]
def get_active_plugin_ids(self):
return ['a']
class DM:
width = DISPLAY_W
height = DISPLAY_H
@@ -962,9 +954,6 @@ class TestBudgetUsesMeasuredGaps:
def get_grouped_content_for_composition(self):
return [('rows', selected)]
def get_active_plugin_ids(self):
return ['rows']
class DM:
width = DISPLAY_W
height = DISPLAY_H
@@ -1258,9 +1247,6 @@ class TestContinuousExtension:
def get_grouped_content_for_composition(self):
return groups[0] if groups else []
def get_active_plugin_ids(self):
return [pid for pid, _ in (groups[0] if groups else [])]
def take_next_group(self, count=None, offscreen_only=False):
self.calls.append(offscreen_only)
if self._i >= len(groups):
@@ -1450,9 +1436,6 @@ class TestDeferredDraining:
def get_grouped_content_for_composition(self):
return [('seed', [Image.new('RGB', (600, DISPLAY_H), (255, 255, 255))])]
def get_active_plugin_ids(self):
return ['seed']
def take_next_group(self, count=None, offscreen_only=False):
if self._served:
return []
test/test_vegas_elements_layout.py
-3
View File
@@ -49,9 +49,6 @@ class _Stream:
def get_grouped_content_for_composition(self):
return self.groups[0]
def get_active_plugin_ids(self):
return [pid for pid, _ in self.groups[0]]
def take_next_group(self, count=None, offscreen_only=False):
self.i += 1
return self.groups[self.i] if self.i < len(self.groups) else []
test/test_vegas_live_apply.py
-3
View File
@@ -43,9 +43,6 @@ class _Stream:
def get_grouped_content_for_composition(self):
return self.groups[0]
def get_active_plugin_ids(self):
return ["p"]
def take_next_group(self, count=None, offscreen_only=False):
self.i += 1
return self.groups[self.i] if self.i < len(self.groups) else []
test/test_vegas_live_integration.py
+1 -1
View File
@@ -189,7 +189,7 @@ def test_with_live_refresh_off_the_same_run_is_plain_content(dm, tmp_path, ticke
start_width = pipeline.scroll_helper.total_scroll_width
assert _run_until(
coordinator,
lambda: pipeline.stats.get('extensions', 0) >= 1, seconds=10.0)
lambda: pipeline.extensions >= 1, seconds=10.0)
assert pipeline.live_records() == ()
assert np.asarray(pipeline.scroll_helper.cached_array).shape[1] > 0
assert start_width > 0
test/test_vegas_live_lifecycle.py
-3
View File
@@ -46,9 +46,6 @@ class _Stream:
def get_grouped_content_for_composition(self):
return self.first
def get_active_plugin_ids(self):
return ["p"]
def take_next_group(self, count=None, offscreen_only=False):
self.taken += 1
return [("q", [_image(40, 99)])]
test/test_vegas_live_property.py
-3
View File
@@ -67,9 +67,6 @@ class _Stream:
def get_grouped_content_for_composition(self):
return self.groups[0]
def get_active_plugin_ids(self):
return []
def take_next_group(self, count=None, offscreen_only=False):
self.i += 1
return self.groups[self.i] if self.i < len(self.groups) else []
test/test_vegas_prepared_blocks.py
-3
View File
@@ -65,9 +65,6 @@ class _Stream:
def get_grouped_content_for_composition(self):
return self.groups[0]
def get_active_plugin_ids(self):
return []
def take_next_group(self, count=None, offscreen_only=False):
self.i += 1
return self.groups[self.i] if self.i < len(self.groups) else []
test/test_vegas_static_mode.py
+2 -6
View File
@@ -52,9 +52,6 @@ class FakeStream:
def get_static_layout(self):
return [(pid, w is None) for pid, w in self.layout]
def get_active_plugin_ids(self):
return [pid for pid, _ in self.layout]
def is_static_plugin(self, plugin_id):
return plugin_id in self.statics
@@ -163,7 +160,6 @@ class TestStreamManagerSkipsStaticContent:
sm.plugin_manager = SimpleNamespace(plugins=plugins)
sm.plugin_adapter = MagicMock()
sm.plugin_adapter.get_content.return_value = [block(10)]
sm.stats = {'segments_fetched': 0, 'fetch_errors': 0}
sm.refresh = lambda: None
return sm
@@ -213,13 +209,13 @@ class TestCoordinatorStaticPause:
coord._live_priority_active = False
coord._live_priority_check = None
coord._interrupt_check = None
coord.stats = {'static_pauses': 0}
return coord
def _plugin(self):
plugin = MagicMock()
plugin.plugin_id = 'clock'
plugin.get_display_duration.return_value = 0
# A moment: zero would pause 15 s, as the rotation shows it.
plugin.get_display_duration.return_value = 0.01
return plugin
def test_trigger_comes_from_the_pipeline(self):
test/test_vegas_static_pause_duration.py
+197
View File
@@ -0,0 +1,197 @@
"""A Vegas static pause lasts as long as the rotation shows the plugin.
The pause asked the plugin for get_display_duration() and compared the
answer with the clock. Several plugins (clock-simple, calendar, countdown)
return their display_duration setting as it is in config.json, so one saved
as "20" or null -- the raw config editor, a hand edit -- reached that
comparison as a string or None. The TypeError went to the pause's broad
except, which ended the pause: the plugin flashed up and the scroll went on,
at every one of its turns. inf paused until something interrupted it, and
NaN, False, 0 or a negative number ended the pause at once.
The pause now reads the answer the way the rotation does since #739, with
the same helper (base_plugin.finite_seconds): a numeric string counts;
anything else that is not a finite number, or a raise, gets the rotation's
30 s; a number at or below zero gets its 15 s.
"""
import logging
import os
import threading
from types import SimpleNamespace
from unittest.mock import MagicMock
os.environ.setdefault("EMULATOR", "true")
import pytest
from src.vegas_mode import coordinator
NOT_NUMBERS = [None, '', 'twenty', True, False, float('nan'), float('inf'),
'inf', '1e400', [20], {'seconds': 20}]
NOT_ABOVE_ZERO = [0, -5, '-5', '0']
NUMBERS = [('20', 20.0), (' 7.5 ', 7.5), (12, 12.0), (12.5, 12.5)]
class FakeClock:
"""time.monotonic/time.sleep for the pause loop: sleeping moves the clock."""
#: A pause still going after this long never ends (inf did that).
LIMIT = 3600.0
def __init__(self):
self.now = 0.0
def monotonic(self):
return self.now
def sleep(self, seconds):
self.now += seconds
if self.now > self.LIMIT:
raise RuntimeError("the static pause never ended")
@pytest.fixture
def clock(monkeypatch):
fake = FakeClock()
monkeypatch.setattr(coordinator, 'time', fake)
return fake
def _plugin(duration, plugin_id='clock-simple'):
plugin = MagicMock()
plugin.plugin_id = plugin_id
plugin.get_display_duration.return_value = duration
return plugin
def _coord(*plugins):
coord = coordinator.VegasModeCoordinator.__new__(coordinator.VegasModeCoordinator)
coord.render_pipeline = MagicMock()
coord.render_pipeline.get_scroll_position.return_value = 0
coord.display_manager = MagicMock()
locks = {plugin.plugin_id: threading.Lock() for plugin in plugins}
coord.plugin_manager = SimpleNamespace(get_plugin_lock=locks.__getitem__)
coord._state_lock = threading.Lock()
coord._static_pause_active = False
coord._saved_scroll_position = None
coord._should_stop = False
coord._live_priority_active = False
coord._live_priority_check = None
coord._interrupt_check = None
coord.stats = {'static_pauses': 0}
return coord
def _pause(coord, plugin, clock):
"""One static pause: (whether it completed, how long it lasted)."""
start = clock.now
completed = coord._handle_static_pause(plugin)
return completed, clock.now - start
class TestPauseLength:
@pytest.mark.parametrize('value, seconds', NUMBERS)
def test_numbers_and_numeric_strings_are_used(self, clock, value, seconds):
plugin = _plugin(value)
completed, lasted = _pause(_coord(plugin), plugin, clock)
assert completed is True
assert lasted == pytest.approx(seconds, abs=0.15)
@pytest.mark.parametrize('value', NOT_NUMBERS, ids=repr)
def test_anything_but_a_finite_number_pauses_for_30s(self, clock, value):
plugin = _plugin(value)
completed, lasted = _pause(_coord(plugin), plugin, clock)
assert completed is True
assert lasted == pytest.approx(30.0, abs=0.15)
plugin.display.assert_called_once_with(force_clear=True)
@pytest.mark.parametrize('value', NOT_ABOVE_ZERO, ids=repr)
def test_a_number_not_above_zero_pauses_for_15s(self, clock, value):
plugin = _plugin(value)
completed, lasted = _pause(_coord(plugin), plugin, clock)
assert completed is True
assert lasted == pytest.approx(15.0, abs=0.15)
def test_a_raising_get_display_duration_pauses_for_30s(self, clock):
plugin = _plugin(None)
plugin.get_display_duration.side_effect = KeyError('display_duration')
completed, lasted = _pause(_coord(plugin), plugin, clock)
assert completed is True
assert lasted == pytest.approx(30.0, abs=0.15)
def test_a_good_value_after_a_bad_one_is_used(self, clock):
plugin = _plugin(None)
coord = _coord(plugin)
assert _pause(coord, plugin, clock)[1] == pytest.approx(30.0, abs=0.15)
plugin.get_display_duration.return_value = 45
assert _pause(coord, plugin, clock)[1] == pytest.approx(45.0, abs=0.15)
def test_the_pause_can_still_be_interrupted(self, clock):
plugin = _plugin('twenty')
coord = _coord(plugin)
coord._interrupt_check = lambda: clock.now >= 5
completed, lasted = _pause(coord, plugin, clock)
assert completed is False
assert lasted == pytest.approx(5.0, abs=0.15)
class TestWarning:
def test_logged_once_per_plugin(self, clock, caplog):
clock_plugin = _plugin('twenty')
calendar = _plugin(None, plugin_id='calendar')
coord = _coord(clock_plugin, calendar)
with caplog.at_level(logging.WARNING, logger='src.vegas_mode.coordinator'):
for _ in range(3):
for plugin in (clock_plugin, calendar):
coord._handle_static_pause(plugin)
warnings = [r.getMessage() for r in caplog.records
if 'display duration' in r.getMessage()]
assert len(warnings) == 2
assert any('clock-simple' in m and "'twenty'" in m for m in warnings)
assert any('calendar' in m and 'None' in m for m in warnings)
class TestFiniteSeconds:
"""The shared rule: what counts as a number of seconds."""
@pytest.mark.parametrize('value, seconds', NUMBERS + [(0, 0.0), ('-5', -5.0)])
def test_numbers_and_numeric_strings(self, value, seconds):
from src.plugin_system.base_plugin import finite_seconds
result = finite_seconds(value)
assert result == seconds and isinstance(result, float)
@pytest.mark.parametrize('value', NOT_NUMBERS + [pytest.param(10 ** 400, id='10**400')],
ids=repr)
def test_anything_else_is_none(self, value):
from src.plugin_system.base_plugin import finite_seconds
assert finite_seconds(value) is None
def _rotation_seconds(plugin):
"""How long the rotation shows ``plugin`` (no dynamic duration, no
Rotation & Durations override): the two calls run() makes for a screen.
"""
from src.display_controller import DisplayController
dc = object.__new__(DisplayController)
dc.config = {}
dc.plugin_modes = {'mode': plugin}
return dc._resolve_durations(plugin, 'mode', dc._get_display_duration('mode'), False)[1]
class TestSameAsTheRotation:
"""The pause and the rotation share finite_seconds; this pins their
fallbacks (30 s, 15 s) to each other too."""
@pytest.mark.parametrize('value', [value for value, _ in NUMBERS]
+ NOT_NUMBERS + NOT_ABOVE_ZERO, ids=repr)
def test_the_pause_lasts_as_long_as_the_rotation_shows_it(self, clock, value):
plugin = _plugin(value)
expected = _rotation_seconds(plugin)
assert _pause(_coord(plugin), plugin, clock)[1] == pytest.approx(expected, abs=0.15)
def test_a_raise_too(self, clock):
plugin = _plugin(None)
plugin.get_display_duration.side_effect = KeyError('display_duration')
expected = _rotation_seconds(plugin)
assert _pause(_coord(plugin), plugin, clock)[1] == pytest.approx(expected, abs=0.15)
test/test_web_smoke.py
+1 -1
View File
@@ -174,7 +174,7 @@ def test_widget_bundle_is_served_and_requested(client):
assert resp.mimetype == "application/javascript"
body = resp.get_data(as_text=True)
assert body.index("/* registry.js */") < body.index("/* notification.js */")
assert "/* json-file-manager.js */" in body
assert "/* plugin-loader.js */" in body
def test_durations_page_groups_by_plugin(client):
test/web_interface/integration/test_plugin_operations.py
+1 -26
View File
@@ -66,31 +66,6 @@ class TestPluginOperationsIntegration(unittest.TestCase):
history = self.operation_history.get_history(plugin_id=plugin_id)
self.assertEqual([r.operation_type for r in history], ["install"])
def test_update_operation_flow(self):
"""Test complete update operation flow."""
plugin_id = "test-plugin"
# Enqueue update operation
operation_id = self.operation_queue.enqueue_operation(
OperationType.UPDATE,
plugin_id,
{"from_version": "1.0.0", "to_version": "2.0.0"}
)
self.assertIsNotNone(operation_id)
# Record in history
self.operation_history.record_operation(
operation_type="update",
plugin_id=plugin_id,
status="in_progress",
operation_id=operation_id
)
# Verify history
history = self.operation_history.get_history(plugin_id=plugin_id)
self.assertEqual([r.operation_type for r in history], ["update"])
def test_uninstall_operation_flow(self):
"""Test complete uninstall operation flow."""
plugin_id = "test-plugin"
@@ -162,7 +137,7 @@ class TestPluginOperationsIntegration(unittest.TestCase):
# The prevention only works for truly concurrent (pending/running) operations
try:
op2_id = self.operation_queue.enqueue_operation(
OperationType.UPDATE,
OperationType.UNINSTALL,
plugin_id
)
# If no exception, the first operation may have completed already
test/web_interface/test_api_v3_helpers.py
-39
View File
@@ -22,7 +22,6 @@ from web_interface.blueprints.api_v3 import ( # noqa: E402
_is_plugin_update_available,
_coerce_to_bool,
deep_merge,
_parse_form_value,
_get_schema_property,
_set_nested_value,
_SKIP_FIELD,
@@ -116,44 +115,6 @@ class TestDeepMerge:
assert result["keep"] is base["keep"] # untouched subtree is shared
class TestParseFormValue:
def test_boolean_strings(self):
assert _parse_form_value("true") is True
assert _parse_form_value("False") is False
def test_null_like_strings(self):
assert _parse_form_value("null") is None
assert _parse_form_value("none") is None
assert _parse_form_value("") is None
def test_none_passthrough(self):
assert _parse_form_value(None) is None
def test_numbers(self):
assert _parse_form_value("42") == 42
assert isinstance(_parse_form_value("42"), int)
assert _parse_form_value("3.5") == 3.5
assert isinstance(_parse_form_value("3.5"), float)
def test_json_array_parsed_before_numbers(self):
# RGB arrays like "[255, 0, 0]" must come back as lists.
assert _parse_form_value("[255, 0, 0]") == [255, 0, 0]
def test_json_object(self):
assert _parse_form_value('{"a": 1}') == {"a": 1}
def test_malformed_json_falls_back_to_string(self):
assert _parse_form_value("[not json") == "[not json"
def test_plain_string_returned_unstripped(self):
# The original value (not the stripped copy) is returned.
assert _parse_form_value(" hello ") == " hello "
def test_non_string_passthrough(self):
assert _parse_form_value(7) == 7
assert _parse_form_value([1, 2]) == [1, 2]
class TestGetSchemaProperty:
SCHEMA: ClassVar[Dict[str, Any]] = {
"properties": {
test/web_interface/test_api_v3_refresh_rate.py
+73
View File
@@ -0,0 +1,73 @@
"""GET /api/v3/config/refresh-rate: the cap, the measured rate, a cap to hold."""
import json
from unittest.mock import MagicMock
import pytest
from flask import Flask
from web_interface.blueprints.api_v3 import api_v3
@pytest.fixture
def client(monkeypatch, tmp_path):
stats = tmp_path / "stats.json"
monkeypatch.setattr("src.common.frame_timing.default_stats_path", lambda: str(stats))
manager = MagicMock()
manager.load_config.return_value = {
"display": {"hardware": {"limit_refresh_rate_hz": 120}}}
monkeypatch.setattr(api_v3, "config_manager", manager, raising=False)
app = Flask(__name__)
app.register_blueprint(api_v3, url_prefix="/api/v3")
c = app.test_client()
c.stats_path = stats
return c
def _get(client):
body = client.get("/api/v3/config/refresh-rate").get_json()
assert body["status"] == "success"
return body["data"]
def test_nothing_measured_yet(client):
data = _get(client)
assert data == {"planned_hz": 120.0, "measured_hz": None, "shortfall": None}
def test_a_panel_short_of_its_cap_gets_a_cap_it_can_hold(client):
client.stats_path.write_text(json.dumps(
{"measured_refresh_hz": 110.4, "planned_refresh_hz": 120.0}))
data = _get(client)
assert data["measured_hz"] == 110.4
assert data["shortfall"]["suggested_cap_hz"] == 100
assert data["shortfall"]["slow_percent"] == 8
def test_a_panel_at_its_cap_has_no_shortfall(client):
client.stats_path.write_text(json.dumps(
{"measured_refresh_hz": 121.3, "planned_refresh_hz": 120.0}))
assert _get(client)["shortfall"] is None
def test_a_file_written_under_another_cap_is_stale(client):
# The cap was changed to 120 but the display still runs under 100 Hz.
client.stats_path.write_text(json.dumps(
{"measured_refresh_hz": 99.9, "planned_refresh_hz": 100.0}))
data = _get(client)
assert data["measured_hz"] is None
assert data["shortfall"] is None
def test_a_file_from_a_display_too_old_to_record_its_cap_still_counts(client):
client.stats_path.write_text(json.dumps({"measured_refresh_hz": 110.4}))
assert _get(client)["shortfall"]["suggested_cap_hz"] == 100
def test_a_measurement_recorded_without_a_planned_rate_is_not_a_panel(client):
# The emulator and the fallback canvas write the key as null: their frames
# are not paced by a panel, so a rate under the cap is no shortfall.
client.stats_path.write_text(json.dumps(
{"measured_refresh_hz": 60.0, "planned_refresh_hz": None}))
data = _get(client)
assert data["measured_hz"] is None
assert data["shortfall"] is None
test/web_interface/test_error_handler.py
+34 -30
View File
@@ -1,24 +1,23 @@
"""
Tests for the response builders in src/web_interface/error_handler.py and
the success path in src/web_interface/api_helpers.py.
Tests for the response builders in src/web_interface/api_helpers.py.
describe_exception() in the same module is already covered by
describe_exception() in error_handler.py is already covered by
test/test_web_error_detail.py and is not duplicated here.
Regression coverage for one fixed bug: create_success_response used
Regression coverage for one fixed bug: the success builder used
truthiness for `message` and `metadata` while using `is not None` for
`data`, so an explicitly-passed "" or {} was silently dropped —
api_helpers.success_response() repeated the same gate, which is the path
every api_v3 endpoint actually calls.
success_response() repeated the same gate, which is the path every api_v3
endpoint actually calls.
"""
import pytest
from flask import Flask
from src.web_interface.api_helpers import exception_error_response, success_response
from src.web_interface.error_handler import (
create_error_response,
create_success_response,
from src.web_interface.api_helpers import (
error_response,
exception_error_response,
success_response,
)
from src.web_interface.errors import ErrorCode, WebInterfaceError
@@ -28,23 +27,23 @@ def app():
return Flask(__name__)
class TestCreateErrorResponse:
class TestErrorResponse:
def test_returns_response_and_status_tuple(self, app):
with app.test_request_context():
response, status = create_error_response(
response, status = error_response(
ErrorCode.CONFIG_SAVE_FAILED, "could not save")
assert status == 500
assert response.get_json()["message"] == "could not save"
def test_status_code_passthrough(self, app):
with app.test_request_context():
_, status = create_error_response(
_, status = error_response(
ErrorCode.INVALID_INPUT, "bad", status_code=400)
assert status == 400
def test_body_matches_the_error_dataclass(self, app):
with app.test_request_context():
response, _ = create_error_response(
response, _ = error_response(
ErrorCode.NETWORK_ERROR, "offline",
details="connection refused", context={"url": "http://x"})
expected = WebInterfaceError(
@@ -54,12 +53,12 @@ class TestCreateErrorResponse:
def test_none_context_produces_no_context_key(self, app):
with app.test_request_context():
response, _ = create_error_response(ErrorCode.SYSTEM_ERROR, "boom")
response, _ = error_response(ErrorCode.SYSTEM_ERROR, "boom")
assert "context" not in response.get_json()
def test_suggested_fixes_passed_through(self, app):
with app.test_request_context():
response, _ = create_error_response(
response, _ = error_response(
ErrorCode.SYSTEM_ERROR, "boom", suggested_fixes=["Try again"])
assert response.get_json()["suggested_fixes"] == ["Try again"]
@@ -74,7 +73,6 @@ class TestExceptionErrorResponse:
@staticmethod
def _by_hand(exc, code, with_context):
from src.web_interface.api_helpers import error_response
error = WebInterfaceError.from_exception(exc, code)
if with_context:
return error_response(error.error_code, error.message,
@@ -117,39 +115,45 @@ class TestExceptionErrorResponse:
assert "context" not in response.get_json()
class TestCreateSuccessResponse:
def _success_body(**kwargs):
"""success_response()'s body, outside any request timing."""
with Flask(__name__).test_request_context():
return success_response(**kwargs).get_json()
class TestSuccessResponseBody:
def test_bare_success(self):
assert create_success_response() == {"status": "success"}
assert _success_body() == {"status": "success"}
def test_data_included(self):
assert create_success_response(data={"a": 1})["data"] == {"a": 1}
assert _success_body(data={"a": 1})["data"] == {"a": 1}
@pytest.mark.parametrize("falsy", [0, "", False, {}, []])
def test_falsy_data_is_still_included(self, falsy):
assert create_success_response(data=falsy)["data"] == falsy
assert _success_body(data=falsy)["data"] == falsy
def test_none_data_omitted(self):
assert "data" not in create_success_response(data=None)
assert "data" not in _success_body(data=None)
def test_message_included(self):
assert create_success_response(message="done")["message"] == "done"
assert _success_body(message="done")["message"] == "done"
def test_empty_message_is_still_included(self):
# Regression: `if message:` dropped an explicitly-passed "".
assert create_success_response(message="")["message"] == ""
assert _success_body(message="")["message"] == ""
def test_none_message_omitted(self):
assert "message" not in create_success_response(message=None)
assert "message" not in _success_body(message=None)
def test_metadata_included(self):
assert create_success_response(metadata={"v": 1})["metadata"] == {"v": 1}
assert _success_body(metadata={"v": 1})["metadata"] == {"v": 1}
def test_empty_metadata_is_still_included(self):
# Regression: `if metadata:` dropped an explicitly-passed {}.
assert create_success_response(metadata={})["metadata"] == {}
assert _success_body(metadata={})["metadata"] == {}
def test_none_metadata_omitted(self):
assert "metadata" not in create_success_response(metadata=None)
assert "metadata" not in _success_body(metadata=None)
class TestSuccessResponseHelper:
@@ -162,8 +166,8 @@ class TestSuccessResponseHelper:
def test_explicit_empty_metadata_survives_the_wrapper(self, app):
# Regression: the wrapper re-gated metadata on truthiness after
# create_success_response had already included it, so {} was
# dropped again on the way out.
# the body builder had already included it, so {} was dropped
# again on the way out.
with app.test_request_context():
body = success_response(data=None, metadata={}).get_json()
assert body["metadata"] == {}
test/web_interface/test_plugin_operation_queue.py
+2 -17
View File
@@ -55,7 +55,7 @@ class TestPluginOperationQueue(unittest.TestCase):
# behavior may differ. For this test, we'll verify the mechanism exists.
try:
self.queue.enqueue_operation(
OperationType.UPDATE,
OperationType.UNINSTALL,
"test-plugin"
)
# If no exception, the first operation may have completed
@@ -64,21 +64,6 @@ class TestPluginOperationQueue(unittest.TestCase):
# Expected behavior - concurrent operation prevented
pass
def test_operation_cancellation(self):
"""Test cancelling a pending operation."""
operation_id = self.queue.enqueue_operation(
OperationType.INSTALL,
"test-plugin"
)
# Cancel operation
success = self.queue.cancel_operation(operation_id)
self.assertTrue(success)
# Check status
operation = self.queue.get_operation_status(operation_id)
self.assertEqual(operation.status, OperationStatus.CANCELLED)
def test_operation_history(self):
"""Test operation history tracking."""
# Enqueue and complete an operation
@@ -92,7 +77,7 @@ class TestPluginOperationQueue(unittest.TestCase):
time.sleep(0.5)
# Check history
history = self.queue.get_operation_history(limit=10)
history = self.queue._operation_history
self.assertGreater(len(history), 0)
# Find our operation in history
test/web_interface/test_state_reconciliation.py
-5
View File
@@ -100,7 +100,6 @@ class TestStateReconciliation(unittest.TestCase):
inconsistency = result.inconsistencies_found[0]
self.assertEqual(inconsistency.plugin_id, "plugin1")
self.assertEqual(inconsistency.inconsistency_type, InconsistencyType.PLUGIN_MISSING_IN_CONFIG)
self.assertTrue(inconsistency.can_auto_fix)
self.assertEqual(inconsistency.fix_action, FixAction.AUTO_FIX)
def test_plugin_missing_on_disk(self):
@@ -115,7 +114,6 @@ class TestStateReconciliation(unittest.TestCase):
inconsistency = result.inconsistencies_found[0]
self.assertEqual(inconsistency.plugin_id, "plugin1")
self.assertEqual(inconsistency.inconsistency_type, InconsistencyType.PLUGIN_MISSING_ON_DISK)
self.assertFalse(inconsistency.can_auto_fix)
self.assertEqual(inconsistency.fix_action, FixAction.MANUAL_FIX_REQUIRED)
def test_enabled_but_not_loaded_is_reported_not_fixed(self):
@@ -139,7 +137,6 @@ class TestStateReconciliation(unittest.TestCase):
inconsistency = result.inconsistencies_found[0]
self.assertEqual(inconsistency.inconsistency_type, InconsistencyType.PLUGIN_ENABLED_MISMATCH)
self.assertEqual(inconsistency.fix_action, FixAction.NO_ACTION)
self.assertFalse(inconsistency.can_auto_fix)
self.assertIn("No module named", inconsistency.description)
self.assertEqual(result.inconsistencies_fixed, [])
self.assertEqual(result.inconsistencies_manual, [])
@@ -410,7 +407,6 @@ class TestStateReconciliationUnrecoverable(unittest.TestCase):
# Still one inconsistency, still no install attempt, no new registry fetch
self.assertEqual(len(result.inconsistencies_found), 1)
inc = result.inconsistencies_found[0]
self.assertFalse(inc.can_auto_fix)
self.assertEqual(inc.fix_action, FixAction.MANUAL_FIX_REQUIRED)
self.store_manager.install_plugin.assert_not_called()
self.store_manager.fetch_registry.assert_not_called()
@@ -458,7 +454,6 @@ class TestStateReconciliationUnrecoverable(unittest.TestCase):
self.assertEqual(len(result.inconsistencies_found), 1)
inc = result.inconsistencies_found[0]
self.assertFalse(inc.can_auto_fix)
self.assertEqual(inc.fix_action, FixAction.MANUAL_FIX_REQUIRED)
self.store_manager.install_plugin.assert_not_called()
test/web_interface/test_web_process_runs_no_plugin_code.py
+12 -12
View File
@@ -119,7 +119,7 @@ class Web:
self.config_manager.template_path = str(tmp_path / "no-template.json")
self.schema_manager = SchemaManager(plugins_dir=self.plugins_dir, project_root=tmp_path,
config_manager=self.config_manager)
self.catalog = PluginCatalog(self.plugins_dir, self.config_manager, self.schema_manager)
self.catalog = PluginCatalog(self.plugins_dir)
api = self.api = api_v3_module.api_v3
api.config_manager = self.config_manager
@@ -242,7 +242,8 @@ class TestTheWebProcessNeverRunsAPlugin:
_bump_version(web, "1.1.0")
body = web.post("/api/v3/plugins/update", {"plugin_id": PLUGIN_ID})
assert body["data"]["update_status"] == "updated"
assert web.catalog.get_installed_version(PLUGIN_ID) == "1.1.0"
# The route rescans the catalog, which now has the new manifest.
assert web.catalog.get_manifest(PLUGIN_ID)["version"] == "1.1.0"
assert web.ran() == []
def test_installing_it(self, web):
@@ -497,16 +498,15 @@ class TestCatalogReadsWhatIsInstalled:
assert expected, f"no plugins under {root}"
before = set(sys.modules)
schema_manager = SchemaManager(plugins_dir=root, project_root=PROJECT_ROOT)
catalog = PluginCatalog(root, schema_manager=schema_manager)
catalog = PluginCatalog(root)
assert set(catalog.discover_plugins()) == set(expected)
for plugin_id, (plugin_dir, manifest) in expected.items():
assert catalog.get_manifest(plugin_id) == manifest
assert catalog.get_plugin_directory(plugin_id) == str(plugin_dir)
assert catalog.get_installed_version(plugin_id) == manifest.get("version", "")
assert catalog.get_plugin_display_modes(plugin_id) == manifest.get("display_modes", [])
if (plugin_dir / "config_schema.json").exists():
schema = catalog.get_schema(plugin_id, use_cache=False)
schema = schema_manager.load_schema(plugin_id, use_cache=False)
assert isinstance(schema, dict) and "properties" in schema, plugin_id
imported = [name for name in set(sys.modules) - before
@@ -537,16 +537,16 @@ class TestCatalogReadsWhatIsInstalled:
assert catalog.discover_plugins() == []
assert catalog.get_manifest("ci-fixture-plugin") is None
def test_enabled_follows_the_display_rule(self, tmp_path):
def test_enabled_follows_the_display_rule(self, api_v3_module):
config = MagicMock()
config.load_config.return_value = {"a": {"enabled": True}, "b": {}, "c": "junk"}
catalog = PluginCatalog(tmp_path, config_manager=config)
assert catalog.is_enabled("a") is True
api_v3_module.api_v3.config_manager = config
enabled = api_v3_module._plugin_enabled_in_config
assert enabled("a") is True
# The display runs a plugin only when its section says so.
assert catalog.is_enabled("b") is False
assert catalog.is_enabled("c") is False
assert catalog.is_enabled("missing") is False
assert catalog.get_config("c") == {}
assert enabled("b") is False
assert enabled("c") is False
assert enabled("missing") is False
def test_it_has_nothing_that_runs_a_plugin(self, tmp_path):
catalog = PluginCatalog(tmp_path)