mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-01 16:58:06 +00:00
ChuckBuildsandClaude Sonnet 5
809d676d3b
The sports scoreboards read `getattr(self, 'global_config', {})` to find a
shared scroll frame rate, but nothing ever set that attribute: the loader
constructs plugins with only plugin_id/config/display_manager/cache_manager/
plugin_manager (plugin_loader.py:671), `global_config` appears nowhere in
src/, no plugin manager assigns it, and BasePlugin has no __getattr__ to
synthesize it. The lookup always returned {}, so the ten scroll_display.py
copies that thread target_fps through to ScrollHelper could never fire on any
core. There was also no global target_fps to find -- the only one in the
template is display.vegas_scroll.target_fps, which is Vegas-scoped.
Adds the missing half:
- `BasePlugin.global_config` resolves the full config via
plugin_manager.config_manager, then cache_manager.config_manager, then {}.
Same order the sports timezone helpers already use. Exceptions are swallowed
to debug so an unreadable config can never stop a plugin loading, and a
non-dict result is rejected rather than handed to callers that will .get()
it and feed the result to numeric code.
- A top-level `target_fps` (default 100), exposed on the General tab and
validated 30-200 on save to match ScrollHelper.set_target_fps -- which
clamps silently, so a rejected save reports a value that would otherwise
appear to save and then behave differently.
The property has a setter deliberately. news, stock-news, ledmatrix-stocks,
ledmatrix-elections, ledmatrix-leaderboard and nfl-draft all assign
`self.global_config = config.get('global', {})`; without a setter that raises
"property has no setter" and those six plugins stop loading. Reproduced, then
pinned with a test.
target_fps is also kept out of the `is_general_update` key list: that branch
treats a missing web_display_autostart as an unchecked box, so counting a
target_fps-only POST as a General save would silently switch autostart off.
Verified end to end: config.json -> BasePlugin.global_config ->
scroll_display's existing block -> ScrollHelper.target_fps 120 -> 100, with no
plugin-side change needed. Suite 1441 passed; the 4 failures
(test_display_dirty_tracking, test_web_api::test_get_system_status, two in
test_state_reconciliation) are pre-existing and reproduce identically on a
clean tree.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ
360 lines
16 KiB
Python
360 lines
16 KiB
Python
import time
|
|
from unittest.mock import MagicMock, patch
|
|
from pathlib import Path
|
|
from src.plugin_system.plugin_manager import PluginManager
|
|
from src.plugin_system.plugin_state import PluginState
|
|
from src.plugin_system.resource_monitor import PluginResourceMonitor
|
|
|
|
class TestPluginManager:
|
|
"""Test PluginManager functionality."""
|
|
|
|
def test_init(self, mock_config_manager, mock_display_manager, mock_cache_manager):
|
|
"""Test PluginManager initialization."""
|
|
with patch('src.plugin_system.plugin_manager.ensure_directory_permissions'):
|
|
pm = PluginManager(
|
|
plugins_dir="plugins",
|
|
config_manager=mock_config_manager,
|
|
display_manager=mock_display_manager,
|
|
cache_manager=mock_cache_manager
|
|
)
|
|
assert pm.plugins_dir == Path("plugins")
|
|
assert pm.config_manager == mock_config_manager
|
|
assert pm.display_manager == mock_display_manager
|
|
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'), \
|
|
patch('src.plugin_system.plugin_manager.PluginManager._scan_directory_for_plugins'), \
|
|
patch('src.plugin_system.plugin_manager.PluginLoader') as MockLoader, \
|
|
patch('src.plugin_system.plugin_manager.SchemaManager'):
|
|
|
|
pm = PluginManager(
|
|
plugins_dir="plugins",
|
|
config_manager=mock_config_manager,
|
|
display_manager=mock_display_manager,
|
|
cache_manager=mock_cache_manager
|
|
)
|
|
|
|
# Setup mocks
|
|
pm.plugin_manifests = {"test_plugin": {"id": "test_plugin", "name": "Test Plugin"}}
|
|
|
|
mock_loader = MockLoader.return_value
|
|
mock_loader.find_plugin_directory.return_value = Path("plugins/test_plugin")
|
|
mock_loader.load_plugin.return_value = (MagicMock(), MagicMock())
|
|
|
|
# Test loading
|
|
result = pm.load_plugin("test_plugin")
|
|
|
|
assert result is True
|
|
assert "test_plugin" in pm.plugin_modules
|
|
# PluginManager sets state to ENABLED after successful load
|
|
assert pm.state_manager.get_state("test_plugin") == PluginState.ENABLED
|
|
|
|
def test_load_plugin_missing_manifest(self, mock_config_manager, mock_display_manager, mock_cache_manager):
|
|
"""Test loading plugin with missing manifest."""
|
|
with patch('src.plugin_system.plugin_manager.ensure_directory_permissions'):
|
|
pm = PluginManager(
|
|
plugins_dir="plugins",
|
|
config_manager=mock_config_manager,
|
|
display_manager=mock_display_manager,
|
|
cache_manager=mock_cache_manager
|
|
)
|
|
|
|
# No manifest in pm.plugin_manifests
|
|
result = pm.load_plugin("non_existent_plugin")
|
|
|
|
assert result is False
|
|
assert pm.state_manager.get_state("non_existent_plugin") == PluginState.ERROR
|
|
|
|
def test_run_scheduled_updates_calls_update_with_resource_monitor(
|
|
self, mock_config_manager, mock_display_manager, mock_cache_manager
|
|
):
|
|
"""Regression test: run_scheduled_updates() must actually call a
|
|
plugin's update() when self.resource_monitor is set (as it is in
|
|
every real deployment -- display_controller.py and web_interface/
|
|
app.py both assign a real PluginResourceMonitor after construction).
|
|
|
|
Previously, the resource_monitor branch wrapped the call in a
|
|
function stored as a *class* attribute on a dynamically-built type
|
|
(`type('obj', (object,), {'update': monitored_update})()`), which
|
|
the descriptor protocol turns into a bound method on access --
|
|
silently passing the synthetic instance as an implicit first
|
|
argument to monitored_update(), which takes none. Every plugin's
|
|
scheduled update failed with "monitored_update() takes 0 positional
|
|
arguments but 1 was given" and was silently swallowed into a
|
|
circuit-breaker retry loop that never succeeded, so plugin data
|
|
(scores, odds, etc.) never refreshed.
|
|
"""
|
|
with patch('src.plugin_system.plugin_manager.ensure_directory_permissions'):
|
|
pm = PluginManager(
|
|
plugins_dir="plugins",
|
|
config_manager=mock_config_manager,
|
|
display_manager=mock_display_manager,
|
|
cache_manager=mock_cache_manager
|
|
)
|
|
|
|
plugin_instance = MagicMock()
|
|
plugin_instance.enabled = True
|
|
plugin_instance.update = MagicMock()
|
|
|
|
pm.plugins["test_plugin"] = plugin_instance
|
|
pm.plugin_manifests["test_plugin"] = {"update_interval": 10}
|
|
pm.state_manager.set_state("test_plugin", PluginState.ENABLED)
|
|
# Plain MagicMock, not the mock_cache_manager fixture: this test
|
|
# is about run_scheduled_updates() actually invoking update()
|
|
# through the resource-monitor wrapper, not about
|
|
# PluginResourceMonitor's own cache-backed metrics persistence
|
|
# (which calls cache_manager.get(..., memory_ttl=...) --
|
|
# a kwarg the fixture's mock_get() doesn't accept).
|
|
pm.resource_monitor = PluginResourceMonitor(MagicMock())
|
|
|
|
pm.run_scheduled_updates(current_time=time.time())
|
|
|
|
# Updates now execute on the background worker (the scheduler
|
|
# returns immediately) — wait for completion before asserting.
|
|
deadline = time.time() + 5
|
|
while (plugin_instance.update.call_count == 0
|
|
and time.time() < deadline):
|
|
time.sleep(0.02)
|
|
pm.stop_update_worker()
|
|
|
|
plugin_instance.update.assert_called_once()
|
|
assert "test_plugin" in pm.plugin_last_update
|
|
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."""
|
|
|
|
def test_execute_display_success(self):
|
|
"""Test successful display execution."""
|
|
from src.plugin_system.plugin_executor import PluginExecutor
|
|
executor = PluginExecutor()
|
|
|
|
mock_plugin = MagicMock()
|
|
mock_plugin.display.return_value = True
|
|
|
|
result = executor.execute_display(mock_plugin, "test_plugin")
|
|
|
|
assert result is True
|
|
mock_plugin.display.assert_called_once()
|
|
|
|
def test_execute_display_exception(self):
|
|
"""Test display execution with exception."""
|
|
from src.plugin_system.plugin_executor import PluginExecutor
|
|
executor = PluginExecutor()
|
|
|
|
mock_plugin = MagicMock()
|
|
mock_plugin.display.side_effect = Exception("Test error")
|
|
|
|
result = executor.execute_display(mock_plugin, "test_plugin")
|
|
|
|
assert result is False
|
|
|
|
def test_execute_update_timeout(self):
|
|
"""Test update execution timeout."""
|
|
# Using a very short timeout for testing
|
|
from src.plugin_system.plugin_executor import PluginExecutor
|
|
executor = PluginExecutor(default_timeout=0.01)
|
|
|
|
mock_plugin = MagicMock()
|
|
def slow_update():
|
|
time.sleep(0.05)
|
|
mock_plugin.update.side_effect = slow_update
|
|
|
|
result = executor.execute_update(mock_plugin, "test_plugin")
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestPluginHealth:
|
|
"""Test plugin health monitoring."""
|
|
|
|
def test_circuit_breaker(self, mock_cache_manager):
|
|
"""Test circuit breaker activation."""
|
|
from src.plugin_system.plugin_health import PluginHealthTracker
|
|
tracker = PluginHealthTracker(cache_manager=mock_cache_manager, failure_threshold=3, cooldown_period=60)
|
|
|
|
plugin_id = "test_plugin"
|
|
|
|
# Initial state
|
|
assert tracker.should_skip_plugin(plugin_id) is False
|
|
|
|
# Failures
|
|
tracker.record_failure(plugin_id, Exception("Error 1"))
|
|
assert tracker.should_skip_plugin(plugin_id) is False
|
|
|
|
tracker.record_failure(plugin_id, Exception("Error 2"))
|
|
assert tracker.should_skip_plugin(plugin_id) is False
|
|
|
|
tracker.record_failure(plugin_id, Exception("Error 3"))
|
|
# Should trip now
|
|
assert tracker.should_skip_plugin(plugin_id) is True
|
|
|
|
# Recovery (simulate timeout - need to update health state correctly)
|
|
if plugin_id in tracker._health_state:
|
|
tracker._health_state[plugin_id]["last_failure"] = time.time() - 61
|
|
tracker._health_state[plugin_id]["circuit_state"] = "closed"
|
|
assert tracker.should_skip_plugin(plugin_id) is False
|
|
|
|
|
|
class TestBasePlugin:
|
|
"""Test BasePlugin functionality."""
|
|
|
|
def test_dynamic_duration_defaults(self, mock_display_manager, mock_cache_manager):
|
|
"""Test default dynamic duration behavior."""
|
|
from src.plugin_system.base_plugin import BasePlugin
|
|
|
|
# Concrete implementation for testing
|
|
class ConcretePlugin(BasePlugin):
|
|
def update(self): pass
|
|
def display(self, force_clear=False): pass
|
|
|
|
config = {"enabled": True}
|
|
plugin = ConcretePlugin("test", config, mock_display_manager, mock_cache_manager, None)
|
|
|
|
assert plugin.supports_dynamic_duration() is False
|
|
assert plugin.get_dynamic_duration_cap() is None
|
|
assert plugin.is_cycle_complete() is True
|
|
|
|
def test_live_priority_config(self, mock_display_manager, mock_cache_manager):
|
|
"""Test live priority configuration."""
|
|
from src.plugin_system.base_plugin import BasePlugin
|
|
|
|
class ConcretePlugin(BasePlugin):
|
|
def update(self): pass
|
|
def display(self, force_clear=False): pass
|
|
|
|
config = {"enabled": True, "live_priority": True}
|
|
plugin = ConcretePlugin("test", config, mock_display_manager, mock_cache_manager, None)
|
|
|
|
assert plugin.has_live_priority() is True
|
|
|
|
|
|
class TestBasePluginGlobalConfig:
|
|
"""global_config exposes device-wide settings that self.config cannot.
|
|
|
|
The sports scoreboards read `getattr(self, 'global_config', {})` to find
|
|
the shared target_fps; before this property existed nothing ever set that
|
|
attribute, so the lookup silently returned {} and the setting could never
|
|
take effect on any core.
|
|
"""
|
|
|
|
@staticmethod
|
|
def _plugin(display_manager, cache_manager, plugin_manager=None):
|
|
from src.plugin_system.base_plugin import BasePlugin
|
|
|
|
class ConcretePlugin(BasePlugin):
|
|
def update(self): pass
|
|
def display(self, force_clear=False): pass
|
|
|
|
return ConcretePlugin(
|
|
"test", {"enabled": True}, display_manager, cache_manager, plugin_manager
|
|
)
|
|
|
|
@staticmethod
|
|
def _manager_with(config):
|
|
"""A stand-in manager exposing config_manager.get_config()."""
|
|
manager = MagicMock()
|
|
manager.config_manager.get_config.return_value = config
|
|
return manager
|
|
|
|
def test_reads_config_from_plugin_manager(self, mock_display_manager, mock_cache_manager):
|
|
plugin = self._plugin(
|
|
mock_display_manager, mock_cache_manager,
|
|
self._manager_with({"target_fps": 100}),
|
|
)
|
|
assert plugin.global_config["target_fps"] == 100
|
|
|
|
def test_falls_back_to_cache_manager(self, mock_display_manager):
|
|
# The core that hangs config_manager off the cache manager instead.
|
|
cache_manager = self._manager_with({"target_fps": 75})
|
|
plugin = self._plugin(mock_display_manager, cache_manager, plugin_manager=None)
|
|
assert plugin.global_config["target_fps"] == 75
|
|
|
|
def test_plugin_manager_wins_over_cache_manager(self, mock_display_manager):
|
|
plugin = self._plugin(
|
|
mock_display_manager,
|
|
self._manager_with({"target_fps": 75}),
|
|
self._manager_with({"target_fps": 100}),
|
|
)
|
|
assert plugin.global_config["target_fps"] == 100
|
|
|
|
def test_returns_empty_dict_when_no_config_manager(self, mock_display_manager):
|
|
# Plain objects: no config_manager attribute at all.
|
|
plugin = self._plugin(mock_display_manager, object(), object())
|
|
assert plugin.global_config == {}
|
|
|
|
def test_unreadable_config_does_not_raise(self, mock_display_manager):
|
|
# A plugin must still load when the config on disk is broken.
|
|
broken = MagicMock()
|
|
broken.config_manager.get_config.side_effect = OSError("unreadable")
|
|
plugin = self._plugin(mock_display_manager, broken, broken)
|
|
assert plugin.global_config == {}
|
|
|
|
def test_non_dict_config_is_rejected(self, mock_display_manager):
|
|
# A stub or half-built manager can return a non-mapping; handing that
|
|
# back would blow up later in numeric code, far from the cause.
|
|
plugin = self._plugin(
|
|
mock_display_manager, object(), self._manager_with("not-a-dict")
|
|
)
|
|
assert plugin.global_config == {}
|
|
|
|
def test_missing_property_degrades_to_default(self, mock_display_manager, mock_cache_manager):
|
|
# How plugins actually call it, so a plugin written against this core
|
|
# still loads on one that predates the property.
|
|
plugin = self._plugin(mock_display_manager, mock_cache_manager, object())
|
|
assert getattr(plugin, "global_config", {}).get("target_fps") is None
|
|
|
|
def test_plugin_may_still_assign_global_config(self, mock_display_manager, mock_cache_manager):
|
|
# news, stock-news, ledmatrix-stocks, ledmatrix-elections,
|
|
# ledmatrix-leaderboard and nfl-draft all do exactly this. Without a
|
|
# setter the property raises "has no setter" and those plugins stop
|
|
# loading entirely.
|
|
from src.plugin_system.base_plugin import BasePlugin
|
|
|
|
class AssigningPlugin(BasePlugin):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.global_config = self.config.get("global", {})
|
|
|
|
def update(self): pass
|
|
def display(self, force_clear=False): pass
|
|
|
|
plugin = AssigningPlugin(
|
|
"news", {"enabled": True, "global": {"scroll_speed": 2}},
|
|
mock_display_manager, mock_cache_manager, self._manager_with({"target_fps": 100}),
|
|
)
|
|
# The plugin's own value wins over the resolved config.
|
|
assert plugin.global_config == {"scroll_speed": 2}
|
|
|
|
def test_template_ships_a_global_target_fps(self):
|
|
# The plumbing is useless if the setting isn't in the shipped config.
|
|
import json
|
|
with open("config/config.template.json") as fh:
|
|
template = json.load(fh)
|
|
assert isinstance(template.get("target_fps"), int)
|