mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-01 16:58:06 +00:00
ChuckBuildsandClaude Opus 4.8
fe863602e3
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
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
|