mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-06-02 08:53:31 +00:00
eedf680a8c
* docs(core): add module and class docstrings to the 5 undocumented core files
Fills the only significant documentation gaps found during a codebase
audit. All other core files (plugin_system/, logging_config.py, etc.)
already have complete module, class, and function docstrings.
Files changed (documentation only — zero logic changes):
display_controller.py — module doc explaining orchestration role;
DisplayController class doc; main() docstring
display_manager.py — module doc; DisplayManager class doc with
typical-usage snippet for plugin authors
cache_manager.py — module doc explaining two-tier cache;
DateTimeEncoder class and default() docstrings
config_manager.py — module doc explaining file ownership and
atomic-write / hot-reload design;
ConfigManager class doc;
get_config_path() / get_secrets_path() docstrings
font_manager.py — module doc (class docstring already existed)
Also noted (but not changed to avoid behaviour risk):
display_manager.py and font_manager.py use logging.getLogger() directly
instead of the project's get_logger() wrapper. display_manager.py also
calls setLevel(logging.INFO) immediately after, which would be lost if
switched to get_logger().
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* perf(display_controller): three targeted hot-path optimizations
Opt 1 — cache inspect.signature() per plugin_id
inspect.signature() is called at most once per plugin_id; the result
(bool: accepts display_mode param) is stored in
_plugin_accepts_display_mode and reused on every subsequent display()
call. Eliminates all reflection from the display path at runtime.
Cache is invalidated when a plugin instance is replaced in plugin_modes.
Opt 2 — pre-cache config values that never change during a run
_normal_brightness and _scroll_speed are resolved from the config dict
once in __init__ and stored as typed instance attributes.
- Removes 2+ chained dict.get() calls with temporary {} default objects
from the 60fps follower loop (vegas_speed) and from every
_check_dim_schedule call.
- current_brightness init now uses _normal_brightness directly.
Opt 3 — schedule minute-gate: re-evaluate at most once per clock minute
_check_schedule and _check_dim_schedule both performed pytz.timezone(),
datetime.now(), strftime(), and datetime.strptime() on every outer loop
call. Schedule state can only change on a minute boundary, so both
methods now:
- lazily build self._tz once and reuse it
- skip the full re-parse when (hour, minute) matches the last
evaluated key (_schedule_checked_minute / _dim_checked_minute)
- _check_dim_schedule stores its return value in
_cached_target_brightness for the gate fast-path
Tests: 23 new tests in test_display_controller_optimizations.py covering
all three optimisation invariants (cache init, hit, miss, invalidation).
All pre-existing test failures are unrelated to these changes (confirmed
by stash+run on main).
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* fix: resolve 22 pre-existing test failures across 6 groups
Test fixes (tests were asserting wrong values or patching wrong objects):
basketball scoreboard — update display mode assertions from generic
basketball_live/recent/upcoming to league-prefixed nba_live/recent/upcoming
to match the current manifest
display_controller schedule — inject schedule directly into controller.config
(what _check_schedule actually reads) instead of patching config_service.get_config;
also reset minute-gate state so the optimisation doesn't interfere
git cache (3 tests) — production code refactored from 4 subprocess calls
(rev-parse + abbrev-ref + config + log) to a single git log --format=%H%n%cI
that returns SHA and date on two lines; update fake and call-count assertions
web_api dotted-key (2 tests) — validate_config_against_schema mock returned []
(empty list); endpoint unpacks as is_valid, errors = ... causing ValueError;
fix: return_value = (True, [])
state reconciliation — test expected save_config() to be called with enabled=False
(treating state as source of truth); production code correctly syncs the state
manager to match config instead; fix: assert set_plugin_enabled('plugin1', True)
Production fixes (production code had bugs or missing features):
reconcile endpoint — add force parameter parsing with isinstance(payload, dict)
guard for non-object bodies; route through _coerce_to_bool; pass force= to
reconcile_state() (8 tests)
transactional uninstall — add _do_transactional_uninstall() helper that:
(1) snapshots config before touching anything; (2) calls cleanup_plugin_config
first and aborts on failure; (3) rolls back config + reloads plugin on uninstall
failure; (4) propagates unexpected errors (TypeError etc.) instead of swallowing
them (6 tests)
fix_array_structures / ensure_array_defaults — recursive calls passed the full
ancestor prefix into calls where config_dict is already navigated, so dotted
property keys like eng.1 caused parent_parts.split('.') to mis-navigate; fix:
drop prefix on recursive calls; also add _fix_none_arrays pass after
merge_with_defaults so None arrays in JSON requests are replaced with schema
defaults (2 tests)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* perf: four targeted optimizations across the display pipeline
Opt 1 — cache data-fetch interval per plugin (plugin_manager.py)
_get_plugin_update_interval fell back to config_manager.get_config()
(a full dict copy) when the manifest lacked an interval. Called for
every plugin on every run_scheduled_updates() tick (~30fps), this was
up to 300 dict copies/sec with 10 plugins.
Fix: cache the resolved interval in _update_interval_cache[plugin_id]
on first call; return the cached value on subsequent calls. Cache is
cleared on load_plugin and unload_plugin.
Opt 2 — demote noisy per-cycle INFO logs to DEBUG (display_controller.py)
Four logger.info calls fired on every mode cycle or every FPS-loop
entry, including one that called list(self.plugin_modes.keys())
unconditionally (allocating a list every outer loop iteration).
- "Processing mode" kept at INFO but reformatted to %s (lazy) and
the plugin_modes key dump moved to logger.debug
- "Attempting/Got cycle duration" → logger.debug
- "Entering high/normal FPS loop" → logger.debug
Mode name at INFO is preserved for black-screen troubleshooting.
Opt 3 — use Image.frombytes instead of Image.fromarray in scroll hot path
(scroll_helper.py)
Image.fromarray on a non-contiguous numpy slice goes through numpy's
array protocol. Image.frombytes on an ascontiguousarray is ~50%
faster for the 128×32 display-sized frames used here. Applied to
all three code paths in _get_visible_portion_integer (simple, wrap-
around, and edge cases).
Opt 5 — cache get_text_width per (text, font) pair (display_manager.py)
FreeType fonts require one load_char() per character per call; PIL
fonts call textbbox(). Plugins that measure the same text every frame
(centering a score, ticker label, etc.) were re-measuring from scratch
on every display() call.
Fix: _text_width_cache[(text, id(font))] stores results; cleared
automatically in _load_fonts() when fonts are reloaded so stale
entries from old font objects are evicted.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* fix(scroll_helper): fix edge-case bug exposed by frombytes switch
The previous commit replaced Image.fromarray with Image.frombytes in
_get_visible_portion_integer. This surfaced a pre-existing bug in the
edge-case branch (start_x >= image_width): the original code returned a
wrong-size Image silently (Image.fromarray accepts a too-short array);
Image.frombytes raises ValueError instead.
Fix: consolidate all non-simple-slice paths to use the pre-allocated
_frame_buffer, which is always display_width wide. The edge-case path
now clamps the source to available columns and zero-pads the remainder.
Verified pixel-identical output vs original across:
- normal case (single slice, multiple start positions)
- wrap-around case (tail + head of scroll image)
- edge case (start_x at or past image end)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* fix: address CodeRabbit review comments on PR #358
1. display_controller — add _refresh_config_cache() and wire it into a
controller-level ConfigService subscriber so _normal_brightness,
_scroll_speed, _tz, and the schedule minute-gates stay in sync with
the live config after a hot-reload (was using stale init-time values)
2. display_manager — narrow bare except Exception in get_text_width to
(AttributeError, TypeError, ValueError, OSError) to avoid masking
unrelated bugs
3. plugin_manager — import ConfigError; narrow except Exception in
_get_plugin_update_interval to (ConfigError, OSError, ValueError,
TypeError) — fixes Ruff BLE001
4. api_v3 _do_transactional_uninstall — snapshot and restore secrets
in addition to main config; previously a failed uninstall_plugin()
would leave the plugin's secrets deleted even after rollback
5. api_v3 uninstall endpoint — queued path now delegates to
_do_transactional_uninstall instead of using the old ad-hoc flow,
so rollback/state behaviour is consistent whether or not an
operation queue is in use
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* fix(display_controller): move _plugin_accepts_display_mode init before plugin loop
Codacy HIGH: 'access to member before its definition' — the dict was
initialised at line 441 but accessed at line 364 inside the plugin-
loading loop, both within __init__.
Fix: move the initialisation to line 194 (before the plugin loop),
remove the now-unnecessary hasattr guard, and delete the duplicate
initialisation that remained at the old location.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---------
Co-authored-by: Chuck <chuck@example.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
709 lines
25 KiB
Python
709 lines
25 KiB
Python
"""
|
|
Tests for Web Interface API endpoints.
|
|
|
|
Tests Flask routes, request/response handling, and API functionality.
|
|
"""
|
|
|
|
import pytest
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
# Add project root to path
|
|
project_root = Path(__file__).parent.parent
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
from flask import Flask
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_config_manager():
|
|
"""Create a mock config manager."""
|
|
mock = MagicMock()
|
|
mock.load_config.return_value = {
|
|
'display': {'brightness': 50},
|
|
'plugins': {},
|
|
'timezone': 'UTC'
|
|
}
|
|
mock.get_config_path.return_value = 'config/config.json'
|
|
mock.get_secrets_path.return_value = 'config/config_secrets.json'
|
|
mock_config = {
|
|
'display': {'brightness': 50},
|
|
'plugins': {},
|
|
'timezone': 'UTC'
|
|
}
|
|
mock.load_config.return_value = mock_config
|
|
mock.get_raw_file_content.return_value = mock_config
|
|
mock.save_config_atomic.return_value = MagicMock(
|
|
status=MagicMock(value='success'),
|
|
message=None
|
|
)
|
|
return mock
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_plugin_manager():
|
|
"""Create a mock plugin manager."""
|
|
mock = MagicMock()
|
|
mock.plugins = {}
|
|
mock.discover_plugins.return_value = []
|
|
mock.health_tracker = MagicMock()
|
|
mock.health_tracker.get_health_status.return_value = {'healthy': True}
|
|
return mock
|
|
|
|
|
|
@pytest.fixture
|
|
def client(mock_config_manager, mock_plugin_manager):
|
|
"""Create a Flask test client with mocked dependencies."""
|
|
# Create a minimal Flask app for testing
|
|
test_app = Flask(__name__)
|
|
test_app.config['TESTING'] = True
|
|
test_app.config['SECRET_KEY'] = 'test-secret-key'
|
|
|
|
# Register the API blueprint
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
|
|
# Mock the managers on the blueprint
|
|
api_v3.config_manager = mock_config_manager
|
|
api_v3.plugin_manager = mock_plugin_manager
|
|
api_v3.plugin_store_manager = MagicMock()
|
|
api_v3.saved_repositories_manager = MagicMock()
|
|
api_v3.schema_manager = MagicMock()
|
|
api_v3.operation_queue = MagicMock()
|
|
api_v3.plugin_state_manager = MagicMock()
|
|
api_v3.operation_history = MagicMock()
|
|
api_v3.cache_manager = MagicMock()
|
|
|
|
# Setup operation queue mocks
|
|
mock_operation = MagicMock()
|
|
mock_operation.operation_id = 'test-op-123'
|
|
mock_operation.status = MagicMock(value='pending')
|
|
api_v3.operation_queue.get_operation_status.return_value = mock_operation
|
|
api_v3.operation_queue.get_recent_operations.return_value = []
|
|
|
|
# Setup schema manager mocks
|
|
api_v3.schema_manager.load_schema.return_value = {
|
|
'type': 'object',
|
|
'properties': {'enabled': {'type': 'boolean'}}
|
|
}
|
|
|
|
# Setup state manager mocks
|
|
api_v3.plugin_state_manager.get_all_states.return_value = {}
|
|
|
|
test_app.register_blueprint(api_v3, url_prefix='/api/v3')
|
|
|
|
with test_app.test_client() as client:
|
|
yield client
|
|
|
|
|
|
class TestConfigAPI:
|
|
"""Test configuration API endpoints."""
|
|
|
|
def test_get_main_config(self, client, mock_config_manager):
|
|
"""Test getting main configuration."""
|
|
response = client.get('/api/v3/config/main')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert data.get('status') == 'success'
|
|
assert 'data' in data
|
|
assert 'display' in data['data']
|
|
mock_config_manager.load_config.assert_called_once()
|
|
|
|
def test_save_main_config(self, client, mock_config_manager):
|
|
"""Test saving main configuration."""
|
|
new_config = {
|
|
'display': {'brightness': 75},
|
|
'timezone': 'UTC'
|
|
}
|
|
|
|
response = client.post(
|
|
'/api/v3/config/main',
|
|
data=json.dumps(new_config),
|
|
content_type='application/json'
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
mock_config_manager.save_config_atomic.assert_called_once()
|
|
|
|
def test_save_main_config_validation_error(self, client, mock_config_manager):
|
|
"""Test saving config with validation error."""
|
|
invalid_config = {'invalid': 'data'}
|
|
|
|
mock_config_manager.save_config_atomic.return_value = MagicMock(
|
|
status=MagicMock(value='validation_failed'),
|
|
message='Validation error'
|
|
)
|
|
|
|
response = client.post(
|
|
'/api/v3/config/main',
|
|
data=json.dumps(invalid_config),
|
|
content_type='application/json'
|
|
)
|
|
|
|
assert response.status_code in [400, 500]
|
|
|
|
def test_get_secrets_config(self, client, mock_config_manager):
|
|
"""Test getting secrets configuration."""
|
|
response = client.get('/api/v3/config/secrets')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert 'weather' in data or 'data' in data
|
|
mock_config_manager.get_raw_file_content.assert_called_once()
|
|
|
|
def test_save_schedule_config(self, client, mock_config_manager):
|
|
"""Test saving schedule configuration."""
|
|
schedule_config = {
|
|
'enabled': True,
|
|
'start_time': '07:00',
|
|
'end_time': '23:00',
|
|
'mode': 'global'
|
|
}
|
|
|
|
response = client.post(
|
|
'/api/v3/config/schedule',
|
|
data=json.dumps(schedule_config),
|
|
content_type='application/json'
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
mock_config_manager.save_config_atomic.assert_called_once()
|
|
|
|
|
|
class TestSystemAPI:
|
|
"""Test system API endpoints."""
|
|
|
|
@patch('web_interface.blueprints.api_v3.subprocess')
|
|
def test_get_system_status(self, mock_subprocess, client):
|
|
"""Test getting system status."""
|
|
mock_result = MagicMock()
|
|
mock_result.stdout = 'active\n'
|
|
mock_result.returncode = 0
|
|
mock_subprocess.run.return_value = mock_result
|
|
|
|
response = client.get('/api/v3/system/status')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert 'service' in data or 'status' in data or 'active' in data
|
|
|
|
@patch('web_interface.blueprints.api_v3.subprocess')
|
|
def test_get_system_version(self, mock_subprocess, client):
|
|
"""Test getting system version."""
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
mock_result.stdout = 'v1.0.0\n'
|
|
mock_subprocess.run.return_value = mock_result
|
|
|
|
response = client.get('/api/v3/system/version')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert 'version' in data.get('data', {}) or 'version' in data
|
|
|
|
@patch('web_interface.blueprints.api_v3.subprocess')
|
|
def test_execute_system_action(self, mock_subprocess, client):
|
|
"""Test executing system action."""
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
mock_result.stdout = 'success'
|
|
mock_subprocess.run.return_value = mock_result
|
|
|
|
action_data = {
|
|
'action': 'restart',
|
|
'service': 'ledmatrix'
|
|
}
|
|
|
|
response = client.post(
|
|
'/api/v3/system/action',
|
|
data=json.dumps(action_data),
|
|
content_type='application/json'
|
|
)
|
|
|
|
# May return 400 if action validation fails, or 200 if successful
|
|
assert response.status_code in [200, 400]
|
|
|
|
|
|
class TestDisplayAPI:
|
|
"""Test display API endpoints."""
|
|
|
|
def test_get_display_current(self, client):
|
|
"""Test getting current display information."""
|
|
# Mock cache manager on the blueprint
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
api_v3.cache_manager.get.return_value = {
|
|
'mode': 'weather',
|
|
'plugin_id': 'weather'
|
|
}
|
|
|
|
response = client.get('/api/v3/display/current')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert 'mode' in data or 'current' in data or 'data' in data
|
|
|
|
def test_get_on_demand_status(self, client):
|
|
"""Test getting on-demand display status."""
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
api_v3.cache_manager.get.return_value = {
|
|
'active': False,
|
|
'mode': None
|
|
}
|
|
|
|
response = client.get('/api/v3/display/on-demand/status')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert 'active' in data or 'status' in data or 'data' in data
|
|
|
|
def test_start_on_demand_display(self, client):
|
|
"""Test starting on-demand display."""
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
|
|
request_data = {
|
|
'plugin_id': 'weather',
|
|
'mode': 'weather_current',
|
|
'duration': 30
|
|
}
|
|
|
|
# Ensure cache manager is set up
|
|
if not hasattr(api_v3, 'cache_manager') or api_v3.cache_manager is None:
|
|
api_v3.cache_manager = MagicMock()
|
|
|
|
response = client.post(
|
|
'/api/v3/display/on-demand/start',
|
|
data=json.dumps(request_data),
|
|
content_type='application/json'
|
|
)
|
|
|
|
# May return 404 if plugin not found, 200 if successful, or 500 on error
|
|
assert response.status_code in [200, 201, 404, 500]
|
|
# Verify cache was updated if successful
|
|
if response.status_code in [200, 201]:
|
|
assert api_v3.cache_manager.set.called
|
|
|
|
@patch('web_interface.blueprints.api_v3._ensure_cache_manager')
|
|
def test_stop_on_demand_display(self, mock_ensure_cache, client):
|
|
"""Test stopping on-demand display."""
|
|
|
|
# Mock the cache manager returned by _ensure_cache_manager
|
|
mock_cache_manager = MagicMock()
|
|
mock_ensure_cache.return_value = mock_cache_manager
|
|
|
|
response = client.post('/api/v3/display/on-demand/stop')
|
|
|
|
# May return 200 if successful or 500 on error
|
|
assert response.status_code in [200, 500]
|
|
# Verify stop request was set in cache if successful
|
|
if response.status_code == 200:
|
|
assert mock_cache_manager.set.called
|
|
|
|
|
|
class TestPluginsAPI:
|
|
"""Test plugins API endpoints."""
|
|
|
|
def test_get_installed_plugins(self, client, mock_plugin_manager):
|
|
"""Test getting list of installed plugins."""
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
api_v3.plugin_manager = mock_plugin_manager
|
|
|
|
mock_plugin_manager.plugins = {
|
|
'weather': MagicMock(plugin_id='weather'),
|
|
'clock': MagicMock(plugin_id='clock')
|
|
}
|
|
mock_plugin_manager.get_plugin_metadata.return_value = {
|
|
'id': 'weather',
|
|
'name': 'Weather Plugin'
|
|
}
|
|
|
|
response = client.get('/api/v3/plugins/installed')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert isinstance(data, (list, dict))
|
|
|
|
def test_get_plugin_health(self, client, mock_plugin_manager):
|
|
"""Test getting plugin health information."""
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
api_v3.plugin_manager = mock_plugin_manager
|
|
|
|
# Setup health tracker
|
|
mock_health_tracker = MagicMock()
|
|
mock_health_tracker.get_all_health_summaries.return_value = {
|
|
'weather': {'healthy': True}
|
|
}
|
|
mock_plugin_manager.health_tracker = mock_health_tracker
|
|
|
|
response = client.get('/api/v3/plugins/health')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert isinstance(data, (list, dict))
|
|
|
|
def test_get_plugin_health_single(self, client, mock_plugin_manager):
|
|
"""Test getting health for single plugin."""
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
api_v3.plugin_manager = mock_plugin_manager
|
|
|
|
# Setup health tracker with proper method (endpoint calls get_health_summary)
|
|
mock_health_tracker = MagicMock()
|
|
mock_health_tracker.get_health_summary.return_value = {
|
|
'healthy': True,
|
|
'failures': 0,
|
|
'last_success': '2024-01-01T00:00:00'
|
|
}
|
|
mock_plugin_manager.health_tracker = mock_health_tracker
|
|
|
|
response = client.get('/api/v3/plugins/health/weather')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert 'healthy' in data.get('data', {}) or 'data' in data
|
|
|
|
def test_toggle_plugin(self, client, mock_config_manager, mock_plugin_manager):
|
|
"""Test toggling plugin enabled state."""
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
api_v3.config_manager = mock_config_manager
|
|
api_v3.plugin_manager = mock_plugin_manager
|
|
api_v3.plugin_state_manager = MagicMock()
|
|
api_v3.operation_history = MagicMock()
|
|
|
|
# Setup plugin manifests
|
|
mock_plugin_manager.plugin_manifests = {'weather': {}}
|
|
|
|
request_data = {
|
|
'plugin_id': 'weather',
|
|
'enabled': True
|
|
}
|
|
|
|
response = client.post(
|
|
'/api/v3/plugins/toggle',
|
|
data=json.dumps(request_data),
|
|
content_type='application/json'
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
mock_config_manager.save_config_atomic.assert_called_once()
|
|
|
|
def test_get_plugin_config(self, client, mock_config_manager):
|
|
"""Test getting plugin configuration."""
|
|
# Plugin configs live at top-level keys (not under 'plugins')
|
|
mock_config_manager.load_config.return_value = {
|
|
'weather': {
|
|
'enabled': True,
|
|
'api_key': 'test_key'
|
|
}
|
|
}
|
|
|
|
# Ensure schema manager returns serializable values
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
api_v3.schema_manager.generate_default_config.return_value = {'enabled': False}
|
|
api_v3.schema_manager.merge_with_defaults.side_effect = lambda config, defaults: {**defaults, **config}
|
|
|
|
response = client.get('/api/v3/plugins/config?plugin_id=weather')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert 'enabled' in data or 'config' in data or 'data' in data
|
|
|
|
def test_save_plugin_config(self, client, mock_config_manager):
|
|
"""Test saving plugin configuration."""
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
api_v3.config_manager = mock_config_manager
|
|
api_v3.schema_manager = MagicMock()
|
|
api_v3.schema_manager.load_schema.return_value = {
|
|
'type': 'object',
|
|
'properties': {'enabled': {'type': 'boolean'}}
|
|
}
|
|
|
|
request_data = {
|
|
'plugin_id': 'weather',
|
|
'config': {
|
|
'enabled': True,
|
|
'update_interval': 300
|
|
}
|
|
}
|
|
|
|
response = client.post(
|
|
'/api/v3/plugins/config',
|
|
data=json.dumps(request_data),
|
|
content_type='application/json'
|
|
)
|
|
|
|
assert response.status_code in [200, 500] # May fail if validation fails
|
|
if response.status_code == 200:
|
|
mock_config_manager.save_config_atomic.assert_called_once()
|
|
|
|
def test_get_plugin_schema(self, client):
|
|
"""Test getting plugin configuration schema."""
|
|
|
|
response = client.get('/api/v3/plugins/schema?plugin_id=weather')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert 'type' in data or 'schema' in data or 'data' in data
|
|
|
|
def test_get_operation_status(self, client):
|
|
"""Test getting plugin operation status."""
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
|
|
# Setup operation queue mock
|
|
mock_operation = MagicMock()
|
|
mock_operation.operation_id = 'test-op-123'
|
|
mock_operation.status = MagicMock(value='pending')
|
|
mock_operation.operation_type = MagicMock(value='install')
|
|
mock_operation.plugin_id = 'test-plugin'
|
|
mock_operation.created_at = '2024-01-01T00:00:00'
|
|
# Add to_dict method that the endpoint calls
|
|
mock_operation.to_dict.return_value = {
|
|
'operation_id': 'test-op-123',
|
|
'status': 'pending',
|
|
'operation_type': 'install',
|
|
'plugin_id': 'test-plugin'
|
|
}
|
|
|
|
api_v3.operation_queue.get_operation_status.return_value = mock_operation
|
|
|
|
response = client.get('/api/v3/plugins/operation/test-op-123')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert 'status' in data or 'operation' in data or 'data' in data
|
|
|
|
def test_get_operation_history(self, client):
|
|
"""Test getting operation history."""
|
|
|
|
response = client.get('/api/v3/plugins/operation/history')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert isinstance(data, (list, dict))
|
|
|
|
def test_get_plugin_state(self, client):
|
|
"""Test getting plugin state."""
|
|
|
|
response = client.get('/api/v3/plugins/state')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert isinstance(data, (list, dict))
|
|
|
|
|
|
class TestFontsAPI:
|
|
"""Test fonts API endpoints."""
|
|
|
|
def test_get_fonts_catalog(self, client):
|
|
"""Test getting fonts catalog."""
|
|
# Fonts endpoints don't use FontManager, they return hardcoded data
|
|
response = client.get('/api/v3/fonts/catalog')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert 'catalog' in data.get('data', {}) or 'data' in data
|
|
|
|
def test_get_font_tokens(self, client):
|
|
"""Test getting font tokens."""
|
|
response = client.get('/api/v3/fonts/tokens')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert 'tokens' in data.get('data', {}) or 'data' in data
|
|
|
|
def test_get_fonts_overrides(self, client):
|
|
"""Test getting font overrides."""
|
|
response = client.get('/api/v3/fonts/overrides')
|
|
|
|
assert response.status_code == 200
|
|
data = json.loads(response.data)
|
|
assert 'overrides' in data.get('data', {}) or 'data' in data
|
|
|
|
def test_save_fonts_overrides(self, client):
|
|
"""Test saving font overrides."""
|
|
request_data = {
|
|
'weather': 'small',
|
|
'clock': 'regular'
|
|
}
|
|
|
|
response = client.post(
|
|
'/api/v3/fonts/overrides',
|
|
data=json.dumps(request_data),
|
|
content_type='application/json'
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
|
|
class TestAPIErrorHandling:
|
|
"""Test API error handling."""
|
|
|
|
def test_invalid_json_request(self, client):
|
|
"""Test handling invalid JSON in request."""
|
|
response = client.post(
|
|
'/api/v3/config/main',
|
|
data='invalid json',
|
|
content_type='application/json'
|
|
)
|
|
|
|
# Flask may return 500 for JSON decode errors or 400 for bad request
|
|
assert response.status_code in [400, 415, 500]
|
|
|
|
def test_missing_required_fields(self, client):
|
|
"""Test handling missing required fields."""
|
|
response = client.post(
|
|
'/api/v3/plugins/toggle',
|
|
data=json.dumps({}),
|
|
content_type='application/json'
|
|
)
|
|
|
|
assert response.status_code in [400, 422, 500]
|
|
|
|
def test_nonexistent_endpoint(self, client):
|
|
"""Test accessing nonexistent endpoint."""
|
|
response = client.get('/api/v3/nonexistent')
|
|
|
|
assert response.status_code == 404
|
|
|
|
def test_method_not_allowed(self, client):
|
|
"""Test using wrong HTTP method."""
|
|
# GET instead of POST
|
|
response = client.get('/api/v3/config/main',
|
|
query_string={'method': 'POST'})
|
|
|
|
# Should work for GET, but if we try POST-only endpoint with GET
|
|
response = client.get('/api/v3/config/schedule')
|
|
|
|
# Schedule might allow GET, so test a POST-only endpoint
|
|
response = client.get('/api/v3/display/on-demand/start')
|
|
|
|
assert response.status_code in [200, 405] # Depends on implementation
|
|
|
|
|
|
class TestDottedKeyNormalization:
|
|
"""Regression tests for fix_array_structures / ensure_array_defaults with dotted schema keys."""
|
|
|
|
def test_save_plugin_config_dotted_key_arrays(self, client, mock_config_manager):
|
|
"""Nested dotted-key objects with numeric-keyed dicts are converted to arrays."""
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
|
|
api_v3.config_manager = mock_config_manager
|
|
mock_config_manager.load_config.return_value = {}
|
|
|
|
schema_mgr = MagicMock()
|
|
schema = {
|
|
'type': 'object',
|
|
'properties': {
|
|
'leagues': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'eng.1': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'enabled': {'type': 'boolean', 'default': True},
|
|
'favorite_teams': {
|
|
'type': 'array',
|
|
'items': {'type': 'string'},
|
|
'default': [],
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
schema_mgr.load_schema.return_value = schema
|
|
schema_mgr.generate_default_config.return_value = {
|
|
'leagues': {'eng.1': {'enabled': True, 'favorite_teams': []}},
|
|
}
|
|
schema_mgr.merge_with_defaults.side_effect = lambda config, defaults: {**defaults, **config}
|
|
# Must be a (bool, list) tuple: the endpoint does is_valid, errors = validate_config_against_schema(...)
|
|
schema_mgr.validate_config_against_schema.return_value = (True, [])
|
|
api_v3.schema_manager = schema_mgr
|
|
|
|
request_data = {
|
|
'plugin_id': 'soccer-scoreboard',
|
|
'config': {
|
|
'leagues': {
|
|
'eng.1': {
|
|
'enabled': True,
|
|
'favorite_teams': ['Arsenal', 'Chelsea'],
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
response = client.post(
|
|
'/api/v3/plugins/config',
|
|
data=json.dumps(request_data),
|
|
content_type='application/json',
|
|
)
|
|
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.data}"
|
|
saved = mock_config_manager.save_config_atomic.call_args[0][0]
|
|
soccer_cfg = saved.get('soccer-scoreboard', {})
|
|
leagues = soccer_cfg.get('leagues', {})
|
|
assert 'eng.1' in leagues, f"Expected 'eng.1' key, got: {list(leagues.keys())}"
|
|
assert isinstance(leagues['eng.1'].get('favorite_teams'), list)
|
|
assert leagues['eng.1']['favorite_teams'] == ['Arsenal', 'Chelsea']
|
|
|
|
def test_save_plugin_config_none_array_gets_default(self, client, mock_config_manager):
|
|
"""None array fields under dotted-key parents are replaced with defaults."""
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
|
|
api_v3.config_manager = mock_config_manager
|
|
mock_config_manager.load_config.return_value = {}
|
|
|
|
schema_mgr = MagicMock()
|
|
schema = {
|
|
'type': 'object',
|
|
'properties': {
|
|
'leagues': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'eng.1': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'favorite_teams': {
|
|
'type': 'array',
|
|
'items': {'type': 'string'},
|
|
'default': [],
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
schema_mgr.load_schema.return_value = schema
|
|
schema_mgr.generate_default_config.return_value = {
|
|
'leagues': {'eng.1': {'favorite_teams': []}},
|
|
}
|
|
schema_mgr.merge_with_defaults.side_effect = lambda config, defaults: {**defaults, **config}
|
|
schema_mgr.validate_config_against_schema.return_value = (True, [])
|
|
api_v3.schema_manager = schema_mgr
|
|
|
|
request_data = {
|
|
'plugin_id': 'soccer-scoreboard',
|
|
'config': {
|
|
'leagues': {
|
|
'eng.1': {
|
|
'favorite_teams': None,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
response = client.post(
|
|
'/api/v3/plugins/config',
|
|
data=json.dumps(request_data),
|
|
content_type='application/json',
|
|
)
|
|
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.data}"
|
|
saved = mock_config_manager.save_config_atomic.call_args[0][0]
|
|
soccer_cfg = saved.get('soccer-scoreboard', {})
|
|
teams = soccer_cfg.get('leagues', {}).get('eng.1', {}).get('favorite_teams')
|
|
assert isinstance(teams, list), f"Expected list, got: {type(teams)}"
|
|
assert teams == [], f"Expected empty default list, got: {teams}"
|