refactor: delete dead Python code in the core (and stop storing Wi-Fi passwords) (#608)

* refactor(plugins): remove the no-op PluginHealthMonitor

Its monitor loop did nothing (`if callbacks: pass`), register_health_check
had no callers and api_v3.health_monitor was never read by any route. The
live health data comes from PluginHealthTracker, which is untouched.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor(store): drop the never-set uninstall tombstones

Nothing in production called mark_recently_uninstalled, so the
reconciler's was_recently_uninstalled check was always False. The
persistent uninstall registry is what actually stops resurrection; the
reconciler test now exercises that gate instead.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor(common): delete unused config/display/game helpers, utils and error_handler

Nothing in core, the web UI, scripts or the plugin monorepo imports
config_helper, display_helper, game_helper, utils or error_handler; only
their own tests did. The error_handler re-exports leave src.common's
__all__; APIHelper, TextHelper, ScrollHelper, LogoHelper and the adaptive
layout exports are unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor(config): drop ConfigService's unused versioning and save API

ConfigVersion, get_version/get_version_history/get_version_config,
rollback, save_config, reload, get_plugin_config and the backward-compat
load_config/get_config_path/get_secrets_path had no callers. The display
controller only uses get_config, subscribe, unsubscribe and shutdown,
plus the file watcher. Change detection now compares against the
current checksum instead of the last history entry.

The subscriber tests asserted `callback.called or True`; they now
reload the way the watcher does and assert the notification.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor(plugins): drop unread plugin state history and callbacks

plugin_state.PluginStateManager kept a bounded per-plugin transition
history that only get_state_history (tests only) read; get_state_info
reports a separate lifetime count, which stays. set_error_info and
record_display had no callers, and set_state_with_error's `error`
argument only fed the history.

The web-side state_manager.PluginStateManager loses
subscribe_to_state_changes, _notify_callbacks, set_plugin_error and
get_state_version, none of which had callers; with no subscribers the
old-state copy in update_plugin_state went with them.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor(plugins): remove unused PluginManager methods and attribute guards

update_all_plugins was only called by a test (the display loop uses
run_scheduled_updates); get_plugin_health_metrics,
get_plugin_resource_metrics and get_plugin_state had no callers; and
plugin_modules was written but never read. plugin_directories is now
initialised in __init__, so the hasattr() guards around it go.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor(plugins): remove unused executor, loader, store and package helpers

- PluginExecutor.execute_safe: no callers.
- PluginLoader._parse_semver: only its own tests; compatibility.parse_semver
  is the live copy and test_compatibility.py already covers it.
- PluginStoreManager.get_installed_plugin_info: no callers.
- PluginResourceMonitor._local: never read.
- src.plugin_system.get_store_manager and __api_version__: no importers in
  core, scripts or the plugin monorepo.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(wifi): stop storing Wi-Fi passwords in wifi_config.json

WiFiManager appended every joined network's SSID and password, in
plaintext, to saved_networks in config/wifi_config.json, and nothing
(web UI, backup restore, scripts) ever read them back: NetworkManager
keeps its own credentials. The writes are gone, and loading the config
now drops any saved_networks key and rewrites the file, so passwords
already on disk are scrubbed.

Also removes _check_dnsmasq_conflict (never called) and _detect_trixie,
whose result only reached one log line, along with the
NM_CONNECTIONS_PATHS constant only it used.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor(display): remove unreachable and unused DisplayController code

- _follower_rebuild_scroll_image: never called.
- mode_duration (never read) and last_mode_change (write-only).
- The `chosen_cap <= 0` branch: chosen_cap is either the minimum of
  caps already filtered to > 0 or DEFAULT_DYNAMIC_DURATION_CAP (180).
- The `max_duration < min_duration` branch directly after
  `max_duration = max(min_duration, max_duration)`.
- The circuit-breaker branch's `display_result = False` and
  `manager_to_display = None`: the first is overwritten a few lines
  later, the second is already None there.
- The bool-to-bool conversion of execute_display's result, which is
  always a bool.
- The `loaded_plugins` lookup in _update_modules: PluginManager has no
  such attribute, so it always fell through to `plugins`.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor(vegas): remove unused config update, boundary finder and refresh

VegasModeConfig.update had no callers outside its own tests (the
coordinator rebuilds the config with from_config on a change);
geometry.find_item_boundary and StreamManager._refresh_plugin_content
had no callers at all.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor(run): drop the debug block that pretended to import the plugin system

In debug mode run.py put src/plugin_system itself on sys.path and printed
"Plugin system import successful" without importing anything. Nothing
imports plugin_system modules by bare name, so the path entry did
nothing either.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test: delete tests that test nothing

- test/plugins/test_{basketball_scoreboard,calendar,clock_simple,
  odds_ticker,soccer_scoreboard,text_display}.py skip everywhere the named
  plugins are not installed, including CI (LEDMATRIX_PLUGINS_DIR holds only
  the fixture plugin); test_plugin_matrix.py already covers every
  discovered plugin. Their PluginTestBase and the fixtures only it used
  (plugins_dir, mock_display_manager, mock_cache_manager,
  mock_plugin_manager, base_plugin_config in test/plugins/conftest.py) go
  with them.
- test_plugin_system.py: test_discover_plugins (body was `pass`) and
  test_dependency_check (a comment), plus the test_plugin_manager fixture
  only the former requested.
- test_display_manager.py: test_draw_image asserted that an image it had
  just assigned was not None.
- test_display_controller.py: the rotation and schedule-override tests
  re-implemented the run-loop arithmetic inline and asserted on their own
  result without calling the controller.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test: expect one plugin_last_update success stamp after update_all_plugins

EveryStampRecordsACompletion required at least two success-path stamps;
the second was update_all_plugins, removed as test-only. The worker and
synchronous paths share the remaining stamp in _execute_update_now, and
the check that every stamp calls _note_update_completed is unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Chuck
2026-09-23 12:36:26 -04:00
committed by GitHub
co-authored by Claude Opus 5.5
parent e1ce7189f1
commit 84afa9d64f
57 changed files with 273 additions and 6254 deletions
-30
View File
@@ -354,35 +354,6 @@ def test_config_with_plugins(test_config):
return config
@pytest.fixture
def test_plugin_manager(mock_config_manager, mock_display_manager, mock_cache_manager):
"""Create a test PluginManager instance."""
from unittest.mock import patch, MagicMock
import tempfile
from pathlib import Path
# Create temporary plugin directory
with tempfile.TemporaryDirectory() as tmpdir:
plugin_dir = Path(tmpdir) / "plugins"
plugin_dir.mkdir()
with patch('src.plugin_system.plugin_manager.PluginManager') as MockPM:
pm = MagicMock()
pm.plugins = {}
pm.plugin_manifests = {}
pm.loaded_plugins = {}
pm.plugin_last_update = {}
pm.discover_plugins = MagicMock(return_value=[])
pm.load_plugin = MagicMock(return_value=True)
pm.unload_plugin = MagicMock(return_value=True)
pm.get_plugin = MagicMock(return_value=None)
pm.plugin_executor = MagicMock()
pm.health_tracker = None
pm.resource_monitor = None
MockPM.return_value = pm
yield pm
@pytest.fixture
def test_display_controller(mock_config_manager, mock_display_manager, mock_cache_manager,
test_config_with_plugins, emulator_mode):
@@ -406,7 +377,6 @@ def test_display_controller(mock_config_manager, mock_display_manager, mock_cach
mock_pm.load_plugin = MagicMock(return_value=True)
mock_pm.get_plugin = MagicMock(return_value=None)
mock_pm.plugins = {}
mock_pm.loaded_plugins = {}
mock_pm.plugin_manifests = {}
mock_pm.plugin_last_update = {}
mock_pm.plugin_executor = MagicMock()
-95
View File
@@ -7,7 +7,6 @@ import os
import sys
import json
from pathlib import Path
from unittest.mock import MagicMock, Mock
from typing import Any, Dict
# Add project root to path
@@ -19,100 +18,6 @@ if str(project_root) not in sys.path:
os.environ['EMULATOR'] = 'true'
@pytest.fixture
def plugins_dir() -> Path:
"""Get the plugins directory path.
Honors LEDMATRIX_PLUGINS_DIR (first entry) when set — the same override
test_plugin_matrix.py uses, so CI can point every plugin suite at the
bundled fixture plugins. Otherwise checks plugins/ first, then falls
back to plugin-repos/ for monorepo development environments.
"""
env = os.environ.get('LEDMATRIX_PLUGINS_DIR')
if env:
first = env.split(os.pathsep)[0]
if first:
return Path(first)
plugins_path = project_root / 'plugins'
plugin_repos_path = project_root / 'plugin-repos'
# Prefer plugins/ if it has actual plugin directories
if plugins_path.exists():
try:
has_plugins = any(
p for p in plugins_path.iterdir()
if p.is_dir() and not p.name.startswith('.')
)
if has_plugins:
return plugins_path
except PermissionError:
pass
if plugin_repos_path.exists():
return plugin_repos_path
return plugins_path
@pytest.fixture
def mock_display_manager() -> Any:
"""Create a mock DisplayManager for plugin tests."""
mock = MagicMock()
mock.width = 128
mock.height = 32
mock.clear = Mock()
mock.draw_text = Mock()
mock.draw_image = Mock()
mock.update_display = Mock()
mock.get_font = Mock(return_value=None)
# Some plugins access matrix.width/height
mock.matrix = MagicMock()
mock.matrix.width = 128
mock.matrix.height = 32
return mock
@pytest.fixture
def mock_cache_manager() -> Any:
"""Create a mock CacheManager for plugin tests."""
mock = MagicMock()
mock._memory_cache = {}
def mock_get(key: str, max_age: int = 300) -> Any:
return mock._memory_cache.get(key)
def mock_set(key: str, data: Any, ttl: int = None) -> None:
mock._memory_cache[key] = data
def mock_clear(key: str = None) -> None:
if key:
mock._memory_cache.pop(key, None)
else:
mock._memory_cache.clear()
mock.get = Mock(side_effect=mock_get)
mock.set = Mock(side_effect=mock_set)
mock.clear = Mock(side_effect=mock_clear)
return mock
@pytest.fixture
def mock_plugin_manager() -> Any:
"""Create a mock PluginManager for plugin tests."""
mock = MagicMock()
mock.plugins = {}
mock.plugin_manifests = {}
return mock
@pytest.fixture
def base_plugin_config() -> Dict[str, Any]:
"""Base configuration for plugins."""
return {
'enabled': True,
'update_interval': 300
}
def load_plugin_manifest(plugin_id: str, plugins_dir: Path) -> Dict[str, Any]:
"""Load plugin manifest.json."""
manifest_path = plugins_dir / plugin_id / 'manifest.json'
@@ -1,95 +0,0 @@
"""
Integration tests for basketball-scoreboard plugin.
Requires the real plugin to be installed (plugins/ or plugin-repos/,
or the dir named by LEDMATRIX_PLUGINS_DIR) — on machines without it,
every test here skips by design. CI covers plugin safety with the
bundled fixture plugin via test_plugin_matrix.py instead.
"""
import pytest
from test.plugins.test_plugin_base import PluginTestBase
class TestBasketballScoreboardPlugin(PluginTestBase):
"""Test basketball-scoreboard plugin integration."""
@pytest.fixture
def plugin_id(self):
return 'basketball-scoreboard'
def test_manifest_exists(self, plugin_id):
"""Test that plugin manifest exists."""
super().test_manifest_exists(plugin_id)
def test_manifest_has_required_fields(self, plugin_id):
"""Test that manifest has all required fields."""
super().test_manifest_has_required_fields(plugin_id)
def test_plugin_can_be_loaded(self, plugin_id):
"""Test that plugin module can be loaded."""
super().test_plugin_can_be_loaded(plugin_id)
def test_plugin_class_exists(self, plugin_id):
"""Test that plugin class exists."""
super().test_plugin_class_exists(plugin_id)
def test_plugin_can_be_instantiated(self, plugin_id):
"""Test that plugin can be instantiated."""
super().test_plugin_can_be_instantiated(plugin_id)
def test_plugin_has_required_methods(self, plugin_id):
"""Test that plugin has required methods."""
super().test_plugin_has_required_methods(plugin_id)
def test_plugin_update_method(self, plugin_id):
"""Test that plugin update() method works."""
super().test_plugin_update_method(plugin_id)
def test_plugin_display_method(self, plugin_id):
"""Test that plugin display() method works."""
super().test_plugin_display_method(plugin_id)
def test_plugin_has_display_modes(self, plugin_id):
"""Test that plugin has display modes."""
manifest = self.load_plugin_manifest(plugin_id)
assert 'display_modes' in manifest
# Manifest uses league-prefixed modes (nba_, wnba_, ncaam_, ncaaw_)
assert 'nba_live' in manifest['display_modes']
assert 'nba_recent' in manifest['display_modes']
assert 'nba_upcoming' in manifest['display_modes']
def test_plugin_has_get_display_modes(self, plugin_id):
"""Test that plugin can return display modes."""
manifest = self.load_plugin_manifest(plugin_id)
plugin_dir = self.plugins_dir / plugin_id
entry_point = manifest['entry_point']
class_name = manifest['class_name']
module = self.plugin_loader.load_module(
plugin_id=plugin_id,
plugin_dir=plugin_dir,
entry_point=entry_point
)
plugin_class = self.plugin_loader.get_plugin_class(
plugin_id=plugin_id,
module=module,
class_name=class_name
)
config = self.base_config.copy()
plugin_instance = self.plugin_loader.instantiate_plugin(
plugin_id=plugin_id,
plugin_class=plugin_class,
config=config,
display_manager=self.mock_display_manager,
cache_manager=self.mock_cache_manager,
plugin_manager=self.mock_plugin_manager
)
# Check if plugin has get_display_modes method
if hasattr(plugin_instance, 'get_display_modes'):
modes = plugin_instance.get_display_modes()
assert isinstance(modes, list)
assert len(modes) > 0
-63
View File
@@ -1,63 +0,0 @@
"""
Integration tests for calendar plugin.
Requires the real plugin to be installed (plugins/ or plugin-repos/,
or the dir named by LEDMATRIX_PLUGINS_DIR) — on machines without it,
every test here skips by design. CI covers plugin safety with the
bundled fixture plugin via test_plugin_matrix.py instead.
"""
import pytest
from test.plugins.test_plugin_base import PluginTestBase
class TestCalendarPlugin(PluginTestBase):
"""Test calendar plugin integration."""
@pytest.fixture
def plugin_id(self):
return 'calendar'
def test_manifest_exists(self, plugin_id):
"""Test that plugin manifest exists."""
super().test_manifest_exists(plugin_id)
def test_manifest_has_required_fields(self, plugin_id):
"""Test that manifest has all required fields."""
super().test_manifest_has_required_fields(plugin_id)
def test_plugin_can_be_loaded(self, plugin_id):
"""Test that plugin module can be loaded."""
super().test_plugin_can_be_loaded(plugin_id)
def test_plugin_class_exists(self, plugin_id):
"""Test that plugin class exists."""
super().test_plugin_class_exists(plugin_id)
def test_plugin_can_be_instantiated(self, plugin_id):
"""Test that plugin can be instantiated."""
# Calendar plugin may need credentials, but instantiation should work
super().test_plugin_can_be_instantiated(plugin_id)
def test_plugin_has_required_methods(self, plugin_id):
"""Test that plugin has required methods."""
super().test_plugin_has_required_methods(plugin_id)
def test_plugin_update_method(self, plugin_id):
"""Test that plugin update() method works."""
# Calendar requires Google API credentials, so this may skip
super().test_plugin_update_method(plugin_id)
def test_plugin_display_method(self, plugin_id):
"""Test that plugin display() method works."""
super().test_plugin_display_method(plugin_id)
def test_plugin_has_display_modes(self, plugin_id):
"""Test that plugin has display modes."""
manifest = self.load_plugin_manifest(plugin_id)
assert 'display_modes' in manifest
assert 'calendar' in manifest['display_modes']
def test_config_schema_valid(self, plugin_id):
"""Test that config schema is valid."""
super().test_config_schema_valid(plugin_id)
-103
View File
@@ -1,103 +0,0 @@
"""
Integration tests for clock-simple plugin.
Requires the real plugin to be installed (plugins/ or plugin-repos/,
or the dir named by LEDMATRIX_PLUGINS_DIR) — on machines without it,
every test here skips by design. CI covers plugin safety with the
bundled fixture plugin via test_plugin_matrix.py instead.
"""
import pytest
from test.plugins.test_plugin_base import PluginTestBase
class TestClockSimplePlugin(PluginTestBase):
"""Test clock-simple plugin integration."""
@pytest.fixture
def plugin_id(self):
return 'clock-simple'
def test_manifest_exists(self, plugin_id):
"""Test that plugin manifest exists."""
super().test_manifest_exists(plugin_id)
def test_manifest_has_required_fields(self, plugin_id):
"""Test that manifest has all required fields."""
super().test_manifest_has_required_fields(plugin_id)
def test_plugin_can_be_loaded(self, plugin_id):
"""Test that plugin module can be loaded."""
super().test_plugin_can_be_loaded(plugin_id)
def test_plugin_class_exists(self, plugin_id):
"""Test that plugin class exists."""
super().test_plugin_class_exists(plugin_id)
def test_plugin_can_be_instantiated(self, plugin_id):
"""Test that plugin can be instantiated."""
super().test_plugin_can_be_instantiated(plugin_id)
def test_plugin_has_required_methods(self, plugin_id):
"""Test that plugin has required methods."""
super().test_plugin_has_required_methods(plugin_id)
def test_plugin_update_method(self, plugin_id):
"""Test that plugin update() method works."""
# Clock doesn't need external APIs, so this should always work
super().test_plugin_update_method(plugin_id)
def test_plugin_display_method(self, plugin_id):
"""Test that plugin display() method works."""
super().test_plugin_display_method(plugin_id)
def test_plugin_has_display_modes(self, plugin_id):
"""Test that plugin has display modes."""
manifest = self.load_plugin_manifest(plugin_id)
assert 'display_modes' in manifest
assert 'clock-simple' in manifest['display_modes']
def test_clock_displays_time(self, plugin_id):
"""Test that clock plugin actually displays time."""
manifest = self.load_plugin_manifest(plugin_id)
plugin_dir = self.plugins_dir / plugin_id
entry_point = manifest['entry_point']
class_name = manifest['class_name']
module = self.plugin_loader.load_module(
plugin_id=plugin_id,
plugin_dir=plugin_dir,
entry_point=entry_point
)
plugin_class = self.plugin_loader.get_plugin_class(
plugin_id=plugin_id,
module=module,
class_name=class_name
)
config = self.base_config.copy()
config['timezone'] = 'UTC'
config['time_format'] = '12h'
config['show_date'] = True
plugin_instance = self.plugin_loader.instantiate_plugin(
plugin_id=plugin_id,
plugin_class=plugin_class,
config=config,
display_manager=self.mock_display_manager,
cache_manager=self.mock_cache_manager,
plugin_manager=self.mock_plugin_manager
)
# Update and display
plugin_instance.update()
plugin_instance.display(force_clear=True)
# Verify time was formatted
assert hasattr(plugin_instance, 'current_time')
assert plugin_instance.current_time is not None
# Verify display was called
assert self.mock_display_manager.clear.called
assert self.mock_display_manager.update_display.called
-62
View File
@@ -1,62 +0,0 @@
"""
Integration tests for odds-ticker plugin.
Requires the real plugin to be installed (plugins/ or plugin-repos/,
or the dir named by LEDMATRIX_PLUGINS_DIR) — on machines without it,
every test here skips by design. CI covers plugin safety with the
bundled fixture plugin via test_plugin_matrix.py instead.
"""
import pytest
from test.plugins.test_plugin_base import PluginTestBase
class TestOddsTickerPlugin(PluginTestBase):
"""Test odds-ticker plugin integration."""
@pytest.fixture
def plugin_id(self):
return 'odds-ticker'
def test_manifest_exists(self, plugin_id):
"""Test that plugin manifest exists."""
super().test_manifest_exists(plugin_id)
def test_manifest_has_required_fields(self, plugin_id):
"""Test that manifest has all required fields."""
super().test_manifest_has_required_fields(plugin_id)
def test_plugin_can_be_loaded(self, plugin_id):
"""Test that plugin module can be loaded."""
super().test_plugin_can_be_loaded(plugin_id)
def test_plugin_class_exists(self, plugin_id):
"""Test that plugin class exists."""
super().test_plugin_class_exists(plugin_id)
def test_plugin_can_be_instantiated(self, plugin_id):
"""Test that plugin can be instantiated."""
super().test_plugin_can_be_instantiated(plugin_id)
def test_plugin_has_required_methods(self, plugin_id):
"""Test that plugin has required methods."""
super().test_plugin_has_required_methods(plugin_id)
def test_plugin_update_method(self, plugin_id):
"""Test that plugin update() method works."""
# Odds ticker may need API access, but should handle gracefully
super().test_plugin_update_method(plugin_id)
def test_plugin_display_method(self, plugin_id):
"""Test that plugin display() method works."""
super().test_plugin_display_method(plugin_id)
def test_plugin_has_display_modes(self, plugin_id):
"""Test that plugin has display modes."""
manifest = self.load_plugin_manifest(plugin_id)
assert 'display_modes' in manifest
assert 'odds_ticker' in manifest['display_modes']
def test_config_schema_valid(self, plugin_id):
"""Test that config schema is valid."""
super().test_config_schema_valid(plugin_id)
-305
View File
@@ -1,305 +0,0 @@
"""
Base test class for plugin integration tests.
Provides common test functionality for all plugins.
"""
import pytest
import json
from typing import Dict, Any
from src.plugin_system.plugin_loader import PluginLoader
from src.plugin_system.base_plugin import BasePlugin
class PluginTestBase:
"""Base class for plugin integration tests."""
@pytest.fixture(autouse=True)
def setup_base(self, plugins_dir, mock_display_manager, mock_cache_manager,
mock_plugin_manager, base_plugin_config):
"""Setup base fixtures for all plugin tests."""
self.plugins_dir = plugins_dir
self.mock_display_manager = mock_display_manager
self.mock_cache_manager = mock_cache_manager
self.mock_plugin_manager = mock_plugin_manager
self.base_config = base_plugin_config
self.plugin_loader = PluginLoader()
def load_plugin_manifest(self, plugin_id: str) -> Dict[str, Any]:
"""Load plugin manifest.json."""
manifest_path = self.plugins_dir / plugin_id / 'manifest.json'
if not manifest_path.exists():
pytest.skip(f"Manifest not found for {plugin_id}")
with open(manifest_path, 'r') as f:
return json.load(f)
def load_plugin_config_schema(self, plugin_id: str) -> Dict[str, Any]:
"""Load plugin config_schema.json if it exists."""
schema_path = self.plugins_dir / plugin_id / 'config_schema.json'
if schema_path.exists():
with open(schema_path, 'r') as f:
return json.load(f)
return None
def test_manifest_exists(self, plugin_id: str):
"""Test that plugin manifest exists and is valid JSON."""
manifest = self.load_plugin_manifest(plugin_id)
assert manifest is not None
assert 'id' in manifest
assert manifest['id'] == plugin_id
assert 'class_name' in manifest
# entry_point is optional - default to 'manager.py' if missing
if 'entry_point' not in manifest:
manifest['entry_point'] = 'manager.py'
def test_manifest_has_required_fields(self, plugin_id: str):
"""Test that manifest has all required fields."""
manifest = self.load_plugin_manifest(plugin_id)
# Core required fields
required_fields = ['id', 'name', 'description', 'author', 'class_name']
for field in required_fields:
assert field in manifest, f"Manifest missing required field: {field}"
assert manifest[field], f"Manifest field {field} is empty"
# entry_point is required but some plugins may not have it explicitly
# If missing, assume it's 'manager.py'
if 'entry_point' not in manifest:
manifest['entry_point'] = 'manager.py'
def test_plugin_can_be_loaded(self, plugin_id: str):
"""Test that plugin module can be loaded."""
manifest = self.load_plugin_manifest(plugin_id)
plugin_dir = self.plugins_dir / plugin_id
entry_point = manifest.get('entry_point', 'manager.py')
module = self.plugin_loader.load_module(
plugin_id=plugin_id,
plugin_dir=plugin_dir,
entry_point=entry_point
)
assert module is not None
assert hasattr(module, manifest['class_name'])
def test_plugin_class_exists(self, plugin_id: str):
"""Test that plugin class exists in module."""
manifest = self.load_plugin_manifest(plugin_id)
plugin_dir = self.plugins_dir / plugin_id
entry_point = manifest.get('entry_point', 'manager.py')
class_name = manifest['class_name']
module = self.plugin_loader.load_module(
plugin_id=plugin_id,
plugin_dir=plugin_dir,
entry_point=entry_point
)
plugin_class = self.plugin_loader.get_plugin_class(
plugin_id=plugin_id,
module=module,
class_name=class_name
)
assert plugin_class is not None
assert issubclass(plugin_class, BasePlugin)
def test_plugin_can_be_instantiated(self, plugin_id: str):
"""Test that plugin can be instantiated with mock dependencies."""
manifest = self.load_plugin_manifest(plugin_id)
plugin_dir = self.plugins_dir / plugin_id
entry_point = manifest.get('entry_point', 'manager.py')
class_name = manifest['class_name']
module = self.plugin_loader.load_module(
plugin_id=plugin_id,
plugin_dir=plugin_dir,
entry_point=entry_point
)
plugin_class = self.plugin_loader.get_plugin_class(
plugin_id=plugin_id,
module=module,
class_name=class_name
)
# Merge base config with plugin-specific defaults
config = self.base_config.copy()
plugin_instance = self.plugin_loader.instantiate_plugin(
plugin_id=plugin_id,
plugin_class=plugin_class,
config=config,
display_manager=self.mock_display_manager,
cache_manager=self.mock_cache_manager,
plugin_manager=self.mock_plugin_manager
)
assert plugin_instance is not None
assert plugin_instance.plugin_id == plugin_id
assert plugin_instance.enabled == config.get('enabled', True)
def test_plugin_has_required_methods(self, plugin_id: str):
"""Test that plugin has required BasePlugin methods."""
manifest = self.load_plugin_manifest(plugin_id)
plugin_dir = self.plugins_dir / plugin_id
entry_point = manifest.get('entry_point', 'manager.py')
class_name = manifest['class_name']
module = self.plugin_loader.load_module(
plugin_id=plugin_id,
plugin_dir=plugin_dir,
entry_point=entry_point
)
plugin_class = self.plugin_loader.get_plugin_class(
plugin_id=plugin_id,
module=module,
class_name=class_name
)
config = self.base_config.copy()
plugin_instance = self.plugin_loader.instantiate_plugin(
plugin_id=plugin_id,
plugin_class=plugin_class,
config=config,
display_manager=self.mock_display_manager,
cache_manager=self.mock_cache_manager,
plugin_manager=self.mock_plugin_manager
)
# Check required methods exist
assert hasattr(plugin_instance, 'update')
assert hasattr(plugin_instance, 'display')
assert callable(plugin_instance.update)
assert callable(plugin_instance.display)
def test_plugin_update_method(self, plugin_id: str):
"""Test that plugin update() method can be called without errors."""
manifest = self.load_plugin_manifest(plugin_id)
plugin_dir = self.plugins_dir / plugin_id
entry_point = manifest.get('entry_point', 'manager.py')
class_name = manifest['class_name']
module = self.plugin_loader.load_module(
plugin_id=plugin_id,
plugin_dir=plugin_dir,
entry_point=entry_point
)
plugin_class = self.plugin_loader.get_plugin_class(
plugin_id=plugin_id,
module=module,
class_name=class_name
)
config = self.base_config.copy()
plugin_instance = self.plugin_loader.instantiate_plugin(
plugin_id=plugin_id,
plugin_class=plugin_class,
config=config,
display_manager=self.mock_display_manager,
cache_manager=self.mock_cache_manager,
plugin_manager=self.mock_plugin_manager
)
# Call update() - should not raise exceptions
# Some plugins may need API keys, but they should handle that gracefully
try:
plugin_instance.update()
except Exception as e:
# If it's a missing API key or similar, that's acceptable for integration tests
error_msg = str(e).lower()
if 'api' in error_msg or 'key' in error_msg or 'auth' in error_msg or 'credential' in error_msg:
pytest.skip(f"Plugin requires API credentials: {e}")
else:
raise
def test_plugin_display_method(self, plugin_id: str):
"""Test that plugin display() method can be called without errors."""
manifest = self.load_plugin_manifest(plugin_id)
plugin_dir = self.plugins_dir / plugin_id
entry_point = manifest.get('entry_point', 'manager.py')
class_name = manifest['class_name']
module = self.plugin_loader.load_module(
plugin_id=plugin_id,
plugin_dir=plugin_dir,
entry_point=entry_point
)
plugin_class = self.plugin_loader.get_plugin_class(
plugin_id=plugin_id,
module=module,
class_name=class_name
)
config = self.base_config.copy()
plugin_instance = self.plugin_loader.instantiate_plugin(
plugin_id=plugin_id,
plugin_class=plugin_class,
config=config,
display_manager=self.mock_display_manager,
cache_manager=self.mock_cache_manager,
plugin_manager=self.mock_plugin_manager
)
# Some plugins need matrix attribute on display_manager (set before update)
if not hasattr(self.mock_display_manager, 'matrix'):
from unittest.mock import MagicMock
self.mock_display_manager.matrix = MagicMock()
self.mock_display_manager.matrix.width = 128
self.mock_display_manager.matrix.height = 32
# Call update() first if needed
try:
plugin_instance.update()
except Exception as e:
error_msg = str(e).lower()
if 'api' in error_msg or 'key' in error_msg or 'auth' in error_msg:
pytest.skip(f"Plugin requires API credentials: {e}")
# Some plugins need a mode set before display
# Try to set a mode if the plugin has that capability
if hasattr(plugin_instance, 'set_mode') and manifest.get('display_modes'):
try:
first_mode = manifest['display_modes'][0]
plugin_instance.set_mode(first_mode)
except Exception:
pass # If set_mode doesn't exist or fails, continue
# Call display() - should not raise exceptions
try:
plugin_instance.display(force_clear=True)
except Exception as e:
# Some plugins may need specific setup - if it's a mode issue, that's acceptable
error_msg = str(e).lower()
if 'mode' in error_msg or 'manager' in error_msg:
# This is acceptable - plugin needs proper mode setup
pass
else:
raise
# Verify display_manager methods were called (if display succeeded)
# Some plugins may not call these if they skip display due to missing data
# So we just verify the method was callable without exceptions
assert hasattr(plugin_instance, 'display')
def test_plugin_has_display_modes(self, plugin_id: str):
"""Test that plugin has display modes defined."""
manifest = self.load_plugin_manifest(plugin_id)
assert 'display_modes' in manifest
assert isinstance(manifest['display_modes'], list)
assert len(manifest['display_modes']) > 0
def test_config_schema_valid(self, plugin_id: str):
"""Test that config schema is valid JSON if it exists."""
schema = self.load_plugin_config_schema(plugin_id)
if schema is not None:
assert isinstance(schema, dict)
# Schema should have 'type' field for JSON Schema
assert 'type' in schema or 'properties' in schema
-94
View File
@@ -1,94 +0,0 @@
"""
Integration tests for soccer-scoreboard plugin.
Requires the real plugin to be installed (plugins/ or plugin-repos/,
or the dir named by LEDMATRIX_PLUGINS_DIR) — on machines without it,
every test here skips by design. CI covers plugin safety with the
bundled fixture plugin via test_plugin_matrix.py instead.
"""
import pytest
from test.plugins.test_plugin_base import PluginTestBase
class TestSoccerScoreboardPlugin(PluginTestBase):
"""Test soccer-scoreboard plugin integration."""
@pytest.fixture
def plugin_id(self):
return 'soccer-scoreboard'
def test_manifest_exists(self, plugin_id):
"""Test that plugin manifest exists."""
super().test_manifest_exists(plugin_id)
def test_manifest_has_required_fields(self, plugin_id):
"""Test that manifest has all required fields."""
super().test_manifest_has_required_fields(plugin_id)
def test_plugin_can_be_loaded(self, plugin_id):
"""Test that plugin module can be loaded."""
super().test_plugin_can_be_loaded(plugin_id)
def test_plugin_class_exists(self, plugin_id):
"""Test that plugin class exists."""
super().test_plugin_class_exists(plugin_id)
def test_plugin_can_be_instantiated(self, plugin_id):
"""Test that plugin can be instantiated."""
super().test_plugin_can_be_instantiated(plugin_id)
def test_plugin_has_required_methods(self, plugin_id):
"""Test that plugin has required methods."""
super().test_plugin_has_required_methods(plugin_id)
def test_plugin_update_method(self, plugin_id):
"""Test that plugin update() method works."""
super().test_plugin_update_method(plugin_id)
def test_plugin_display_method(self, plugin_id):
"""Test that plugin display() method works."""
super().test_plugin_display_method(plugin_id)
def test_plugin_has_display_modes(self, plugin_id):
"""Test that plugin has display modes."""
manifest = self.load_plugin_manifest(plugin_id)
assert 'display_modes' in manifest
assert 'soccer_live' in manifest['display_modes']
assert 'soccer_recent' in manifest['display_modes']
assert 'soccer_upcoming' in manifest['display_modes']
def test_plugin_has_get_display_modes(self, plugin_id):
"""Test that plugin can return display modes."""
manifest = self.load_plugin_manifest(plugin_id)
plugin_dir = self.plugins_dir / plugin_id
entry_point = manifest['entry_point']
class_name = manifest['class_name']
module = self.plugin_loader.load_module(
plugin_id=plugin_id,
plugin_dir=plugin_dir,
entry_point=entry_point
)
plugin_class = self.plugin_loader.get_plugin_class(
plugin_id=plugin_id,
module=module,
class_name=class_name
)
config = self.base_config.copy()
plugin_instance = self.plugin_loader.instantiate_plugin(
plugin_id=plugin_id,
plugin_class=plugin_class,
config=config,
display_manager=self.mock_display_manager,
cache_manager=self.mock_cache_manager,
plugin_manager=self.mock_plugin_manager
)
# Check if plugin has get_display_modes method
if hasattr(plugin_instance, 'get_display_modes'):
modes = plugin_instance.get_display_modes()
assert isinstance(modes, list)
assert len(modes) > 0
-114
View File
@@ -1,114 +0,0 @@
"""
Integration tests for text-display plugin.
Requires the real plugin to be installed (plugins/ or plugin-repos/,
or the dir named by LEDMATRIX_PLUGINS_DIR) — on machines without it,
every test here skips by design. CI covers plugin safety with the
bundled fixture plugin via test_plugin_matrix.py instead.
"""
import pytest
from unittest.mock import MagicMock
from test.plugins.test_plugin_base import PluginTestBase
class TestTextDisplayPlugin(PluginTestBase):
"""Test text-display plugin integration."""
@pytest.fixture
def plugin_id(self):
return 'text-display'
def test_manifest_exists(self, plugin_id):
"""Test that plugin manifest exists."""
super().test_manifest_exists(plugin_id)
def test_manifest_has_required_fields(self, plugin_id):
"""Test that manifest has all required fields."""
super().test_manifest_has_required_fields(plugin_id)
def test_plugin_can_be_loaded(self, plugin_id):
"""Test that plugin module can be loaded."""
super().test_plugin_can_be_loaded(plugin_id)
def test_plugin_class_exists(self, plugin_id):
"""Test that plugin class exists."""
super().test_plugin_class_exists(plugin_id)
def test_plugin_can_be_instantiated(self, plugin_id):
"""Test that plugin can be instantiated."""
super().test_plugin_can_be_instantiated(plugin_id)
def test_plugin_has_required_methods(self, plugin_id):
"""Test that plugin has required methods."""
super().test_plugin_has_required_methods(plugin_id)
def test_plugin_update_method(self, plugin_id):
"""Test that plugin update() method works."""
# Text display doesn't need external APIs
super().test_plugin_update_method(plugin_id)
def test_plugin_display_method(self, plugin_id):
"""Test that plugin display() method works."""
super().test_plugin_display_method(plugin_id)
def test_plugin_has_display_modes(self, plugin_id):
"""Test that plugin has display modes."""
manifest = self.load_plugin_manifest(plugin_id)
assert 'display_modes' in manifest
assert 'text_display' in manifest['display_modes']
def test_text_display_shows_text(self, plugin_id):
"""Test that text display plugin actually displays text."""
manifest = self.load_plugin_manifest(plugin_id)
plugin_dir = self.plugins_dir / plugin_id
entry_point = manifest.get('entry_point', 'manager.py')
class_name = manifest['class_name']
module = self.plugin_loader.load_module(
plugin_id=plugin_id,
plugin_dir=plugin_dir,
entry_point=entry_point
)
plugin_class = self.plugin_loader.get_plugin_class(
plugin_id=plugin_id,
module=module,
class_name=class_name
)
config = self.base_config.copy()
config['text'] = 'Test Message'
config['scroll'] = False
config['text_color'] = [255, 255, 255]
config['background_color'] = [0, 0, 0]
# Mock display_manager.matrix to have width/height attributes
if not hasattr(self.mock_display_manager, 'matrix'):
self.mock_display_manager.matrix = MagicMock()
self.mock_display_manager.matrix.width = 128
self.mock_display_manager.matrix.height = 32
plugin_instance = self.plugin_loader.instantiate_plugin(
plugin_id=plugin_id,
plugin_class=plugin_class,
config=config,
display_manager=self.mock_display_manager,
cache_manager=self.mock_cache_manager,
plugin_manager=self.mock_plugin_manager
)
# Update and display
plugin_instance.update()
plugin_instance.display(force_clear=True)
# Verify text was set
assert plugin_instance.text == 'Test Message'
# Verify display was called (may be called via image assignment)
assert (self.mock_display_manager.update_display.called or
hasattr(self.mock_display_manager, 'image'))
def test_config_schema_valid(self, plugin_id):
"""Test that config schema is valid."""
super().test_config_schema_valid(plugin_id)
-253
View File
@@ -1,253 +0,0 @@
"""
Tests for src/common/config_helper.py — pins the ConfigHelper contract.
Covers: load/save round trips (missing/malformed files return {} rather
than raising, non-ASCII preserved via ensure_ascii=False, top-level JSON
lists returned as-is), dot-notation get/set including the silent-failure
contract when an intermediate key holds a non-dict, merge_configs deep
semantics with NO aliasing of the base config (the fixed bug — the old
shallow copy let mutations of the merged result leak into base's nested
dicts), simplified schema validation including the caught-TypeError path
when a schema 'type' is given as a string, plugin config key conventions
('{plugin_id}_config', enabled defaults True), and required-key checks
where a key present with value None counts as present.
"""
import json
import pytest
from src.common.config_helper import ConfigHelper
@pytest.fixture
def helper():
return ConfigHelper()
class TestLoadConfig:
def test_missing_file_returns_empty_dict(self, helper, tmp_path):
assert helper.load_config(tmp_path / "nope.json") == {}
def test_malformed_json_returns_empty_dict(self, helper, tmp_path):
path = tmp_path / "bad.json"
path.write_text("{ this is not json", encoding="utf-8")
assert helper.load_config(path) == {}
def test_top_level_list_returned_as_is(self, helper, tmp_path):
# load_config does not enforce a dict shape: a JSON list comes
# straight back. Pinned as a characterization of current behavior.
path = tmp_path / "list.json"
path.write_text("[1, 2, 3]", encoding="utf-8")
assert helper.load_config(path) == [1, 2, 3]
class TestSaveConfig:
def test_round_trip(self, helper, tmp_path):
path = tmp_path / "config.json"
config = {'display': {'hardware': {'rows': 32}}, 'timezone': 'UTC'}
assert helper.save_config(config, path) is True
assert helper.load_config(path) == config
def test_creates_parent_directories(self, helper, tmp_path):
path = tmp_path / "deep" / "nested" / "config.json"
assert helper.save_config({'a': 1}, path) is True
assert path.exists()
assert helper.load_config(path) == {'a': 1}
def test_non_ascii_survives_round_trip(self, helper, tmp_path):
path = tmp_path / "config.json"
config = {'city': 'Zürich', 'note': 'météo ☀'}
assert helper.save_config(config, path) is True
assert helper.load_config(path) == config
# ensure_ascii=False: characters are written raw, not \u-escaped
assert 'Zürich' in path.read_text(encoding='utf-8')
def test_directory_path_returns_false_not_raise(self, helper, tmp_path):
assert helper.save_config({'a': 1}, tmp_path) is False
class TestGetConfigValue:
def test_dot_notation_hit(self, helper):
config = {'display': {'hardware': {'rows': 32}}}
assert helper.get_config_value(config, 'display.hardware.rows') == 32
def test_missing_returns_default(self, helper):
sentinel = object()
assert helper.get_config_value({}, 'display.rows', default=sentinel) is sentinel
def test_intermediate_non_dict_returns_default(self, helper):
config = {'display': 'not-a-dict'}
assert helper.get_config_value(config, 'display.hardware.rows', default=64) == 64
def test_required_missing_raises_keyerror(self, helper):
with pytest.raises(KeyError):
helper.get_config_value({}, 'display.rows', required=True)
class TestSetConfigValue:
def test_sets_top_level(self, helper):
config = {}
helper.set_config_value(config, 'timezone', 'UTC')
assert config == {'timezone': 'UTC'}
def test_auto_creates_intermediates(self, helper):
config = {}
helper.set_config_value(config, 'display.hardware.rows', 32)
assert config == {'display': {'hardware': {'rows': 32}}}
def test_silent_failure_on_non_dict_intermediate(self, helper):
# 'a' exists but holds an int; the assignment attempt raises
# TypeError internally, which set_config_value swallows and logs.
# The config is left unchanged — pinned silent-failure contract.
config = {'a': 5}
helper.set_config_value(config, 'a.b', 1)
assert config == {'a': 5}
class TestMergeConfigs:
def test_nested_dicts_merge_recursively(self, helper):
base = {'display': {'rows': 32, 'cols': 64}, 'timezone': 'UTC'}
override = {'display': {'cols': 128, 'brightness': 90}}
merged = helper.merge_configs(base, override)
assert merged == {
'display': {'rows': 32, 'cols': 128, 'brightness': 90},
'timezone': 'UTC',
}
def test_scalar_override_wins_over_dict(self, helper):
merged = helper.merge_configs({'display': {'rows': 32}}, {'display': 7})
assert merged['display'] == 7
def test_dict_override_wins_over_scalar(self, helper):
merged = helper.merge_configs({'display': 7}, {'display': {'rows': 32}})
assert merged['display'] == {'rows': 32}
def test_no_aliasing_of_base(self, helper):
# Post-fix: merge deep-copies base, so mutating the result never
# leaks back into the caller's base config.
base = {'display': {'x': 1}}
merged = helper.merge_configs(base, {})
assert merged['display'] is not base['display']
merged['display']['x'] = 99
assert base['display']['x'] == 1
def test_inputs_unchanged(self, helper):
base = {'a': {'b': 1}}
override = {'a': {'c': 2}}
helper.merge_configs(base, override)
assert base == {'a': {'b': 1}}
assert override == {'a': {'c': 2}}
def test_no_aliasing_of_override_values(self, helper):
# The non-recursive branch must deep-copy the override value too:
# mutating a merged-in list or dict must not reach back into
# override_config.
override = {'teams': ['A', 'B'], 'nested': {'x': [1]}}
merged = helper.merge_configs({}, override)
merged['teams'].append('C')
merged['nested']['x'].append(2)
assert override == {'teams': ['A', 'B'], 'nested': {'x': [1]}}
class TestValidateConfig:
def test_no_schema_dict_is_valid(self, helper):
assert helper.validate_config({'a': 1}) is True
def test_no_schema_list_is_invalid(self, helper):
assert helper.validate_config([1, 2]) is False
def test_required_key_missing_is_invalid(self, helper):
schema = {'rows': {'required': True, 'type': int}}
assert helper.validate_config({}, schema) is False
def test_optional_key_missing_is_valid(self, helper):
schema = {'rows': {'required': False, 'type': int}}
assert helper.validate_config({}, schema) is True
def test_wrong_type_is_invalid(self, helper):
schema = {'rows': {'type': int}}
assert helper.validate_config({'rows': 'thirty-two'}, schema) is False
assert helper.validate_config({'rows': 32}, schema) is True
def test_allowed_values_violation_is_invalid(self, helper):
schema = {'mode': {'allowed_values': ['clock', 'weather']}}
assert helper.validate_config({'mode': 'stocks'}, schema) is False
assert helper.validate_config({'mode': 'clock'}, schema) is True
def test_string_type_in_schema_is_invalid_via_typeerror(self, helper):
# 'type' given as the STRING "int" makes isinstance() raise
# TypeError; validate_config catches it and returns False rather
# than raising. Pinned characterization.
schema = {'rows': {'type': 'int'}}
assert helper.validate_config({'rows': 32}, schema) is False
class TestPluginConfigHelpers:
def test_get_plugin_config_uses_suffixed_key(self, helper):
plugin_cfg = {'enabled': True, 'display_duration': 30}
assert helper.get_plugin_config({'clock_config': plugin_cfg}, 'clock') == plugin_cfg
def test_get_plugin_config_bare_id_key_not_found(self, helper):
# Only '{plugin_id}_config' is consulted — a bare 'clock' section
# is invisible to this helper. Pinned key contract.
assert helper.get_plugin_config({'clock': {'enabled': True}}, 'clock') == {}
def test_create_default_config_wraps_in_suffixed_key(self, helper):
defaults = {'enabled': True}
assert helper.create_default_config('clock', defaults) == {'clock_config': defaults}
def test_is_plugin_enabled_defaults_true_for_unknown(self, helper):
assert helper.is_plugin_enabled({}, 'clock') is True
def test_is_plugin_enabled_false_when_disabled(self, helper):
config = {'clock_config': {'enabled': False}}
assert helper.is_plugin_enabled(config, 'clock') is False
def test_is_plugin_enabled_ignores_bare_id_key(self, helper):
# Disabled under the wrong key -> still reported enabled (default).
config = {'clock': {'enabled': False}}
assert helper.is_plugin_enabled(config, 'clock') is True
class TestSportsAndDisplayHelpers:
def test_get_display_config(self, helper):
display = {'hardware': {'rows': 32}}
assert helper.get_display_config({'display': display}) == display
assert helper.get_display_config({}) == {}
def test_get_sports_config_uses_scoreboard_suffix(self, helper):
sport_cfg = {'favorite_teams': ['TB']}
config = {'football_scoreboard': sport_cfg}
assert helper.get_sports_config(config, 'football') == sport_cfg
assert helper.get_sports_config(config, 'hockey') == {}
def test_get_favorite_teams(self, helper):
config = {'football_scoreboard': {'favorite_teams': ['TB', 'DAL']}}
assert helper.get_favorite_teams(config, 'football') == ['TB', 'DAL']
assert helper.get_favorite_teams({}, 'football') == []
def test_get_display_modes(self, helper):
modes = {'live': True, 'recent': False}
config = {'football_scoreboard': {'display_modes': modes}}
assert helper.get_display_modes(config, 'football') == modes
assert helper.get_display_modes({}, 'football') == {}
class TestValidateRequiredKeys:
def test_returns_missing_subset(self, helper):
config = {'a': 1, 'c': {'d': 2}}
missing = helper.validate_required_keys(config, ['a', 'b', 'c.d', 'c.e'])
assert missing == ['b', 'c.e']
def test_dot_notation_present(self, helper):
config = {'display': {'hardware': {'rows': 32}}}
assert helper.validate_required_keys(config, ['display.hardware.rows']) == []
def test_empty_requirements(self, helper):
assert helper.validate_required_keys({'a': 1}, []) == []
def test_present_with_none_counts_as_present(self, helper):
# _has_key checks key membership, not truthiness — a key set to
# None is NOT reported missing. Pinned semantics.
assert helper.validate_required_keys({'a': None}, ['a']) == []
+22 -14
View File
@@ -108,12 +108,13 @@ class TestConfigService:
with open(config_path, 'w') as f:
json.dump(current_config, f)
# Trigger reload manually - should detect change and notify
service.reload()
# Reload the way the file watcher does - should detect change and notify
assert service._load_config() is True
# Check callback was called (may be called during init or reload)
# The callback should be called if config actually changed
assert callback.called or True # May not be called if checksum matches
callback.assert_called_once()
old_config, new_config = callback.call_args[0]
assert old_config['display']['brightness'] == 50
assert new_config['display']['brightness'] == 75
def test_plugin_specific_subscriber(self, config_manager):
"""Test plugin-specific subscriber notification."""
@@ -128,19 +129,17 @@ class TestConfigService:
config_path = config_manager.config_path
with open(config_path, 'r') as f:
current_config = json.load(f)
if 'plugins' not in current_config:
current_config['plugins'] = {}
if 'weather' not in current_config['plugins']:
current_config['plugins']['weather'] = {}
current_config['plugins']['weather']['enabled'] = False # Change value
current_config['weather'] = {'enabled': False} # Change value
with open(config_path, 'w') as f:
json.dump(current_config, f)
# Trigger reload manually - should detect change and notify
service.reload()
# Reload the way the file watcher does - should detect change and notify
assert service._load_config() is True
# Check callback was called if config changed
assert callback.called or True # May not be called if checksum matches
callback.assert_called_once()
old_plugin_config, new_plugin_config = callback.call_args[0]
assert new_plugin_config['enabled'] is False
assert new_plugin_config['api_key'] == 'secret_key'
def test_config_merging(self, config_manager):
"""Test config merging logic via ConfigService."""
@@ -151,6 +150,15 @@ class TestConfigService:
assert "weather" in config
assert config["weather"]["api_key"] == "secret_key"
def test_unchanged_config_does_not_notify(self, config_manager):
"""Reloading an unchanged config must not notify subscribers."""
service = ConfigService(config_manager, enable_hot_reload=False)
callback = MagicMock()
service.subscribe(callback)
assert service._load_config() is False
callback.assert_not_called()
def test_shutdown(self, config_manager):
"""Test proper shutdown."""
service = ConfigService(config_manager, enable_hot_reload=True)
-54
View File
@@ -16,46 +16,6 @@ class TestDisplayControllerInitialization:
assert test_display_controller.available_modes == []
class TestDisplayControllerModeRotation:
"""Test display mode rotation logic."""
def test_basic_rotation(self, test_display_controller):
"""Test basic mode rotation."""
controller = test_display_controller
controller.available_modes = ["mode1", "mode2", "mode3"]
controller.current_mode_index = 0
controller.current_display_mode = "mode1"
# Simulate rotation
controller.current_mode_index = (controller.current_mode_index + 1) % len(controller.available_modes)
controller.current_display_mode = controller.available_modes[controller.current_mode_index]
assert controller.current_display_mode == "mode2"
assert controller.current_mode_index == 1
# Rotate again
controller.current_mode_index = (controller.current_mode_index + 1) % len(controller.available_modes)
controller.current_display_mode = controller.available_modes[controller.current_mode_index]
assert controller.current_display_mode == "mode3"
# Rotate back to start
controller.current_mode_index = (controller.current_mode_index + 1) % len(controller.available_modes)
controller.current_display_mode = controller.available_modes[controller.current_mode_index]
assert controller.current_display_mode == "mode1"
def test_rotation_with_single_mode(self, test_display_controller):
"""Test rotation with only one mode."""
controller = test_display_controller
controller.available_modes = ["mode1"]
controller.current_mode_index = 0
controller.current_mode_index = (controller.current_mode_index + 1) % len(controller.available_modes)
assert controller.current_mode_index == 0
class TestDisplayControllerOnDemand:
"""Test on-demand request handling."""
@@ -93,20 +53,6 @@ class TestDisplayControllerOnDemand:
assert controller.on_demand_active is False
assert controller.on_demand_mode is None
assert controller.on_demand_last_event == "expired"
def test_on_demand_schedule_override(self, test_display_controller):
"""Test that on-demand overrides schedule."""
controller = test_display_controller
controller.is_display_active = False
controller.on_demand_active = True
# Logic in run() loop handles this, so we simulate it
if controller.on_demand_active and not controller.is_display_active:
controller.on_demand_schedule_override = True
controller.is_display_active = True
assert controller.is_display_active is True
assert controller.on_demand_schedule_override is True
class TestDisplayControllerLivePriority:
-307
View File
@@ -1,307 +0,0 @@
"""Tests for src/common/display_helper.py (DisplayHelper).
Pure-PIL tests, no hardware or mocks required. Pixel assertions rely on
getbbox()/getpixel() rather than exact text pixel counts, because the
default-font metrics vary across Pillow versions.
These tests pin the FIXED behaviors on this branch:
- draw_error_message / draw_no_data_message return a rendered image
(they previously crashed with AttributeError),
- draw_scorebug_layout draws period/status/clock as one combined top
line (previously overprinted at the same y),
- draw_ticker_layout draws at x=0 (previously started at
x=display_width, i.e. entirely off-canvas -> blank frames).
"""
from PIL import Image, ImageDraw, ImageFont
from src.common.display_helper import DisplayHelper
def default_font():
return ImageFont.load_default()
def make_helper(width=128, height=32):
return DisplayHelper(width, height)
class TestCreateBaseImage:
def test_default_is_black_rgb_display_sized(self):
helper = make_helper()
img = helper.create_base_image()
assert img.size == (128, 32)
assert img.mode == 'RGB'
assert img.getpixel((0, 0)) == (0, 0, 0)
assert img.getpixel((127, 31)) == (0, 0, 0)
# Entirely black -> no bounding box in luminance
assert img.convert('L').getbbox() is None
def test_custom_background_color(self):
helper = make_helper()
img = helper.create_base_image(background_color=(10, 20, 30))
assert img.getpixel((0, 0)) == (10, 20, 30)
assert img.getpixel((64, 16)) == (10, 20, 30)
def test_mode_rgba_is_honored(self):
helper = make_helper()
img = helper.create_base_image(mode='RGBA')
assert img.mode == 'RGBA'
assert img.size == (128, 32)
class TestCreateOverlay:
def test_overlay_is_transparent_rgba(self):
helper = make_helper()
overlay = helper.create_overlay()
assert overlay.mode == 'RGBA'
assert overlay.size == (128, 32)
assert overlay.getpixel((0, 0)) == (0, 0, 0, 0)
assert overlay.getpixel((127, 31)) == (0, 0, 0, 0)
class TestCompositeImages:
def test_rgb_inputs_are_upconverted_and_result_is_rgba(self):
helper = make_helper()
base = Image.new('RGB', (128, 32), (0, 0, 0))
overlay = Image.new('RGB', (128, 32), (255, 0, 0))
result = helper.composite_images(base, overlay)
assert result.mode == 'RGBA'
assert result.size == base.size
# RGB->RGBA conversion yields a fully opaque overlay
assert result.getpixel((0, 0)) == (255, 0, 0, 255)
def test_transparent_overlay_leaves_base_visible(self):
helper = make_helper()
base = Image.new('RGB', (128, 32), (5, 6, 7))
overlay = helper.create_overlay()
result = helper.composite_images(base, overlay)
assert result.mode == 'RGBA'
assert result.getpixel((64, 16)) == (5, 6, 7, 255)
class TestScorebugLayout:
def test_full_game_data_renders(self):
helper = make_helper()
font = default_font()
fonts = {'time': font, 'status': font, 'score': font, 'team': font}
game_data = {
'home_score': 3, 'away_score': 2,
'home_abbr': 'NYY', 'away_abbr': 'BOS',
'status_text': 'LIVE', 'period_text': 'T9', 'clock': '2:30',
}
img = helper.draw_scorebug_layout(game_data, fonts)
assert img.mode == 'RGB'
assert img.size == (128, 32)
assert img.convert('L').getbbox() is not None
def test_empty_game_data_uses_defaults_without_raising(self):
helper = make_helper()
font = default_font()
fonts = {'time': font, 'status': font, 'score': font, 'team': font}
img = helper.draw_scorebug_layout({}, fonts)
assert img.mode == 'RGB'
assert img.size == (128, 32)
# Defaults '0'/'HOME'/'AWAY' actually render something
assert img.convert('L').getbbox() is not None
def test_empty_fonts_dict_falls_back_to_default_font(self):
# Pin: fonts={} must not raise — PIL falls back to the default
# font when font=None is passed through.
helper = make_helper()
img = helper.draw_scorebug_layout(
{'status_text': 'FINAL', 'period_text': 'Q4', 'clock': '0:00'}, {})
assert img.size == (128, 32)
assert img.convert('L').getbbox() is not None
def test_top_line_is_one_combined_centered_draw(self):
# FIXED behavior: period/status/clock are joined into a single
# top line drawn once at y=1 instead of three overprinted draws.
helper = make_helper()
calls = []
original = helper._draw_centered_text
def spy(draw, text, font, y_position):
calls.append({'text': text, 'y_position': y_position})
original(draw, text, font, y_position)
helper._draw_centered_text = spy
font = default_font()
fonts = {'time': font, 'status': font, 'score': font, 'team': font}
helper.draw_scorebug_layout(
{'period_text': 'Q4', 'status_text': 'LIVE', 'clock': '2:30'},
fonts)
top_calls = [c for c in calls if c['y_position'] == 1]
assert len(top_calls) == 1
text = top_calls[0]['text']
assert 'Q4' in text
assert 'LIVE' in text
assert '2:30' in text
def test_no_top_line_when_all_parts_empty(self):
helper = make_helper()
calls = []
original = helper._draw_centered_text
def spy(draw, text, font, y_position):
calls.append(y_position)
original(draw, text, font, y_position)
helper._draw_centered_text = spy
font = default_font()
helper.draw_scorebug_layout({}, {'score': font, 'team': font})
assert 1 not in calls # no combined top line drawn
def test_logo_positions_bleed_off_edges(self):
# Home logo pastes at x = width - logo.width + 10 (right edge,
# bleeding off-screen right); away at x = -10 (bleeding left).
helper = make_helper()
home_logo = Image.new('RGBA', (20, 20), (0, 0, 255, 255)) # blue
away_logo = Image.new('RGBA', (20, 20), (255, 0, 0, 255)) # red
# Empty abbrs/status so text can't land on the probed pixels.
game_data = {'home_abbr': '', 'away_abbr': ''}
font = default_font()
img = helper.draw_scorebug_layout(game_data, {'score': font},
home_logo=home_logo,
away_logo=away_logo)
# center_y = 16; logos span y 6..25 -> probe y=16 at both edges.
assert img.getpixel((0, 16)) == (255, 0, 0) # away (left edge)
assert img.getpixel((127, 16)) == (0, 0, 255) # home (right edge)
# And the off-screen parts are truly clipped: image is still 128 wide
assert img.size == (128, 32)
class TestTickerLayout:
def test_frame_is_not_blank(self):
# FIXED behavior: text now starts at x=0. Previously it was drawn
# at x=display_width, entirely off-canvas, so frames were blank.
helper = make_helper()
img = helper.draw_ticker_layout('HELLO WORLD', default_font())
assert img.size == (128, 32)
assert img.mode == 'RGB'
assert img.convert('L').getbbox() is not None
def test_text_starts_at_left_edge(self):
helper = make_helper()
img = helper.draw_ticker_layout('HELLO', default_font())
bbox = img.convert('L').getbbox()
assert bbox is not None
# Text is positioned at x=0 (outline extends 1px left, clipped),
# so ink begins hugging the left edge. Allow a couple of pixels of
# slack for font-dependent left-side bearing.
assert bbox[0] <= 2
def test_scroll_speed_does_not_affect_frame(self):
# Pin: scroll_speed is accepted for API compatibility only.
helper = make_helper()
font = default_font()
img1 = helper.draw_ticker_layout('SCROLLING', font, scroll_speed=1)
img5 = helper.draw_ticker_layout('SCROLLING', font, scroll_speed=5)
assert img1.tobytes() == img5.tobytes()
def test_custom_colors(self):
helper = make_helper()
img = helper.draw_ticker_layout('X', default_font(),
background_color=(0, 0, 40),
text_color=(0, 255, 0))
assert img.getpixel((127, 0)) == (0, 0, 40) # background corner
colors = {img.getpixel((x, y))
for x in range(img.width) for y in range(img.height)}
# Text color appears somewhere (anti-aliasing may blend it, so
# check for a green-dominant pixel rather than the exact color).
assert any(g > 150 and r < 100 for (r, g, b) in colors)
class TestCenteredText:
def test_renders_centered_text_on_background(self):
helper = make_helper()
img = helper.draw_centered_text('HI', default_font(),
background_color=(0, 0, 60),
text_color=(255, 255, 0))
assert img.size == (128, 32)
assert img.convert('L').getbbox() is not None
# Corners stay pure background
assert img.getpixel((0, 0)) == (0, 0, 60)
assert img.getpixel((127, 0)) == (0, 0, 60)
assert img.getpixel((0, 31)) == (0, 0, 60)
assert img.getpixel((127, 31)) == (0, 0, 60)
class TestErrorAndNoDataMessages:
def test_draw_error_message_returns_rendered_image(self):
# FIXED behavior: used to crash with AttributeError; now returns
# a rendered image on a dark red background.
helper = make_helper()
img = helper.draw_error_message('Boom')
assert img.size == (128, 32)
assert img.mode == 'RGB'
assert img.convert('L').getbbox() is not None
assert img.getpixel((0, 0)) == (50, 0, 0) # dark red background
def test_draw_error_message_default_text(self):
helper = make_helper()
img = helper.draw_error_message()
assert img.size == (128, 32)
assert img.getpixel((127, 31)) == (50, 0, 0)
def test_draw_no_data_message_returns_rendered_image(self):
helper = make_helper()
img = helper.draw_no_data_message()
assert img.size == (128, 32)
assert img.mode == 'RGB'
assert img.convert('L').getbbox() is not None
assert img.getpixel((0, 0)) == (0, 0, 0) # black background
class TestDrawTextWithOutline:
def test_fill_color_appears_in_output(self):
helper = make_helper()
img = Image.new('RGB', (40, 20), (0, 0, 255))
draw = ImageDraw.Draw(img)
helper._draw_text_with_outline(draw, 'X', (5, 2), default_font(),
fill=(255, 0, 0))
pixels = {img.getpixel((x, y))
for x in range(img.width) for y in range(img.height)}
# Anti-aliased fonts blend edge pixels, so look for red-dominant
# (fill) and near-black (outline) pixels rather than exact colors.
assert any(r > 150 and g < 50 for (r, g, b) in pixels) # fill
assert any(max(p) < 80 for p in pixels) # outline
def test_default_fill_is_white(self):
helper = make_helper()
img = Image.new('RGB', (40, 20), (0, 0, 255))
draw = ImageDraw.Draw(img)
helper._draw_text_with_outline(draw, 'X', (5, 2), default_font())
pixels = {img.getpixel((x, y))
for x in range(img.width) for y in range(img.height)}
# White-dominant pixel present (exact white may be anti-aliased)
assert any(r > 200 and g > 200 for (r, g, b) in pixels)
class TestOrientationAndDimensions:
def test_landscape_display(self):
helper = DisplayHelper(128, 32)
assert helper.is_landscape() is True
assert helper.is_portrait() is False
def test_portrait_display(self):
helper = DisplayHelper(32, 128)
assert helper.is_portrait() is True
assert helper.is_landscape() is False
def test_square_display_is_neither(self):
# Pin: a square display is neither portrait nor landscape.
helper = DisplayHelper(64, 64)
assert helper.is_portrait() is False
assert helper.is_landscape() is False
def test_get_center_position(self):
assert DisplayHelper(128, 32).get_center_position() == (64, 16)
def test_get_center_position_floors_odd_dimensions(self):
assert DisplayHelper(65, 33).get_center_position() == (32, 16)
def test_get_display_dimensions(self):
assert DisplayHelper(128, 32).get_display_dimensions() == (128, 32)
assert DisplayHelper(64, 64).get_display_dimensions() == (64, 64)
-17
View File
@@ -1,7 +1,6 @@
import os
import pytest
from unittest.mock import MagicMock, patch
from PIL import ImageDraw
# display_manager imports the hardware rgbmatrix module at import time unless
# EMULATOR=true. Use the emulator (same convention as
@@ -106,22 +105,6 @@ class TestDisplayManagerDrawing:
assert dm.image.convert("L").getbbox() is not None, \
"draw_text lit no pixels"
def test_draw_image(self, test_config, mock_rgb_matrix):
"""Test image drawing."""
with patch.dict('os.environ', {'EMULATOR': 'false'}):
dm = DisplayManager(test_config)
# DisplayManager doesn't have draw_image method
# It uses SetImage on canvas in update_display()
# Just verify DisplayManager can handle image operations
from PIL import Image
test_image = Image.new('RGB', (64, 32))
dm.image = test_image
dm.draw = ImageDraw.Draw(dm.image)
# Verify image was set
assert dm.image is not None
class TestDisplayManagerResourceManagement:
+1 -89
View File
@@ -1,11 +1,5 @@
import logging
import json
from src.exceptions import CacheError, ConfigError, PluginError, DisplayError
from src.common.error_handler import (
handle_file_operation,
handle_json_operation,
safe_execute
)
class TestCustomExceptions:
"""Test custom exception classes."""
@@ -37,85 +31,3 @@ class TestCustomExceptions:
# DisplayError includes context in string representation
assert "Display not found" in str(error)
assert error.context.get('display_mode') == 'adafruit'
class TestErrorHandlerUtilities:
"""Test error handler utilities."""
def test_handle_file_operation_read_success(self, tmp_path):
"""Test successful file read."""
test_file = tmp_path / "test.txt"
test_file.write_text("test content")
result = handle_file_operation(
lambda: test_file.read_text(),
"Read failed",
logging.getLogger(__name__),
default=""
)
assert result == "test content"
def test_handle_file_operation_read_failure(self, tmp_path):
"""Test file read failure."""
non_existent = tmp_path / "nonexistent.txt"
result = handle_file_operation(
lambda: non_existent.read_text(),
"Read failed",
logging.getLogger(__name__),
default="fallback"
)
assert result == "fallback"
def test_handle_json_operation_success(self, tmp_path):
"""Test successful JSON parse."""
test_file = tmp_path / "test.json"
test_file.write_text('{"key": "value"}')
result = handle_json_operation(
lambda: json.loads(test_file.read_text()),
"JSON parse failed",
logging.getLogger(__name__),
default={}
)
assert result == {"key": "value"}
def test_handle_json_operation_failure(self, tmp_path):
"""Test JSON parse failure."""
test_file = tmp_path / "invalid.json"
test_file.write_text('invalid json {')
result = handle_json_operation(
lambda: json.loads(test_file.read_text()),
"JSON parse failed",
logging.getLogger(__name__),
default={"default": True}
)
assert result == {"default": True}
def test_safe_execute_success(self):
"""Test successful execution with safe_execute."""
def success_func():
return "success"
result = safe_execute(
success_func,
"Execution failed",
logging.getLogger(__name__),
default="failed"
)
assert result == "success"
def test_safe_execute_failure(self):
"""Test failure handling with safe_execute."""
def failing_func():
raise ValueError("Something went wrong")
result = safe_execute(
failing_func,
"Execution failed",
logging.getLogger(__name__),
default="fallback"
)
assert result == "fallback"
-317
View File
@@ -1,317 +0,0 @@
"""
Tests for src/common/game_helper.py
Covers GameHelper: extract_game_details, filter_*, sort_games_by_time,
process_games, get_game_summary, and all private helpers.
"""
import logging
import pytest
from datetime import datetime, timezone, timedelta
from src.common.game_helper import GameHelper
def _make_logger() -> logging.Logger:
return logging.getLogger("test_game_helper")
def _make_espn_event(
state: str = "in",
home_abbr: str = "LAL",
away_abbr: str = "BOS",
home_score: str = "105",
away_score: str = "98",
date_str: str = "2024-01-15T20:00:00Z",
period: int = 4,
status_name: str = "STATUS_IN_PROGRESS",
home_record: str = "30-10",
away_record: str = "25-15",
event_id: str = "game-1",
) -> dict:
return {
"id": event_id,
"date": date_str,
"competitions": [
{
"status": {
"type": {
"state": state,
"shortDetail": "Q4 2:30",
"name": status_name,
},
"period": period,
"displayClock": "2:30",
},
"competitors": [
{
"homeAway": "home",
"id": "h1",
"team": {"abbreviation": home_abbr, "displayName": f"{home_abbr} Team"},
"score": home_score,
"records": [{"summary": home_record}],
},
{
"homeAway": "away",
"id": "a1",
"team": {"abbreviation": away_abbr, "displayName": f"{away_abbr} Team"},
"score": away_score,
"records": [{"summary": away_record}],
},
],
}
],
}
@pytest.fixture
def helper():
return GameHelper(timezone_str="UTC", logger=_make_logger())
# ---------------------------------------------------------------------------
# extract_game_details
# ---------------------------------------------------------------------------
class TestExtractGameDetails:
def test_live_game(self, helper):
event = _make_espn_event(state="in")
result = helper.extract_game_details(event)
assert result is not None
assert result["is_live"] is True
assert result["is_final"] is False
assert result["is_upcoming"] is False
def test_final_game(self, helper):
event = _make_espn_event(state="post")
result = helper.extract_game_details(event)
assert result["is_final"] is True
def test_upcoming_game(self, helper):
event = _make_espn_event(state="pre")
result = helper.extract_game_details(event)
assert result["is_upcoming"] is True
def test_halftime_detection(self, helper):
event = _make_espn_event(state="halftime", status_name="STATUS_HALFTIME")
result = helper.extract_game_details(event)
assert result["is_halftime"] is True
def test_basic_fields_present(self, helper):
event = _make_espn_event()
result = helper.extract_game_details(event)
for key in ("id", "home_abbr", "away_abbr", "home_score", "away_score",
"home_record", "away_record", "start_time_utc"):
assert key in result
def test_team_abbreviations(self, helper):
event = _make_espn_event(home_abbr="MIA", away_abbr="PHX")
result = helper.extract_game_details(event)
assert result["home_abbr"] == "MIA"
assert result["away_abbr"] == "PHX"
def test_scores_as_strings(self, helper):
event = _make_espn_event(home_score="110", away_score="99")
result = helper.extract_game_details(event)
assert result["home_score"] == "110"
assert result["away_score"] == "99"
def test_returns_none_on_empty(self, helper):
assert helper.extract_game_details({}) is None
assert helper.extract_game_details(None) is None
def test_returns_none_when_no_competitors(self, helper):
event = _make_espn_event()
event["competitions"][0]["competitors"] = []
assert helper.extract_game_details(event) is None
def test_date_z_suffix_parsed(self, helper):
event = _make_espn_event(date_str="2024-06-01T19:30:00Z")
result = helper.extract_game_details(event)
assert result["start_time_utc"] is not None
assert result["start_time_utc"].tzinfo is not None
def test_zero_zero_record_suppressed(self, helper):
event = _make_espn_event(home_record="0-0", away_record="0-0-0")
result = helper.extract_game_details(event)
assert result["home_record"] == ""
assert result["away_record"] == ""
def test_basketball_sport_fields(self, helper):
event = _make_espn_event(period=3)
result = helper.extract_game_details(event, sport="basketball")
assert result["period_text"] == "Q3"
assert "clock" in result
def test_basketball_overtime_period(self, helper):
event = _make_espn_event(period=5)
result = helper.extract_game_details(event, sport="basketball")
assert result["period_text"] == "OT1"
def test_football_sport_fields(self, helper):
event = _make_espn_event(period=2)
result = helper.extract_game_details(event, sport="football")
assert result["period_text"] == "Q2"
def test_hockey_sport_fields_period_1(self, helper):
event = _make_espn_event(period=1)
result = helper.extract_game_details(event, sport="hockey")
assert result["period_text"] == "P1"
def test_hockey_sport_fields_ot(self, helper):
event = _make_espn_event(period=4)
result = helper.extract_game_details(event, sport="hockey")
assert result["period_text"] == "OT1"
def test_baseball_sport_fields(self, helper):
event = _make_espn_event(period=7)
result = helper.extract_game_details(event, sport="baseball")
assert result["period_text"] == "INN 7"
# ---------------------------------------------------------------------------
# Filter methods
# ---------------------------------------------------------------------------
class TestFilterMethods:
def _make_games(self):
now = datetime.now(timezone.utc)
return [
{"is_live": True, "is_final": False, "is_upcoming": False, "home_abbr": "LAL", "away_abbr": "BOS", "start_time_utc": now},
{"is_live": False, "is_final": True, "is_upcoming": False, "home_abbr": "MIA", "away_abbr": "PHX", "start_time_utc": now - timedelta(hours=3)},
{"is_live": False, "is_final": False, "is_upcoming": True, "home_abbr": "DAL", "away_abbr": "CHI", "start_time_utc": now + timedelta(hours=2)},
]
def test_filter_live_games(self, helper):
games = self._make_games()
result = helper.filter_live_games(games)
assert len(result) == 1
assert result[0]["home_abbr"] == "LAL"
def test_filter_final_games(self, helper):
games = self._make_games()
result = helper.filter_final_games(games)
assert len(result) == 1
assert result[0]["home_abbr"] == "MIA"
def test_filter_upcoming_games(self, helper):
games = self._make_games()
result = helper.filter_upcoming_games(games)
assert len(result) == 1
assert result[0]["home_abbr"] == "DAL"
def test_filter_favorite_teams_match(self, helper):
games = self._make_games()
result = helper.filter_favorite_teams(games, ["LAL"])
assert len(result) == 1
assert result[0]["home_abbr"] == "LAL"
def test_filter_favorite_teams_empty_list_returns_all(self, helper):
games = self._make_games()
result = helper.filter_favorite_teams(games, [])
assert len(result) == 3
def test_filter_favorite_teams_away_match(self, helper):
games = self._make_games()
result = helper.filter_favorite_teams(games, ["BOS"])
assert len(result) == 1
def test_filter_recent_games_within_window(self, helper):
now = datetime.now(timezone.utc)
games = [
{"start_time_utc": now - timedelta(days=2), "is_final": True},
{"start_time_utc": now - timedelta(days=10), "is_final": True},
]
result = helper.filter_recent_games(games, days_back=7)
assert len(result) == 1
def test_filter_recent_games_all_within(self, helper):
now = datetime.now(timezone.utc)
games = [
{"start_time_utc": now - timedelta(days=1)},
{"start_time_utc": now - timedelta(days=3)},
]
result = helper.filter_recent_games(games, days_back=7)
assert len(result) == 2
def test_sort_games_ascending(self, helper):
now = datetime.now(timezone.utc)
games = [
{"start_time_utc": now + timedelta(hours=2), "id": "late"},
{"start_time_utc": now + timedelta(hours=1), "id": "early"},
]
result = helper.sort_games_by_time(games)
assert result[0]["id"] == "early"
def test_sort_games_descending(self, helper):
now = datetime.now(timezone.utc)
games = [
{"start_time_utc": now + timedelta(hours=1), "id": "early"},
{"start_time_utc": now + timedelta(hours=2), "id": "late"},
]
result = helper.sort_games_by_time(games, reverse=True)
assert result[0]["id"] == "late"
# ---------------------------------------------------------------------------
# process_games
# ---------------------------------------------------------------------------
class TestProcessGames:
def test_processes_valid_events(self, helper):
events = [
_make_espn_event(event_id="1"),
_make_espn_event(event_id="2"),
]
result = helper.process_games(events)
assert len(result) == 2
def test_skips_invalid_events(self, helper):
events = [
_make_espn_event(event_id="1"),
{}, # invalid
]
result = helper.process_games(events)
assert len(result) == 1
def test_empty_events(self, helper):
assert helper.process_games([]) == []
# ---------------------------------------------------------------------------
# get_game_summary
# ---------------------------------------------------------------------------
class TestGetGameSummary:
def test_live_summary(self, helper):
game = {
"home_abbr": "LAL", "away_abbr": "BOS",
"home_score": "105", "away_score": "98",
"status_text": "Q4 2:30",
"is_live": True, "is_final": False,
}
summary = helper.get_game_summary(game)
assert "BOS" in summary
assert "LAL" in summary
assert "98" in summary
assert "105" in summary
def test_final_summary(self, helper):
game = {
"home_abbr": "LAL", "away_abbr": "BOS",
"home_score": "110", "away_score": "102",
"status_text": "Final",
"is_live": False, "is_final": True,
}
summary = helper.get_game_summary(game)
assert "Final" in summary
def test_upcoming_summary(self, helper):
game = {
"home_abbr": "LAL", "away_abbr": "BOS",
"home_score": "0", "away_score": "0",
"status_text": "7:30 PM",
"is_live": False, "is_final": False,
}
summary = helper.get_game_summary(game)
assert "7:30 PM" in summary
-307
View File
@@ -1,307 +0,0 @@
"""
Tests for src/plugin_system/health_monitor.py
Covers PluginHealthMonitor: get_plugin_health_status, get_plugin_health_metrics,
get_all_plugin_health, _get_recovery_suggestions, start/stop_monitoring,
register_health_check.
"""
import pytest
from unittest.mock import MagicMock, patch
from datetime import datetime
from src.plugin_system.health_monitor import (
PluginHealthMonitor,
HealthStatus,
HealthMetrics,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _make_health_tracker(
summary: dict | None = None,
all_summaries: dict | None = None,
):
"""Return a mock PluginHealthTracker."""
tracker = MagicMock()
tracker.get_health_summary.return_value = summary
tracker.get_all_health_summaries.return_value = all_summaries or {}
return tracker
def _healthy_summary() -> dict:
return {
"success_rate": 100.0,
"circuit_state": "closed",
"consecutive_failures": 0,
"total_failures": 0,
"total_successes": 50,
"last_success_time": datetime.now().isoformat(),
"last_error": None,
}
def _degraded_summary() -> dict:
return {
"success_rate": 40.0, # 60% error rate
"circuit_state": "closed",
"consecutive_failures": 3,
"total_failures": 6,
"total_successes": 4,
"last_success_time": None,
"last_error": "timeout occurred",
}
def _unhealthy_summary() -> dict:
return {
"success_rate": 10.0, # 90% error rate
"circuit_state": "open",
"consecutive_failures": 10,
"total_failures": 9,
"total_successes": 1,
"last_success_time": None,
"last_error": "ImportError: missing module",
}
@pytest.fixture
def monitor():
tracker = _make_health_tracker(_healthy_summary())
return PluginHealthMonitor(health_tracker=tracker)
# ---------------------------------------------------------------------------
# get_plugin_health_status
# ---------------------------------------------------------------------------
class TestGetPluginHealthStatus:
def test_healthy_status(self):
tracker = _make_health_tracker(_healthy_summary())
monitor = PluginHealthMonitor(tracker)
status = monitor.get_plugin_health_status("plugin_a")
assert status == HealthStatus.HEALTHY
def test_degraded_status(self):
tracker = _make_health_tracker(_degraded_summary())
monitor = PluginHealthMonitor(tracker, degraded_threshold=0.5, unhealthy_threshold=0.8)
status = monitor.get_plugin_health_status("plugin_b")
assert status == HealthStatus.DEGRADED
def test_unhealthy_status(self):
tracker = _make_health_tracker(_unhealthy_summary())
monitor = PluginHealthMonitor(tracker, unhealthy_threshold=0.8)
status = monitor.get_plugin_health_status("plugin_c")
assert status == HealthStatus.UNHEALTHY
def test_open_circuit_breaker_is_unhealthy(self):
summary = _healthy_summary()
summary["circuit_state"] = "open"
tracker = _make_health_tracker(summary)
monitor = PluginHealthMonitor(tracker)
status = monitor.get_plugin_health_status("plugin_d")
assert status == HealthStatus.UNHEALTHY
def test_unknown_when_no_tracker(self):
monitor = PluginHealthMonitor(health_tracker=None)
status = monitor.get_plugin_health_status("plugin_e")
assert status == HealthStatus.UNKNOWN
def test_unknown_when_no_summary(self):
tracker = _make_health_tracker(None)
monitor = PluginHealthMonitor(tracker)
status = monitor.get_plugin_health_status("plugin_f")
assert status == HealthStatus.UNKNOWN
# ---------------------------------------------------------------------------
# get_plugin_health_metrics
# ---------------------------------------------------------------------------
class TestGetPluginHealthMetrics:
def test_healthy_metrics(self):
tracker = _make_health_tracker(_healthy_summary())
monitor = PluginHealthMonitor(tracker)
metrics = monitor.get_plugin_health_metrics("plugin_a")
assert isinstance(metrics, HealthMetrics)
assert metrics.status == HealthStatus.HEALTHY
assert metrics.success_rate == pytest.approx(1.0)
assert metrics.error_rate == pytest.approx(0.0)
def test_degraded_metrics(self):
tracker = _make_health_tracker(_degraded_summary())
monitor = PluginHealthMonitor(tracker, degraded_threshold=0.5, unhealthy_threshold=0.8)
metrics = monitor.get_plugin_health_metrics("plugin_b")
assert metrics.status == HealthStatus.DEGRADED
assert metrics.consecutive_failures == 3
def test_unhealthy_metrics(self):
tracker = _make_health_tracker(_unhealthy_summary())
monitor = PluginHealthMonitor(tracker, unhealthy_threshold=0.8)
metrics = monitor.get_plugin_health_metrics("plugin_c")
assert metrics.status == HealthStatus.UNHEALTHY
assert metrics.circuit_breaker_state == "open"
assert metrics.last_error is not None
def test_metrics_without_tracker(self):
monitor = PluginHealthMonitor(health_tracker=None)
metrics = monitor.get_plugin_health_metrics("plugin_d")
assert metrics.status == HealthStatus.UNKNOWN
assert metrics.plugin_id == "plugin_d"
def test_metrics_without_summary(self):
tracker = _make_health_tracker(None)
monitor = PluginHealthMonitor(tracker)
metrics = monitor.get_plugin_health_metrics("plugin_e")
assert metrics.status == HealthStatus.UNKNOWN
def test_last_successful_update_parsed(self):
summary = _healthy_summary()
summary["last_success_time"] = "2024-06-01T12:00:00"
tracker = _make_health_tracker(summary)
monitor = PluginHealthMonitor(tracker)
metrics = monitor.get_plugin_health_metrics("plugin_a")
assert metrics.last_successful_update is not None
assert isinstance(metrics.last_successful_update, datetime)
def test_invalid_last_success_time_handled(self):
summary = _healthy_summary()
summary["last_success_time"] = "not-a-date"
tracker = _make_health_tracker(summary)
monitor = PluginHealthMonitor(tracker)
# Should not raise
metrics = monitor.get_plugin_health_metrics("plugin_a")
assert metrics.last_successful_update is None
def test_total_successes_failures(self):
tracker = _make_health_tracker(_degraded_summary())
monitor = PluginHealthMonitor(tracker, degraded_threshold=0.5, unhealthy_threshold=0.8)
metrics = monitor.get_plugin_health_metrics("plugin_b")
assert metrics.total_failures == 6
assert metrics.total_successes == 4
# ---------------------------------------------------------------------------
# get_all_plugin_health
# ---------------------------------------------------------------------------
class TestGetAllPluginHealth:
def test_returns_empty_without_tracker(self):
monitor = PluginHealthMonitor(health_tracker=None)
result = monitor.get_all_plugin_health()
assert result == {}
def test_returns_metrics_for_each_plugin(self):
all_summaries = {
"plugin_a": _healthy_summary(),
"plugin_b": _degraded_summary(),
}
tracker = MagicMock()
tracker.get_all_health_summaries.return_value = all_summaries
tracker.get_health_summary.side_effect = lambda pid: all_summaries.get(pid)
monitor = PluginHealthMonitor(tracker, degraded_threshold=0.5, unhealthy_threshold=0.8)
result = monitor.get_all_plugin_health()
assert "plugin_a" in result
assert "plugin_b" in result
assert isinstance(result["plugin_a"], HealthMetrics)
def test_returns_empty_when_no_summaries(self):
tracker = _make_health_tracker(all_summaries={})
monitor = PluginHealthMonitor(tracker)
result = monitor.get_all_plugin_health()
assert result == {}
# ---------------------------------------------------------------------------
# _get_recovery_suggestions
# ---------------------------------------------------------------------------
class TestGetRecoverySuggestions:
def test_healthy_plugin_suggestion(self):
tracker = _make_health_tracker(_healthy_summary())
monitor = PluginHealthMonitor(tracker)
suggestions = monitor._get_recovery_suggestions("p", _healthy_summary(), HealthStatus.HEALTHY)
assert any("healthy" in s.lower() for s in suggestions)
def test_unhealthy_suggestions(self):
tracker = _make_health_tracker(_unhealthy_summary())
monitor = PluginHealthMonitor(tracker, unhealthy_threshold=0.8)
suggestions = monitor._get_recovery_suggestions("p", _unhealthy_summary(), HealthStatus.UNHEALTHY)
assert len(suggestions) > 0
assert any("unhealthy" in s.lower() for s in suggestions)
def test_open_circuit_breaker_suggestion(self):
summary = _unhealthy_summary()
summary["circuit_state"] = "open"
tracker = _make_health_tracker(summary)
monitor = PluginHealthMonitor(tracker, unhealthy_threshold=0.8)
suggestions = monitor._get_recovery_suggestions("p", summary, HealthStatus.UNHEALTHY)
assert any("circuit" in s.lower() for s in suggestions)
def test_timeout_error_suggestion(self):
summary = _degraded_summary()
summary["last_error"] = "connection timeout occurred"
tracker = _make_health_tracker(summary)
monitor = PluginHealthMonitor(tracker, degraded_threshold=0.5, unhealthy_threshold=0.8)
suggestions = monitor._get_recovery_suggestions("p", summary, HealthStatus.DEGRADED)
assert any("timeout" in s.lower() for s in suggestions)
def test_import_error_suggestion(self):
summary = _unhealthy_summary()
summary["last_error"] = "ImportError: missing module"
tracker = _make_health_tracker(summary)
monitor = PluginHealthMonitor(tracker, unhealthy_threshold=0.8)
suggestions = monitor._get_recovery_suggestions("p", summary, HealthStatus.UNHEALTHY)
assert any("dependencies" in s.lower() or "import" in s.lower() or "missing" in s.lower()
for s in suggestions)
def test_permission_error_suggestion(self):
summary = _unhealthy_summary()
summary["last_error"] = "permission denied to access resource"
tracker = _make_health_tracker(summary)
monitor = PluginHealthMonitor(tracker, unhealthy_threshold=0.8)
suggestions = monitor._get_recovery_suggestions("p", summary, HealthStatus.UNHEALTHY)
assert any("permission" in s.lower() for s in suggestions)
def test_degraded_suggestions_include_error_rate(self):
tracker = _make_health_tracker(_degraded_summary())
monitor = PluginHealthMonitor(tracker, degraded_threshold=0.5, unhealthy_threshold=0.8)
suggestions = monitor._get_recovery_suggestions("p", _degraded_summary(), HealthStatus.DEGRADED)
assert any("%" in s for s in suggestions)
# ---------------------------------------------------------------------------
# start / stop monitoring
# ---------------------------------------------------------------------------
class TestMonitorLifecycle:
def test_start_monitoring(self, monitor):
monitor.start_monitoring()
try:
assert monitor._monitor_thread is not None
assert monitor._monitor_thread.is_alive()
finally:
monitor.stop_monitoring()
def test_stop_monitoring(self, monitor):
monitor.start_monitoring()
monitor.stop_monitoring()
# Thread should no longer be alive
assert not monitor._monitor_thread.is_alive()
def test_double_start_no_duplicate_threads(self, monitor):
monitor.start_monitoring()
try:
thread1 = monitor._monitor_thread
monitor.start_monitoring() # should be idempotent
assert monitor._monitor_thread is thread1
finally:
monitor.stop_monitoring()
def test_register_health_check(self, monitor):
callback = MagicMock()
monitor.register_health_check(callback)
assert callback in monitor._health_check_callbacks
+1 -4
View File
@@ -61,10 +61,7 @@ def tiny_floor(monkeypatch):
def _controller(plugin_ids, executor):
c = DisplayController.__new__(DisplayController)
c.plugin_manager = Mock()
# Both attributes, because _update_modules reads
# `loaded_plugins or plugins` and an empty dict is falsy.
c.plugin_manager.loaded_plugins = {pid: Mock() for pid in plugin_ids}
c.plugin_manager.plugins = dict(c.plugin_manager.loaded_plugins)
c.plugin_manager.plugins = {pid: Mock() for pid in plugin_ids}
c.plugin_manager.plugin_executor = executor
c.plugin_manager.plugin_last_update = {}
c.plugin_manager.health_tracker = None
-11
View File
@@ -16,17 +16,6 @@ def _warnings(caplog):
return [r for r in caplog.records if r.levelno == logging.WARNING]
class TestParseSemver:
def test_basic(self, loader):
assert loader._parse_semver("3.1.0") == (3, 1, 0)
assert loader._parse_semver("v2.0") == (2, 0, 0)
assert loader._parse_semver("2.0.0-beta.1") == (2, 0, 0)
def test_unparseable(self, loader):
assert loader._parse_semver(None) is None
assert loader._parse_semver(123) is None
class TestWarnIfIncompatible:
def test_warns_when_plugin_needs_newer_core(self, loader, caplog, monkeypatch):
import src
-166
View File
@@ -1,166 +0,0 @@
"""Plugin state history must not grow without bound.
`PluginStateManager` recorded every state transition in a per-plugin list and
never trimmed it. The only code that removed entries was `clear_state()`, called
solely from `PluginManager.unload_plugin()`, so a plugin that stays loaded --
i.e. normal operation -- never released a single entry.
The list is written on the hot scheduling path. Every update cycle appends
twice: `_reserve_for_update()` sets RUNNING and `_finish()` sets ENABLED back
again. At the default 60-second update interval that is 2,880 entries per
plugin per day, and nothing ever reads the entries -- `get_state_info()` only
takes their `len()`. It is pure dead weight.
Measured against the unpatched class, ten plugins on a 60s interval retain
864,010 transitions after thirty simulated days, for 231 MB of heap. On a 1 GB
Pi that is fatal on its own, and the failure is not a clean OOM: once
MemAvailable falls far enough, fork() starts returning ENOMEM, so sshd accepts
connections and closes them before its banner while the kernel still answers
pings. The board looks like a hardware fault and needs a power cycle.
These tests pin the cap, the retention order, and the one piece of behaviour the
cap must not change: `state_history_count` is surfaced through the web API, so
it has to keep reporting the lifetime total rather than plateauing at the cap.
"""
import os
import sys
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from src.plugin_system.plugin_state import ( # noqa: E402
MAX_STATE_HISTORY_PER_PLUGIN,
PluginState,
PluginStateManager,
)
def _cycle_updates(manager, plugin_id, cycles):
"""Drive the real scheduling path: RUNNING on reserve, ENABLED on finish."""
for _ in range(cycles):
manager.set_state(plugin_id, PluginState.RUNNING)
manager.set_state(plugin_id, PluginState.ENABLED)
def test_state_history_is_capped():
"""A day of updates must not retain a day of transitions."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
# One simulated day at the default 60s update interval.
_cycle_updates(manager, "clock", 1440)
history = manager.get_state_history("clock")
assert len(history) <= MAX_STATE_HISTORY_PER_PLUGIN, (
f"history grew to {len(history)} entries; it is never trimmed"
)
def test_state_history_keeps_the_most_recent_transitions():
"""Trimming drops the oldest entries, not the newest."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
_cycle_updates(manager, "clock", MAX_STATE_HISTORY_PER_PLUGIN)
history = manager.get_state_history("clock")
# The scheduling cycle ends on ENABLED, so the newest entry is the
# RUNNING -> ENABLED half of the last cycle.
assert history[-1]["from"] == PluginState.RUNNING.value
assert history[-1]["to"] == PluginState.ENABLED.value
# And the very first ENABLED transition has aged out.
assert history[0]["from"] != PluginState.UNLOADED.value
def test_state_history_count_reports_lifetime_total():
"""The count exposed through the API must not plateau at the cap.
`get_state_info()['state_history_count']` is surfaced by the web UI. Capping
the retained list must not turn it into "entries we happen to still hold".
"""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
total = 1
cycles = MAX_STATE_HISTORY_PER_PLUGIN * 2
_cycle_updates(manager, "clock", cycles)
total += cycles * 2
info = manager.get_state_info("clock")
assert info["state_history_count"] == total
assert len(manager.get_state_history("clock")) <= MAX_STATE_HISTORY_PER_PLUGIN
def test_error_transitions_are_capped_too():
"""set_state_with_error() appends to the same list and needs the same cap."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
for _ in range(MAX_STATE_HISTORY_PER_PLUGIN * 2):
manager.set_state_with_error(
"clock",
PluginState.ENABLED,
{"reason": "update timeout"},
error=RuntimeError("boom"),
)
assert len(manager.get_state_history("clock")) <= MAX_STATE_HISTORY_PER_PLUGIN
def test_history_is_isolated_per_plugin():
"""The cap is per plugin, not shared across the manager."""
manager = PluginStateManager()
for plugin_id in ("clock", "weather"):
manager.set_state(plugin_id, PluginState.ENABLED)
_cycle_updates(manager, plugin_id, 50)
assert len(manager.get_state_history("clock")) == 101
assert len(manager.get_state_history("weather")) == 101
def test_get_state_history_returns_a_copy():
"""Callers must not be able to mutate the manager's internal history."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
history = manager.get_state_history("clock")
history.clear()
assert len(manager.get_state_history("clock")) == 1
def test_get_state_history_entries_are_copies():
"""Copying the outer list is not enough -- the entries are handed out too.
A caller holding a returned transition must not be able to rewrite the
manager's record of what happened.
"""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
entry = manager.get_state_history("clock")[0]
entry["to"] = "tampered"
entry["error"] = "injected"
stored = manager.get_state_history("clock")[0]
assert stored["to"] == PluginState.ENABLED.value
assert stored["error"] is None
def test_clear_state_drops_history():
"""Unloading a plugin still releases everything it accumulated."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
_cycle_updates(manager, "clock", 10)
manager.clear_state("clock")
assert manager.get_state_history("clock") == []
assert manager.get_state_info("clock")["state_history_count"] == 0
if __name__ == "__main__":
sys.exit(pytest.main([__file__, "-v"]))
-209
View File
@@ -1,209 +0,0 @@
"""Retention is bounded by age first and by count second.
The cap added in the parent change is a flat entry count, and an entry count
answers the wrong question. What a reader wants from this history is "the last
couple of hours"; how many transitions that is depends entirely on the
plugin's update interval, which on a real board spans 2s to 3600s. A flat 200
entries is 4.2 days of history for the slowest plugin and 3.3 minutes for the
fastest -- so the plugin churning hardest, the one actually worth looking at,
keeps the least.
Trimming by age makes the retained window comparable whatever the cadence, and
the count then serves only as a memory ceiling for pollers fast enough to
produce thousands of transitions inside that window.
"""
import time
import pytest
from src.plugin_system.plugin_state import (
PluginState,
PluginStateManager,
MAX_STATE_HISTORY_PER_PLUGIN,
STATE_HISTORY_MAX_AGE_SECONDS,
)
class FakeClock:
"""A monotonic clock the test drives, so no test has to sleep."""
def __init__(self):
self.t = 1000.0
def __call__(self):
return self.t
def advance(self, seconds):
self.t += seconds
@pytest.fixture
def clock(monkeypatch):
c = FakeClock()
monkeypatch.setattr("src.plugin_system.plugin_state.time.monotonic", c)
return c
def _cycle(manager, plugin_id, clock, interval, cycles):
"""One update cycle: RUNNING on reserve, ENABLED on finish."""
for _ in range(cycles):
manager.set_state(plugin_id, PluginState.RUNNING)
manager.set_state(plugin_id, PluginState.ENABLED)
clock.advance(interval)
def test_transitions_older_than_the_window_are_dropped(clock):
m = PluginStateManager()
_cycle(m, "clock", clock, interval=60, cycles=10)
assert len(m.get_state_history("clock")) == 20
# Nothing happens for longer than the window, then one more cycle.
clock.advance(STATE_HISTORY_MAX_AGE_SECONDS + 1)
_cycle(m, "clock", clock, interval=60, cycles=1)
assert len(m.get_state_history("clock")) == 2, (
"only the transitions inside the window should survive")
def test_every_plugin_keeps_the_same_WINDOW_not_the_same_COUNT(clock):
"""The point of the age policy, stated as the property that distinguishes it.
Run both plugins for three times the retention window. Under a flat count
cap the slow one would still be holding transitions from hours before the
window, because it never produces enough entries to evict them. Under the
age policy each plugin retains its own last two hours and no more --
different entry counts, same span of time.
"""
window = STATE_HISTORY_MAX_AGE_SECONDS
m = PluginStateManager()
_cycle(m, "slow", clock, interval=60, cycles=(3 * window) // 60)
slow = len(m.get_state_history("slow"))
# Assert the property directly rather than a derived count. The guarantee
# is about the SPAN of retained history, not its age against the current
# clock: trimming happens on append, so a plugin that has gone quiet keeps
# its last window until it writes again. That is intentional -- it is
# bounded either way, and a lazy trim costs nothing on the hot path.
stamps = [stamp for stamp, _ in m._state_history["slow"]]
assert stamps[-1] - stamps[0] <= window, (
f"retained history spans {stamps[-1] - stamps[0]:.0f}s, "
f"window is {window}s")
assert slow < 2 * ((3 * window) // 60), (
f"slow plugin kept {slow} entries -- three windows' worth was retained")
clock.t = 1000.0
_cycle(m, "fast", clock, interval=2, cycles=(3 * window) // 2)
fast = len(m.get_state_history("fast"))
# Different counts, and the fast poller keeps more of them -- under a flat
# count cap these would be equal and the fast one would cover minutes.
assert fast > slow, f"fast={fast} slow={slow}"
def test_the_count_ceiling_still_bounds_a_fast_poller(clock):
"""Age alone would let a 2s plugin hold 7,200 entries."""
m = PluginStateManager()
_cycle(m, "flights", clock, interval=2, cycles=STATE_HISTORY_MAX_AGE_SECONDS)
assert len(m.get_state_history("flights")) <= MAX_STATE_HISTORY_PER_PLUGIN
def test_a_burst_inside_the_window_is_capped_not_kept(clock):
"""Transitions with no time between them still cannot grow without bound."""
m = PluginStateManager()
for _ in range(MAX_STATE_HISTORY_PER_PLUGIN * 3):
m.set_state("flapping", PluginState.RUNNING) # clock never advances
assert len(m.get_state_history("flapping")) <= MAX_STATE_HISTORY_PER_PLUGIN
def test_ageing_out_does_not_disturb_the_lifetime_count(clock):
m = PluginStateManager()
_cycle(m, "clock", clock, interval=60, cycles=10)
clock.advance(STATE_HISTORY_MAX_AGE_SECONDS + 1)
_cycle(m, "clock", clock, interval=60, cycles=1)
assert len(m.get_state_history("clock")) == 2
assert m.get_state_info("clock")["state_history_count"] == 22, (
"the lifetime total must survive trimming, it is the flap signal")
def test_the_surviving_entries_are_the_recent_ones(clock):
m = PluginStateManager()
_cycle(m, "clock", clock, interval=60, cycles=5)
clock.advance(STATE_HISTORY_MAX_AGE_SECONDS + 1)
m.set_state("clock", PluginState.ERROR)
history = m.get_state_history("clock")
assert [h["to"] for h in history] == ["error"]
def test_a_monotonic_clock_is_used_not_the_wall_clock(clock):
"""A DST shift or NTP step must not flush the history.
The trim reads time.monotonic(); the human-readable datetime inside each
transition is for display only.
"""
m = PluginStateManager()
_cycle(m, "clock", clock, interval=60, cycles=3)
before = len(m.get_state_history("clock"))
import datetime as real_datetime
class ShiftedDatetime(real_datetime.datetime):
@classmethod
def now(cls, tz=None):
return real_datetime.datetime(1999, 1, 1) # clock jumps backwards
import src.plugin_system.plugin_state as ps
original = ps.datetime
ps.datetime = ShiftedDatetime
try:
m.set_state("clock", PluginState.ENABLED)
finally:
ps.datetime = original
assert len(m.get_state_history("clock")) == before + 1, (
"a wall-clock jump must not trim anything")
def test_get_state_info_is_a_consistent_snapshot():
"""An unload running concurrently must not be observed half-done.
Each field used to be read under its own lock, so clear_state() could
interleave: 'state' read before the removal, 'state_history_count' after,
handing a caller a plugin that is ENABLED with zero transitions. The whole
payload is now built in one critical section.
"""
import threading
m = PluginStateManager()
for _ in range(50):
m.set_state("clock", PluginState.RUNNING)
m.set_state("clock", PluginState.ENABLED)
inconsistent = []
stop = threading.Event()
def reader():
while not stop.is_set():
info = m.get_state_info("clock")
# Either fully present or fully cleared -- never a live state with
# a wiped count.
if info["state"] != PluginState.UNLOADED.value and \
info["state_history_count"] == 0:
inconsistent.append(info)
return
def clearer():
for _ in range(200):
for _ in range(20):
m.set_state("clock", PluginState.ENABLED)
m.clear_state("clock")
t = threading.Thread(target=reader, daemon=True)
t.start()
clearer()
stop.set()
t.join(timeout=5)
assert not inconsistent, f"observed a torn snapshot: {inconsistent[:1]}"
+119
View File
@@ -0,0 +1,119 @@
"""Plugin state transitions are counted, not stored.
`PluginStateManager` used to keep every transition in a per-plugin history on
the hot scheduling path (RUNNING on reserve, ENABLED on finish), but nothing
ever read the entries -- `get_state_info()` only reported how many there were.
It now keeps just that lifetime count, which `state_history_count` surfaces
through the web API.
"""
import os
import sys
import threading
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from src.plugin_system.plugin_state import ( # noqa: E402
PluginState,
PluginStateManager,
)
def _cycle_updates(manager, plugin_id, cycles):
"""Drive the real scheduling path: RUNNING on reserve, ENABLED on finish."""
for _ in range(cycles):
manager.set_state(plugin_id, PluginState.RUNNING)
manager.set_state(plugin_id, PluginState.ENABLED)
def test_state_history_count_reports_lifetime_total():
"""`get_state_info()['state_history_count']` counts every transition."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
cycles = 4000
_cycle_updates(manager, "clock", cycles)
info = manager.get_state_info("clock")
assert info["state_history_count"] == 1 + cycles * 2
def test_error_transitions_are_counted():
"""set_state_with_error() is a transition too."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
manager.set_state_with_error(
"clock", PluginState.ENABLED, {"reason": "update timeout"}
)
info = manager.get_state_info("clock")
assert info["state_history_count"] == 2
assert info["error_info"] == {"reason": "update timeout"}
def test_count_is_isolated_per_plugin():
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
_cycle_updates(manager, "clock", 50)
manager.set_state("weather", PluginState.ENABLED)
assert manager.get_state_info("clock")["state_history_count"] == 101
assert manager.get_state_info("weather")["state_history_count"] == 1
def test_clear_state_drops_the_count():
"""Unloading a plugin still releases everything it accumulated."""
manager = PluginStateManager()
manager.set_state("clock", PluginState.ENABLED)
_cycle_updates(manager, "clock", 10)
manager.clear_state("clock")
info = manager.get_state_info("clock")
assert info["state"] == PluginState.UNLOADED.value
assert info["state_history_count"] == 0
def test_get_state_info_is_a_consistent_snapshot():
"""An unload running concurrently must not be observed half-done.
Each field used to be read under its own lock, so clear_state() could
interleave: 'state' read before the removal, 'state_history_count' after,
handing a caller a plugin that is ENABLED with zero transitions. The whole
payload is now built in one critical section.
"""
m = PluginStateManager()
_cycle_updates(m, "clock", 50)
inconsistent = []
stop = threading.Event()
def reader():
while not stop.is_set():
info = m.get_state_info("clock")
# Either fully present or fully cleared -- never a live state with
# a wiped count.
if info["state"] != PluginState.UNLOADED.value and \
info["state_history_count"] == 0:
inconsistent.append(info)
return
def clearer():
for _ in range(200):
for _ in range(20):
m.set_state("clock", PluginState.ENABLED)
m.clear_state("clock")
t = threading.Thread(target=reader, daemon=True)
t.start()
clearer()
stop.set()
t.join(timeout=5)
assert not inconsistent, f"observed a torn snapshot: {inconsistent[:1]}"
if __name__ == "__main__":
sys.exit(pytest.main([__file__, "-v"]))
+1 -22
View File
@@ -23,17 +23,6 @@ class TestPluginManager:
assert pm.cache_manager == mock_cache_manager
assert pm.plugins == {}
def test_discover_plugins(self, test_plugin_manager):
"""Test plugin discovery."""
pm = test_plugin_manager
# Mock _scan_directory_for_plugins since we can't easily create real files in fixture
pm._scan_directory_for_plugins = MagicMock(return_value=["plugin1", "plugin2"])
# We need to call the real discover_plugins method, not the mock from the fixture
# But the fixture mocks the whole class instance.
# Let's create a real instance with mocked dependencies for this test
pass # Handled by separate test below
def test_load_plugin_success(self, mock_config_manager, mock_display_manager, mock_cache_manager):
"""Test successful plugin loading."""
with patch('src.plugin_system.plugin_manager.ensure_directory_permissions'), \
@@ -59,7 +48,7 @@ class TestPluginManager:
result = pm.load_plugin("test_plugin")
assert result is True
assert "test_plugin" in pm.plugin_modules
assert "test_plugin" in pm.plugins
# PluginManager sets state to ENABLED after successful load
assert pm.state_manager.get_state("test_plugin") == PluginState.ENABLED
@@ -136,16 +125,6 @@ class TestPluginManager:
assert pm.state_manager.get_state("test_plugin") == PluginState.ENABLED
class TestPluginLoader:
"""Test PluginLoader functionality."""
def test_dependency_check(self):
"""Test dependency checking logic."""
# Covered by test_plugin_loader.py's install_dependencies tests,
# which exercise requirements_has_real_deps/requirements_are_satisfied
# and the pip subprocess fallback.
class TestPluginExecutor:
"""Test PluginExecutor functionality."""
-11
View File
@@ -183,17 +183,6 @@ class TestNoConcurrentUpdate:
f"update() ran {plugin.max_concurrent}x concurrently on the "
"synchronous path")
def test_update_all_plugins_never_overlaps(self, pm):
plugin = OverlapDetectingPlugin(update_seconds=0.25)
_install(pm, plugin)
_widen_check_then_act_window(pm)
_hammer(pm.update_all_plugins, threads=8)
assert plugin.max_concurrent == 1, (
f"update() ran {plugin.max_concurrent}x concurrently via "
"update_all_plugins()")
def test_async_path_never_overlaps(self, pm):
plugin = OverlapDetectingPlugin(update_seconds=0.2)
plugin_id = _install(pm, plugin)
+9 -49
View File
@@ -1,10 +1,8 @@
"""
Tests for the caching and tombstone behaviors added to PluginStoreManager
to fix the plugin-list slowness and the uninstall-resurrection bugs.
Tests for the caching behaviors added to PluginStoreManager to fix the
plugin-list slowness and the uninstall-resurrection bugs.
Coverage targets:
- ``mark_recently_uninstalled`` / ``was_recently_uninstalled`` lifecycle and
TTL expiry.
- ``_get_local_git_info`` mtime-gated cache: ``git`` subprocesses only run
when ``.git/HEAD`` mtime changes.
- ``fetch_registry`` stale-cache fallback on network failure.
@@ -20,29 +18,6 @@ from unittest.mock import patch, MagicMock
from src.plugin_system.store_manager import PluginStoreManager
class TestUninstallTombstone(unittest.TestCase):
def setUp(self):
self._tmp = TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.sm = PluginStoreManager(plugins_dir=self._tmp.name)
def test_unmarked_plugin_is_not_recent(self):
self.assertFalse(self.sm.was_recently_uninstalled("foo"))
def test_marking_makes_it_recent(self):
self.sm.mark_recently_uninstalled("foo")
self.assertTrue(self.sm.was_recently_uninstalled("foo"))
def test_tombstone_expires_after_ttl(self):
self.sm._uninstall_tombstone_ttl = 0.05
self.sm.mark_recently_uninstalled("foo")
self.assertTrue(self.sm.was_recently_uninstalled("foo"))
time.sleep(0.1)
self.assertFalse(self.sm.was_recently_uninstalled("foo"))
# Expired entry should also be pruned from the dict.
self.assertNotIn("foo", self.sm._uninstall_tombstones)
class TestPersistentUninstallRegistry(unittest.TestCase):
"""Regression tests for the persistent uninstall registry that stops a
core `git pull` update from resurrecting built-in plugins the user
@@ -590,18 +565,13 @@ class TestStaleOnErrorFallbacks(unittest.TestCase):
class TestInstallUpdateUninstallInvariants(unittest.TestCase):
"""Regression guard: the caching and tombstone work added in this PR
must not break the install / update / uninstall code paths.
"""Regression guard: the caching work added in this PR must not break
the install / update / uninstall code paths.
Specifically:
- ``install_plugin`` bypasses commit/manifest caches via force_refresh,
so the 5→30 min TTL bump cannot cause users to install a stale commit.
- ``update_plugin`` does the same.
- The uninstall tombstone is only honored by the state reconciler, not
by explicit ``install_plugin`` calls — so a user can uninstall and
immediately reinstall from the store UI without the tombstone getting
in the way.
- ``was_recently_uninstalled`` is not touched by ``install_plugin``.
"""
def setUp(self):
@@ -649,10 +619,8 @@ class TestInstallUpdateUninstallInvariants(unittest.TestCase):
self.assertTrue(manifest_calls, "manifest fetch was not called")
self.assertTrue(manifest_calls[0][3], "force_refresh=True did not reach _fetch_manifest_from_github")
def test_install_plugin_is_not_blocked_by_tombstone(self):
"""A tombstone must only gate the reconciler, not explicit installs.
Uses a complete, valid manifest stub and a no-op dependency
def test_install_plugin_runs_to_completion(self):
"""Uses a complete, valid manifest stub and a no-op dependency
installer so ``install_plugin`` runs all the way through to a
True return. Anything less (e.g. swallowing exceptions) would
hide real regressions in the install path.
@@ -664,11 +632,6 @@ class TestInstallUpdateUninstallInvariants(unittest.TestCase):
}
self.sm.registry_cache_time = time.time()
# Mark it recently uninstalled (simulates a user who just clicked
# uninstall and then immediately clicked install again).
self.sm.mark_recently_uninstalled("bar")
self.assertTrue(self.sm.was_recently_uninstalled("bar"))
# Stub the heavy bits so install_plugin can run without network.
self.sm._get_github_repo_info = lambda url: {
"default_branch": "main", "stars": 0,
@@ -701,17 +664,14 @@ class TestInstallUpdateUninstallInvariants(unittest.TestCase):
self.sm._install_via_git = fake_install_via_git
# No exception-swallowing: if install_plugin fails for ANY reason
# unrelated to the tombstone, the test fails loudly.
# No exception-swallowing: if install_plugin fails for ANY reason,
# the test fails loudly.
result = self.sm.install_plugin("bar")
self.assertTrue(
result,
"install_plugin returned False — the tombstone should not gate "
"explicit installs and all other stubs should allow success.",
"install_plugin returned False — all stubs should allow success.",
)
# Tombstone survives install (harmless — nothing reads it for installed plugins).
self.assertTrue(self.sm.was_recently_uninstalled("bar"))
class TestRegistryStaleCacheFallback(unittest.TestCase):
+4 -2
View File
@@ -150,9 +150,11 @@ class EveryStampRecordsACompletion(unittest.TestCase):
if assigns_time:
stamps.append(node)
# The worker and synchronous paths share one stamp, in
# _execute_update_now's _finish().
self.assertGreaterEqual(
len(stamps), 2,
"expected the worker and inline success paths to stamp the time; "
len(stamps), 1,
"expected the update success path to stamp the time; "
"if this drops, the search below is looking at the wrong thing")
for stamp in stamps:
-329
View File
@@ -1,329 +0,0 @@
"""
Tests for src/common/utils.py
Covers all pure utility functions: normalize_team_abbreviation, format_time,
format_date, get_timezone, validate_dimensions, parse_team_abbreviation,
format_score, format_period, is_live_game, is_final_game, is_upcoming_game,
sanitize_filename, truncate_text, parse_boolean.
"""
import pytest
from datetime import datetime, timezone
import pytz
from src.common.utils import (
normalize_team_abbreviation,
format_time,
format_date,
get_timezone,
validate_dimensions,
parse_team_abbreviation,
format_score,
format_period,
is_live_game,
is_final_game,
is_upcoming_game,
sanitize_filename,
truncate_text,
parse_boolean,
)
# ---------------------------------------------------------------------------
# normalize_team_abbreviation
# ---------------------------------------------------------------------------
class TestNormalizeTeamAbbreviation:
def test_basic_uppercase(self):
assert normalize_team_abbreviation("lal") == "LAL"
def test_strips_spaces(self):
assert normalize_team_abbreviation(" KC ") == "KC"
def test_replaces_ampersand(self):
assert normalize_team_abbreviation("TA&M") == "TAANDM"
def test_removes_internal_spaces(self):
assert normalize_team_abbreviation("A B") == "AB"
def test_removes_hyphens(self):
assert normalize_team_abbreviation("A-B") == "AB"
def test_empty_string_returns_empty(self):
assert normalize_team_abbreviation("") == ""
def test_none_returns_empty(self):
assert normalize_team_abbreviation(None) == ""
# ---------------------------------------------------------------------------
# format_time / format_date
# ---------------------------------------------------------------------------
class TestFormatTime:
def _utc_dt(self, hour=20, minute=30):
return datetime(2024, 1, 15, hour, minute, 0, tzinfo=timezone.utc)
def test_formats_utc_to_utc(self):
dt = self._utc_dt(20, 30)
result = format_time(dt, timezone_str="UTC")
# 20:30 UTC → "8:30PM" (leading zero stripped)
assert "8:30PM" in result or "8:30 PM" in result or result != ""
def test_naive_datetime_treated_as_utc(self):
dt = datetime(2024, 1, 15, 12, 0, 0) # naive
result = format_time(dt, timezone_str="UTC")
assert result != ""
def test_invalid_timezone_returns_empty(self):
dt = self._utc_dt()
result = format_time(dt, timezone_str="Invalid/TZ")
assert result == ""
def test_eastern_timezone(self):
dt = self._utc_dt(20, 0) # 8 PM UTC = 3 PM ET
result = format_time(dt, timezone_str="America/New_York")
assert result != ""
class TestFormatDate:
def test_formats_date(self):
dt = datetime(2024, 6, 15, 18, 0, 0, tzinfo=timezone.utc)
result = format_date(dt, timezone_str="UTC")
assert "June" in result or "15" in result
def test_naive_datetime(self):
dt = datetime(2024, 3, 10, 12, 0, 0)
result = format_date(dt, timezone_str="UTC")
assert result != ""
def test_invalid_timezone_returns_empty(self):
dt = datetime(2024, 6, 15, 18, 0, 0, tzinfo=timezone.utc)
result = format_date(dt, timezone_str="BadZone/Here")
assert result == ""
# ---------------------------------------------------------------------------
# get_timezone
# ---------------------------------------------------------------------------
class TestGetTimezone:
def test_valid_timezone(self):
tz = get_timezone("America/New_York")
assert tz is not None
def test_utc(self):
tz = get_timezone("UTC")
assert tz is pytz.utc or str(tz) == "UTC"
def test_invalid_returns_utc(self):
tz = get_timezone("Not/ATimezone")
assert tz is pytz.utc
# ---------------------------------------------------------------------------
# validate_dimensions
# ---------------------------------------------------------------------------
class TestValidateDimensions:
def test_valid(self):
assert validate_dimensions(64, 32) is True
def test_zero_width(self):
assert validate_dimensions(0, 32) is False
def test_zero_height(self):
assert validate_dimensions(64, 0) is False
def test_negative(self):
assert validate_dimensions(-1, 32) is False
def test_too_large(self):
assert validate_dimensions(1001, 32) is False
def test_max_valid(self):
assert validate_dimensions(1000, 1000) is True
def test_non_integer(self):
assert validate_dimensions("64", 32) is False # type: ignore[arg-type]
# ---------------------------------------------------------------------------
# parse_team_abbreviation
# ---------------------------------------------------------------------------
class TestParseTeamAbbreviation:
def test_empty_string(self):
assert parse_team_abbreviation("") == ""
def test_none_returns_empty(self):
assert parse_team_abbreviation(None) == ""
def test_extracts_uppercase(self):
result = parse_team_abbreviation("LAL")
assert result == "LAL"
def test_fallback_first_three(self):
# text without recognisable 2-4 char uppercase block
result = parse_team_abbreviation("ab")
assert len(result) <= 3
# ---------------------------------------------------------------------------
# format_score
# ---------------------------------------------------------------------------
class TestFormatScore:
def test_format_score(self):
assert format_score(14, 7) == "7-14"
def test_format_score_strings(self):
assert format_score("21", "14") == "14-21"
def test_zero_zero(self):
assert format_score(0, 0) == "0-0"
# ---------------------------------------------------------------------------
# format_period
# ---------------------------------------------------------------------------
class TestFormatPeriod:
def test_basketball_q1(self):
assert format_period(1, "basketball") == "Q1"
def test_basketball_q4(self):
assert format_period(4, "basketball") == "Q4"
def test_basketball_ot1(self):
assert format_period(5, "basketball") == "OT1"
def test_basketball_ot2(self):
assert format_period(6, "basketball") == "OT2"
def test_football_q1(self):
assert format_period(1, "football") == "Q1"
def test_football_ot(self):
assert format_period(5, "football") == "OT1"
def test_hockey_p1(self):
assert format_period(1, "hockey") == "P1"
def test_hockey_p3(self):
assert format_period(3, "hockey") == "P3"
def test_hockey_ot(self):
assert format_period(4, "hockey") == "OT1"
def test_baseball_inning(self):
assert format_period(7, "baseball") == "INN 7"
def test_unknown_sport(self):
result = format_period(2, "unknown")
assert "2" in result
# ---------------------------------------------------------------------------
# is_live_game / is_final_game / is_upcoming_game
# ---------------------------------------------------------------------------
class TestGameStatusHelpers:
def test_is_live_game_true(self):
assert is_live_game("In Progress") is True
assert is_live_game("halftime") is True
assert is_live_game("overtime") is True
def test_is_live_game_false(self):
assert is_live_game("Final") is False
assert is_live_game("Scheduled") is False
def test_is_final_game_true(self):
assert is_final_game("Final") is True
assert is_final_game("COMPLETED") is True
def test_is_final_game_false(self):
assert is_final_game("In Progress") is False
def test_is_upcoming_game_true(self):
assert is_upcoming_game("Scheduled") is True
assert is_upcoming_game("upcoming") is True
def test_is_upcoming_game_false(self):
assert is_upcoming_game("Final") is False
assert is_upcoming_game("In Progress") is False
# ---------------------------------------------------------------------------
# sanitize_filename
# ---------------------------------------------------------------------------
class TestSanitizeFilename:
def test_removes_invalid_chars(self):
result = sanitize_filename('file<>:"/\\|?*.txt')
assert "<" not in result
assert ">" not in result
assert ":" not in result
def test_collapses_underscores(self):
result = sanitize_filename("file___name")
assert "__" not in result
def test_strips_leading_trailing(self):
result = sanitize_filename("_file_")
assert not result.startswith("_")
assert not result.endswith("_")
def test_normal_filename_unchanged(self):
result = sanitize_filename("my_logo")
assert result == "my_logo"
# ---------------------------------------------------------------------------
# truncate_text
# ---------------------------------------------------------------------------
class TestTruncateText:
def test_no_truncation_needed(self):
assert truncate_text("hello", 10) == "hello"
def test_truncation_adds_suffix(self):
result = truncate_text("hello world", 8)
assert result.endswith("...")
assert len(result) == 8
def test_exact_length(self):
assert truncate_text("hello", 5) == "hello"
def test_custom_suffix(self):
result = truncate_text("hello world", 8, suffix="~")
assert result.endswith("~")
# ---------------------------------------------------------------------------
# parse_boolean
# ---------------------------------------------------------------------------
class TestParseBoolean:
def test_true_bool(self):
assert parse_boolean(True) is True
def test_false_bool(self):
assert parse_boolean(False) is False
def test_int_1(self):
assert parse_boolean(1) is True
def test_int_0(self):
assert parse_boolean(0) is False
def test_string_true(self):
for val in ("true", "True", "TRUE", "1", "yes", "on", "enabled"):
assert parse_boolean(val) is True, f"Expected True for {val!r}"
def test_string_false(self):
for val in ("false", "False", "0", "no", "off", "disabled"):
assert parse_boolean(val) is False, f"Expected False for {val!r}"
def test_none_returns_false(self):
assert parse_boolean(None) is False # type: ignore[arg-type]
+1 -46
View File
@@ -2,7 +2,7 @@
Tests for src/vegas_mode/config.py
Covers VegasModeConfig: from_config, to_dict, get_frame_interval,
is_plugin_included, get_ordered_plugins, validate, update.
is_plugin_included, get_ordered_plugins, validate.
"""
import pytest
@@ -263,48 +263,3 @@ class TestValidate:
cfg = VegasModeConfig(scroll_speed=0.1, target_fps=5)
errors = cfg.validate()
assert len(errors) >= 2
# ---------------------------------------------------------------------------
# update
# ---------------------------------------------------------------------------
class TestUpdate:
def _wrap(self, **kwargs) -> dict:
return {"display": {"vegas_scroll": kwargs}}
def test_update_enabled(self):
cfg = VegasModeConfig(enabled=False)
cfg.update(self._wrap(enabled=True))
assert cfg.enabled is True
def test_update_scroll_speed(self):
cfg = VegasModeConfig(scroll_speed=50.0)
cfg.update(self._wrap(scroll_speed=90.0))
assert cfg.scroll_speed == 90.0
def test_update_separator_width(self):
cfg = VegasModeConfig(separator_width=32)
cfg.update(self._wrap(separator_width=8))
assert cfg.separator_width == 8
def test_update_plugin_order(self):
cfg = VegasModeConfig(plugin_order=[])
cfg.update(self._wrap(plugin_order=["x", "y"]))
assert cfg.plugin_order == ["x", "y"]
def test_update_excluded_plugins(self):
cfg = VegasModeConfig()
cfg.update(self._wrap(excluded_plugins=["skip_me"]))
assert "skip_me" in cfg.excluded_plugins
def test_update_ignores_missing_keys(self):
cfg = VegasModeConfig(scroll_speed=50.0)
cfg.update(self._wrap(target_fps=80)) # only fps, not speed
assert cfg.scroll_speed == 50.0
assert cfg.target_fps == 80
def test_empty_update_no_change(self):
cfg = VegasModeConfig(scroll_speed=50.0)
cfg.update({})
assert cfg.scroll_speed == 50.0
-5
View File
@@ -626,11 +626,6 @@ class TestConfigSurface:
assert restored.trim_threshold == 20
assert restored.lead_in_width == 64
def test_update_applies_new_keys(self):
cfg = VegasModeConfig()
cfg.update({'display': {'vegas_scroll': {'content_padding': 16}}})
assert cfg.content_padding == 16
@pytest.mark.parametrize('overrides,bad_key', [
({'trim_threshold': 300}, 'trim_threshold'),
({'trim_threshold': -1}, 'trim_threshold'),
+1 -2
View File
@@ -21,8 +21,7 @@ before tagging:
python scripts/check_release_version.py v3.2.0
Note: `src.plugin_system.__version__` is deliberately NOT checked. That module
versions the *plugin API* (it sits beside `__api_version__` and is documented as
such), which moves independently of the core version.
versions the *plugin API*, which moves independently of the core version.
"""
import re
+61 -4
View File
@@ -10,6 +10,7 @@ Scenarios covered:
3. iptables rules and ip_forward are reverted when the AP is torn down.
4. LED matrix message includes the SSID, 'No password', and the setup URL.
5. Known AP profile names are deleted before the new profile is created.
6. Wi-Fi passwords are not written to wifi_config.json, and old ones are scrubbed.
"""
from __future__ import annotations
@@ -62,7 +63,6 @@ def wifi_config(tmp_path: Path) -> Path:
"ap_ssid": "LEDMatrix-Setup",
"ap_channel": 7,
"auto_enable_ap_mode": True,
"saved_networks": [],
}
p = cfg_dir / "wifi_config.json"
p.write_text(json.dumps(cfg))
@@ -75,8 +75,7 @@ def manager(wifi_config: Path, tmp_path: Path) -> WiFiManager:
WiFiManager with all system calls stubbed out during construction and the
ip_forward save file redirected to a per-test temporary path.
"""
with patch("src.wifi_manager.subprocess.run", return_value=_ok(stdout="wlan0\n")), \
patch.object(WiFiManager, "_detect_trixie", return_value=False):
with patch("src.wifi_manager.subprocess.run", return_value=_ok(stdout="wlan0\n")):
mgr = WiFiManager(config_path=wifi_config)
# Force clean, deterministic state regardless of what __init__ inferred
@@ -85,7 +84,6 @@ def manager(wifi_config: Path, tmp_path: Path) -> WiFiManager:
mgr.has_hostapd = False
mgr.has_dnsmasq = False
mgr.has_iwlist = False
mgr._is_trixie = False
# Redirect the ip_forward save file to tmp so tests never share state
mgr._IP_FORWARD_SAVE_PATH = tmp_path / "ip_fwd_saved"
return mgr
@@ -332,3 +330,62 @@ def test_existing_ap_profiles_deleted_before_new_profile_created(manager: WiFiMa
assert del_indices, "Expected 'nmcli connection delete' calls"
assert max(del_indices) < min(add_indices), \
"All connection deletions must complete before the new profile is created"
# ---------------------------------------------------------------------------
# 6. Wi-Fi passwords are not kept in wifi_config.json
# ---------------------------------------------------------------------------
@pytest.mark.unit
def test_loading_scrubs_plaintext_saved_networks(wifi_config: Path) -> None:
"""Older versions wrote every joined network's password to the config in
plaintext and never read it back. Loading must remove it from disk."""
cfg = json.loads(wifi_config.read_text())
cfg["saved_networks"] = [
{"ssid": "HomeNet", "password": "hunter22", "saved_at": 0},
]
wifi_config.write_text(json.dumps(cfg))
with patch("src.wifi_manager.subprocess.run", return_value=_ok(stdout="wlan0\n")):
mgr = WiFiManager(config_path=wifi_config)
assert "saved_networks" not in mgr.config
assert "hunter22" not in wifi_config.read_text()
on_disk = json.loads(wifi_config.read_text())
assert "saved_networks" not in on_disk
# Everything else survives the scrub.
assert on_disk["ap_ssid"] == "LEDMatrix-Setup"
assert on_disk["auto_enable_ap_mode"] is True
@pytest.mark.unit
def test_default_config_has_no_saved_networks(tmp_path: Path) -> None:
config_path = tmp_path / "config" / "wifi_config.json"
config_path.parent.mkdir()
with patch("src.wifi_manager.subprocess.run", return_value=_ok(stdout="wlan0\n")):
WiFiManager(config_path=config_path)
assert "saved_networks" not in json.loads(config_path.read_text())
@pytest.mark.unit
def test_connecting_does_not_store_the_password(manager: WiFiManager) -> None:
commands = []
def fake_run(cmd, *args, **kwargs):
commands.append(cmd)
# No existing profile for the SSID, so a new connection is created.
if cmd[:3] == ["nmcli", "connection", "show"] and "HomeNet" in cmd:
return _fail()
return _ok(stdout="")
with patch("src.wifi_manager.subprocess.run", side_effect=fake_run), \
patch("src.wifi_manager.time.sleep"), \
patch.object(manager, "_show_led_message"):
manager._connect_nmcli("HomeNet", "hunter22")
assert ["nmcli", "device", "wifi", "connect", "HomeNet", "password", "hunter22"] in commands, \
"the new-connection path was not reached"
assert "hunter22" not in json.dumps(manager.config)
assert "hunter22" not in manager.config_path.read_text()
@@ -366,7 +366,6 @@ class TestStateReconciliationUnrecoverable(unittest.TestCase):
self.store_manager = Mock()
self.store_manager.fetch_registry.return_value = {"plugins": []}
self.store_manager.install_plugin.return_value = False
self.store_manager.was_recently_uninstalled.return_value = False
# A bare Mock() returns a truthy Mock for is_plugin_uninstalled(),
# which reads as "persistently uninstalled" and skips auto-repair
# entirely — these tests need the repair path to run.
@@ -441,9 +440,9 @@ class TestStateReconciliationUnrecoverable(unittest.TestCase):
self.assertNotIn("ghost", self.reconciler._unrecoverable_missing_on_disk)
self.store_manager.install_plugin.assert_not_called()
def test_recently_uninstalled_skips_auto_repair(self):
"""A freshly-uninstalled plugin must not be resurrected by the reconciler."""
self.store_manager.was_recently_uninstalled.return_value = True
def test_persistently_uninstalled_skips_auto_repair(self):
"""A plugin the user uninstalled must not be resurrected by the reconciler."""
self.store_manager.is_plugin_uninstalled.return_value = True
self.store_manager.fetch_registry.return_value = {
"plugins": [{"id": "ghost"}]
}