mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-01 16:58:06 +00:00
3872a68ff7
* Surface plugin update availability on the plugins page
The plugin manager page already showed each installed plugin's version and
had an Update button, but nothing told users an update actually existed —
they had to guess. The store manager already compares the installed
manifest version against the registry's latest_version for its reinstall
decision; this surfaces that same signal in the UI.
- api_v3 /plugins/installed now returns `latest_version` (from the registry
cache, no extra network call) and an `update_available` flag computed by a
new semver-aware helper `_is_plugin_update_available`. A locally modified
plugin whose version is ahead of the registry is not flagged.
- The installed-plugin card shows "vX.Y.Z available" next to the installed
version, and the Update button becomes emphasized ("Update to vX.Y.Z" with
a gentle pulse) when a newer version is published — mirroring the app's own
update banner styling.
- Added tests for the helper and the endpoint fields.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012EW8yDbsk8EceDpq6Hjgqj
* Catch InvalidVersion specifically in plugin update comparison
Address review feedback: the version comparison caught a blind `except
Exception` (ruff BLE001). Split the two failure modes and catch each
specifically — ImportError for a missing `packaging` (a core dependency)
and InvalidVersion for an unparseable version string — while preserving
the existing "surface the mismatch" fallback behavior. Added a test for
the unparseable-version path.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012EW8yDbsk8EceDpq6Hjgqj
---------
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
873 lines
33 KiB
Python
873 lines
33 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_save_double_sided_settings(self, client, mock_config_manager):
|
|
"""Double-sided form fields are persisted under display.double_sided."""
|
|
response = client.post(
|
|
'/api/v3/config/main',
|
|
data={
|
|
'double_sided_enabled': 'true',
|
|
'double_sided_copies': '2',
|
|
'double_sided_axis': 'vertical',
|
|
},
|
|
content_type='application/x-www-form-urlencoded',
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
saved = mock_config_manager.save_config_atomic.call_args[0][0]
|
|
assert saved['display']['double_sided'] == {
|
|
'enabled': True, 'copies': 2, 'axis': 'vertical',
|
|
}
|
|
|
|
def test_save_double_sided_unchecked_disables(self, client, mock_config_manager):
|
|
"""An omitted 'enabled' checkbox is saved as disabled, not left stale."""
|
|
response = client.post(
|
|
'/api/v3/config/main',
|
|
data={'double_sided_copies': '4', 'double_sided_axis': 'horizontal'},
|
|
content_type='application/x-www-form-urlencoded',
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
ds = mock_config_manager.save_config_atomic.call_args[0][0]['display']['double_sided']
|
|
assert ds['enabled'] is False
|
|
assert ds['copies'] == 4
|
|
|
|
def test_save_double_sided_invalid_copies_rejected(self, client, mock_config_manager):
|
|
"""copies < 2 is rejected with a 400 before any save."""
|
|
response = client.post(
|
|
'/api/v3/config/main',
|
|
data={'double_sided_enabled': 'true', 'double_sided_copies': '1'},
|
|
content_type='application/x-www-form-urlencoded',
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
mock_config_manager.save_config_atomic.assert_not_called()
|
|
|
|
def test_save_double_sided_invalid_axis_rejected(self, client, mock_config_manager):
|
|
"""An unknown axis is rejected with a 400 before any save."""
|
|
response = client.post(
|
|
'/api/v3/config/main',
|
|
data={'double_sided_enabled': 'true', 'double_sided_axis': 'diagonal'},
|
|
content_type='application/x-www-form-urlencoded',
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
mock_config_manager.save_config_atomic.assert_not_called()
|
|
|
|
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_installed_plugins_report_update_available(self, client, mock_plugin_manager):
|
|
"""Installed-plugin entries surface latest_version + update_available
|
|
by comparing the on-disk manifest version to the registry."""
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
api_v3.plugin_manager = mock_plugin_manager
|
|
# No on-disk manifest to merge — keep the version we hand in below.
|
|
mock_plugin_manager.plugins_dir = '/nonexistent-plugins-dir'
|
|
mock_plugin_manager.get_all_plugin_info.return_value = [
|
|
{'id': 'weather', 'name': 'Weather', 'version': '1.0.0'}
|
|
]
|
|
# Avoid touching plugin instances (Vegas hooks, enabled fallback).
|
|
mock_plugin_manager.get_plugin.return_value = None
|
|
# Registry advertises a newer version than the installed one.
|
|
api_v3.plugin_store_manager.get_registry_info.return_value = {
|
|
'verified': True, 'latest_version': '1.2.0'
|
|
}
|
|
|
|
response = client.get('/api/v3/plugins/installed')
|
|
|
|
assert response.status_code == 200
|
|
payload = json.loads(response.data)
|
|
entry = payload['data']['plugins'][0]
|
|
assert entry['version'] == '1.0.0'
|
|
assert entry['latest_version'] == '1.2.0'
|
|
assert entry['update_available'] is True
|
|
|
|
def test_installed_plugins_no_update_when_current(self, client, mock_plugin_manager):
|
|
"""No update is flagged when installed version matches the registry."""
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
api_v3.plugin_manager = mock_plugin_manager
|
|
mock_plugin_manager.plugins_dir = '/nonexistent-plugins-dir'
|
|
mock_plugin_manager.get_all_plugin_info.return_value = [
|
|
{'id': 'weather', 'name': 'Weather', 'version': '1.2.0'}
|
|
]
|
|
mock_plugin_manager.get_plugin.return_value = None
|
|
api_v3.plugin_store_manager.get_registry_info.return_value = {
|
|
'verified': True, 'latest_version': '1.2.0'
|
|
}
|
|
|
|
response = client.get('/api/v3/plugins/installed')
|
|
|
|
assert response.status_code == 200
|
|
entry = json.loads(response.data)['data']['plugins'][0]
|
|
assert entry['latest_version'] == '1.2.0'
|
|
assert entry['update_available'] is False
|
|
|
|
def test_is_plugin_update_available_helper(self):
|
|
"""Unit-level checks for the semver-aware update comparison."""
|
|
from web_interface.blueprints.api_v3 import _is_plugin_update_available
|
|
assert _is_plugin_update_available('1.0.0', '1.0.1') is True
|
|
assert _is_plugin_update_available('1.0.1', '1.0.1') is False
|
|
# Local build ahead of the registry must not be flagged.
|
|
assert _is_plugin_update_available('2.0.0', '1.9.9') is False
|
|
# Missing either side yields no signal.
|
|
assert _is_plugin_update_available('', '1.0.0') is False
|
|
assert _is_plugin_update_available('1.0.0', '') is False
|
|
# Unparseable version differing from the installed one surfaces the
|
|
# mismatch rather than hiding a possible update.
|
|
assert _is_plugin_update_available('1.0.0', 'not-a-semver') is True
|
|
|
|
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}"
|
|
|
|
|
|
class TestPluginHealthRoutes:
|
|
"""Phase 1: /plugins/health and /plugins/metrics build per-installed-id so
|
|
they surface cross-process data persisted by the display service."""
|
|
|
|
def test_health_route_builds_per_installed_id(self, client, mock_plugin_manager):
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
from src.plugin_system.plugin_health import PluginHealthTracker
|
|
|
|
cache = MagicMock()
|
|
cache.get.return_value = None
|
|
api_v3.plugin_manager = mock_plugin_manager
|
|
mock_plugin_manager.plugin_manifests = {'p1': {}, 'p2': {}}
|
|
mock_plugin_manager.health_tracker = PluginHealthTracker(cache)
|
|
|
|
resp = client.get('/api/v3/plugins/health')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()['data']
|
|
assert set(data.keys()) == {'p1', 'p2'}
|
|
assert data['p1']['is_healthy'] is True
|
|
assert data['p1']['degraded'] is False
|
|
|
|
def test_health_route_reports_not_available_without_tracker(self, client, mock_plugin_manager):
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
api_v3.plugin_manager = mock_plugin_manager
|
|
mock_plugin_manager.health_tracker = None
|
|
|
|
resp = client.get('/api/v3/plugins/health')
|
|
assert resp.status_code == 200
|
|
body = resp.get_json()
|
|
assert body['data'] == {}
|
|
assert 'not available' in body['message'].lower()
|
|
|
|
def test_metrics_route_builds_per_installed_id(self, client, mock_plugin_manager):
|
|
from web_interface.blueprints.api_v3 import api_v3
|
|
from src.plugin_system.resource_monitor import PluginResourceMonitor
|
|
|
|
cache = MagicMock()
|
|
cache.get.return_value = None
|
|
api_v3.plugin_manager = mock_plugin_manager
|
|
mock_plugin_manager.plugin_manifests = {'p1': {}}
|
|
mock_plugin_manager.resource_monitor = PluginResourceMonitor(
|
|
cache, enable_monitoring=False
|
|
)
|
|
|
|
resp = client.get('/api/v3/plugins/metrics')
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()['data']
|
|
assert 'p1' in data
|
|
assert data['p1']['call_count'] == 0
|