mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-13 02:26:37 +00:00
Merge audit16-core/configsave into the September 16 core audit fixes
# Conflicts: # src/plugin_system/plugin_manager.py
This commit is contained in:
@@ -0,0 +1,288 @@
|
||||
"""POST /config/main: a partial JSON body changes only what it sends.
|
||||
|
||||
The settings forms post every field, and a browser leaves an unchecked box out,
|
||||
so for a form a missing checkbox means False. JSON API clients send only what
|
||||
they change. Treating their missing keys as unchecked meant:
|
||||
|
||||
- the MQTT bridge's Home Assistant brightness slider (``{"brightness": N}``)
|
||||
turned off disable_hardware_pulsing, inverse_colors, show_refresh_rate and
|
||||
use_short_date_format on every change;
|
||||
- the documented timezone/location update turned off web_display_autostart and
|
||||
weekly automatic updates.
|
||||
|
||||
The v3 forms post JSON as well (htmx json-enc), so they mark themselves with a
|
||||
hidden ``__form_section`` input; these tests pin both halves of that contract.
|
||||
|
||||
Also here: the Vegas cycle-time fields no longer land in display_durations
|
||||
(and a blank one no longer 400s the Display save), the Raw JSON editor starts
|
||||
auto-update setup like the General form does, and both schedule POSTs accept
|
||||
the per-day shape their GETs return.
|
||||
"""
|
||||
|
||||
import copy
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from test._api_v3_test_helpers import api_v3_client, api_v3_module # noqa: F401
|
||||
|
||||
REPO = Path(__file__).resolve().parent.parent
|
||||
PARTIALS = REPO / 'web_interface' / 'templates' / 'v3' / 'partials'
|
||||
|
||||
STORED = {
|
||||
'web_display_autostart': True,
|
||||
'auto_update': {'enabled': True},
|
||||
'timezone': 'America/Chicago',
|
||||
'plugin_system': {'auto_discover': True, 'auto_load_enabled': True,
|
||||
'development_mode': True},
|
||||
'display': {
|
||||
'hardware': {'rows': 32, 'cols': 64, 'chain_length': 2, 'brightness': 90,
|
||||
'disable_hardware_pulsing': True, 'inverse_colors': True,
|
||||
'show_refresh_rate': True},
|
||||
'runtime': {'gpio_slowdown': 4},
|
||||
'use_short_date_format': True,
|
||||
'double_sided': {'enabled': True, 'copies': 2, 'axis': 'horizontal'},
|
||||
'vegas_scroll': {'enabled': True, 'auto_trim': True,
|
||||
'dynamic_duration_enabled': True,
|
||||
'continuous_scroll': True, 'smooth_scroll': True},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def saved(api_v3_module, monkeypatch):
|
||||
captured = {}
|
||||
api_v3_module.api_v3.config_manager.load_config.side_effect = \
|
||||
lambda *a, **k: copy.deepcopy(STORED)
|
||||
|
||||
def fake_save(_manager, config, **_kwargs):
|
||||
captured['config'] = config
|
||||
return True, ''
|
||||
|
||||
monkeypatch.setattr(api_v3_module, '_save_config_atomic', fake_save)
|
||||
from web_interface import auto_update
|
||||
monkeypatch.setattr(auto_update, 'start_setup_if_needed', lambda *a, **k: None)
|
||||
return captured
|
||||
|
||||
|
||||
def _post_json(client, body):
|
||||
return client.post('/api/v3/config/main', data=json.dumps(body),
|
||||
content_type='application/json')
|
||||
|
||||
|
||||
class TestJsonPartialSaves:
|
||||
def test_mqtt_bridge_brightness_changes_only_brightness(self, api_v3_client, saved):
|
||||
# integrations/mqtt_bridge/ledmatrix_mqtt_bridge.py set_brightness
|
||||
resp = _post_json(api_v3_client, {'brightness': 40})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
display = saved['config']['display']
|
||||
assert display['hardware']['brightness'] == 40
|
||||
assert display['hardware']['disable_hardware_pulsing'] is True
|
||||
assert display['hardware']['inverse_colors'] is True
|
||||
assert display['hardware']['show_refresh_rate'] is True
|
||||
assert display['use_short_date_format'] is True
|
||||
|
||||
def test_timezone_only_keeps_autostart_and_auto_update(self, api_v3_client, saved):
|
||||
resp = _post_json(api_v3_client, {'timezone': 'UTC'})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
config = saved['config']
|
||||
assert config['timezone'] == 'UTC'
|
||||
assert config['web_display_autostart'] is True
|
||||
assert config['auto_update'] == {'enabled': True}
|
||||
|
||||
def test_location_only_keeps_autostart_and_auto_update(self, api_v3_client, saved):
|
||||
resp = _post_json(api_v3_client, {'city': 'Paris', 'country': 'FR'})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
config = saved['config']
|
||||
assert config['location'] == {'city': 'Paris', 'country': 'FR'}
|
||||
assert config['web_display_autostart'] is True
|
||||
assert config['auto_update'] == {'enabled': True}
|
||||
|
||||
@pytest.mark.parametrize('key', ['auto_discover', 'auto_load_enabled', 'development_mode'])
|
||||
def test_a_legacy_plugin_system_toggle_is_not_a_general_save(self, api_v3_client, saved, key):
|
||||
# These left the General form; a client still sending one must not
|
||||
# have the general-settings checkboxes treated as unchecked.
|
||||
resp = api_v3_client.post('/api/v3/config/main', data={key: 'on'},
|
||||
content_type='application/x-www-form-urlencoded')
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert saved['config']['web_display_autostart'] is True
|
||||
assert saved['config']['auto_update'] == {'enabled': True}
|
||||
|
||||
def test_vegas_speed_only_keeps_vegas_toggles(self, api_v3_client, saved):
|
||||
resp = _post_json(api_v3_client, {'vegas_scroll_speed': 80})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
vegas = saved['config']['display']['vegas_scroll']
|
||||
assert vegas['scroll_speed'] == 80
|
||||
for key in ('enabled', 'auto_trim', 'dynamic_duration_enabled',
|
||||
'continuous_scroll', 'smooth_scroll'):
|
||||
assert vegas[key] is True, key
|
||||
|
||||
def test_double_sided_axis_only_keeps_enabled(self, api_v3_client, saved):
|
||||
resp = _post_json(api_v3_client, {'double_sided_axis': 'horizontal'})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert saved['config']['display']['double_sided']['enabled'] is True
|
||||
|
||||
def test_json_can_still_turn_a_checkbox_off(self, api_v3_client, saved):
|
||||
resp = _post_json(api_v3_client, {'inverse_colors': False, 'auto_update_enabled': False})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert saved['config']['display']['hardware']['inverse_colors'] is False
|
||||
assert saved['config']['display']['hardware']['disable_hardware_pulsing'] is True
|
||||
assert saved['config']['auto_update'] == {'enabled': False}
|
||||
|
||||
def test_json_with_charset_is_still_json(self, api_v3_client, saved):
|
||||
resp = api_v3_client.post('/api/v3/config/main', data=json.dumps({'brightness': 41}),
|
||||
content_type='application/json; charset=utf-8')
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert saved['config']['display']['hardware']['brightness'] == 41
|
||||
assert saved['config']['display']['hardware']['inverse_colors'] is True
|
||||
|
||||
def test_a_non_object_body_is_refused(self, api_v3_client, saved):
|
||||
assert _post_json(api_v3_client, [1, 2]).status_code == 400
|
||||
|
||||
|
||||
class TestFormSavesStillUncheck:
|
||||
"""Unchecking a box in the UI must still save false."""
|
||||
|
||||
def test_marked_json_form_post_unchecks_missing_display_boxes(self, api_v3_client, saved):
|
||||
resp = _post_json(api_v3_client, {'__form_section': 'display', 'brightness': '40',
|
||||
'vegas_scroll_speed': '50'})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
display = saved['config']['display']
|
||||
for key in ('disable_hardware_pulsing', 'inverse_colors', 'show_refresh_rate'):
|
||||
assert display['hardware'][key] is False, key
|
||||
assert display['use_short_date_format'] is False
|
||||
assert display['vegas_scroll']['enabled'] is False
|
||||
assert '__form_section' not in saved['config']
|
||||
|
||||
def test_marked_json_form_post_keeps_checked_boxes(self, api_v3_client, saved):
|
||||
resp = _post_json(api_v3_client, {'__form_section': 'display', 'brightness': '40',
|
||||
'inverse_colors': 'on'})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert saved['config']['display']['hardware']['inverse_colors'] is True
|
||||
assert saved['config']['display']['hardware']['show_refresh_rate'] is False
|
||||
|
||||
def test_marked_general_form_unchecks_autostart_and_auto_update(self, api_v3_client, saved):
|
||||
resp = _post_json(api_v3_client, {'__form_section': 'general', 'timezone': 'UTC'})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert saved['config']['web_display_autostart'] is False
|
||||
assert saved['config']['auto_update'] == {'enabled': False}
|
||||
|
||||
def test_form_encoded_post_unchecks_missing_boxes(self, api_v3_client, saved):
|
||||
resp = api_v3_client.post('/api/v3/config/main', data={'brightness': '40'},
|
||||
content_type='application/x-www-form-urlencoded')
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert saved['config']['display']['hardware']['inverse_colors'] is False
|
||||
|
||||
@pytest.mark.parametrize('partial', ['general.html', 'display.html', 'durations.html'])
|
||||
def test_every_config_main_form_carries_the_marker(self, partial):
|
||||
html = (PARTIALS / partial).read_text(encoding='utf-8')
|
||||
form = re.search(r'<form hx-post="/api/v3/config/main".*?</form>', html, re.S)
|
||||
assert form, f'{partial} no longer posts to /config/main'
|
||||
assert re.search(r'<input type="hidden" name="__form_section" value="\w+">',
|
||||
form.group(0)), f'{partial} form lost its __form_section marker'
|
||||
|
||||
|
||||
class TestVegasCycleDurations:
|
||||
def test_cycle_durations_are_not_display_durations(self, api_v3_client, saved):
|
||||
resp = api_v3_client.post('/api/v3/config/main', data={
|
||||
'vegas_scroll_enabled': 'on', 'vegas_min_cycle_duration': '90',
|
||||
'vegas_max_cycle_duration': '300'})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
display = saved['config']['display']
|
||||
assert display['vegas_scroll']['min_cycle_duration'] == 90
|
||||
assert display['vegas_scroll']['max_cycle_duration'] == 300
|
||||
durations = display.get('display_durations', {})
|
||||
assert 'vegas_min_cycle_duration' not in durations
|
||||
assert 'vegas_max_cycle_duration' not in durations
|
||||
assert 'vegas_min_cycle_duration' not in saved['config']
|
||||
|
||||
def test_blank_cycle_duration_does_not_reject_the_save(self, api_v3_client, saved):
|
||||
resp = api_v3_client.post('/api/v3/config/main', data={
|
||||
'vegas_scroll_enabled': 'on', 'vegas_scroll_speed': '60',
|
||||
'vegas_min_cycle_duration': ''})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert saved['config']['display']['vegas_scroll']['scroll_speed'] == 60
|
||||
|
||||
def test_real_display_durations_still_save(self, api_v3_client, saved):
|
||||
resp = _post_json(api_v3_client, {'clock_duration': '45'})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert saved['config']['display']['display_durations']['clock_duration'] == 45
|
||||
|
||||
|
||||
class TestRawSaveStartsAutoUpdateSetup:
|
||||
@pytest.fixture
|
||||
def raw_env(self, api_v3_module, monkeypatch):
|
||||
cm = api_v3_module.api_v3.config_manager
|
||||
cm.get_raw_file_content.return_value = {'timezone': 'UTC', 'auto_update': {'enabled': False}}
|
||||
calls = []
|
||||
from web_interface import auto_update
|
||||
monkeypatch.setattr(auto_update, 'start_setup_if_needed',
|
||||
lambda was, config: calls.append((was, config)) or 'Setup note.')
|
||||
return cm, calls
|
||||
|
||||
def test_enabling_from_raw_json_calls_setup(self, api_v3_client, raw_env):
|
||||
cm, calls = raw_env
|
||||
body = {'timezone': 'UTC', 'auto_update': {'enabled': True}}
|
||||
resp = api_v3_client.post('/api/v3/config/raw/main', json=body)
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
cm.save_raw_file_content.assert_called_once_with('main', body)
|
||||
assert calls == [(False, body)]
|
||||
assert 'Setup note.' in resp.get_json()['message']
|
||||
|
||||
def test_previously_enabled_is_passed_through(self, api_v3_client, raw_env):
|
||||
cm, calls = raw_env
|
||||
cm.get_raw_file_content.return_value = {'auto_update': {'enabled': True}}
|
||||
body = {'auto_update': {'enabled': True}}
|
||||
assert api_v3_client.post('/api/v3/config/raw/main', json=body).status_code == 200
|
||||
assert calls == [(True, body)]
|
||||
|
||||
|
||||
class TestSchedulesAcceptTheirGetShape:
|
||||
PER_DAY = {
|
||||
'enabled': True, 'mode': 'per-day', 'dim_brightness': 20,
|
||||
'days': {
|
||||
'monday': {'enabled': True, 'start_time': '21:15', 'end_time': '06:45'},
|
||||
'tuesday': {'enabled': False},
|
||||
'wednesday': {'enabled': True, 'start_time': '22:00', 'end_time': '05:30'},
|
||||
'thursday': {'enabled': True, 'start_time': '20:00', 'end_time': '07:00'},
|
||||
'friday': {'enabled': True, 'start_time': '23:00', 'end_time': '08:00'},
|
||||
'saturday': {'enabled': True, 'start_time': '23:30', 'end_time': '09:00'},
|
||||
'sunday': {'enabled': True, 'start_time': '21:00', 'end_time': '07:00'},
|
||||
},
|
||||
}
|
||||
|
||||
@pytest.fixture
|
||||
def store(self, api_v3_module, monkeypatch):
|
||||
state = {'config': {}}
|
||||
api_v3_module.api_v3.config_manager.load_config.side_effect = \
|
||||
lambda *a, **k: copy.deepcopy(state['config'])
|
||||
|
||||
def fake_save(_manager, config, **_kwargs):
|
||||
state['config'] = copy.deepcopy(config)
|
||||
return True, ''
|
||||
|
||||
monkeypatch.setattr(api_v3_module, '_save_config_atomic', fake_save)
|
||||
return state
|
||||
|
||||
@pytest.mark.parametrize('route,section,extra', [
|
||||
('/api/v3/config/dim-schedule', 'dim_schedule', {'dim_brightness': 20}),
|
||||
('/api/v3/config/schedule', 'schedule', {}),
|
||||
])
|
||||
def test_get_output_posts_back_unchanged(self, api_v3_client, store, route, section, extra):
|
||||
body = {k: v for k, v in self.PER_DAY.items() if k != 'dim_brightness'}
|
||||
body.update(extra)
|
||||
store['config'] = {section: copy.deepcopy(body)}
|
||||
read = api_v3_client.get(route).get_json()['data']
|
||||
resp = api_v3_client.post(route, json=read)
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert store['config'][section]['days'] == body['days']
|
||||
|
||||
def test_flat_form_keys_still_work(self, api_v3_client, store):
|
||||
body = {'enabled': True, 'mode': 'per-day', 'dim_brightness': 25,
|
||||
'monday_enabled': 'on', 'monday_start': '19:00', 'monday_end': '06:00'}
|
||||
resp = api_v3_client.post('/api/v3/config/dim-schedule', json=body)
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert store['config']['dim_schedule']['days']['monday'] == {
|
||||
'enabled': True, 'start_time': '19:00', 'end_time': '06:00'}
|
||||
@@ -832,7 +832,9 @@ class TestSettingsSave:
|
||||
|
||||
def test_unchecked_toggle_on_general_save_disables(self, api_client):
|
||||
client, cm, setup_calls = api_client
|
||||
resp = client.post('/api/v3/config/main', json={'timezone': 'UTC'})
|
||||
# The General form marks its posts; an unchecked box is then absent.
|
||||
resp = client.post('/api/v3/config/main',
|
||||
json={'__form_section': 'general', 'timezone': 'UTC'})
|
||||
assert resp.status_code == 200
|
||||
assert self._saved(cm)['auto_update'] == {'enabled': False}
|
||||
assert setup_calls == [True]
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
"""Every "is this top-level key a plugin?" decision uses src/core_config_keys.py.
|
||||
|
||||
#589 moved plugin-state reconciliation onto the shared list, but three private
|
||||
copies stayed behind and did not know about auto_update, dim_schedule, sync,
|
||||
location, target_fps, ...:
|
||||
|
||||
- StartupValidator warned "Plugin 'auto_update' is enabled but not found in
|
||||
plugins directory" on every display start with auto-update or a dim
|
||||
schedule switched on;
|
||||
- SchemaManager.detect_config_key_collisions let a plugin id collide with
|
||||
those sections silently;
|
||||
- ConfigManager.cleanup_orphaned_plugin_configs deleted display, schedule,
|
||||
auto_update, ... as orphans, and validate_all_plugin_configs validated core
|
||||
sections as plugins.
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from src.config_manager import ConfigManager
|
||||
from src.core_config_keys import CORE_CONFIG_KEYS, CORE_SECRETS_KEYS
|
||||
from src.plugin_system.schema_manager import SchemaManager
|
||||
from src.startup_validator import StartupValidator
|
||||
|
||||
REPO = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
class TestStartupValidator:
|
||||
def test_enabled_core_sections_are_not_missing_plugins(self):
|
||||
config = {
|
||||
'display': {'hardware': {}},
|
||||
'auto_update': {'enabled': True},
|
||||
'dim_schedule': {'enabled': True},
|
||||
'schedule': {'enabled': True},
|
||||
'gone-plugin': {'enabled': True},
|
||||
}
|
||||
cm = MagicMock()
|
||||
cm.get_config.return_value = config
|
||||
pm = MagicMock()
|
||||
pm.discover_plugins.return_value = []
|
||||
validator = StartupValidator(cm, plugin_manager=pm, cache_manager=MagicMock())
|
||||
validator.warnings = []
|
||||
validator.errors = []
|
||||
validator._validate_plugins()
|
||||
missing = [w for w in validator.warnings if 'not found in plugins directory' in w]
|
||||
assert missing == ["Plugin 'gone-plugin' is enabled but not found in plugins directory"]
|
||||
|
||||
|
||||
class TestReservedPluginIds:
|
||||
@pytest.mark.parametrize('key', sorted(CORE_CONFIG_KEYS))
|
||||
def test_every_core_key_is_reserved(self, key):
|
||||
collisions = SchemaManager().detect_config_key_collisions([key])
|
||||
assert [c['type'] for c in collisions] == ['reserved_key_collision']
|
||||
|
||||
def test_an_ordinary_plugin_id_is_not(self):
|
||||
assert SchemaManager().detect_config_key_collisions(['clock-simple']) == []
|
||||
|
||||
|
||||
class TestConfigManagerHelpers:
|
||||
@pytest.fixture
|
||||
def manager(self, tmp_path):
|
||||
config = {key: {'enabled': True} for key in ('display', 'schedule', 'auto_update',
|
||||
'dim_schedule', 'sync', 'location')}
|
||||
config.update({'timezone': 'UTC', 'installed': {'enabled': True},
|
||||
'orphan': {'enabled': True}})
|
||||
secrets = {'github': {'token': 't'}, 'youtube': {'api_key': 'k'},
|
||||
'installed': {'api_key': 'a'}, 'orphan': {'api_key': 'o'}}
|
||||
(tmp_path / 'config.json').write_text(json.dumps(config))
|
||||
(tmp_path / 'secrets.json').write_text(json.dumps(secrets))
|
||||
manager = ConfigManager(config_path=str(tmp_path / 'config.json'),
|
||||
secrets_path=str(tmp_path / 'secrets.json'))
|
||||
manager.template_path = str(tmp_path / 'no-template.json')
|
||||
return manager
|
||||
|
||||
def test_cleanup_removes_only_real_orphans(self, manager, tmp_path):
|
||||
removed = manager.cleanup_orphaned_plugin_configs(['installed'])
|
||||
assert removed == ['orphan']
|
||||
config = json.loads((tmp_path / 'config.json').read_text())
|
||||
assert {'display', 'schedule', 'auto_update', 'dim_schedule', 'sync',
|
||||
'location', 'timezone', 'installed'} == set(config)
|
||||
secrets = json.loads((tmp_path / 'secrets.json').read_text())
|
||||
assert set(secrets) == {'github', 'youtube', 'installed'}
|
||||
|
||||
def test_validate_all_skips_core_sections(self, manager):
|
||||
schema_manager = MagicMock()
|
||||
schema_manager.load_schema.return_value = None
|
||||
results = manager.validate_all_plugin_configs(schema_manager)
|
||||
assert set(results) == {'installed', 'orphan'}
|
||||
|
||||
|
||||
def test_core_secrets_keys_are_not_core_config_keys():
|
||||
assert not (CORE_SECRETS_KEYS & CORE_CONFIG_KEYS)
|
||||
|
||||
|
||||
@pytest.mark.parametrize('relpath', [
|
||||
'src/startup_validator.py',
|
||||
'src/config_manager.py',
|
||||
'src/plugin_system/schema_manager.py',
|
||||
])
|
||||
def test_no_private_copy_of_the_list_remains(relpath):
|
||||
"""The old copies all started with this literal; a new copy likely would too."""
|
||||
source = (REPO / relpath).read_text(encoding='utf-8')
|
||||
assert not re.search(r"""\[\s*['"]display['"]\s*,\s*['"]schedule['"]""", source), \
|
||||
f'{relpath} has a private core-key list again; use src/core_config_keys.py'
|
||||
@@ -0,0 +1,205 @@
|
||||
"""One preparation for a plugin's config, wherever the config comes from.
|
||||
|
||||
A plugin's stored section becomes the config it runs with through
|
||||
schema_manager.prepare_plugin_config: legacy booleans (#588) read as
|
||||
``{"enabled": ...}`` objects, then schema and core defaults filled in. Loading
|
||||
did that; hot reload handed plugins the raw section instead (a legacy
|
||||
``dynamic_duration: true`` came back as a boolean, turning dynamic duration
|
||||
off), and the dev tools each built configs their own way:
|
||||
|
||||
- dev_server read only top-level defaults and let a schema ``enabled: false``
|
||||
override its forced ``enabled: True``;
|
||||
- check_plugin/render_plugin (build_full_config) merged overrides with
|
||||
dict.update, so ``{"nhl": {"enabled": true}}`` dropped every other nhl
|
||||
default;
|
||||
- the harness and the device disagreed on object and array defaults.
|
||||
|
||||
The web saves and GET are covered in
|
||||
test/web_interface/test_plugin_config_json_saves.py.
|
||||
"""
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from src.plugin_system.plugin_manager import PluginManager
|
||||
from src.plugin_system.schema_manager import (
|
||||
CORE_PLUGIN_PROPERTIES, SchemaManager, extract_schema_defaults,
|
||||
)
|
||||
from src.plugin_system.testing import loading
|
||||
|
||||
REPO = Path(__file__).resolve().parent.parent
|
||||
|
||||
SCHEMA = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"enabled": {"type": "boolean", "default": False},
|
||||
"global": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"dynamic_duration": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"enabled": {"type": "boolean", "default": True},
|
||||
"max_duration_seconds": {"type": "integer", "default": 300},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
"nhl": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"enabled": {"type": "boolean", "default": False},
|
||||
"favorite_teams": {"type": "array", "default": ["TB"]},
|
||||
"show_records": {"type": "boolean", "default": True},
|
||||
},
|
||||
},
|
||||
"colors": {
|
||||
"type": "object",
|
||||
"default": {"text": [255, 255, 255]},
|
||||
"properties": {"text": {"type": "array", "default": [1, 2, 3]}},
|
||||
},
|
||||
"feeds": {"type": "array", "items": {"type": "string"}},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def plugin_dir(tmp_path):
|
||||
pdir = tmp_path / "plugins" / "demo"
|
||||
pdir.mkdir(parents=True)
|
||||
(pdir / "config_schema.json").write_text(json.dumps(SCHEMA))
|
||||
(pdir / "manifest.json").write_text(json.dumps({"id": "demo"}))
|
||||
return pdir
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def plugin_manager(plugin_dir, tmp_path):
|
||||
manager = PluginManager.__new__(PluginManager) # skip the heavy constructor
|
||||
manager.logger = MagicMock()
|
||||
manager.schema_manager = SchemaManager(plugins_dir=plugin_dir.parent, project_root=tmp_path)
|
||||
return manager
|
||||
|
||||
|
||||
class TestPluginManagerPreparation:
|
||||
def test_legacy_boolean_and_defaults(self, plugin_manager):
|
||||
prepared = plugin_manager.prepare_plugin_config(
|
||||
"demo", {"enabled": True, "global": {"dynamic_duration": True}})
|
||||
assert prepared["global"]["dynamic_duration"] == {
|
||||
"enabled": True, "max_duration_seconds": 300}
|
||||
assert prepared["nhl"]["favorite_teams"] == ["TB"]
|
||||
assert prepared["display_duration"] == 15
|
||||
|
||||
def test_does_not_mutate_the_raw_section(self, plugin_manager):
|
||||
raw = {"global": {"dynamic_duration": True}}
|
||||
plugin_manager.prepare_plugin_config("demo", raw)
|
||||
assert raw == {"global": {"dynamic_duration": True}}
|
||||
|
||||
def test_never_raises(self, plugin_manager):
|
||||
plugin_manager.schema_manager = MagicMock()
|
||||
plugin_manager.schema_manager.load_schema.return_value = SCHEMA
|
||||
plugin_manager.schema_manager.prepare_plugin_config.side_effect = RuntimeError("boom")
|
||||
prepared = plugin_manager.prepare_plugin_config("demo", {"global": {"dynamic_duration": False}})
|
||||
assert prepared["global"]["dynamic_duration"] == {"enabled": False}
|
||||
|
||||
|
||||
class TestHotReload:
|
||||
def test_on_config_change_gets_the_prepared_config(self, test_display_controller, plugin_manager):
|
||||
controller = test_display_controller
|
||||
plugin = MagicMock()
|
||||
plugin.modes = ["demo"]
|
||||
pm = controller.plugin_manager
|
||||
pm.discover_plugins.return_value = ["demo"]
|
||||
pm.load_plugin.return_value = True
|
||||
pm.plugin_manifests = {}
|
||||
pm.get_plugin.side_effect = lambda pid: plugin if pid == "demo" else None
|
||||
pm.prepare_plugin_config.side_effect = plugin_manager.prepare_plugin_config
|
||||
controller.config_service.get_config = lambda: {"demo": {"enabled": True}}
|
||||
controller._reconcile_enabled_plugins()
|
||||
|
||||
callback = controller._plugin_config_callbacks["demo"]
|
||||
callback({"enabled": True}, {"enabled": True, "global": {"dynamic_duration": True}})
|
||||
|
||||
new_config = plugin.on_config_change.call_args[0][0]
|
||||
assert new_config["global"]["dynamic_duration"] == {
|
||||
"enabled": True, "max_duration_seconds": 300}
|
||||
assert new_config["nhl"]["show_records"] is True
|
||||
|
||||
def test_raw_section_still_delivered_without_a_preparer(self, test_display_controller):
|
||||
controller = test_display_controller
|
||||
plugin = MagicMock()
|
||||
plugin.modes = ["demo"]
|
||||
pm = controller.plugin_manager
|
||||
pm.discover_plugins.return_value = ["demo"]
|
||||
pm.load_plugin.return_value = True
|
||||
pm.plugin_manifests = {}
|
||||
pm.get_plugin.side_effect = lambda pid: plugin if pid == "demo" else None
|
||||
pm.prepare_plugin_config.return_value = None
|
||||
controller.config_service.get_config = lambda: {"demo": {"enabled": True}}
|
||||
controller._reconcile_enabled_plugins()
|
||||
|
||||
raw = {"enabled": False}
|
||||
controller._plugin_config_callbacks["demo"]({}, raw)
|
||||
plugin.on_config_change.assert_called_once_with(raw)
|
||||
|
||||
|
||||
class TestDevToolsMatchTheDevice:
|
||||
def test_harness_defaults_are_the_device_defaults(self, plugin_dir):
|
||||
assert loading.load_config_defaults(plugin_dir) == extract_schema_defaults(SCHEMA)
|
||||
defaults = loading.load_config_defaults(plugin_dir)
|
||||
# An object's own default wins, as on a device; arrays start empty
|
||||
assert defaults["colors"] == {"text": [255, 255, 255]}
|
||||
assert defaults["feeds"] == []
|
||||
|
||||
def test_nested_override_keeps_sibling_defaults(self, plugin_dir):
|
||||
config = loading.build_full_config(plugin_dir, cli_config={"nhl": {"enabled": True}})
|
||||
assert config["nhl"] == {"enabled": True, "favorite_teams": ["TB"], "show_records": True}
|
||||
|
||||
def test_spec_and_cli_overrides_both_deep_merge(self, plugin_dir):
|
||||
config = loading.build_full_config(
|
||||
plugin_dir, spec={"config": {"nhl": {"show_records": False}}},
|
||||
cli_config={"nhl": {"enabled": True}})
|
||||
assert config["nhl"] == {"enabled": True, "favorite_teams": ["TB"], "show_records": False}
|
||||
|
||||
def test_build_config_matches_the_device_load(self, plugin_dir, plugin_manager):
|
||||
overrides = {"enabled": True, "global": {"dynamic_duration": False}}
|
||||
assert loading.build_config(plugin_dir, overrides) == \
|
||||
plugin_manager.prepare_plugin_config("demo", overrides)
|
||||
|
||||
def test_core_defaults_are_present(self, plugin_dir):
|
||||
config = loading.build_full_config(plugin_dir)
|
||||
assert config["enabled"] is True
|
||||
assert config["display_duration"] == CORE_PLUGIN_PROPERTIES["display_duration"]["default"]
|
||||
assert config["live_priority"] is False
|
||||
|
||||
def test_render_plugin_matrix_config(self, plugin_dir, monkeypatch):
|
||||
from src.plugin_system.testing import harness
|
||||
seen = {}
|
||||
|
||||
def fake_render_size(plugin_id, manifest, pdir, config, *args, **kwargs):
|
||||
seen["config"] = config
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(harness, "_render_size", fake_render_size)
|
||||
harness.render_plugin_matrix("demo", plugin_dir, config={"nhl": {"enabled": True}},
|
||||
sizes=[(64, 32)], run_update=False)
|
||||
assert seen["config"]["enabled"] is True, "a schema enabled:false must not win"
|
||||
assert seen["config"]["nhl"]["favorite_teams"] == ["TB"]
|
||||
|
||||
def test_dev_server_render_config(self, plugin_dir, monkeypatch):
|
||||
spec = importlib.util.spec_from_file_location("dev_server_under_test",
|
||||
REPO / "scripts" / "dev_server.py")
|
||||
dev_server = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(dev_server)
|
||||
monkeypatch.setattr(dev_server, "find_plugin_dir", lambda pid: plugin_dir)
|
||||
monkeypatch.setattr(dev_server, "_trusted_plugin_dir", lambda d: plugin_dir)
|
||||
|
||||
_, _, config, _, _ = dev_server._parse_render_request(
|
||||
{"plugin_id": "demo", "config": {"nhl": {"enabled": True}}})
|
||||
assert config["enabled"] is True, "a schema enabled:false must not win"
|
||||
assert config["nhl"] == {"enabled": True, "favorite_teams": ["TB"], "show_records": True}
|
||||
assert config["global"]["dynamic_duration"]["max_duration_seconds"] == 300
|
||||
assert dev_server.load_config_defaults(plugin_dir) == extract_schema_defaults(SCHEMA)
|
||||
@@ -0,0 +1,258 @@
|
||||
"""JSON plugin-config saves store what the device would load, and only what was sent.
|
||||
|
||||
Runs a real ConfigManager and SchemaManager over tmp_path, like
|
||||
test_api_v3_secret_roundtrip.py, so assertions are on config.json itself.
|
||||
|
||||
- POST /plugins/config with a partial ``config`` reset every unsent setting to
|
||||
its schema default: the JSON path built on defaults, not the stored section.
|
||||
- Legacy booleans (#588) were normalized only at load, so posting back what
|
||||
GET /plugins/config returned failed validation.
|
||||
- The JSON save's filter kept only enabled/display_duration/live_priority, so
|
||||
a submitted skin, skin_options or vegas_* tuning key was silently dropped.
|
||||
- Plugin sections posted to /config/main skipped all of that and were stored
|
||||
verbatim, including values /plugins/config rejects.
|
||||
"""
|
||||
|
||||
import json
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from flask import Flask
|
||||
|
||||
from src.config_manager import ConfigManager
|
||||
from src.plugin_system.schema_manager import SchemaManager
|
||||
from web_interface.blueprints.api_v3 import api_v3
|
||||
|
||||
PLUGIN_ID = "demo"
|
||||
|
||||
SCHEMA = {
|
||||
"type": "object",
|
||||
"additionalProperties": False,
|
||||
"properties": {
|
||||
"enabled": {"type": "boolean", "default": True},
|
||||
"city": {"type": "string", "default": "Dallas"},
|
||||
"update_interval": {"type": "integer", "default": 300, "minimum": 30},
|
||||
"api_key": {"type": "string", "x-secret": True, "default": ""},
|
||||
"display": {
|
||||
"type": "object",
|
||||
"additionalProperties": False,
|
||||
"properties": {
|
||||
"brightness": {"type": "integer", "default": 50},
|
||||
"show_icons": {"type": "boolean", "default": True},
|
||||
},
|
||||
},
|
||||
"global": {
|
||||
"type": "object",
|
||||
"additionalProperties": False,
|
||||
"properties": {
|
||||
"dynamic_duration": {
|
||||
"type": "object",
|
||||
"additionalProperties": False,
|
||||
"properties": {
|
||||
"enabled": {"type": "boolean", "default": True},
|
||||
"min_duration_seconds": {"type": "integer", "default": 30},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
STORED = {
|
||||
"enabled": True,
|
||||
"city": "Paris",
|
||||
"update_interval": 600,
|
||||
"display": {"brightness": 80, "show_icons": False},
|
||||
"global": {"dynamic_duration": {"enabled": False, "min_duration_seconds": 45}},
|
||||
"vegas_width_pct": 60,
|
||||
"skin": "neon",
|
||||
}
|
||||
|
||||
_ATTRS = ('config_manager', 'plugin_manager', 'plugin_store_manager',
|
||||
'plugin_state_manager', 'saved_repositories_manager', 'schema_manager',
|
||||
'operation_queue', 'operation_history', 'cache_manager')
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def env(tmp_path):
|
||||
config_file = tmp_path / "config.json"
|
||||
secrets_file = tmp_path / "config_secrets.json"
|
||||
plugins_dir = tmp_path / "plugins"
|
||||
plugin_dir = plugins_dir / PLUGIN_ID
|
||||
plugin_dir.mkdir(parents=True)
|
||||
(plugin_dir / "config_schema.json").write_text(json.dumps(SCHEMA))
|
||||
(plugin_dir / "manifest.json").write_text(json.dumps({"id": PLUGIN_ID}))
|
||||
|
||||
sentinel = object()
|
||||
originals = {name: getattr(api_v3, name, sentinel) for name in _ATTRS}
|
||||
|
||||
config_manager = ConfigManager(config_path=str(config_file),
|
||||
secrets_path=str(secrets_file))
|
||||
config_manager.template_path = str(tmp_path / "no-template.json")
|
||||
plugin_manager = MagicMock()
|
||||
plugin_manager.plugin_manifests = {PLUGIN_ID: {"id": PLUGIN_ID}}
|
||||
plugin_manager.plugins_dir = plugins_dir
|
||||
plugin_manager.get_plugin.return_value = None
|
||||
|
||||
for name in _ATTRS:
|
||||
setattr(api_v3, name, MagicMock())
|
||||
api_v3.config_manager = config_manager
|
||||
api_v3.schema_manager = SchemaManager(plugins_dir=plugins_dir, project_root=tmp_path)
|
||||
api_v3.plugin_manager = plugin_manager
|
||||
api_v3.operation_queue = None
|
||||
|
||||
app = Flask(__name__)
|
||||
app.config["TESTING"] = True
|
||||
app.register_blueprint(api_v3, url_prefix="/api/v3")
|
||||
|
||||
class Env:
|
||||
client = app.test_client()
|
||||
|
||||
@staticmethod
|
||||
def store(section):
|
||||
config_file.write_text(json.dumps({"timezone": "UTC", PLUGIN_ID: section}))
|
||||
|
||||
@staticmethod
|
||||
def main():
|
||||
return json.loads(config_file.read_text())
|
||||
|
||||
@staticmethod
|
||||
def stored():
|
||||
return json.loads(config_file.read_text())[PLUGIN_ID]
|
||||
|
||||
@staticmethod
|
||||
def secrets():
|
||||
return json.loads(secrets_file.read_text()) if secrets_file.exists() else {}
|
||||
|
||||
@staticmethod
|
||||
def save(config):
|
||||
return app.test_client().post("/api/v3/plugins/config",
|
||||
json={"plugin_id": PLUGIN_ID, "config": config})
|
||||
|
||||
Env.store(json.loads(json.dumps(STORED)))
|
||||
yield Env
|
||||
|
||||
for name, original in originals.items():
|
||||
if original is sentinel:
|
||||
if hasattr(api_v3, name):
|
||||
delattr(api_v3, name)
|
||||
else:
|
||||
setattr(api_v3, name, original)
|
||||
|
||||
|
||||
class TestPartialJsonSave:
|
||||
def test_unsent_settings_keep_their_values(self, env):
|
||||
resp = env.save({"enabled": True})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
stored = env.stored()
|
||||
assert stored["city"] == "Paris"
|
||||
assert stored["update_interval"] == 600
|
||||
assert stored["display"] == {"brightness": 80, "show_icons": False}
|
||||
assert stored["global"]["dynamic_duration"] == {"enabled": False, "min_duration_seconds": 45}
|
||||
|
||||
def test_a_nested_partial_keeps_its_siblings(self, env):
|
||||
resp = env.save({"display": {"brightness": 20}})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert env.stored()["display"] == {"brightness": 20, "show_icons": False}
|
||||
|
||||
def test_a_sent_setting_still_changes(self, env):
|
||||
assert env.save({"city": "Lyon", "enabled": False}).status_code == 200
|
||||
stored = env.stored()
|
||||
assert stored["city"] == "Lyon" and stored["enabled"] is False
|
||||
|
||||
def test_an_invalid_sent_value_is_still_rejected(self, env):
|
||||
resp = env.save({"update_interval": 5})
|
||||
assert resp.status_code == 400
|
||||
assert env.stored()["update_interval"] == 600
|
||||
|
||||
def test_config_must_be_an_object(self, env):
|
||||
assert env.save(["city"]).status_code == 400
|
||||
|
||||
|
||||
class TestCoreOwnedKeysSurviveTheFilter:
|
||||
def test_submitted_core_keys_are_stored(self, env):
|
||||
env.store({"enabled": True, "city": "Paris"})
|
||||
resp = env.save({
|
||||
"vegas_width_pct": 50, "vegas_overflow": "truncate",
|
||||
"vegas_max_width_screens": 2, "skin": "retro",
|
||||
"skin_options": {"accent": "#ff0000"}, "live_priority": True,
|
||||
})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
stored = env.stored()
|
||||
assert stored["vegas_width_pct"] == 50
|
||||
assert stored["vegas_overflow"] == "truncate"
|
||||
assert stored["vegas_max_width_screens"] == 2
|
||||
assert stored["skin"] == "retro"
|
||||
assert stored["skin_options"] == {"accent": "#ff0000"}
|
||||
assert stored["live_priority"] is True
|
||||
|
||||
def test_stored_core_keys_survive_an_unrelated_save(self, env):
|
||||
assert env.save({"city": "Nice"}).status_code == 200
|
||||
stored = env.stored()
|
||||
assert stored["vegas_width_pct"] == 60 and stored["skin"] == "neon"
|
||||
|
||||
def test_a_non_core_unknown_key_is_still_filtered(self, env):
|
||||
assert env.save({"not_in_schema": 1}).status_code == 200
|
||||
assert "not_in_schema" not in env.stored()
|
||||
|
||||
|
||||
class TestLegacyBooleans:
|
||||
LEGACY = {"enabled": True, "city": "Paris", "global": {"dynamic_duration": True}}
|
||||
|
||||
def test_get_returns_the_object_shape(self, env):
|
||||
env.store(dict(self.LEGACY))
|
||||
data = env.client.get(f"/api/v3/plugins/config?plugin_id={PLUGIN_ID}").get_json()["data"]
|
||||
assert data["global"]["dynamic_duration"] == {"enabled": True, "min_duration_seconds": 30}
|
||||
|
||||
def test_get_output_posts_back(self, env):
|
||||
env.store(dict(self.LEGACY))
|
||||
data = env.client.get(f"/api/v3/plugins/config?plugin_id={PLUGIN_ID}").get_json()["data"]
|
||||
resp = env.save(data)
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert env.stored()["global"]["dynamic_duration"] == {"enabled": True, "min_duration_seconds": 30}
|
||||
|
||||
def test_an_unrelated_json_save_upgrades_the_stored_boolean(self, env):
|
||||
env.store(dict(self.LEGACY))
|
||||
resp = env.save({"city": "Lyon"})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert env.stored()["global"]["dynamic_duration"] == {"enabled": True, "min_duration_seconds": 30}
|
||||
|
||||
def test_a_posted_legacy_boolean_is_read_as_the_object(self, env):
|
||||
resp = env.save({"global": {"dynamic_duration": False}})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert env.stored()["global"]["dynamic_duration"]["enabled"] is False
|
||||
|
||||
|
||||
class TestPluginSectionsInConfigMain:
|
||||
def _post(self, env, body):
|
||||
return env.client.post("/api/v3/config/main", json=body)
|
||||
|
||||
def test_a_value_plugins_config_rejects_is_rejected_here_too(self, env):
|
||||
resp = self._post(env, {PLUGIN_ID: {"update_interval": "abc"}})
|
||||
assert resp.status_code == 400
|
||||
assert env.stored()["update_interval"] == 600
|
||||
assert env.save({"update_interval": "abc"}).status_code == 400
|
||||
|
||||
def test_nothing_is_saved_when_a_plugin_section_fails(self, env):
|
||||
resp = self._post(env, {"timezone": "Europe/Paris", PLUGIN_ID: {"update_interval": 1}})
|
||||
assert resp.status_code == 400
|
||||
assert env.main()["timezone"] == "UTC"
|
||||
assert env.stored()["update_interval"] == 600
|
||||
|
||||
def test_partial_section_merges_and_keeps_core_keys(self, env):
|
||||
resp = self._post(env, {PLUGIN_ID: {"city": "Lyon", "vegas_overflow": "rotate"}})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
stored = env.stored()
|
||||
assert stored["city"] == "Lyon"
|
||||
assert stored["display"] == {"brightness": 80, "show_icons": False}
|
||||
assert stored["vegas_overflow"] == "rotate"
|
||||
assert stored["vegas_width_pct"] == 60 and stored["skin"] == "neon"
|
||||
|
||||
def test_secrets_still_go_to_the_secrets_file(self, env):
|
||||
resp = self._post(env, {PLUGIN_ID: {"api_key": "s3cret"}})
|
||||
assert resp.status_code == 200, resp.get_json()
|
||||
assert "api_key" not in env.stored() or env.stored()["api_key"] in ("", None)
|
||||
assert env.secrets()[PLUGIN_ID]["api_key"] == "s3cret"
|
||||
|
||||
def test_a_non_object_section_is_rejected(self, env):
|
||||
assert self._post(env, {PLUGIN_ID: "on"}).status_code == 400
|
||||
Reference in New Issue
Block a user