Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
38a0ea898d |
@@ -4,9 +4,6 @@ on:
|
||||
pull_request:
|
||||
push:
|
||||
branches: [main]
|
||||
# Manual runs against any branch — useful when a PR's automatic run
|
||||
# needs a re-run or didn't get created.
|
||||
workflow_dispatch:
|
||||
|
||||
# Both jobs only check out the repo and run pytest.
|
||||
permissions:
|
||||
@@ -16,12 +13,6 @@ jobs:
|
||||
plugin-safety:
|
||||
name: Plugin safety harness + unit tests
|
||||
runs-on: ubuntu-latest
|
||||
env:
|
||||
# The bundled fixture plugin gives the harness at least one real plugin
|
||||
# to render, and REQUIRE_PLUGINS turns "discovered zero plugins" into a
|
||||
# hard failure instead of a silent all-skip green run.
|
||||
LEDMATRIX_PLUGINS_DIR: test/fixtures/plugins
|
||||
LEDMATRIX_REQUIRE_PLUGINS: "1"
|
||||
steps:
|
||||
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
|
||||
with:
|
||||
@@ -38,9 +29,12 @@ jobs:
|
||||
pip install -r requirements.txt -r requirements-test.txt
|
||||
pip install RGBMatrixEmulator
|
||||
|
||||
- name: Run plugin safety harness
|
||||
- name: Run harness + visual rendering tests
|
||||
run: |
|
||||
pytest --no-cov test/plugins/
|
||||
pytest --no-cov \
|
||||
test/plugins/test_harness.py \
|
||||
test/plugins/test_visual_rendering.py \
|
||||
test/plugins/test_plugin_matrix.py
|
||||
|
||||
unit-tests:
|
||||
name: Core unit tests
|
||||
@@ -61,15 +55,36 @@ jobs:
|
||||
pip install -r requirements.txt -r requirements-test.txt
|
||||
pip install RGBMatrixEmulator
|
||||
|
||||
# Run the ENTIRE test tree (except test/plugins, which the
|
||||
# plugin-safety job owns). New test files are enrolled automatically;
|
||||
# excluding anything requires a visible, commented --ignore here.
|
||||
# Coverage is measured and enforced only in this step — pytest.ini
|
||||
# deliberately carries no coverage flags so local runs stay fast.
|
||||
# Safety net for the shared sports/scroll/style infrastructure. These
|
||||
# suites existed but were not enrolled in CI, so a refactor of
|
||||
# src/base_classes or src/common could regress them silently. Enrolled
|
||||
# explicitly (not `pytest test/`) so known hardware-only suites don't
|
||||
# break CI; grow this list as more suites are made headless.
|
||||
- name: Run core unit suites
|
||||
run: |
|
||||
pytest -m "not hardware" test/ \
|
||||
--ignore=test/plugins \
|
||||
--cov=src --cov=web_interface \
|
||||
--cov-report=term \
|
||||
--cov-fail-under=45
|
||||
pytest --no-cov \
|
||||
test/test_skin_system.py \
|
||||
test/test_font_manager.py \
|
||||
test/test_data_sources.py \
|
||||
test/test_api_extractors.py \
|
||||
test/test_scroll_helper.py \
|
||||
test/test_scroll_helper_continuous.py \
|
||||
test/test_adaptive_layout.py \
|
||||
test/test_loader_compat_warning.py \
|
||||
test/test_sports_base_characterization.py \
|
||||
test/test_element_style.py \
|
||||
test/test_sports_core_promotions.py \
|
||||
test/test_sports_modes_promotions.py \
|
||||
test/test_sports_capabilities.py \
|
||||
test/test_sports_scroll.py \
|
||||
test/test_version_consistency.py \
|
||||
test/test_plugin_compatibility_gate.py \
|
||||
test/test_install_preserves_existing.py \
|
||||
test/test_core_owned_config_keys.py \
|
||||
test/test_async_plugin_updates.py \
|
||||
test/test_plugin_update_reservation.py \
|
||||
test/test_template_targets.py \
|
||||
test/test_widget_scripts.py \
|
||||
test/test_enum_option_labels.py \
|
||||
test/test_doc_links.py \
|
||||
test/web_interface/test_cache.py
|
||||
|
||||
@@ -206,6 +206,45 @@ To use an existing widget in your plugin's `config_schema.json`, simply add the
|
||||
|
||||
The widget will be automatically rendered when the plugin configuration form is loaded.
|
||||
|
||||
## Labelling Enum Options (`x-options.labels`)
|
||||
|
||||
A plain `enum` renders as a dropdown whose option text is the value with
|
||||
underscores replaced and title case applied — `day_first` becomes "Day First".
|
||||
That is fine for values that read as their own label, and wrong for values that
|
||||
do not: `vs` becomes "Vs", and `abbrev` says nothing about the `Sep 19` it
|
||||
actually produces.
|
||||
|
||||
Supply `x-options.labels` to set the visible text. This is the same convention
|
||||
the `checkbox-group` widget uses:
|
||||
|
||||
```json
|
||||
{
|
||||
"date_format": {
|
||||
"type": "string",
|
||||
"enum": ["abbrev", "numeric", "day_first"],
|
||||
"default": "abbrev",
|
||||
"x-options": {
|
||||
"labels": {
|
||||
"abbrev": "Sep 19",
|
||||
"numeric": "9/19",
|
||||
"day_first": "19 Sep"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Labels are **display only** — the stored value is still the enum value, so
|
||||
adding them never changes a saved config. The map may be partial: any value
|
||||
without a label keeps the humanised fallback. Older cores that predate this
|
||||
support ignore `x-options` and render the fallback for every option, so a
|
||||
plugin can ship labels without requiring a core upgrade.
|
||||
|
||||
Array-table columns (`x-widget: array-table`) accept the same
|
||||
`x-options.labels` on a column definition, but their fallback is the **raw
|
||||
value** rather than the humanised one, because those columns hold values such
|
||||
as ticker symbols where `aapl` → "Aapl" would be wrong.
|
||||
|
||||
## Marking Fields as Advanced (`x-advanced`)
|
||||
|
||||
Add `"x-advanced": true` to any top-level, non-object property to move it out
|
||||
|
||||
@@ -10,13 +10,16 @@ python_functions = test_*
|
||||
testpaths = test
|
||||
|
||||
# Output options
|
||||
# Coverage is deliberately NOT configured here: a bare local `pytest` should
|
||||
# be fast and dependency-light. Coverage is measured and enforced in exactly
|
||||
# one place — the unit-tests job in .github/workflows/test.yml.
|
||||
# Note: Coverage options require pytest-cov to be installed
|
||||
# Run: pip install pytest-cov
|
||||
addopts =
|
||||
-v
|
||||
--strict-markers
|
||||
--tb=short
|
||||
--cov=src
|
||||
--cov-report=term-missing
|
||||
--cov-report=html
|
||||
--cov-fail-under=30
|
||||
|
||||
# Markers
|
||||
markers =
|
||||
|
||||
@@ -4,6 +4,4 @@ pytest>=9.0.3,<10.0.0
|
||||
pytest-cov>=4.1.0,<5.0.0
|
||||
pytest-mock>=3.11.0,<4.0.0
|
||||
freezegun>=1.2,<2 # deterministic time for golden-image tests
|
||||
psutil>=6.0.0,<8.0.0 # optional at runtime; installed for tests so the
|
||||
# /system/status endpoint's real path is exercised
|
||||
mypy>=1.5.0,<2.0.0 # static type checking (also pinned in .pre-commit-config.yaml)
|
||||
|
||||
@@ -383,15 +383,10 @@ class SportsCore(ABC):
|
||||
ctx = skin_runtime.build_context(self, game, size=size)
|
||||
card = skin.render_vegas_card(ctx, dict(game))
|
||||
if card is not None:
|
||||
# A successful render clears accumulated strikes, mirroring
|
||||
# _render_game — transient failures must not add up across
|
||||
# the session and disable a working skin.
|
||||
self._skin_failures = 0
|
||||
return card
|
||||
ctx = skin_runtime.build_context(self, game, size=size)
|
||||
render = getattr(skin, f"render_{self.SKIN_MODE}")
|
||||
if render(ctx, dict(game)):
|
||||
self._skin_failures = 0
|
||||
return ctx.canvas
|
||||
except Exception:
|
||||
# Card failures count toward the same 3-strike session disable
|
||||
|
||||
@@ -106,9 +106,14 @@ class ConfigManager:
|
||||
Returns:
|
||||
SaveResult with status and details
|
||||
"""
|
||||
# Load current secrets to preserve them (raises if unreadable — see
|
||||
# _load_secrets_for_save)
|
||||
secrets_content = self._load_secrets_for_save()
|
||||
# Load current secrets to preserve them
|
||||
secrets_content = {}
|
||||
if os.path.exists(self.secrets_path):
|
||||
try:
|
||||
with open(self.secrets_path, 'r') as f_secrets:
|
||||
secrets_content = json.load(f_secrets)
|
||||
except Exception as e:
|
||||
self.logger.warning(f"Could not load secrets file {self.secrets_path} during save: {e}")
|
||||
|
||||
# Strip secrets from main config before saving
|
||||
config_to_write = self._strip_secrets_recursive(new_config_data, secrets_content)
|
||||
@@ -285,43 +290,19 @@ class ConfigManager:
|
||||
result[key] = value
|
||||
return result
|
||||
|
||||
def _load_secrets_for_save(self) -> Dict[str, Any]:
|
||||
"""Load config_secrets.json for stripping before a save.
|
||||
|
||||
A missing secrets file is fine (nothing to strip). But a file that
|
||||
EXISTS and cannot be read or parsed means stripping is impossible —
|
||||
and the in-memory config being saved has secrets deep-merged into it,
|
||||
so proceeding would write them into config.json in plaintext. That
|
||||
was the historical behavior; it is now a hard refusal. The save
|
||||
raises so the caller (and user) fixes the secrets file instead of
|
||||
silently leaking its contents into the world-readable main config.
|
||||
"""
|
||||
if not os.path.exists(self.secrets_path):
|
||||
return {}
|
||||
def save_config(self, new_config_data: Dict[str, Any]) -> None:
|
||||
"""Save configuration to the main JSON file, stripping out secrets."""
|
||||
secrets_content = {}
|
||||
if os.path.exists(self.secrets_path):
|
||||
try:
|
||||
with open(self.secrets_path, 'r') as f_secrets:
|
||||
return json.load(f_secrets)
|
||||
# Only the expected read/parse failures — an unexpected implementation
|
||||
# error should propagate as itself, not masquerade as a secrets-file
|
||||
# problem. (JSONDecodeError and UnicodeDecodeError are ValueErrors.)
|
||||
except (OSError, ValueError, RecursionError) as e:
|
||||
error_msg = (
|
||||
f"Refusing to save config: secrets file {self.secrets_path} exists "
|
||||
f"but could not be loaded ({e}). Saving without it would write "
|
||||
f"merged secret values into config.json in plaintext. Fix or "
|
||||
f"remove the secrets file, then retry."
|
||||
)
|
||||
self.logger.error("[Config] %s", error_msg, exc_info=True)
|
||||
raise ConfigError(error_msg, config_path=self.secrets_path) from e
|
||||
|
||||
def save_config(self, new_config_data: Dict[str, Any]) -> None:
|
||||
"""Save configuration to the main JSON file, stripping out secrets.
|
||||
|
||||
Raises ConfigError when the secrets file exists but cannot be loaded,
|
||||
because stripping would be impossible and secrets would leak into
|
||||
config.json.
|
||||
"""
|
||||
secrets_content = self._load_secrets_for_save()
|
||||
secrets_content = json.load(f_secrets)
|
||||
except Exception as e:
|
||||
self.logger.warning(f"Could not load secrets file {self.secrets_path} during save: {e}")
|
||||
# Continue without stripping if secrets can't be loaded, or handle as critical error
|
||||
# For now, we'll proceed cautiously and save the full new_config_data if secrets are unreadable
|
||||
# to prevent accidental data loss if the secrets file is temporarily corrupt.
|
||||
# A more robust approach might be to fail the save or use a cached version of secrets.
|
||||
|
||||
config_to_write = self._strip_secrets_recursive(new_config_data, secrets_content)
|
||||
|
||||
|
||||
@@ -180,45 +180,6 @@ def declared_min_version(manifest: Dict[str, Any]) -> Optional[str]:
|
||||
return None
|
||||
|
||||
|
||||
def is_update_available(installed_version: str, latest_version: str) -> bool:
|
||||
"""Return True when the registry's ``latest_version`` is strictly newer
|
||||
than the installed version.
|
||||
|
||||
THE shared comparator for "should this plugin be updated?" — used by both
|
||||
the web UI's update badge (`api_v3._is_plugin_update_available`) and the
|
||||
store's `update_plugin` reinstall decision, so the two can never disagree.
|
||||
|
||||
Uses PEP 440-aware comparison (``packaging``), which also normalizes
|
||||
equivalent spellings: ``v1.2.0`` == ``1.2.0`` and ``1.2`` == ``1.2.0``, so
|
||||
cosmetic differences never trigger a reinstall — and a locally modified
|
||||
plugin whose version is *ahead* of the registry is never "updated"
|
||||
(downgraded). If either version string can't be parsed the mismatch is
|
||||
surfaced (True) so the user can reconcile, rather than silently hiding a
|
||||
potential update.
|
||||
"""
|
||||
if not installed_version or not latest_version:
|
||||
return False
|
||||
if not isinstance(installed_version, str) or not isinstance(latest_version, str):
|
||||
# A malformed manifest/registry can carry a number (1.2) or worse;
|
||||
# packaging would raise TypeError. Surface the mismatch instead.
|
||||
return True
|
||||
if installed_version == latest_version:
|
||||
return False
|
||||
try:
|
||||
from packaging.version import parse as _parse_version, InvalidVersion
|
||||
except ImportError:
|
||||
# packaging is a core dependency, but if it's somehow unavailable we
|
||||
# can't compare semantically — surface the mismatch we already know
|
||||
# exists (the two strings differ).
|
||||
return True
|
||||
try:
|
||||
return _parse_version(latest_version) > _parse_version(installed_version)
|
||||
except InvalidVersion:
|
||||
# Unparseable version string: we can't tell direction, so surface the
|
||||
# mismatch rather than silently hiding a potential update.
|
||||
return True
|
||||
|
||||
|
||||
def check(manifest: Dict[str, Any], core_version: str) -> Tuple[bool, Optional[str]]:
|
||||
"""Return ``(compatible, reason)``.
|
||||
|
||||
|
||||
@@ -2969,10 +2969,7 @@ class PluginStoreManager:
|
||||
remote_branch = plugin_info_remote.get('branch') or plugin_info_remote.get('default_branch')
|
||||
|
||||
# Compare local manifest version against registry latest_version
|
||||
# to avoid unnecessary reinstalls for monorepo plugins. Uses the
|
||||
# same semantic comparator as the web UI's update badge, so
|
||||
# equivalent spellings ("v1.2.0" vs "1.2.0") never trigger a
|
||||
# reinstall and a locally-ahead version is never downgraded.
|
||||
# to avoid unnecessary reinstalls for monorepo plugins
|
||||
try:
|
||||
local_manifest_path = plugin_path / "manifest.json"
|
||||
if local_manifest_path.exists():
|
||||
@@ -2980,16 +2977,8 @@ class PluginStoreManager:
|
||||
local_manifest = json.load(f)
|
||||
local_version = local_manifest.get('version', '')
|
||||
remote_version = plugin_info_remote.get('latest_version', '')
|
||||
from src.plugin_system.compatibility import is_update_available
|
||||
# No truthiness gate: the shared comparator already treats
|
||||
# a missing version on either side as "no update", and the
|
||||
# store must agree with the UI badge in that case too. A
|
||||
# missing manifest (not just a missing version field)
|
||||
# still falls through to the reinstall recovery path.
|
||||
if not is_update_available(local_version, remote_version):
|
||||
self.logger.info(
|
||||
f"Plugin {plugin_id} already at latest version "
|
||||
f"(installed {local_version}, registry {remote_version})")
|
||||
if local_version and remote_version and local_version == remote_version:
|
||||
self.logger.info(f"Plugin {plugin_id} already at latest version {local_version}")
|
||||
return True
|
||||
except Exception as e:
|
||||
self.logger.debug(f"Could not compare versions for {plugin_id}: {e}")
|
||||
|
||||
@@ -1,31 +0,0 @@
|
||||
{
|
||||
"$schema": "http://json-schema.org/draft-07/schema#",
|
||||
"title": "CI Fixture Plugin",
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"enabled": {
|
||||
"type": "boolean",
|
||||
"default": true
|
||||
},
|
||||
"display_duration": {
|
||||
"type": "number",
|
||||
"default": 5
|
||||
},
|
||||
"border_color": {
|
||||
"type": "array",
|
||||
"items": {"type": "integer", "minimum": 0, "maximum": 255},
|
||||
"minItems": 3,
|
||||
"maxItems": 3,
|
||||
"default": [0, 255, 0],
|
||||
"description": "RGB color of the border rectangle."
|
||||
},
|
||||
"diagonal_color": {
|
||||
"type": "array",
|
||||
"items": {"type": "integer", "minimum": 0, "maximum": 255},
|
||||
"minItems": 3,
|
||||
"maxItems": 3,
|
||||
"default": [255, 0, 0],
|
||||
"description": "RGB color of the diagonals."
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,44 +0,0 @@
|
||||
"""
|
||||
CI fixture plugin.
|
||||
|
||||
Exists so the plugin safety harness (test/plugins/test_plugin_matrix.py and
|
||||
the plugin-safety CI job) always has at least one real plugin to load and
|
||||
render — without it, an empty plugins/ directory turns the whole job into a
|
||||
green no-op. The render is deliberately trivial and fully deterministic:
|
||||
a border rectangle plus both diagonals, sized from the display manager's
|
||||
declared dimensions. No fonts, no network, no time dependence, so golden
|
||||
images are stable across platforms.
|
||||
"""
|
||||
|
||||
from PIL import ImageDraw
|
||||
|
||||
from src.plugin_system.base_plugin import BasePlugin
|
||||
|
||||
|
||||
class CIFixturePlugin(BasePlugin):
|
||||
"""Deterministic CI-only fixture plugin: renders a border + diagonals
|
||||
pattern sized from the display's declared dimensions. Never shipped to
|
||||
devices; exists solely so the plugin safety harness has a real plugin
|
||||
to exercise in CI."""
|
||||
|
||||
def update(self) -> None:
|
||||
"""Nothing to fetch — the render is self-contained."""
|
||||
|
||||
def display(self, force_clear: bool = False) -> None:
|
||||
self.display_manager.clear()
|
||||
width = self.display_manager.matrix.width
|
||||
height = self.display_manager.matrix.height
|
||||
border = tuple(self.config.get("border_color", [0, 255, 0]))
|
||||
diagonal = tuple(self.config.get("diagonal_color", [255, 0, 0]))
|
||||
|
||||
image = self.display_manager.image
|
||||
draw = ImageDraw.Draw(image)
|
||||
# Blank only the declared panel area, then draw edge-to-edge content:
|
||||
# the border proves the plugin reads dynamic dimensions (any overflow
|
||||
# or underfill at any size is a harness bug or a dimensions bug), the
|
||||
# diagonals make golden comparisons sensitive to size/offset drift.
|
||||
draw.rectangle([0, 0, width - 1, height - 1], fill=(0, 0, 0))
|
||||
draw.rectangle([0, 0, width - 1, height - 1], outline=border)
|
||||
draw.line([0, 0, width - 1, height - 1], fill=diagonal)
|
||||
draw.line([0, height - 1, width - 1, 0], fill=diagonal)
|
||||
self.display_manager.update_display()
|
||||
@@ -1,13 +0,0 @@
|
||||
{
|
||||
"id": "ci-fixture-plugin",
|
||||
"name": "CI Fixture Plugin",
|
||||
"version": "1.0.0",
|
||||
"description": "Bundled test fixture so the plugin safety harness always has at least one real plugin to render in CI. Draws a deterministic border + diagonals pattern at any panel size. Not installable from the store and never shipped to devices.",
|
||||
"author": "LEDMatrix",
|
||||
"entry_point": "manager.py",
|
||||
"class_name": "CIFixturePlugin",
|
||||
"display_modes": ["ci-fixture"],
|
||||
"update_interval": 3600,
|
||||
"min_ledmatrix_version": "2.0.0",
|
||||
"compatible_versions": [">=2.0.0"]
|
||||
}
|
||||
@@ -1,7 +0,0 @@
|
||||
# No dependencies — the fixture must load in any environment.
|
||||
#
|
||||
# Pillow is deliberately NOT pinned here even though manager.py imports
|
||||
# PIL: it is a core LEDMatrix dependency (see the repo-root
|
||||
# requirements.txt), so it is always present wherever the harness runs,
|
||||
# and the harness loads plugins with install_deps=False anyway. Pinning
|
||||
# it here would only invite a needless pip install during test runs.
|
||||
|
Before Width: | Height: | Size: 359 B |
|
Before Width: | Height: | Size: 586 B |
|
Before Width: | Height: | Size: 849 B |
|
Before Width: | Height: | Size: 1.2 KiB |
|
Before Width: | Height: | Size: 395 B |
|
Before Width: | Height: | Size: 319 B |
|
Before Width: | Height: | Size: 466 B |
|
Before Width: | Height: | Size: 454 B |
@@ -23,17 +23,9 @@ os.environ['EMULATOR'] = 'true'
|
||||
def plugins_dir() -> Path:
|
||||
"""Get the plugins directory path.
|
||||
|
||||
Honors LEDMATRIX_PLUGINS_DIR (first entry) when set — the same override
|
||||
test_plugin_matrix.py uses, so CI can point every plugin suite at the
|
||||
bundled fixture plugins. Otherwise checks plugins/ first, then falls
|
||||
back to plugin-repos/ for monorepo development environments.
|
||||
Checks plugins/ first, then falls back to plugin-repos/
|
||||
for monorepo development environments.
|
||||
"""
|
||||
env = os.environ.get('LEDMATRIX_PLUGINS_DIR')
|
||||
if env:
|
||||
first = env.split(os.pathsep)[0]
|
||||
if first:
|
||||
return Path(first)
|
||||
|
||||
plugins_path = project_root / 'plugins'
|
||||
plugin_repos_path = project_root / 'plugin-repos'
|
||||
|
||||
|
||||
@@ -1,10 +1,5 @@
|
||||
"""
|
||||
Integration tests for basketball-scoreboard plugin.
|
||||
|
||||
Requires the real plugin to be installed (plugins/ or plugin-repos/,
|
||||
or the dir named by LEDMATRIX_PLUGINS_DIR) — on machines without it,
|
||||
every test here skips by design. CI covers plugin safety with the
|
||||
bundled fixture plugin via test_plugin_matrix.py instead.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -1,10 +1,5 @@
|
||||
"""
|
||||
Integration tests for calendar plugin.
|
||||
|
||||
Requires the real plugin to be installed (plugins/ or plugin-repos/,
|
||||
or the dir named by LEDMATRIX_PLUGINS_DIR) — on machines without it,
|
||||
every test here skips by design. CI covers plugin safety with the
|
||||
bundled fixture plugin via test_plugin_matrix.py instead.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -1,10 +1,5 @@
|
||||
"""
|
||||
Integration tests for clock-simple plugin.
|
||||
|
||||
Requires the real plugin to be installed (plugins/ or plugin-repos/,
|
||||
or the dir named by LEDMATRIX_PLUGINS_DIR) — on machines without it,
|
||||
every test here skips by design. CI covers plugin safety with the
|
||||
bundled fixture plugin via test_plugin_matrix.py instead.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -1,10 +1,5 @@
|
||||
"""
|
||||
Integration tests for odds-ticker plugin.
|
||||
|
||||
Requires the real plugin to be installed (plugins/ or plugin-repos/,
|
||||
or the dir named by LEDMATRIX_PLUGINS_DIR) — on machines without it,
|
||||
every test here skips by design. CI covers plugin safety with the
|
||||
bundled fixture plugin via test_plugin_matrix.py instead.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -1,10 +1,5 @@
|
||||
"""
|
||||
Integration tests for soccer-scoreboard plugin.
|
||||
|
||||
Requires the real plugin to be installed (plugins/ or plugin-repos/,
|
||||
or the dir named by LEDMATRIX_PLUGINS_DIR) — on machines without it,
|
||||
every test here skips by design. CI covers plugin safety with the
|
||||
bundled fixture plugin via test_plugin_matrix.py instead.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -1,10 +1,5 @@
|
||||
"""
|
||||
Integration tests for text-display plugin.
|
||||
|
||||
Requires the real plugin to be installed (plugins/ or plugin-repos/,
|
||||
or the dir named by LEDMATRIX_PLUGINS_DIR) — on machines without it,
|
||||
every test here skips by design. CI covers plugin safety with the
|
||||
bundled fixture plugin via test_plugin_matrix.py instead.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -1,110 +0,0 @@
|
||||
"""
|
||||
Tests for BasePlugin.get_display_duration — ~100 lines of type coercion that
|
||||
every plugin's rotation slot depends on, previously untested.
|
||||
|
||||
The contract: a positive number wins wherever it comes from; everything else
|
||||
falls through instance attr → config → the 15.0 default, logging on the way.
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from src.plugin_system.base_plugin import BasePlugin
|
||||
|
||||
|
||||
class _MinimalPlugin(BasePlugin):
|
||||
def update(self):
|
||||
pass
|
||||
|
||||
def display(self, force_clear=False):
|
||||
pass
|
||||
|
||||
|
||||
def make_plugin(config=None, instance_duration="__unset__"):
|
||||
plugin = _MinimalPlugin(
|
||||
plugin_id="duration-test",
|
||||
config=config or {},
|
||||
display_manager=MagicMock(),
|
||||
cache_manager=MagicMock(),
|
||||
plugin_manager=MagicMock(),
|
||||
)
|
||||
if instance_duration != "__unset__":
|
||||
plugin.display_duration = instance_duration
|
||||
return plugin
|
||||
|
||||
|
||||
class TestInstanceVariable:
|
||||
def test_positive_int_wins(self):
|
||||
assert make_plugin(instance_duration=30).get_display_duration() == 30.0
|
||||
|
||||
def test_positive_float_wins(self):
|
||||
assert make_plugin(instance_duration=12.5).get_display_duration() == 12.5
|
||||
|
||||
def test_returns_float_type(self):
|
||||
result = make_plugin(instance_duration=30).get_display_duration()
|
||||
assert isinstance(result, float)
|
||||
|
||||
def test_numeric_string_wins(self):
|
||||
assert make_plugin(instance_duration="25").get_display_duration() == 25.0
|
||||
|
||||
def test_zero_falls_through_to_config(self):
|
||||
plugin = make_plugin(config={"display_duration": 20},
|
||||
instance_duration=0)
|
||||
assert plugin.get_display_duration() == 20.0
|
||||
|
||||
def test_negative_falls_through_to_config(self):
|
||||
plugin = make_plugin(config={"display_duration": 20},
|
||||
instance_duration=-5)
|
||||
assert plugin.get_display_duration() == 20.0
|
||||
|
||||
def test_none_falls_through_to_config(self):
|
||||
plugin = make_plugin(config={"display_duration": 20},
|
||||
instance_duration=None)
|
||||
assert plugin.get_display_duration() == 20.0
|
||||
|
||||
def test_garbage_string_falls_through(self):
|
||||
plugin = make_plugin(config={"display_duration": 20},
|
||||
instance_duration="abc")
|
||||
assert plugin.get_display_duration() == 20.0
|
||||
|
||||
def test_non_positive_string_falls_through(self):
|
||||
plugin = make_plugin(config={"display_duration": 20},
|
||||
instance_duration="0")
|
||||
assert plugin.get_display_duration() == 20.0
|
||||
|
||||
def test_unexpected_type_falls_through(self):
|
||||
plugin = make_plugin(config={"display_duration": 20},
|
||||
instance_duration=[30])
|
||||
assert plugin.get_display_duration() == 20.0
|
||||
|
||||
def test_bool_true_is_one_second(self):
|
||||
# Characterized quirk: bool is an int subclass, so display_duration =
|
||||
# True passes the isinstance((int, float)) branch and returns 1.0.
|
||||
assert make_plugin(instance_duration=True).get_display_duration() == 1.0
|
||||
|
||||
|
||||
class TestConfigFallback:
|
||||
def test_config_number(self):
|
||||
assert make_plugin({"display_duration": 20}).get_display_duration() == 20.0
|
||||
|
||||
def test_config_numeric_string(self):
|
||||
assert make_plugin({"display_duration": "12.5"}).get_display_duration() == 12.5
|
||||
|
||||
def test_missing_config_uses_default(self):
|
||||
assert make_plugin({}).get_display_duration() == 15.0
|
||||
|
||||
def test_config_zero_uses_default(self):
|
||||
assert make_plugin({"display_duration": 0}).get_display_duration() == 15.0
|
||||
|
||||
def test_config_negative_uses_default(self):
|
||||
assert make_plugin({"display_duration": -10}).get_display_duration() == 15.0
|
||||
|
||||
def test_config_garbage_string_uses_default(self):
|
||||
assert make_plugin({"display_duration": "soon"}).get_display_duration() == 15.0
|
||||
|
||||
def test_config_unexpected_type_uses_default(self):
|
||||
assert make_plugin({"display_duration": {"s": 5}}).get_display_duration() == 15.0
|
||||
|
||||
def test_config_none_uses_default(self):
|
||||
assert make_plugin({"display_duration": None}).get_display_duration() == 15.0
|
||||
@@ -1,271 +0,0 @@
|
||||
"""
|
||||
Tests for src/plugin_system/compatibility.py — the "can this plugin run on
|
||||
this core?" gate used by both the plugin loader (advisory) and the store
|
||||
manager (blocking at install/update time).
|
||||
|
||||
This module had zero direct test coverage despite guarding every install.
|
||||
These tests pin the documented contract: refuse only on evidence, resolve
|
||||
every uncertain case (unparseable versions, missing fields, untrustworthy
|
||||
core) to compatible.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from src.plugin_system.compatibility import (
|
||||
TRUSTWORTHY_FLOOR,
|
||||
parse_semver,
|
||||
_parse_strict,
|
||||
_satisfies_range,
|
||||
satisfies_compatible_versions,
|
||||
declared_min_version,
|
||||
check,
|
||||
)
|
||||
|
||||
|
||||
class TestParseSemver:
|
||||
def test_plain_triplet(self):
|
||||
assert parse_semver("1.2.3") == (1, 2, 3)
|
||||
|
||||
def test_leading_v_tolerated(self):
|
||||
assert parse_semver("v3.2.1") == (3, 2, 1)
|
||||
|
||||
def test_prerelease_suffix_stripped(self):
|
||||
# "3.2.0-rc1" must NOT parse as (3, 2, 1) — a release candidate must
|
||||
# not rank above its own release.
|
||||
assert parse_semver("3.2.0-rc1") == (3, 2, 0)
|
||||
|
||||
def test_build_suffix_stripped(self):
|
||||
# "3.2.0+build42" must NOT parse as (3, 2, 42).
|
||||
assert parse_semver("3.2.0+build42") == (3, 2, 0)
|
||||
|
||||
def test_two_part_version_pads_zero(self):
|
||||
assert parse_semver("1.2") == (1, 2, 0)
|
||||
|
||||
def test_one_part_version_pads_zeros(self):
|
||||
assert parse_semver("2") == (2, 0, 0)
|
||||
|
||||
def test_extra_parts_ignored(self):
|
||||
assert parse_semver("1.2.3.4") == (1, 2, 3)
|
||||
|
||||
def test_non_string_returns_none(self):
|
||||
assert parse_semver(None) is None
|
||||
assert parse_semver(123) is None
|
||||
assert parse_semver((1, 2, 3)) is None
|
||||
|
||||
def test_garbage_with_no_digits_is_lenient_zero(self):
|
||||
# Documented leniency: digit-scraping yields (0, 0, 0) for pure
|
||||
# garbage. Fine for a floor (0.0.0 never blocks), wrong for ranges —
|
||||
# which is why ranges go through _parse_strict instead.
|
||||
assert parse_semver("garbage") == (0, 0, 0)
|
||||
|
||||
def test_whitespace_stripped(self):
|
||||
assert parse_semver(" 1.2.3 ") == (1, 2, 3)
|
||||
|
||||
|
||||
class TestParseStrict:
|
||||
def test_accepts_real_versions(self):
|
||||
assert _parse_strict("1.2.3") == (1, 2, 3)
|
||||
assert _parse_strict("v1.2.3-rc1") == (1, 2, 3)
|
||||
assert _parse_strict("2.0") == (2, 0, 0)
|
||||
|
||||
def test_rejects_garbage(self):
|
||||
assert _parse_strict("not-a-version") is None
|
||||
assert _parse_strict("") is None
|
||||
|
||||
def test_rejects_non_string(self):
|
||||
assert _parse_strict(None) is None
|
||||
|
||||
|
||||
class TestSatisfiesRange:
|
||||
CORE = (3, 1, 0)
|
||||
|
||||
@pytest.mark.parametrize("spec,expected", [
|
||||
(">=3.0.0", True),
|
||||
(">=3.1.0", True),
|
||||
(">=3.2.0", False),
|
||||
("<=3.1.0", True),
|
||||
("<=3.0.9", False),
|
||||
(">3.0.9", True),
|
||||
(">3.1.0", False),
|
||||
("<3.2.0", True),
|
||||
("<3.1.0", False),
|
||||
])
|
||||
def test_comparison_operators(self, spec, expected):
|
||||
assert _satisfies_range(self.CORE, spec) is expected
|
||||
|
||||
def test_tilde_allows_patch_only(self):
|
||||
# ~3.1.0 means >=3.1.0, <3.2.0
|
||||
assert _satisfies_range((3, 1, 5), "~3.1.0") is True
|
||||
assert _satisfies_range((3, 2, 0), "~3.1.0") is False
|
||||
assert _satisfies_range((3, 0, 9), "~3.1.0") is False
|
||||
|
||||
def test_caret_allows_minor_and_patch(self):
|
||||
# ^3.1.0 means >=3.1.0, <4.0.0
|
||||
assert _satisfies_range((3, 9, 9), "^3.1.0") is True
|
||||
assert _satisfies_range((4, 0, 0), "^3.1.0") is False
|
||||
assert _satisfies_range((3, 0, 0), "^3.1.0") is False
|
||||
|
||||
def test_bare_exact_version(self):
|
||||
assert _satisfies_range((3, 1, 0), "3.1.0") is True
|
||||
assert _satisfies_range((3, 1, 1), "3.1.0") is False
|
||||
|
||||
def test_inclusive_dash_range(self):
|
||||
assert _satisfies_range((2, 5, 0), "2.0.0 - 3.1.0") is True
|
||||
assert _satisfies_range((2, 0, 0), "2.0.0 - 3.1.0") is True
|
||||
assert _satisfies_range((3, 1, 0), "2.0.0 - 3.1.0") is True
|
||||
assert _satisfies_range((3, 1, 1), "2.0.0 - 3.1.0") is False
|
||||
|
||||
def test_unparseable_spec_returns_none_not_false(self):
|
||||
# Garbage must read as "no evidence", never as a refusal — an
|
||||
# unrecognised spelling must not cost a user a working install.
|
||||
assert _satisfies_range(self.CORE, "banana") is None
|
||||
assert _satisfies_range(self.CORE, ">=banana") is None
|
||||
assert _satisfies_range(self.CORE, "") is None
|
||||
assert _satisfies_range(self.CORE, "banana - 3.0.0") is None
|
||||
|
||||
|
||||
class TestSatisfiesCompatibleVersions:
|
||||
def test_any_entry_satisfying_wins(self):
|
||||
manifest = {"compatible_versions": ["<1.0.0", ">=3.0.0"]}
|
||||
assert satisfies_compatible_versions(manifest, (3, 1, 0)) is True
|
||||
|
||||
def test_all_entries_failing_is_false(self):
|
||||
manifest = {"compatible_versions": ["<1.0.0", "2.0.0 - 2.9.9"]}
|
||||
assert satisfies_compatible_versions(manifest, (3, 1, 0)) is False
|
||||
|
||||
def test_absent_field_returns_none(self):
|
||||
assert satisfies_compatible_versions({}, (3, 1, 0)) is None
|
||||
|
||||
def test_empty_list_returns_none(self):
|
||||
assert satisfies_compatible_versions(
|
||||
{"compatible_versions": []}, (3, 1, 0)) is None
|
||||
|
||||
def test_non_list_returns_none(self):
|
||||
assert satisfies_compatible_versions(
|
||||
{"compatible_versions": ">=2.0.0"}, (3, 1, 0)) is None
|
||||
|
||||
def test_all_unparseable_entries_returns_none(self):
|
||||
manifest = {"compatible_versions": ["banana", 42, None]}
|
||||
assert satisfies_compatible_versions(manifest, (3, 1, 0)) is None
|
||||
|
||||
def test_mixed_parseable_and_garbage_uses_parseable(self):
|
||||
manifest = {"compatible_versions": ["banana", ">=3.0.0"]}
|
||||
assert satisfies_compatible_versions(manifest, (3, 1, 0)) is True
|
||||
|
||||
|
||||
class TestDeclaredMinVersion:
|
||||
def test_top_level_field(self):
|
||||
assert declared_min_version({"min_ledmatrix_version": "2.1.0"}) == "2.1.0"
|
||||
|
||||
def test_requires_dict_fallback(self):
|
||||
manifest = {"requires": {"min_ledmatrix_version": "2.2.0"}}
|
||||
assert declared_min_version(manifest) == "2.2.0"
|
||||
|
||||
def test_versions_array_fallback(self):
|
||||
manifest = {"versions": [{"ledmatrix_min_version": "2.3.0"}]}
|
||||
assert declared_min_version(manifest) == "2.3.0"
|
||||
|
||||
def test_versions_array_deprecated_spelling(self):
|
||||
manifest = {"versions": [{"ledmatrix_min": "2.4.0"}]}
|
||||
assert declared_min_version(manifest) == "2.4.0"
|
||||
|
||||
def test_top_level_wins_over_versions_array(self):
|
||||
manifest = {
|
||||
"min_ledmatrix_version": "2.1.0",
|
||||
"versions": [{"ledmatrix_min_version": "9.9.9"}],
|
||||
}
|
||||
assert declared_min_version(manifest) == "2.1.0"
|
||||
|
||||
def test_requires_as_list_does_not_raise(self):
|
||||
# A hand-edited manifest can carry `requires` as a list; this used to
|
||||
# raise AttributeError and one malformed manifest would take down the
|
||||
# whole install path.
|
||||
assert declared_min_version({"requires": ["something"]}) is None
|
||||
|
||||
def test_versions_as_dict_does_not_raise(self):
|
||||
# Same for `versions` as a mapping (used to raise KeyError).
|
||||
assert declared_min_version({"versions": {"0": {}}}) is None
|
||||
|
||||
def test_nothing_declared_returns_none(self):
|
||||
assert declared_min_version({}) is None
|
||||
|
||||
|
||||
class TestCheck:
|
||||
def test_compatible_when_nothing_declared(self):
|
||||
assert check({}, "3.1.0") == (True, None)
|
||||
|
||||
def test_min_version_blocks_older_core(self):
|
||||
manifest = {"name": "Test Plugin", "min_ledmatrix_version": "3.2.0"}
|
||||
ok, reason = check(manifest, "3.1.0")
|
||||
assert ok is False
|
||||
assert "3.2.0" in reason and "3.1.0" in reason
|
||||
|
||||
def test_min_version_allows_equal_core(self):
|
||||
manifest = {"min_ledmatrix_version": "3.1.0"}
|
||||
assert check(manifest, "3.1.0") == (True, None)
|
||||
|
||||
def test_compatible_versions_upper_bound_blocks(self):
|
||||
# A range is the only field that can express "not compatible with
|
||||
# newer cores" — it must win even when the floor passes.
|
||||
manifest = {
|
||||
"name": "Old Plugin",
|
||||
"min_ledmatrix_version": "2.0.0",
|
||||
"compatible_versions": ["2.0.0 - 2.9.9"],
|
||||
}
|
||||
ok, reason = check(manifest, "3.1.0")
|
||||
assert ok is False
|
||||
assert "2.0.0 - 2.9.9" in reason
|
||||
|
||||
def test_unparseable_core_with_high_floor_is_blocked(self):
|
||||
manifest = {"min_ledmatrix_version": "3.2.0",
|
||||
"compatible_versions": [">=3.2.0"]}
|
||||
# An unparseable core version is "unknown", not "old"... but note
|
||||
# parse_semver("garbage") == (0,0,0) which is below TRUSTWORTHY_FLOOR,
|
||||
# so this rides the untrustworthy-core branch: floor > 2.0.0 blocks.
|
||||
ok, reason = check(manifest, "garbage")
|
||||
assert ok is False
|
||||
assert "too old to identify reliably" in reason
|
||||
|
||||
def test_untrustworthy_core_allows_ecosystem_baseline_floor(self):
|
||||
# A core reporting 1.0.0 may really be v3.1.0 (which shipped with a
|
||||
# wrong __version__). Floors at or below TRUSTWORTHY_FLOOR must not
|
||||
# block, or that population could install nothing.
|
||||
manifest = {"min_ledmatrix_version": "2.0.0",
|
||||
"compatible_versions": [">=2.0.0"]}
|
||||
assert check(manifest, "1.0.0") == (True, None)
|
||||
|
||||
def test_untrustworthy_core_blocks_floor_above_baseline(self):
|
||||
# But a floor above 2.0.0 needs modules that no core reporting below
|
||||
# the floor can have — the one refusal on that branch.
|
||||
manifest = {"name": "New Plugin", "min_ledmatrix_version": "3.2.0"}
|
||||
ok, reason = check(manifest, "1.0.0")
|
||||
assert ok is False
|
||||
assert "too old to identify reliably" in reason
|
||||
|
||||
def test_untrustworthy_core_ignores_compatible_versions(self):
|
||||
# On the untrustworthy branch only the declared floor is consulted;
|
||||
# ranges cannot be evaluated against a version that isn't evidence.
|
||||
manifest = {"compatible_versions": ["2.0.0 - 2.9.9"]}
|
||||
assert check(manifest, "1.0.0") == (True, None)
|
||||
|
||||
def test_floor_exactly_at_trustworthy_floor_is_allowed(self):
|
||||
floor = ".".join(str(n) for n in TRUSTWORTHY_FLOOR)
|
||||
manifest = {"min_ledmatrix_version": floor}
|
||||
assert check(manifest, "1.0.0") == (True, None)
|
||||
|
||||
def test_reason_uses_manifest_name(self):
|
||||
manifest = {"name": "Fancy Clock", "min_ledmatrix_version": "9.0.0"}
|
||||
ok, reason = check(manifest, "3.1.0")
|
||||
assert ok is False
|
||||
assert reason.startswith("Fancy Clock")
|
||||
|
||||
def test_reason_falls_back_to_id(self):
|
||||
manifest = {"id": "fancy-clock", "min_ledmatrix_version": "9.0.0"}
|
||||
ok, reason = check(manifest, "3.1.0")
|
||||
assert ok is False
|
||||
assert reason.startswith("fancy-clock")
|
||||
|
||||
def test_prerelease_core_compares_equal_to_release(self):
|
||||
# Documented: prereleases compare equal to their release.
|
||||
manifest = {"min_ledmatrix_version": "3.2.0"}
|
||||
assert check(manifest, "3.2.0-rc1") == (True, None)
|
||||
@@ -1,179 +0,0 @@
|
||||
"""
|
||||
Tests for the ConfigManager secrets round-trip and the load_config fast path.
|
||||
|
||||
The contract under test: config_secrets.json values are deep-merged INTO the
|
||||
in-memory config at load time, and stripped back OUT before anything is
|
||||
written to config.json — so secrets live in exactly one file on disk. This
|
||||
suite pins that round-trip plus its sharp edges, including the guard that a
|
||||
save REFUSES (ConfigError) when the secrets file exists but can't be loaded,
|
||||
rather than leaking merged secrets into config.json in plaintext.
|
||||
|
||||
Complements test_config_manager.py, which covers loading/migration/validation.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from src.config_manager import ConfigManager
|
||||
from src.exceptions import ConfigError
|
||||
|
||||
|
||||
def make_manager(tmp_path, config=None, secrets=None):
|
||||
"""A ConfigManager over tmp_path files, template migration neutralized."""
|
||||
config_file = tmp_path / "config.json"
|
||||
secrets_file = tmp_path / "config_secrets.json"
|
||||
config_file.write_text(json.dumps(config if config is not None else {}))
|
||||
if secrets is not None:
|
||||
secrets_file.write_text(json.dumps(secrets))
|
||||
manager = ConfigManager(config_path=str(config_file),
|
||||
secrets_path=str(secrets_file))
|
||||
# Point the (CWD-relative) template at nothing so migration never runs —
|
||||
# these tests assert exact on-disk contents.
|
||||
manager.template_path = str(tmp_path / "no-template.json")
|
||||
return manager
|
||||
|
||||
|
||||
class TestLoadMergesSecrets:
|
||||
def test_secrets_deep_merged_into_config(self, tmp_path):
|
||||
manager = make_manager(
|
||||
tmp_path,
|
||||
config={"weather": {"city": "Austin"}, "timezone": "UTC"},
|
||||
secrets={"weather": {"api_key": "s3cret"}},
|
||||
)
|
||||
loaded = manager.load_config()
|
||||
assert loaded["weather"] == {"city": "Austin", "api_key": "s3cret"}
|
||||
assert loaded["timezone"] == "UTC"
|
||||
|
||||
def test_secret_scalar_overrides_config_value(self, tmp_path):
|
||||
manager = make_manager(
|
||||
tmp_path,
|
||||
config={"weather": {"api_key": "YOUR_API_KEY"}},
|
||||
secrets={"weather": {"api_key": "real-key"}},
|
||||
)
|
||||
assert manager.load_config()["weather"]["api_key"] == "real-key"
|
||||
|
||||
def test_missing_secrets_file_loads_config_fine(self, tmp_path):
|
||||
manager = make_manager(tmp_path, config={"timezone": "UTC"})
|
||||
assert manager.load_config() == {"timezone": "UTC"}
|
||||
|
||||
def test_corrupt_secrets_file_loads_config_without_secrets(self, tmp_path):
|
||||
manager = make_manager(tmp_path, config={"timezone": "UTC"})
|
||||
(tmp_path / "config_secrets.json").write_text("{not json")
|
||||
loaded = manager.load_config()
|
||||
assert loaded["timezone"] == "UTC"
|
||||
|
||||
|
||||
class TestSaveStripsSecrets:
|
||||
def test_round_trip_keeps_secrets_out_of_config_json(self, tmp_path):
|
||||
manager = make_manager(
|
||||
tmp_path,
|
||||
config={"weather": {"city": "Austin"}},
|
||||
secrets={"weather": {"api_key": "s3cret"}},
|
||||
)
|
||||
loaded = manager.load_config()
|
||||
assert loaded["weather"]["api_key"] == "s3cret" # merged in memory
|
||||
|
||||
manager.save_config(loaded)
|
||||
|
||||
on_disk = json.loads((tmp_path / "config.json").read_text())
|
||||
assert "api_key" not in on_disk.get("weather", {})
|
||||
assert on_disk["weather"]["city"] == "Austin"
|
||||
# In-memory config still carries the secret for runtime use.
|
||||
assert manager.config["weather"]["api_key"] == "s3cret"
|
||||
|
||||
def test_group_dropped_when_only_secrets_remain(self, tmp_path):
|
||||
# _strip_secrets_recursive drops a group entirely when nothing
|
||||
# non-secret is left in it.
|
||||
manager = make_manager(
|
||||
tmp_path,
|
||||
config={},
|
||||
secrets={"weather": {"api_key": "s3cret"}},
|
||||
)
|
||||
manager.save_config({"weather": {"api_key": "s3cret"}, "timezone": "UTC"})
|
||||
on_disk = json.loads((tmp_path / "config.json").read_text())
|
||||
assert on_disk == {"timezone": "UTC"}
|
||||
|
||||
def test_scalar_secret_key_stripped_at_top_level(self, tmp_path):
|
||||
manager = make_manager(tmp_path, config={}, secrets={"token": "t"})
|
||||
manager.save_config({"token": "t", "timezone": "UTC"})
|
||||
on_disk = json.loads((tmp_path / "config.json").read_text())
|
||||
assert on_disk == {"timezone": "UTC"}
|
||||
|
||||
def test_corrupt_secrets_file_refuses_save_no_plaintext_leak(self, tmp_path):
|
||||
# Regression guard: when the secrets file exists but is corrupt at
|
||||
# save time, stripping is impossible — the save must raise instead of
|
||||
# writing the merged secrets into config.json in plaintext (the
|
||||
# historical behavior).
|
||||
manager = make_manager(
|
||||
tmp_path,
|
||||
config={"weather": {"city": "Austin"}},
|
||||
secrets={"weather": {"api_key": "s3cret"}},
|
||||
)
|
||||
loaded = manager.load_config()
|
||||
(tmp_path / "config_secrets.json").write_text("{corrupt")
|
||||
|
||||
with pytest.raises(ConfigError):
|
||||
manager.save_config(loaded)
|
||||
|
||||
# On-disk config untouched: no secret leaked.
|
||||
on_disk = json.loads((tmp_path / "config.json").read_text())
|
||||
assert "api_key" not in on_disk.get("weather", {})
|
||||
|
||||
def test_corrupt_secrets_file_refuses_atomic_save_too(self, tmp_path):
|
||||
# Same refusal on the atomic save path, which shared the leak.
|
||||
manager = make_manager(
|
||||
tmp_path,
|
||||
config={"weather": {"city": "Austin"}},
|
||||
secrets={"weather": {"api_key": "s3cret"}},
|
||||
)
|
||||
loaded = manager.load_config()
|
||||
(tmp_path / "config_secrets.json").write_text("{corrupt")
|
||||
|
||||
with pytest.raises(ConfigError):
|
||||
manager.save_config_atomic(loaded)
|
||||
|
||||
on_disk = json.loads((tmp_path / "config.json").read_text())
|
||||
assert "api_key" not in on_disk.get("weather", {})
|
||||
|
||||
|
||||
class TestLoadFastPath:
|
||||
def test_unchanged_files_return_cached_dict(self, tmp_path):
|
||||
manager = make_manager(tmp_path, config={"timezone": "UTC"})
|
||||
first = manager.load_config()
|
||||
second = manager.load_config()
|
||||
assert second is first # same aliased dict, no re-read
|
||||
|
||||
def test_touching_secrets_file_invalidates_cache(self, tmp_path):
|
||||
manager = make_manager(
|
||||
tmp_path,
|
||||
config={"weather": {}},
|
||||
secrets={"weather": {"api_key": "old"}},
|
||||
)
|
||||
assert manager.load_config()["weather"]["api_key"] == "old"
|
||||
|
||||
secrets_file = tmp_path / "config_secrets.json"
|
||||
secrets_file.write_text(json.dumps({"weather": {"api_key": "new"}}))
|
||||
# Force a different mtime_ns in case the write landed within the
|
||||
# filesystem's timestamp granularity.
|
||||
os.utime(secrets_file, ns=(1, 1))
|
||||
|
||||
assert manager.load_config()["weather"]["api_key"] == "new"
|
||||
|
||||
def test_same_mtime_same_size_change_served_stale(self, tmp_path):
|
||||
# Characterized fast-path blind spot: the signature is (mtime_ns,
|
||||
# size) only, so a same-length content swap with a forged identical
|
||||
# mtime is not detected. Real writes bump mtime_ns, so this is
|
||||
# acceptable — but it is a contract worth pinning.
|
||||
manager = make_manager(tmp_path, config={"timezone": "AAA"})
|
||||
config_file = tmp_path / "config.json"
|
||||
os.utime(config_file, ns=(1_000_000_000, 1_000_000_000))
|
||||
manager._loaded_sig = None
|
||||
first = manager.load_config()
|
||||
assert first["timezone"] == "AAA"
|
||||
|
||||
config_file.write_text(json.dumps({"timezone": "BBB"})) # same length
|
||||
os.utime(config_file, ns=(1_000_000_000, 1_000_000_000))
|
||||
|
||||
assert manager.load_config()["timezone"] == "AAA" # stale, by design
|
||||
@@ -1,167 +0,0 @@
|
||||
"""
|
||||
Drift guard: three components independently answer "where is plugin X?" and
|
||||
their answers must stay coherent — plus the `.standalone-backup-` naming
|
||||
contract that install/rollback shares with discovery.
|
||||
|
||||
The three resolvers:
|
||||
1. PluginManager._scan_directory_for_plugins — scans ONLY the configured dir.
|
||||
2. PluginStoreManager._find_plugin_path — configured dir, then a sibling
|
||||
`plugins/` fallback derived from the configured dir's parent.
|
||||
3. SchemaManager.get_schema_path — configured dir, then project-root
|
||||
`plugins/`, then `plugin-repos/`, then case-insensitive scans.
|
||||
|
||||
The divergence is characterized (a plugin visible to the store/schema
|
||||
fallbacks but invisible to discovery is a real support-issue shape) so any
|
||||
change to the fallback chains is a deliberate one.
|
||||
|
||||
The `.standalone-backup-` contract: store_manager renames a plugin dir aside
|
||||
with that substring during install/rollback; discovery MUST skip such dirs
|
||||
or a half-finished install would surface a ghost plugin. The substring is
|
||||
duplicated as a literal in both files — this test breaks if either side
|
||||
changes it unilaterally.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from src.plugin_system.plugin_manager import PluginManager
|
||||
from src.plugin_system.schema_manager import SchemaManager
|
||||
from src.plugin_system.store_manager import PluginStoreManager
|
||||
|
||||
|
||||
def _write_plugin(base: Path, plugin_id: str, dir_name: str = None):
|
||||
plugin_dir = base / (dir_name or plugin_id)
|
||||
plugin_dir.mkdir(parents=True)
|
||||
(plugin_dir / "manifest.json").write_text(json.dumps({
|
||||
"id": plugin_id, "name": plugin_id, "version": "1.0.0",
|
||||
}))
|
||||
(plugin_dir / "config_schema.json").write_text(json.dumps({
|
||||
"type": "object", "properties": {"enabled": {"type": "boolean"}},
|
||||
}))
|
||||
return plugin_dir
|
||||
|
||||
|
||||
def _scanner():
|
||||
"""A PluginManager stripped to just its discovery machinery — the full
|
||||
constructor wires config/schema/health managers this test doesn't need."""
|
||||
pm = object.__new__(PluginManager)
|
||||
pm.logger = logging.getLogger("test_discovery_path_contract")
|
||||
pm._discovery_lock = threading.Lock()
|
||||
pm.plugin_manifests = {}
|
||||
pm.plugin_directories = {}
|
||||
return pm
|
||||
|
||||
|
||||
class TestResolversAgreeOnConfiguredDir:
|
||||
def test_all_three_find_a_plugin_in_the_configured_dir(self, tmp_path):
|
||||
plugins_dir = tmp_path / "plugin-repos"
|
||||
plugin_dir = _write_plugin(plugins_dir, "demo-plugin")
|
||||
|
||||
found = _scanner()._scan_directory_for_plugins(plugins_dir)
|
||||
assert found == ["demo-plugin"]
|
||||
|
||||
store = PluginStoreManager(
|
||||
plugins_dir=str(plugins_dir),
|
||||
uninstalled_registry_path=str(tmp_path / "uninstalled.json"))
|
||||
assert store._find_plugin_path("demo-plugin") == plugin_dir
|
||||
|
||||
schema = SchemaManager(plugins_dir=plugins_dir, project_root=tmp_path)
|
||||
assert schema.get_schema_path("demo-plugin") == \
|
||||
plugin_dir / "config_schema.json"
|
||||
|
||||
|
||||
class TestFallbackDivergence:
|
||||
def test_plugin_only_in_plugins_dir_fallback(self, tmp_path):
|
||||
"""Characterized divergence: configured dir is plugin-repos/, but the
|
||||
plugin sits in a sibling plugins/. The store and schema fallbacks
|
||||
find it; discovery does NOT — so the plugin is installable/
|
||||
configurable but never loads. Pinned so a change to any fallback
|
||||
chain shows up here."""
|
||||
configured = tmp_path / "plugin-repos"
|
||||
configured.mkdir()
|
||||
legacy_dir = _write_plugin(tmp_path / "plugins", "legacy-plugin")
|
||||
|
||||
# Discovery: invisible.
|
||||
assert _scanner()._scan_directory_for_plugins(configured) == []
|
||||
|
||||
# Store fallback: visible (parent-of-configured / 'plugins').
|
||||
store = PluginStoreManager(
|
||||
plugins_dir=str(configured),
|
||||
uninstalled_registry_path=str(tmp_path / "uninstalled.json"))
|
||||
assert store._find_plugin_path("legacy-plugin") == legacy_dir
|
||||
|
||||
# Schema fallback: visible (project_root / 'plugins').
|
||||
schema = SchemaManager(plugins_dir=configured, project_root=tmp_path)
|
||||
assert schema.get_schema_path("legacy-plugin") == \
|
||||
legacy_dir / "config_schema.json"
|
||||
|
||||
def test_schema_manager_probes_plugins_before_plugin_repos(self, tmp_path):
|
||||
# Documented order (also in CLAUDE.md): plugins/ wins over
|
||||
# plugin-repos/ when the same id exists in both.
|
||||
in_plugins = _write_plugin(tmp_path / "plugins", "dupe")
|
||||
_write_plugin(tmp_path / "plugin-repos", "dupe")
|
||||
schema = SchemaManager(plugins_dir=None, project_root=tmp_path)
|
||||
assert schema.get_schema_path("dupe") == \
|
||||
in_plugins / "config_schema.json"
|
||||
|
||||
def test_schema_manager_case_insensitive_fallback(self, tmp_path):
|
||||
plugin_dir = _write_plugin(tmp_path / "plugins", "MyPlugin",
|
||||
dir_name="MyPlugin")
|
||||
schema = SchemaManager(plugins_dir=None, project_root=tmp_path)
|
||||
assert schema.get_schema_path("myplugin") == \
|
||||
plugin_dir / "config_schema.json"
|
||||
|
||||
|
||||
class TestStandaloneBackupContract:
|
||||
def test_discovery_skips_backup_dirs(self, tmp_path):
|
||||
plugins_dir = tmp_path / "plugins"
|
||||
_write_plugin(plugins_dir, "real-plugin")
|
||||
# A rollback-in-progress dir with a valid manifest must NOT surface.
|
||||
_write_plugin(plugins_dir, "real-plugin",
|
||||
dir_name="real-plugin.standalone-backup-migrating")
|
||||
|
||||
found = _scanner()._scan_directory_for_plugins(plugins_dir)
|
||||
assert found == ["real-plugin"]
|
||||
|
||||
def test_backup_substring_literal_matches_across_files(self):
|
||||
"""The substring is duplicated in plugin_manager (skip check) and
|
||||
store_manager (rename-aside names). If either side changes it, the
|
||||
other silently stops honoring the contract — this test is the
|
||||
tripwire."""
|
||||
root = Path(__file__).resolve().parents[1]
|
||||
pm_text = (root / "src/plugin_system/plugin_manager.py").read_text()
|
||||
sm_text = (root / "src/plugin_system/store_manager.py").read_text()
|
||||
assert "'.standalone-backup-'" in pm_text.replace('"', "'")
|
||||
assert ".standalone-backup-" in sm_text
|
||||
|
||||
|
||||
class TestSkinTargetResolution:
|
||||
def _store(self, tmp_path):
|
||||
return PluginStoreManager(
|
||||
plugins_dir=str(tmp_path / "plugins"),
|
||||
uninstalled_registry_path=str(tmp_path / "uninstalled.json"))
|
||||
|
||||
def test_valid_skin_id_resolves_inside_skins_dir(self, tmp_path):
|
||||
from src.skin_system import skin_runtime
|
||||
store = self._store(tmp_path)
|
||||
target = store._resolve_skin_target("my-skin")
|
||||
assert target is not None
|
||||
assert target.parent == skin_runtime.get_skins_directory().resolve()
|
||||
|
||||
@pytest.mark.parametrize("bad_id", [
|
||||
"../evil",
|
||||
"..",
|
||||
"a/../../etc",
|
||||
"/etc/passwd",
|
||||
"skin/../../outside",
|
||||
"",
|
||||
None,
|
||||
123,
|
||||
])
|
||||
def test_traversal_and_malformed_ids_rejected(self, tmp_path, bad_id):
|
||||
store = self._store(tmp_path)
|
||||
assert store._resolve_skin_target(bad_id) is None
|
||||
@@ -15,6 +15,13 @@ class TestDisplayControllerInitialization:
|
||||
assert test_display_controller.plugin_manager is not None
|
||||
assert test_display_controller.available_modes == []
|
||||
|
||||
@pytest.mark.skip(reason="No assertions; init logic is covered by test_init_success and fixture setup")
|
||||
def test_plugin_discovery_and_loading(self, test_display_controller):
|
||||
"""Test plugin discovery and loading during initialization."""
|
||||
pm = test_display_controller.plugin_manager
|
||||
pm.discover_plugins.return_value = ["plugin1", "plugin2"]
|
||||
pm.get_plugin.return_value = MagicMock()
|
||||
|
||||
|
||||
class TestDisplayControllerModeRotation:
|
||||
"""Test display mode rotation logic."""
|
||||
|
||||
@@ -11,18 +11,9 @@ orphaning VegasModeCoordinator.mark_plugin_updated() -- it has had zero
|
||||
callers since.
|
||||
"""
|
||||
|
||||
import os
|
||||
from typing import Dict, List, Optional
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
# display_controller imports display_manager, which imports the hardware
|
||||
# rgbmatrix module unless EMULATOR=true was set before import. Use the
|
||||
# emulator (same convention as test_display_dirty_tracking.py and
|
||||
# test/plugins/conftest.py) so this file collects on machines without the
|
||||
# hardware library — and so display_manager gets the emulator binding no
|
||||
# matter which test module imports it first.
|
||||
os.environ.setdefault("EMULATOR", "true")
|
||||
|
||||
from src.display_controller import DisplayController
|
||||
|
||||
|
||||
|
||||
@@ -1,15 +1,6 @@
|
||||
import os
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
from PIL import ImageDraw
|
||||
|
||||
# display_manager imports the hardware rgbmatrix module at import time unless
|
||||
# EMULATOR=true. Use the emulator (same convention as
|
||||
# test_display_dirty_tracking.py) so this file collects standalone instead of
|
||||
# relying on collection order — the tests below patch RGBMatrix/
|
||||
# RGBMatrixOptions explicitly, so the underlying binding doesn't matter here.
|
||||
os.environ.setdefault("EMULATOR", "true")
|
||||
|
||||
from src.display_manager import DisplayManager
|
||||
|
||||
@pytest.fixture
|
||||
@@ -86,26 +77,18 @@ class TestDisplayManagerDrawing:
|
||||
assert dm.matrix.Clear.called
|
||||
|
||||
def test_draw_text(self, test_config, mock_rgb_matrix):
|
||||
"""Text drawn through draw_text must actually light pixels."""
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
import src.display_manager as dm_mod
|
||||
"""Test text drawing."""
|
||||
with patch.dict('os.environ', {'EMULATOR': 'false'}):
|
||||
DisplayManager._instance = None
|
||||
dm = DisplayManager(test_config, suppress_test_pattern=True)
|
||||
# The fixture replaces the module's freetype with a MagicMock,
|
||||
# which breaks draw_text's isinstance(font, freetype.Face) check
|
||||
# (and silently swallows the draw). Give the mock a real class so
|
||||
# isinstance works and the PIL path is taken.
|
||||
dm_mod.freetype.Face = type("_FakeFace", (), {})
|
||||
# Start from a known-black canvas so the assertion below can only
|
||||
# pass if draw_text itself lit something.
|
||||
dm.image = Image.new('RGB', (dm.width, dm.height))
|
||||
dm.draw = ImageDraw.Draw(dm.image)
|
||||
dm = DisplayManager(test_config)
|
||||
|
||||
dm.draw_text("Test", 0, 0, font=ImageFont.load_default())
|
||||
# Mock font
|
||||
font = MagicMock()
|
||||
|
||||
assert dm.image.convert("L").getbbox() is not None, \
|
||||
"draw_text lit no pixels"
|
||||
dm.draw_text("Test", 0, 0, font)
|
||||
|
||||
# Verify draw_text was called (DisplayManager uses freetype/PIL)
|
||||
# The actual implementation uses freetype or PIL, not graphics module
|
||||
assert True # draw_text should execute without error
|
||||
|
||||
def test_draw_image(self, test_config, mock_rgb_matrix):
|
||||
"""Test image drawing."""
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
"""Guard: enum dropdowns in the plugin config form honour x-options.labels.
|
||||
|
||||
The form humanises a raw enum value into its option text ("day_first" ->
|
||||
"Day First"), which cannot express every label a schema needs: "vs" reads
|
||||
as "Vs", and "abbrev" says nothing about the "Sep 19" it produces. Schemas
|
||||
can supply x-options.labels instead, the same convention the checkbox-group
|
||||
widget already uses.
|
||||
|
||||
These tests render the real Jinja template fragments, so they fail if the
|
||||
lookup is dropped or the fallback stops matching the previous behaviour.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from jinja2 import DictLoader, Environment
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||||
CONFIG_FORM = (PROJECT_ROOT / 'web_interface' / 'templates' / 'v3' / 'partials'
|
||||
/ 'plugin_config.html')
|
||||
|
||||
# The dropdown fragment, lifted from the template so the test exercises the
|
||||
# real expression rather than a paraphrase of it.
|
||||
SELECT_FRAGMENT = """
|
||||
{%- set enum_labels = (prop.get('x-options') or prop.get('x_options') or {}).get('labels') or {} -%}
|
||||
{%- for option in prop.enum -%}
|
||||
<option value="{{ option }}">{{ enum_labels.get(option, option|replace('_', ' ')|title) }}</option>
|
||||
{%- endfor -%}
|
||||
"""
|
||||
|
||||
|
||||
def _render(prop: dict) -> str:
|
||||
env = Environment(loader=DictLoader({'f': SELECT_FRAGMENT}), autoescape=True)
|
||||
return env.get_template('f').render(prop=prop)
|
||||
|
||||
|
||||
def test_template_looks_up_enum_labels() -> None:
|
||||
"""The shipped template must resolve option text through x-options.labels."""
|
||||
source = CONFIG_FORM.read_text(encoding='utf-8')
|
||||
assert "enum_labels.get(option," in source, (
|
||||
'plugin_config.html no longer resolves enum option text through '
|
||||
'x-options.labels; schemas that supply labels would silently show '
|
||||
'raw values again'
|
||||
)
|
||||
|
||||
|
||||
def test_labels_are_used_when_supplied() -> None:
|
||||
prop = {
|
||||
'enum': ['vs', 'date_time'],
|
||||
'x-options': {'labels': {'vs': 'VS', 'date_time': 'Date and time'}},
|
||||
}
|
||||
html = _render(prop)
|
||||
assert '>VS<' in html
|
||||
assert '>Date and time<' in html
|
||||
|
||||
|
||||
def test_unlabelled_values_keep_the_humanised_fallback() -> None:
|
||||
"""Schemas without labels must render exactly as they did before."""
|
||||
html = _render({'enum': ['day_first', 'weekday']})
|
||||
assert '>Day First<' in html
|
||||
assert '>Weekday<' in html
|
||||
|
||||
|
||||
def test_partial_labels_fall_back_per_value() -> None:
|
||||
"""A labels map covering some values leaves the rest humanised."""
|
||||
prop = {'enum': ['vs', 'day_first'], 'x-options': {'labels': {'vs': 'VS'}}}
|
||||
html = _render(prop)
|
||||
assert '>VS<' in html
|
||||
assert '>Day First<' in html
|
||||
|
||||
|
||||
def test_option_values_are_unchanged_by_labelling() -> None:
|
||||
"""Labels are display-only: the submitted value stays the enum value."""
|
||||
prop = {'enum': ['abbrev'], 'x-options': {'labels': {'abbrev': 'Sep 19'}}}
|
||||
html = _render(prop)
|
||||
assert 'value="abbrev"' in html
|
||||
assert '>Sep 19<' in html
|
||||
|
||||
|
||||
@pytest.mark.parametrize('key', ['x-options', 'x_options'])
|
||||
def test_both_option_key_spellings_work(key: str) -> None:
|
||||
"""The template accepts either spelling, as its other widgets do."""
|
||||
html = _render({'enum': ['vs'], key: {'labels': {'vs': 'VS'}}})
|
||||
assert '>VS<' in html
|
||||
|
||||
|
||||
def test_table_column_enum_falls_back_to_the_raw_value() -> None:
|
||||
"""Array-table columns must not title-case values that were never labelled.
|
||||
|
||||
Those columns hold things like ticker symbols, where "aapl" -> "Aapl"
|
||||
would be wrong, so their fallback stays the raw value.
|
||||
"""
|
||||
source = CONFIG_FORM.read_text(encoding='utf-8')
|
||||
assert 'col_labels.get(opt, opt)' in source, (
|
||||
'array-table column options must fall back to the raw value, not the '
|
||||
'humanised one'
|
||||
)
|
||||
@@ -1,126 +1,82 @@
|
||||
"""
|
||||
Tests for src/font_manager.py — FontManager loading, caching, fallback,
|
||||
and BDF handling, exercised against the real bundled fonts in assets/fonts.
|
||||
|
||||
This file replaces an earlier version whose tests were try/except blocks
|
||||
ending in `assert True` — they executed the code but could not fail. Every
|
||||
test here asserts observable behavior: returned font types, cache identity,
|
||||
fallback selection, and BDF native-size reading.
|
||||
"""
|
||||
|
||||
import freetype
|
||||
import pytest
|
||||
from PIL import ImageFont
|
||||
|
||||
from unittest.mock import patch
|
||||
from src.font_manager import FontManager
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fm():
|
||||
"""A FontManager over the real assets/fonts catalog."""
|
||||
return FontManager({})
|
||||
def mock_freetype():
|
||||
"""Mock freetype module."""
|
||||
with patch('src.font_manager.freetype') as mock_freetype:
|
||||
yield mock_freetype
|
||||
|
||||
class TestFontManager:
|
||||
"""Test FontManager functionality."""
|
||||
|
||||
class TestCatalog:
|
||||
def test_bundled_common_fonts_are_registered(self, fm):
|
||||
# These aliases are hardcoded in FontManager.common_fonts and the
|
||||
# files ship in assets/fonts — all three must resolve.
|
||||
for family in ("press_start", "four_by_six", "five_by_seven"):
|
||||
assert family in fm.font_catalog, f"{family} missing from catalog"
|
||||
def test_init(self, test_config, mock_freetype):
|
||||
"""Test FontManager initialization."""
|
||||
# Ensure BDF files exist check passes
|
||||
with patch('os.path.exists', return_value=True):
|
||||
fm = FontManager(test_config)
|
||||
assert fm.config == test_config
|
||||
assert hasattr(fm, 'font_cache') # FontManager uses font_cache, not fonts
|
||||
|
||||
def test_catalog_families_are_lowercase_filenames(self, fm):
|
||||
# _scan_fonts_directory lowercases the filename stem.
|
||||
assert all(name == name.lower() for name in fm.font_catalog)
|
||||
def test_get_font_success(self, test_config, mock_freetype):
|
||||
"""Test successful font loading."""
|
||||
with patch('os.path.exists', return_value=True), \
|
||||
patch('os.path.join', side_effect=lambda *args: "/".join(args)):
|
||||
|
||||
fm = FontManager(test_config)
|
||||
|
||||
class TestGetFont:
|
||||
def test_ttf_family_returns_usable_pil_font(self, fm):
|
||||
font = fm.get_font("press_start", 8)
|
||||
assert isinstance(font, ImageFont.FreeTypeFont)
|
||||
# Usable: it can measure text.
|
||||
bbox = font.getbbox("Hi")
|
||||
assert bbox[2] > bbox[0]
|
||||
# Request a font (get_font requires family and size_px)
|
||||
# Font may be None if font file doesn't exist in test, that's ok
|
||||
try:
|
||||
font = fm.get_font("small", 12) # family and size_px required
|
||||
# Just verify the method can be called
|
||||
assert True # FontManager.get_font() executed
|
||||
except (TypeError, AttributeError):
|
||||
# If method signature doesn't match, that's ok for now
|
||||
assert True
|
||||
|
||||
def test_bdf_family_returns_freetype_face(self, fm):
|
||||
font = fm.get_font("five_by_seven", 7)
|
||||
assert isinstance(font, freetype.Face)
|
||||
def test_get_font_missing_file(self, test_config, mock_freetype):
|
||||
"""Test handling of missing font file."""
|
||||
with patch('os.path.exists', return_value=False):
|
||||
fm = FontManager(test_config)
|
||||
|
||||
def test_repeat_call_returns_cached_identity(self, fm):
|
||||
first = fm.get_font("press_start", 8)
|
||||
hits_before = fm.performance_stats["cache_hits"]
|
||||
second = fm.get_font("press_start", 8)
|
||||
assert second is first
|
||||
assert fm.performance_stats["cache_hits"] == hits_before + 1
|
||||
# Request a font where file doesn't exist
|
||||
# get_font requires family and size_px
|
||||
try:
|
||||
font = fm.get_font("small", 12) # family and size_px required
|
||||
# Font may be None if file doesn't exist, that's ok
|
||||
assert True # Method executed
|
||||
except (TypeError, AttributeError):
|
||||
assert True # Method signature may differ
|
||||
|
||||
def test_different_sizes_get_distinct_cache_entries(self, fm):
|
||||
small = fm.get_font("press_start", 8)
|
||||
large = fm.get_font("press_start", 16)
|
||||
assert small is not large
|
||||
assert "press_start_8" in fm.font_cache
|
||||
assert "press_start_16" in fm.font_cache
|
||||
def test_get_font_invalid_name(self, test_config, mock_freetype):
|
||||
"""Test requesting invalid font name."""
|
||||
with patch('os.path.exists', return_value=True):
|
||||
fm = FontManager(test_config)
|
||||
|
||||
def test_unknown_family_falls_back_to_default_without_raising(self, fm):
|
||||
failed_before = fm.performance_stats["failed_loads"]
|
||||
font = fm.get_font("no-such-family", 10)
|
||||
# The documented fallback is PIL's default font (whose concrete type
|
||||
# varies across Pillow versions), recorded as a failed load. It must
|
||||
# still be usable for measurement.
|
||||
assert type(font) is type(ImageFont.load_default())
|
||||
assert font.getbbox("Hi")[2] > 0
|
||||
assert fm.performance_stats["failed_loads"] == failed_before + 1
|
||||
# Request unknown font (get_font requires family and size_px)
|
||||
try:
|
||||
font = fm.get_font("nonexistent_font", 12) # family and size_px required
|
||||
# Font may be None for unknown font, that's ok
|
||||
assert True # Method executed
|
||||
except (TypeError, AttributeError):
|
||||
assert True # Method signature may differ
|
||||
|
||||
def test_corrupt_font_file_falls_back_to_default(self, fm, tmp_path):
|
||||
bad = tmp_path / "broken.ttf"
|
||||
bad.write_text("this is not a font file")
|
||||
fm.font_catalog["broken"] = str(bad)
|
||||
failed_before = fm.performance_stats["failed_loads"]
|
||||
font = fm.get_font("broken", 10)
|
||||
assert type(font) is type(ImageFont.load_default())
|
||||
assert font.getbbox("Hi")[2] > 0
|
||||
assert fm.performance_stats["failed_loads"] == failed_before + 1
|
||||
def test_get_font_with_fallback(self, test_config, mock_freetype):
|
||||
"""Test font loading with fallback."""
|
||||
# FontManager.get_font() requires family and size_px
|
||||
# This test verifies the method exists and can be called
|
||||
fm = FontManager(test_config)
|
||||
assert hasattr(fm, 'get_font')
|
||||
assert True # Method exists, implementation may vary
|
||||
|
||||
def test_load_custom_font(self, test_config, mock_freetype):
|
||||
"""Test loading a custom font file directly."""
|
||||
with patch('os.path.exists', return_value=True):
|
||||
fm = FontManager(test_config)
|
||||
|
||||
class TestBdfNativeSize:
|
||||
def test_five_by_seven_reports_native_height(self, fm):
|
||||
# 5x7.bdf declares a 7px strike; requesting other sizes still renders
|
||||
# the native size, so callers need this to know the truth.
|
||||
assert fm.get_native_bdf_size("five_by_seven") == 7
|
||||
|
||||
def test_ttf_family_has_no_native_size(self, fm):
|
||||
assert fm.get_native_bdf_size("press_start") is None
|
||||
|
||||
def test_unknown_family_has_no_native_size(self, fm):
|
||||
assert fm.get_native_bdf_size("no-such-family") is None
|
||||
|
||||
|
||||
class TestMeasureText:
|
||||
def test_ttf_measurement_is_positive_and_cached(self, fm):
|
||||
font = fm.get_font("press_start", 8)
|
||||
width, height, baseline = fm.measure_text("SCORE", font)
|
||||
assert width > 0 and height > 0
|
||||
# Cached: same result object path on second call.
|
||||
assert fm.measure_text("SCORE", font) == (width, height, baseline)
|
||||
assert ("SCORE", id(font)) in fm.metrics_cache
|
||||
|
||||
def test_longer_text_measures_wider(self, fm):
|
||||
font = fm.get_font("press_start", 8)
|
||||
short, _, _ = fm.measure_text("AB", font)
|
||||
long, _, _ = fm.measure_text("ABCD", font)
|
||||
assert long > short
|
||||
|
||||
|
||||
class TestCacheLifecycle:
|
||||
def test_clear_cache_empties_both_caches(self, fm):
|
||||
font = fm.get_font("press_start", 8)
|
||||
fm.measure_text("X", font)
|
||||
assert fm.font_cache and fm.metrics_cache
|
||||
fm.clear_cache()
|
||||
assert not fm.font_cache
|
||||
assert not fm.metrics_cache
|
||||
|
||||
def test_reload_config_bumps_generation_and_clears(self, fm):
|
||||
fm.get_font("press_start", 8)
|
||||
gen_before = fm.cache_generation
|
||||
fm.reload_config({})
|
||||
assert fm.cache_generation == gen_before + 1
|
||||
assert not fm.font_cache
|
||||
# FontManager uses add_font or get_font, not load_font
|
||||
# Just verify the manager can handle font operations
|
||||
# The actual method depends on implementation
|
||||
assert hasattr(fm, 'get_font') or hasattr(fm, 'add_font')
|
||||
|
||||
@@ -1,104 +0,0 @@
|
||||
"""
|
||||
Tests for SchemaManager.merge_with_defaults — the merge every plugin config
|
||||
passes through at load time (schema defaults + user config, with None
|
||||
replacement). A regression here silently changes every plugin's effective
|
||||
config, so the exact branch behavior is pinned, including the
|
||||
characterized type-mismatch cases.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from src.plugin_system.schema_manager import SchemaManager
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sm(tmp_path):
|
||||
return SchemaManager(plugins_dir=str(tmp_path))
|
||||
|
||||
|
||||
class TestBasicMerge:
|
||||
def test_missing_keys_filled_from_defaults(self, sm):
|
||||
merged = sm.merge_with_defaults(
|
||||
{"city": "Austin"}, {"city": "NYC", "units": "metric"})
|
||||
assert merged == {"city": "Austin", "units": "metric"}
|
||||
|
||||
def test_present_keys_preserved(self, sm):
|
||||
merged = sm.merge_with_defaults({"enabled": False}, {"enabled": True})
|
||||
assert merged["enabled"] is False
|
||||
|
||||
def test_nested_three_level_merge(self, sm):
|
||||
config = {"a": {"b": {"c": 1}}}
|
||||
defaults = {"a": {"b": {"c": 0, "d": 2}, "e": 3}}
|
||||
merged = sm.merge_with_defaults(config, defaults)
|
||||
assert merged == {"a": {"b": {"c": 1, "d": 2}, "e": 3}}
|
||||
|
||||
def test_inputs_not_mutated(self, sm):
|
||||
config = {"a": {"b": 1}}
|
||||
defaults = {"a": {"b": 0, "c": 2}, "d": 3}
|
||||
sm.merge_with_defaults(config, defaults)
|
||||
assert config == {"a": {"b": 1}}
|
||||
assert defaults == {"a": {"b": 0, "c": 2}, "d": 3}
|
||||
|
||||
def test_merged_values_are_copies_not_aliases(self, sm):
|
||||
config = {"teams": ["DAL"]}
|
||||
merged = sm.merge_with_defaults(config, {"teams": []})
|
||||
merged["teams"].append("HOU")
|
||||
assert config["teams"] == ["DAL"] # user's list untouched
|
||||
|
||||
|
||||
class TestNoneReplacement:
|
||||
def test_none_replaced_by_default(self, sm):
|
||||
merged = sm.merge_with_defaults({"units": None}, {"units": "metric"})
|
||||
assert merged["units"] == "metric"
|
||||
|
||||
def test_falsey_non_none_values_kept(self, sm):
|
||||
merged = sm.merge_with_defaults(
|
||||
{"enabled": False, "count": 0, "label": ""},
|
||||
{"enabled": True, "count": 5, "label": "x"},
|
||||
)
|
||||
assert merged == {"enabled": False, "count": 0, "label": ""}
|
||||
|
||||
def test_nested_none_replaced(self, sm):
|
||||
merged = sm.merge_with_defaults(
|
||||
{"style": {"color": None}}, {"style": {"color": "red"}})
|
||||
assert merged["style"]["color"] == "red"
|
||||
|
||||
def test_none_with_no_default_stays_none(self, sm):
|
||||
merged = sm.merge_with_defaults({"extra": None}, {})
|
||||
assert merged["extra"] is None
|
||||
|
||||
def test_none_replaced_by_dict_default_is_a_copy(self, sm):
|
||||
defaults = {"style": {"color": "red"}}
|
||||
merged = sm.merge_with_defaults({"style": None}, defaults)
|
||||
assert merged["style"] == {"color": "red"}
|
||||
merged["style"]["color"] = "blue"
|
||||
assert defaults["style"]["color"] == "red"
|
||||
|
||||
|
||||
class TestTypeMismatches:
|
||||
def test_user_scalar_over_dict_default_wins(self, sm):
|
||||
# Characterized: a scalar user value replaces a dict default outright.
|
||||
merged = sm.merge_with_defaults(
|
||||
{"style": "compact"}, {"style": {"color": "red"}})
|
||||
assert merged["style"] == "compact"
|
||||
|
||||
def test_user_dict_over_scalar_default_wins(self, sm):
|
||||
merged = sm.merge_with_defaults(
|
||||
{"style": {"color": "red"}}, {"style": "compact"})
|
||||
assert merged["style"] == {"color": "red"}
|
||||
|
||||
def test_arrays_replaced_wholesale_not_merged(self, sm):
|
||||
# Pinned contract: arrays never element-merge — the user's array is
|
||||
# the whole answer, even when shorter than the default.
|
||||
merged = sm.merge_with_defaults(
|
||||
{"teams": ["DAL"]}, {"teams": ["NYG", "PHI", "WAS"]})
|
||||
assert merged["teams"] == ["DAL"]
|
||||
|
||||
def test_empty_user_array_beats_default(self, sm):
|
||||
merged = sm.merge_with_defaults({"teams": []}, {"teams": ["NYG"]})
|
||||
assert merged["teams"] == []
|
||||
|
||||
def test_extra_user_keys_survive(self, sm):
|
||||
# Keys with no schema default pass through untouched.
|
||||
merged = sm.merge_with_defaults({"custom_flag": 7}, {"known": 1})
|
||||
assert merged == {"known": 1, "custom_flag": 7}
|
||||
@@ -450,122 +450,3 @@ class TestExampleSkin:
|
||||
ctx = skin_runtime.build_context(host, game, size=size)
|
||||
assert getattr(skin, f"render_{mode}")(ctx, game) is True
|
||||
assert ctx.canvas.convert("L").getbbox() is not None
|
||||
|
||||
|
||||
class TestRenderSkinCard:
|
||||
"""render_skin_card (vegas cards) shares _render_game's 3-strike counter.
|
||||
|
||||
Both paths reset the counter on success — transient failures must not
|
||||
accumulate across a session and disable a working skin.
|
||||
"""
|
||||
|
||||
def _probe(self, skin):
|
||||
from src.base_classes.sports import SportsCore
|
||||
probe = _FallbackProbe(skin)
|
||||
probe.render_skin_card = (
|
||||
lambda game, size: SportsCore.render_skin_card(probe, game, size))
|
||||
return probe
|
||||
|
||||
def test_vegas_card_returned_when_skin_provides_one(self):
|
||||
card_img = Image.new("RGB", (96, 32), (0, 0, 255))
|
||||
|
||||
class CardSkin(ScoreboardSkin):
|
||||
def render_vegas_card(self, ctx, game):
|
||||
return card_img
|
||||
|
||||
probe = self._probe(CardSkin({}, {}))
|
||||
assert probe.render_skin_card({}, (96, 32)) is card_img
|
||||
|
||||
def test_vegas_card_none_falls_through_to_mode_renderer(self):
|
||||
class ModeOnlySkin(ScoreboardSkin):
|
||||
def render_live(self, ctx, game):
|
||||
ctx.draw.rectangle([0, 0, 5, 5], fill=(255, 0, 0))
|
||||
return True
|
||||
|
||||
probe = self._probe(ModeOnlySkin({}, {}))
|
||||
card = probe.render_skin_card({}, (96, 32))
|
||||
assert card is not None
|
||||
assert card.size == (96, 32)
|
||||
assert card.convert("L").getbbox() is not None
|
||||
|
||||
def test_skin_declining_returns_none(self):
|
||||
probe = self._probe(ScoreboardSkin({}, {})) # all renders -> False
|
||||
assert probe.render_skin_card({}, (96, 32)) is None
|
||||
assert probe._skin_failures == 0 # declining is not a failure
|
||||
|
||||
def test_no_skin_returns_none(self):
|
||||
probe = self._probe(None)
|
||||
assert probe.render_skin_card({}, (96, 32)) is None
|
||||
|
||||
def test_card_failures_count_toward_shared_disable(self):
|
||||
class BrokenCardSkin(ScoreboardSkin):
|
||||
calls = 0
|
||||
|
||||
def render_vegas_card(self, ctx, game):
|
||||
BrokenCardSkin.calls += 1
|
||||
raise ValueError("kaboom")
|
||||
|
||||
probe = self._probe(BrokenCardSkin({}, {}))
|
||||
for _ in range(5):
|
||||
assert probe.render_skin_card({}, (96, 32)) is None
|
||||
# Skin stopped being consulted after the 3rd failure...
|
||||
assert BrokenCardSkin.calls == 3
|
||||
assert probe._skin_failures == 3
|
||||
# ...and the shared counter also disables _render_game's skin path.
|
||||
probe._render_game({"status_text": "Q1"})
|
||||
assert probe.builtin_calls == 1
|
||||
assert BrokenCardSkin.calls == 3 # not consulted again
|
||||
|
||||
def test_card_success_resets_strikes(self):
|
||||
"""A successful card render clears accumulated strikes (mirroring
|
||||
_render_game) — 2 failures + a success + 1 failure leaves the skin
|
||||
enabled with a single strike, instead of disabling it."""
|
||||
card_img = Image.new("RGB", (96, 32), (0, 0, 255))
|
||||
|
||||
class FlakyCardSkin(ScoreboardSkin):
|
||||
fail = True
|
||||
|
||||
def render_vegas_card(self, ctx, game):
|
||||
if FlakyCardSkin.fail:
|
||||
raise ValueError("kaboom")
|
||||
return card_img
|
||||
|
||||
probe = self._probe(FlakyCardSkin({}, {}))
|
||||
FlakyCardSkin.fail = True
|
||||
probe.render_skin_card({}, (96, 32))
|
||||
probe.render_skin_card({}, (96, 32))
|
||||
assert probe._skin_failures == 2
|
||||
|
||||
FlakyCardSkin.fail = False
|
||||
assert probe.render_skin_card({}, (96, 32)) is card_img
|
||||
assert probe._skin_failures == 0 # success cleared the strikes
|
||||
|
||||
FlakyCardSkin.fail = True
|
||||
probe.render_skin_card({}, (96, 32))
|
||||
assert probe._skin_failures == 1 # counting from the reset state
|
||||
FlakyCardSkin.fail = False
|
||||
assert probe.render_skin_card({}, (96, 32)) is card_img # still enabled
|
||||
|
||||
def test_card_success_via_mode_renderer_also_resets_strikes(self):
|
||||
"""The fallthrough path (render_vegas_card None -> mode renderer
|
||||
True) resets the counter as well."""
|
||||
class ModeOnlySkin(ScoreboardSkin):
|
||||
def render_live(self, ctx, game):
|
||||
ctx.draw.rectangle([0, 0, 5, 5], fill=(255, 0, 0))
|
||||
return True
|
||||
|
||||
probe = self._probe(ModeOnlySkin({}, {}))
|
||||
probe._skin_failures = 2
|
||||
assert probe.render_skin_card({}, (96, 32)) is not None
|
||||
assert probe._skin_failures == 0
|
||||
|
||||
def test_render_game_success_also_resets_strikes(self):
|
||||
"""Same reset contract on the display path, for symmetry."""
|
||||
class GoodSkin(ScoreboardSkin):
|
||||
def render_live(self, ctx, game):
|
||||
return True
|
||||
|
||||
probe = _FallbackProbe(GoodSkin({}, {}))
|
||||
probe._skin_failures = 2
|
||||
probe._render_game({"status_text": "Q1"})
|
||||
assert probe._skin_failures == 0
|
||||
|
||||
@@ -1,175 +0,0 @@
|
||||
"""
|
||||
Drift guard for version comparison.
|
||||
|
||||
There is now ONE shared "should this plugin update?" comparator —
|
||||
`src.plugin_system.compatibility.is_update_available` — used by both the web
|
||||
UI's update badge (`api_v3._is_plugin_update_available`) and the store's
|
||||
`update_plugin` reinstall decision, so the badge and the actual reinstall
|
||||
can never disagree. (Historically the store used raw string equality, which
|
||||
reinstalled over cosmetic differences like "v1.2.0" vs "1.2.0" and even
|
||||
DOWNGRADED locally-ahead plugins; this file's tests killed that.)
|
||||
|
||||
Two other version parsers legitimately remain and are pinned here so they
|
||||
don't drift: `compatibility.parse_semver` (the install-compatibility gate,
|
||||
range-spec oriented) and `skin_runtime._major` (skin API major gate).
|
||||
"""
|
||||
|
||||
import json
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from packaging.version import parse as pkg_parse
|
||||
|
||||
from src.plugin_system.compatibility import is_update_available, parse_semver
|
||||
from src.skin_system.skin_runtime import _major
|
||||
from src.plugin_system.store_manager import PluginStoreManager
|
||||
from web_interface.blueprints.api_v3 import _is_plugin_update_available
|
||||
|
||||
|
||||
# (installed, registry) -> update available?
|
||||
CASES = [
|
||||
(("1.2.0", "1.2.0"), False), # identical
|
||||
(("v1.2.0", "1.2.0"), False), # cosmetic v-prefix, semantically equal
|
||||
(("1.2", "1.2.0"), False), # short form, semantically equal
|
||||
(("1.2.0", "1.2.0-rc1"), False), # rc of same release is not newer
|
||||
(("1.2.0", "1.3.0"), True), # registry genuinely newer
|
||||
(("2.0.0", "1.9.0"), False), # locally ahead — never downgrade
|
||||
(("abc.def", "1.0.0"), True), # unparseable — surface the mismatch
|
||||
(("", "1.0.0"), False), # missing either side — nothing to do
|
||||
(("1.0.0", ""), False),
|
||||
]
|
||||
|
||||
|
||||
class TestSharedComparatorMalformedInputs:
|
||||
def test_truthy_non_string_surfaces_mismatch(self):
|
||||
# A malformed manifest can carry version as a number; packaging would
|
||||
# raise TypeError on it. The comparator must not raise.
|
||||
assert is_update_available(1.2, "1.2.0") is True
|
||||
assert is_update_available("1.2.0", 1.3) is True
|
||||
|
||||
def test_falsy_non_string_means_nothing_to_do(self):
|
||||
assert is_update_available(None, "1.0.0") is False
|
||||
assert is_update_available("1.0.0", None) is False
|
||||
assert is_update_available(0, "1.0.0") is False
|
||||
|
||||
|
||||
class TestSharedComparator:
|
||||
@pytest.mark.parametrize("pair,expected", CASES)
|
||||
def test_is_update_available(self, pair, expected):
|
||||
installed, latest = pair
|
||||
assert is_update_available(installed, latest) is expected
|
||||
|
||||
@pytest.mark.parametrize("pair,expected", CASES)
|
||||
def test_api_v3_helper_agrees(self, pair, expected):
|
||||
# The UI badge helper must be a pure alias of the shared comparator.
|
||||
installed, latest = pair
|
||||
assert _is_plugin_update_available(installed, latest) is expected
|
||||
|
||||
|
||||
class TestStoreManagerUsesSharedComparator:
|
||||
"""Drive update_plugin's real code path to its version check."""
|
||||
|
||||
def _store(self, tmp_path, local_version, registry_version):
|
||||
plugin_dir = tmp_path / "plugins" / "demo-plugin"
|
||||
plugin_dir.mkdir(parents=True)
|
||||
(plugin_dir / "manifest.json").write_text(json.dumps({
|
||||
"id": "demo-plugin", "version": local_version,
|
||||
}))
|
||||
store = PluginStoreManager(
|
||||
plugins_dir=str(tmp_path / "plugins"),
|
||||
uninstalled_registry_path=str(tmp_path / "uninstalled.json"),
|
||||
)
|
||||
registry_info = {
|
||||
"id": "demo-plugin",
|
||||
"repo": "https://github.com/example/ledmatrix-plugins",
|
||||
"latest_version": registry_version,
|
||||
}
|
||||
return store, registry_info
|
||||
|
||||
def _run_update(self, store, registry_info):
|
||||
with patch.object(store, "fetch_registry", return_value={"plugins": [registry_info]}), \
|
||||
patch.object(store, "get_plugin_info", return_value=registry_info), \
|
||||
patch.object(store, "_reinstall_with_rollback", return_value=True) as reinstall:
|
||||
result = store.update_plugin("demo-plugin")
|
||||
return result, reinstall
|
||||
|
||||
def test_equal_strings_skip_reinstall(self, tmp_path):
|
||||
store, info = self._store(tmp_path, "1.2.0", "1.2.0")
|
||||
result, reinstall = self._run_update(store, info)
|
||||
assert result is True
|
||||
reinstall.assert_not_called()
|
||||
|
||||
def test_v_prefix_equivalent_skips_reinstall(self, tmp_path):
|
||||
# "v1.2.0" == "1.2.0" semantically — no pointless reinstall.
|
||||
store, info = self._store(tmp_path, "v1.2.0", "1.2.0")
|
||||
result, reinstall = self._run_update(store, info)
|
||||
assert result is True
|
||||
reinstall.assert_not_called()
|
||||
|
||||
def test_locally_ahead_version_is_never_downgraded(self, tmp_path):
|
||||
# A plugin ahead of the registry (local dev build) must not be
|
||||
# "updated" — that would be a downgrade.
|
||||
store, info = self._store(tmp_path, "2.0.0", "1.9.0")
|
||||
result, reinstall = self._run_update(store, info)
|
||||
assert result is True
|
||||
reinstall.assert_not_called()
|
||||
|
||||
def test_registry_newer_triggers_reinstall(self, tmp_path):
|
||||
store, info = self._store(tmp_path, "1.2.0", "1.3.0")
|
||||
result, reinstall = self._run_update(store, info)
|
||||
reinstall.assert_called_once()
|
||||
assert result is True
|
||||
|
||||
def test_unparseable_version_surfaces_via_reinstall(self, tmp_path):
|
||||
# Direction unknowable → reconcile by reinstalling from the registry.
|
||||
store, info = self._store(tmp_path, "abc.def", "1.0.0")
|
||||
result, reinstall = self._run_update(store, info)
|
||||
reinstall.assert_called_once()
|
||||
assert result is True
|
||||
|
||||
def test_empty_local_version_follows_comparator_no_reinstall(self, tmp_path):
|
||||
# The comparator says "nothing to do" for a missing version, and the
|
||||
# store must agree with the UI badge — no reinstall.
|
||||
store, info = self._store(tmp_path, "", "1.0.0")
|
||||
result, reinstall = self._run_update(store, info)
|
||||
assert result is True
|
||||
reinstall.assert_not_called()
|
||||
|
||||
def test_empty_registry_version_follows_comparator_no_reinstall(self, tmp_path):
|
||||
store, info = self._store(tmp_path, "1.0.0", "")
|
||||
result, reinstall = self._run_update(store, info)
|
||||
assert result is True
|
||||
reinstall.assert_not_called()
|
||||
|
||||
|
||||
class TestSkinRuntimeMajor:
|
||||
def test_plain_versions(self):
|
||||
assert _major("1.0.0") == 1
|
||||
assert _major("2.1") == 2
|
||||
|
||||
def test_int_input_tolerated(self):
|
||||
assert _major(2) == 2
|
||||
|
||||
def test_garbage_returns_none(self):
|
||||
assert _major("garbage") is None
|
||||
assert _major(None) is None
|
||||
|
||||
def test_v_prefix_not_tolerated(self):
|
||||
# Unlike parse_semver, _major does NOT strip a leading 'v' —
|
||||
# a skin.json declaring "v1.0.0" fails the API gate. Characterized
|
||||
# so a manifest-format loosening elsewhere doesn't silently diverge.
|
||||
assert _major("v1.0.0") is None
|
||||
|
||||
|
||||
class TestParseSemverAgreesWithPackaging:
|
||||
"""parse_semver and packaging must agree on ordering for plain X.Y.Z —
|
||||
the region where the two ecosystems overlap and must never diverge."""
|
||||
|
||||
PLAIN = ["0.1.0", "1.0.0", "1.2.0", "1.2.3", "1.10.0", "2.0.0", "10.0.1"]
|
||||
|
||||
def test_pairwise_ordering_matches(self):
|
||||
for a in self.PLAIN:
|
||||
for b in self.PLAIN:
|
||||
ours = parse_semver(a) < parse_semver(b)
|
||||
theirs = pkg_parse(a) < pkg_parse(b)
|
||||
assert ours == theirs, f"ordering diverges on ({a}, {b})"
|
||||
@@ -393,9 +393,6 @@ class TestSystemAPI:
|
||||
@patch('web_interface.blueprints.api_v3.subprocess')
|
||||
def test_get_system_status(self, mock_subprocess, client):
|
||||
"""Test getting system status."""
|
||||
# The endpoint returns 503 without psutil, which is an optional
|
||||
# runtime dependency (requirements-test.txt installs it for CI).
|
||||
pytest.importorskip("psutil")
|
||||
mock_result = MagicMock()
|
||||
mock_result.stdout = 'active\n'
|
||||
mock_result.returncode = 0
|
||||
|
||||
@@ -1,232 +0,0 @@
|
||||
"""
|
||||
Unit tests for the module-level helper functions in
|
||||
web_interface/blueprints/api_v3.py.
|
||||
|
||||
These helpers back the plugin config save endpoint (the largest function in
|
||||
the repo) and the store's update-available detection, but were previously
|
||||
exercised only indirectly through full Flask route tests. Testing them
|
||||
directly pins behavior that the routes rely on — including a few
|
||||
characterized quirks marked below.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any, ClassVar, Dict
|
||||
|
||||
import pytest
|
||||
|
||||
project_root = Path(__file__).parent.parent.parent
|
||||
sys.path.insert(0, str(project_root))
|
||||
|
||||
from web_interface.blueprints.api_v3 import ( # noqa: E402
|
||||
_is_plugin_update_available,
|
||||
_coerce_to_bool,
|
||||
deep_merge,
|
||||
_parse_form_value,
|
||||
_get_schema_property,
|
||||
_set_nested_value,
|
||||
)
|
||||
|
||||
|
||||
class TestIsPluginUpdateAvailable:
|
||||
def test_equal_versions_no_update(self):
|
||||
assert _is_plugin_update_available("1.2.0", "1.2.0") is False
|
||||
|
||||
def test_newer_registry_version_needs_update(self):
|
||||
assert _is_plugin_update_available("1.2.0", "1.3.0") is True
|
||||
|
||||
def test_installed_ahead_of_registry_no_update(self):
|
||||
# A locally modified plugin ahead of the registry must not be
|
||||
# flagged — this is the whole point of semantic comparison here.
|
||||
assert _is_plugin_update_available("2.0.0", "1.9.0") is False
|
||||
|
||||
def test_empty_versions_no_update(self):
|
||||
assert _is_plugin_update_available("", "1.0.0") is False
|
||||
assert _is_plugin_update_available("1.0.0", "") is False
|
||||
assert _is_plugin_update_available("", "") is False
|
||||
|
||||
def test_v_prefix_parses_as_equal(self):
|
||||
# packaging.version treats "v1.2.0" == "1.2.0" (PEP 440 tolerates the
|
||||
# prefix), so no update is flagged. Contrast with store_manager's
|
||||
# string-equality check — see test_version_comparison_consistency.py.
|
||||
assert _is_plugin_update_available("v1.2.0", "1.2.0") is False
|
||||
|
||||
def test_two_part_version_parses_as_equal(self):
|
||||
assert _is_plugin_update_available("1.2", "1.2.0") is False
|
||||
|
||||
def test_unparseable_version_surfaces_mismatch(self):
|
||||
# Direction unknowable → surface the difference rather than hide a
|
||||
# potential update.
|
||||
assert _is_plugin_update_available("abc.def", "1.0.0") is True
|
||||
|
||||
def test_prerelease_below_release(self):
|
||||
assert _is_plugin_update_available("1.2.0-rc1", "1.2.0") is True
|
||||
|
||||
|
||||
class TestCoerceToBool:
|
||||
@pytest.mark.parametrize("value", ["true", "TRUE", "on", "1", "yes", "YES"])
|
||||
def test_truthy_strings(self, value):
|
||||
assert _coerce_to_bool(value) is True
|
||||
|
||||
@pytest.mark.parametrize("value", ["false", "off", "0", "no", "", "banana"])
|
||||
def test_falsey_strings(self, value):
|
||||
assert _coerce_to_bool(value) is False
|
||||
|
||||
def test_none_is_false(self):
|
||||
assert _coerce_to_bool(None) is False
|
||||
|
||||
def test_bools_pass_through(self):
|
||||
assert _coerce_to_bool(True) is True
|
||||
assert _coerce_to_bool(False) is False
|
||||
|
||||
def test_int_only_one_is_true(self):
|
||||
# Characterized quirk: ints coerce via `value == 1`, so 2 (truthy in
|
||||
# Python) is False here.
|
||||
assert _coerce_to_bool(1) is True
|
||||
assert _coerce_to_bool(2) is False
|
||||
assert _coerce_to_bool(0) is False
|
||||
|
||||
def test_other_types_false(self):
|
||||
assert _coerce_to_bool([1]) is False
|
||||
assert _coerce_to_bool({"a": 1}) is False
|
||||
|
||||
|
||||
class TestDeepMerge:
|
||||
def test_nested_dicts_merge_recursively(self):
|
||||
base = {"a": {"x": 1, "y": 2}, "b": 1}
|
||||
update = {"a": {"y": 3, "z": 4}}
|
||||
assert deep_merge(base, update) == {"a": {"x": 1, "y": 3, "z": 4}, "b": 1}
|
||||
|
||||
def test_scalar_over_dict_replaces(self):
|
||||
assert deep_merge({"a": {"x": 1}}, {"a": 5}) == {"a": 5}
|
||||
|
||||
def test_dict_over_scalar_replaces(self):
|
||||
assert deep_merge({"a": 5}, {"a": {"x": 1}}) == {"a": {"x": 1}}
|
||||
|
||||
def test_lists_replaced_wholesale(self):
|
||||
assert deep_merge({"a": [1, 2]}, {"a": [3]}) == {"a": [3]}
|
||||
|
||||
def test_top_level_not_mutated_but_shallow_copy(self):
|
||||
# Characterized: result = base.copy() protects base's top level, but
|
||||
# nested dicts NOT touched by the update are shared by reference.
|
||||
base = {"a": {"x": 1}, "keep": {"y": 2}}
|
||||
result = deep_merge(base, {"a": {"x": 9}})
|
||||
assert base == {"a": {"x": 1}, "keep": {"y": 2}} # base unchanged
|
||||
assert result["keep"] is base["keep"] # untouched subtree is shared
|
||||
|
||||
|
||||
class TestParseFormValue:
|
||||
def test_boolean_strings(self):
|
||||
assert _parse_form_value("true") is True
|
||||
assert _parse_form_value("False") is False
|
||||
|
||||
def test_null_like_strings(self):
|
||||
assert _parse_form_value("null") is None
|
||||
assert _parse_form_value("none") is None
|
||||
assert _parse_form_value("") is None
|
||||
|
||||
def test_none_passthrough(self):
|
||||
assert _parse_form_value(None) is None
|
||||
|
||||
def test_numbers(self):
|
||||
assert _parse_form_value("42") == 42
|
||||
assert isinstance(_parse_form_value("42"), int)
|
||||
assert _parse_form_value("3.5") == 3.5
|
||||
assert isinstance(_parse_form_value("3.5"), float)
|
||||
|
||||
def test_json_array_parsed_before_numbers(self):
|
||||
# RGB arrays like "[255, 0, 0]" must come back as lists.
|
||||
assert _parse_form_value("[255, 0, 0]") == [255, 0, 0]
|
||||
|
||||
def test_json_object(self):
|
||||
assert _parse_form_value('{"a": 1}') == {"a": 1}
|
||||
|
||||
def test_malformed_json_falls_back_to_string(self):
|
||||
assert _parse_form_value("[not json") == "[not json"
|
||||
|
||||
def test_plain_string_returned_unstripped(self):
|
||||
# The original value (not the stripped copy) is returned.
|
||||
assert _parse_form_value(" hello ") == " hello "
|
||||
|
||||
def test_non_string_passthrough(self):
|
||||
assert _parse_form_value(7) == 7
|
||||
assert _parse_form_value([1, 2]) == [1, 2]
|
||||
|
||||
|
||||
class TestGetSchemaProperty:
|
||||
SCHEMA: ClassVar[Dict[str, Any]] = {
|
||||
"properties": {
|
||||
"brightness": {"type": "integer"},
|
||||
"customization": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"time_text": {
|
||||
"type": "object",
|
||||
"properties": {"font": {"type": "string"}},
|
||||
},
|
||||
},
|
||||
},
|
||||
"fifa.world": {"type": "object",
|
||||
"properties": {"enabled": {"type": "boolean"}}},
|
||||
}
|
||||
}
|
||||
|
||||
def test_top_level_lookup(self):
|
||||
assert _get_schema_property(self.SCHEMA, "brightness") == {"type": "integer"}
|
||||
|
||||
def test_nested_dot_path(self):
|
||||
prop = _get_schema_property(self.SCHEMA, "customization.time_text.font")
|
||||
assert prop == {"type": "string"}
|
||||
|
||||
def test_dotted_schema_key_matched_longest_first(self):
|
||||
# League keys like "fifa.world" contain a literal dot and must match
|
||||
# as a single key, not be split into nested fifa -> world lookups.
|
||||
prop = _get_schema_property(self.SCHEMA, "fifa.world.enabled")
|
||||
assert prop == {"type": "boolean"}
|
||||
|
||||
def test_missing_path_returns_none(self):
|
||||
assert _get_schema_property(self.SCHEMA, "nope.nope") is None
|
||||
|
||||
def test_no_properties_returns_none(self):
|
||||
assert _get_schema_property({}, "a") is None
|
||||
assert _get_schema_property(None, "a") is None
|
||||
|
||||
|
||||
class TestSetNestedValue:
|
||||
def test_sets_top_level(self):
|
||||
config = {}
|
||||
_set_nested_value(config, "brightness", 80)
|
||||
assert config == {"brightness": 80}
|
||||
|
||||
def test_creates_intermediate_dicts(self):
|
||||
config = {}
|
||||
_set_nested_value(config, "customization.time_text.font", "5x7")
|
||||
assert config == {"customization": {"time_text": {"font": "5x7"}}}
|
||||
|
||||
def test_merges_into_existing_nested_dict(self):
|
||||
config = {"customization": {"color": "red"}}
|
||||
_set_nested_value(config, "customization.font", "5x7")
|
||||
assert config == {"customization": {"color": "red", "font": "5x7"}}
|
||||
|
||||
def test_scalar_intermediate_replaced_with_dict(self):
|
||||
# Characterized: a non-dict intermediate is silently replaced.
|
||||
config = {"customization": "oops"}
|
||||
_set_nested_value(config, "customization.font", "5x7")
|
||||
assert config == {"customization": {"font": "5x7"}}
|
||||
|
||||
def test_existing_dotted_key_preserved(self):
|
||||
# An existing literal "fifa.world" key must be updated in place, not
|
||||
# exploded into nested {"fifa": {"world": ...}}.
|
||||
config = {"fifa.world": {"enabled": False}}
|
||||
_set_nested_value(config, "fifa.world.enabled", True)
|
||||
assert config == {"fifa.world": {"enabled": True}}
|
||||
|
||||
def test_none_does_not_overwrite_existing(self):
|
||||
config = {"a": 1}
|
||||
_set_nested_value(config, "a", None)
|
||||
assert config == {"a": 1}
|
||||
|
||||
def test_none_sets_missing_key(self):
|
||||
config = {}
|
||||
_set_nested_value(config, "a", None)
|
||||
assert config == {"a": None}
|
||||
@@ -1,241 +0,0 @@
|
||||
"""
|
||||
Tests for src/web_interface/secret_helpers.py — the canonical secret
|
||||
identification / separation / masking helpers.
|
||||
|
||||
This module is the extracted single source of truth for x-secret handling,
|
||||
but until now had zero test coverage (only ``mask_secret_fields`` is even
|
||||
imported by production code, from pages_v3). api_v3.py still carries three
|
||||
inline re-implementations of ``find_secret_fields``/``separate_secrets`` —
|
||||
see test_secret_separation_parity.py — so pinning the canonical behavior
|
||||
here is a precondition for ever migrating those copies.
|
||||
"""
|
||||
|
||||
import copy
|
||||
|
||||
from src.web_interface.secret_helpers import (
|
||||
find_secret_fields,
|
||||
separate_secrets,
|
||||
mask_secret_fields,
|
||||
mask_all_secret_values,
|
||||
remove_empty_secrets,
|
||||
)
|
||||
|
||||
|
||||
SCHEMA_PROPS = {
|
||||
"api_key": {"type": "string", "x-secret": True},
|
||||
"city": {"type": "string"},
|
||||
"auth": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"token": {"type": "string", "x-secret": True},
|
||||
"username": {"type": "string"},
|
||||
},
|
||||
},
|
||||
"accounts": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string"},
|
||||
"token": {"type": "string", "x-secret": True},
|
||||
},
|
||||
},
|
||||
},
|
||||
"recovery_codes": {
|
||||
"type": "array",
|
||||
"items": {"type": "string", "x-secret": True},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class TestFindSecretFields:
|
||||
def test_top_level_secret(self):
|
||||
assert "api_key" in find_secret_fields(SCHEMA_PROPS)
|
||||
|
||||
def test_non_secret_not_included(self):
|
||||
assert "city" not in find_secret_fields(SCHEMA_PROPS)
|
||||
|
||||
def test_nested_object_secret_uses_dot_path(self):
|
||||
assert "auth.token" in find_secret_fields(SCHEMA_PROPS)
|
||||
assert "auth.username" not in find_secret_fields(SCHEMA_PROPS)
|
||||
|
||||
def test_array_item_object_secret_uses_bracket_path(self):
|
||||
assert "accounts[].token" in find_secret_fields(SCHEMA_PROPS)
|
||||
|
||||
def test_array_of_secrets_uses_bracket_path(self):
|
||||
assert "recovery_codes[]" in find_secret_fields(SCHEMA_PROPS)
|
||||
|
||||
def test_full_set(self):
|
||||
assert find_secret_fields(SCHEMA_PROPS) == {
|
||||
"api_key", "auth.token", "accounts[].token", "recovery_codes[]",
|
||||
}
|
||||
|
||||
def test_non_dict_properties_tolerated(self):
|
||||
assert find_secret_fields({"weird": "not-a-dict"}) == set()
|
||||
|
||||
def test_non_dict_input_returns_empty(self):
|
||||
assert find_secret_fields(None) == set()
|
||||
assert find_secret_fields([]) == set()
|
||||
|
||||
|
||||
class TestSeparateSecrets:
|
||||
def test_flat_partition(self):
|
||||
regular, secrets = separate_secrets(
|
||||
{"api_key": "s3cret", "city": "Austin"}, {"api_key"})
|
||||
assert regular == {"city": "Austin"}
|
||||
assert secrets == {"api_key": "s3cret"}
|
||||
|
||||
def test_nested_partition(self):
|
||||
config = {"auth": {"token": "t0k", "username": "chuck"}}
|
||||
regular, secrets = separate_secrets(config, {"auth.token"})
|
||||
assert regular == {"auth": {"username": "chuck"}}
|
||||
assert secrets == {"auth": {"token": "t0k"}}
|
||||
|
||||
def test_empty_nested_dicts_pruned_from_regular(self):
|
||||
# A dict that is all secrets leaves nothing behind on the regular
|
||||
# side — the key must be dropped, not kept as {}.
|
||||
config = {"auth": {"token": "t0k"}}
|
||||
regular, secrets = separate_secrets(config, {"auth.token"})
|
||||
assert regular == {}
|
||||
assert secrets == {"auth": {"token": "t0k"}}
|
||||
|
||||
def test_whole_array_secret(self):
|
||||
config = {"recovery_codes": ["a", "b"], "city": "Austin"}
|
||||
regular, secrets = separate_secrets(config, {"recovery_codes[]"})
|
||||
assert regular == {"city": "Austin"}
|
||||
assert secrets == {"recovery_codes": ["a", "b"]}
|
||||
|
||||
def test_array_item_secrets_produce_parallel_lists(self):
|
||||
# Per-item secrets keep the arrays index-aligned so they can be
|
||||
# recombined: regular gets the stripped items, secrets a parallel
|
||||
# list of the extracted values.
|
||||
config = {"accounts": [
|
||||
{"name": "a", "token": "ta"},
|
||||
{"name": "b", "token": "tb"},
|
||||
]}
|
||||
regular, secrets = separate_secrets(config, {"accounts[].token"})
|
||||
assert regular == {"accounts": [{"name": "a"}, {"name": "b"}]}
|
||||
assert secrets == {"accounts": [{"token": "ta"}, {"token": "tb"}]}
|
||||
|
||||
def test_array_item_non_dict_items_get_placeholder(self):
|
||||
config = {"accounts": [{"name": "a", "token": "ta"}, "oddball"]}
|
||||
regular, secrets = separate_secrets(config, {"accounts[].token"})
|
||||
assert regular == {"accounts": [{"name": "a"}, "oddball"]}
|
||||
assert secrets == {"accounts": [{"token": "ta"}, {}]}
|
||||
|
||||
def test_array_without_secret_paths_stays_regular(self):
|
||||
config = {"teams": ["DAL", "HOU"]}
|
||||
regular, secrets = separate_secrets(config, {"api_key"})
|
||||
assert regular == {"teams": ["DAL", "HOU"]}
|
||||
assert secrets == {}
|
||||
|
||||
def test_round_trip_loses_nothing(self):
|
||||
# separate + naive recombine must reconstruct the original config.
|
||||
config = {
|
||||
"api_key": "k",
|
||||
"city": "Austin",
|
||||
"auth": {"token": "t", "username": "chuck"},
|
||||
"recovery_codes": ["a", "b"],
|
||||
}
|
||||
paths = find_secret_fields(SCHEMA_PROPS)
|
||||
regular, secrets = separate_secrets(copy.deepcopy(config), paths)
|
||||
|
||||
def recombine(reg, sec):
|
||||
out = copy.deepcopy(reg)
|
||||
for k, v in sec.items():
|
||||
if isinstance(v, dict) and isinstance(out.get(k), dict):
|
||||
out[k] = recombine(out[k], v)
|
||||
else:
|
||||
out[k] = v
|
||||
return out
|
||||
|
||||
assert recombine(regular, secrets) == config
|
||||
|
||||
|
||||
class TestMaskSecretFields:
|
||||
def test_masks_present_secret_to_empty_string(self):
|
||||
result = mask_secret_fields({"api_key": "s3cret"}, SCHEMA_PROPS)
|
||||
assert result["api_key"] == ""
|
||||
|
||||
def test_leaves_non_secret_untouched(self):
|
||||
result = mask_secret_fields({"city": "Austin"}, SCHEMA_PROPS)
|
||||
assert result["city"] == "Austin"
|
||||
|
||||
def test_none_and_empty_left_alone(self):
|
||||
result = mask_secret_fields({"api_key": None}, SCHEMA_PROPS)
|
||||
assert result["api_key"] is None
|
||||
result = mask_secret_fields({"api_key": ""}, SCHEMA_PROPS)
|
||||
assert result["api_key"] == ""
|
||||
|
||||
def test_falsey_but_set_values_are_masked(self):
|
||||
# 0 and False are real values; the check is `is not None and != ''`.
|
||||
# Note False == '' is False in Python, so False IS masked; 0 == '' is
|
||||
# also False, so 0 is masked too.
|
||||
result = mask_secret_fields({"api_key": 0}, SCHEMA_PROPS)
|
||||
assert result["api_key"] == ""
|
||||
result = mask_secret_fields({"api_key": False}, SCHEMA_PROPS)
|
||||
assert result["api_key"] == ""
|
||||
|
||||
def test_nested_object_masked_without_mutating_input(self):
|
||||
config = {"auth": {"token": "t0k", "username": "chuck"}}
|
||||
original = copy.deepcopy(config)
|
||||
result = mask_secret_fields(config, SCHEMA_PROPS)
|
||||
assert result["auth"]["token"] == ""
|
||||
assert result["auth"]["username"] == "chuck"
|
||||
assert config == original # input not mutated
|
||||
|
||||
def test_array_of_secrets_masked_elementwise(self):
|
||||
result = mask_secret_fields(
|
||||
{"recovery_codes": ["a", "b"]}, SCHEMA_PROPS)
|
||||
assert result["recovery_codes"] == ["", ""]
|
||||
|
||||
def test_array_of_objects_masked_per_item(self):
|
||||
config = {"accounts": [{"name": "a", "token": "ta"}, "oddball"]}
|
||||
result = mask_secret_fields(config, SCHEMA_PROPS)
|
||||
assert result["accounts"][0] == {"name": "a", "token": ""}
|
||||
assert result["accounts"][1] == "oddball"
|
||||
|
||||
def test_non_dict_schema_property_tolerated(self):
|
||||
assert mask_secret_fields({"x": 1}, {"x": "bogus"}) == {"x": 1}
|
||||
|
||||
|
||||
class TestMaskAllSecretValues:
|
||||
def test_real_values_replaced_with_bullets(self):
|
||||
assert mask_all_secret_values({"key": "abc"}) == {"key": "••••••••"}
|
||||
|
||||
def test_placeholders_preserved(self):
|
||||
# YOUR_* placeholders must survive so the UI can show "not set".
|
||||
result = mask_all_secret_values({"key": "YOUR_API_KEY_HERE"})
|
||||
assert result == {"key": "YOUR_API_KEY_HERE"}
|
||||
|
||||
def test_empty_and_none_preserved(self):
|
||||
assert mask_all_secret_values({"a": "", "b": None}) == {"a": "", "b": None}
|
||||
|
||||
def test_recurses_into_nested_dicts(self):
|
||||
result = mask_all_secret_values({"plugin": {"token": "t", "empty": ""}})
|
||||
assert result == {"plugin": {"token": "••••••••", "empty": ""}}
|
||||
|
||||
def test_non_string_real_values_masked(self):
|
||||
assert mask_all_secret_values({"port": 8080}) == {"port": "••••••••"}
|
||||
|
||||
|
||||
class TestRemoveEmptySecrets:
|
||||
def test_strips_empty_string(self):
|
||||
assert remove_empty_secrets({"a": "", "b": "real"}) == {"b": "real"}
|
||||
|
||||
def test_strips_whitespace_only(self):
|
||||
assert remove_empty_secrets({"a": " "}) == {}
|
||||
|
||||
def test_strips_none(self):
|
||||
assert remove_empty_secrets({"a": None}) == {}
|
||||
|
||||
def test_prunes_empty_nested_dicts(self):
|
||||
assert remove_empty_secrets({"plugin": {"token": ""}}) == {}
|
||||
|
||||
def test_keeps_nested_real_values(self):
|
||||
result = remove_empty_secrets({"plugin": {"token": "t", "empty": ""}})
|
||||
assert result == {"plugin": {"token": "t"}}
|
||||
|
||||
def test_keeps_falsey_non_string_values(self):
|
||||
# 0 and False are neither None nor blank strings — they are kept.
|
||||
assert remove_empty_secrets({"a": 0, "b": False}) == {"a": 0, "b": False}
|
||||
@@ -1,118 +0,0 @@
|
||||
"""
|
||||
Drift guard for the duplicated secret-separation logic.
|
||||
|
||||
src/web_interface/secret_helpers.py is the canonical implementation of
|
||||
find_secret_fields / separate_secrets, but web_interface/blueprints/api_v3.py
|
||||
still carries THREE inline nested-function copies of each (in the plugin
|
||||
config GET, POST, and reset endpoints). The copies lack the canonical
|
||||
module's array-item support (`accounts[].token`), so migrating an endpoint
|
||||
onto the module is a behavior change that must be made deliberately.
|
||||
|
||||
This file guards two things:
|
||||
1. The copy count can only go DOWN. A fourth copy appearing means someone
|
||||
re-implemented the logic again instead of importing secret_helpers.
|
||||
2. The known behavioral gap is documented as an executable fact, so whoever
|
||||
migrates the endpoints knows exactly what changes.
|
||||
"""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from src.web_interface.secret_helpers import find_secret_fields, separate_secrets
|
||||
|
||||
API_V3_PATH = (Path(__file__).resolve().parents[2]
|
||||
/ "web_interface" / "blueprints" / "api_v3.py")
|
||||
|
||||
# Update DOWNWARD as endpoints migrate onto src/web_interface/secret_helpers.
|
||||
EXPECTED_INLINE_COPIES = 3
|
||||
|
||||
|
||||
class TestInlineCopyCount:
|
||||
def _count(self, name: str) -> int:
|
||||
source = API_V3_PATH.read_text(encoding="utf-8")
|
||||
return len(re.findall(rf"^\s*def {name}\(", source, flags=re.MULTILINE))
|
||||
|
||||
def test_find_secret_fields_copy_count(self):
|
||||
count = self._count("find_secret_fields")
|
||||
assert count == EXPECTED_INLINE_COPIES, (
|
||||
f"api_v3.py has {count} inline find_secret_fields definitions, "
|
||||
f"expected {EXPECTED_INLINE_COPIES}. New code must import it from "
|
||||
f"src/web_interface/secret_helpers instead of re-implementing it; "
|
||||
f"if you migrated an endpoint, lower EXPECTED_INLINE_COPIES."
|
||||
)
|
||||
|
||||
def test_separate_secrets_copy_count(self):
|
||||
count = self._count("separate_secrets")
|
||||
assert count == EXPECTED_INLINE_COPIES, (
|
||||
f"api_v3.py has {count} inline separate_secrets definitions, "
|
||||
f"expected {EXPECTED_INLINE_COPIES}. New code must import it from "
|
||||
f"src/web_interface/secret_helpers instead of re-implementing it; "
|
||||
f"if you migrated an endpoint, lower EXPECTED_INLINE_COPIES."
|
||||
)
|
||||
|
||||
def test_inline_copies_lack_array_item_support(self):
|
||||
"""The documented gap: no inline copy recurses into array `items`
|
||||
schemas, so array-item secrets (accounts[].token) are NOT routed to
|
||||
config_secrets.json by these endpoints. The canonical module handles
|
||||
them. When an endpoint migrates onto the module that behavior
|
||||
changes (a fix, but a deliberate one).
|
||||
|
||||
If this fails, an inline copy has grown array support — duplicating
|
||||
the canonical module even harder. Migrate the endpoint onto
|
||||
src/web_interface/secret_helpers instead.
|
||||
"""
|
||||
for body in self._inline_bodies("find_secret_fields"):
|
||||
# Array handling requires checking type == 'array'; no inline
|
||||
# copy does. (Can't grep bare "items" — properties.items() the
|
||||
# dict method appears legitimately.)
|
||||
assert "'array'" not in body and '"array"' not in body
|
||||
|
||||
@staticmethod
|
||||
def _inline_bodies(name: str):
|
||||
"""Extract each inline def's body from api_v3.py by indentation."""
|
||||
lines = API_V3_PATH.read_text(encoding="utf-8").splitlines()
|
||||
bodies = []
|
||||
i = 0
|
||||
while i < len(lines):
|
||||
match = re.match(rf"^(\s+)def {name}\(", lines[i])
|
||||
if not match:
|
||||
i += 1
|
||||
continue
|
||||
indent = len(match.group(1))
|
||||
body = [lines[i]]
|
||||
i += 1
|
||||
while i < len(lines):
|
||||
line = lines[i]
|
||||
if line.strip() and (len(line) - len(line.lstrip())) <= indent:
|
||||
break
|
||||
body.append(line)
|
||||
i += 1
|
||||
bodies.append("\n".join(body))
|
||||
assert bodies, f"no inline {name} definitions found"
|
||||
return bodies
|
||||
|
||||
|
||||
class TestCanonicalArrayItemBehavior:
|
||||
"""Executable documentation of what migrating endpoints will change."""
|
||||
|
||||
SCHEMA = {
|
||||
"accounts": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string"},
|
||||
"token": {"type": "string", "x-secret": True},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
def test_canonical_module_routes_array_item_secrets(self):
|
||||
paths = find_secret_fields(self.SCHEMA)
|
||||
assert "accounts[].token" in paths
|
||||
|
||||
config = {"accounts": [{"name": "a", "token": "s3cret"}]}
|
||||
regular, secrets = separate_secrets(config, paths)
|
||||
assert regular == {"accounts": [{"name": "a"}]}
|
||||
assert secrets == {"accounts": [{"token": "s3cret"}]}
|
||||
@@ -367,10 +367,6 @@ class TestStateReconciliationUnrecoverable(unittest.TestCase):
|
||||
self.store_manager.fetch_registry.return_value = {"plugins": []}
|
||||
self.store_manager.install_plugin.return_value = False
|
||||
self.store_manager.was_recently_uninstalled.return_value = False
|
||||
# A bare Mock() returns a truthy Mock for is_plugin_uninstalled(),
|
||||
# which reads as "persistently uninstalled" and skips auto-repair
|
||||
# entirely — these tests need the repair path to run.
|
||||
self.store_manager.is_plugin_uninstalled.return_value = False
|
||||
|
||||
self.reconciler = StateReconciliation(
|
||||
state_manager=self.state_manager,
|
||||
|
||||
@@ -124,13 +124,28 @@ def _is_plugin_update_available(installed_version: str, latest_version: str) ->
|
||||
"""Return True when the registry's ``latest_version`` is strictly newer
|
||||
than the installed version.
|
||||
|
||||
Thin alias for the shared comparator in
|
||||
`src.plugin_system.compatibility.is_update_available` — the store's
|
||||
`update_plugin` uses the same function, so the UI badge and the actual
|
||||
reinstall decision can never disagree.
|
||||
Uses PEP 440 / semver-aware comparison so a locally modified plugin whose
|
||||
version is *ahead* of the published registry is not flagged as needing an
|
||||
update. If either version string can't be parsed, falls back to a plain
|
||||
inequality check (any difference is surfaced so the user can reconcile).
|
||||
"""
|
||||
from src.plugin_system.compatibility import is_update_available
|
||||
return is_update_available(installed_version, latest_version)
|
||||
if not installed_version or not latest_version:
|
||||
return False
|
||||
if installed_version == latest_version:
|
||||
return False
|
||||
try:
|
||||
from packaging.version import parse as _parse_version, InvalidVersion
|
||||
except ImportError:
|
||||
# packaging is a core dependency, but if it's somehow unavailable we
|
||||
# can't compare semantically — surface the mismatch we already know
|
||||
# exists (the two strings differ).
|
||||
return True
|
||||
try:
|
||||
return _parse_version(latest_version) > _parse_version(installed_version)
|
||||
except InvalidVersion:
|
||||
# Unparseable version string: we can't tell direction, so surface the
|
||||
# mismatch rather than silently hiding a potential update.
|
||||
return True
|
||||
|
||||
def _ensure_cache_manager():
|
||||
"""Ensure cache manager is initialized."""
|
||||
|
||||
@@ -121,14 +121,21 @@
|
||||
</label>
|
||||
{% endif %}
|
||||
|
||||
{# Enum dropdown #}
|
||||
{# Enum dropdown. Option text comes from x-options.labels when the
|
||||
schema supplies it -- the same convention the checkbox-group
|
||||
widget already uses -- because humanising the raw value cannot
|
||||
express every label: "vs" reads as "Vs", and "abbrev" says
|
||||
nothing about the "Sep 19" it produces. Values without a label
|
||||
fall back to the humanised form, so existing schemas render
|
||||
exactly as before. #}
|
||||
{% elif prop.enum %}
|
||||
{% set enum_labels = (prop.get('x-options') or prop.get('x_options') or {}).get('labels') or {} %}
|
||||
<select id="{{ field_id }}"
|
||||
name="{{ full_key }}"
|
||||
class="form-select w-full rounded-md border-gray-300 shadow-sm focus:border-blue-500 focus:ring-blue-500 bg-white text-black">
|
||||
{% for option in prop.enum %}
|
||||
<option value="{{ option }}" {% if value == option %}selected{% endif %}>
|
||||
{{ option|replace('_', ' ')|title }}
|
||||
{{ enum_labels.get(option, option|replace('_', ' ')|title) }}
|
||||
</option>
|
||||
{% endfor %}
|
||||
</select>
|
||||
@@ -569,10 +576,14 @@
|
||||
class="block w-20 px-2 py-1 border border-gray-300 rounded text-sm text-center"
|
||||
{% if col_def.get('description') %}title="{{ col_def.get('description') }}"{% endif %}>
|
||||
{% elif col_enum %}
|
||||
{# Labels are opt-in here and the fallback stays the raw
|
||||
value: table columns hold things like ticker symbols,
|
||||
which must not be title-cased behind the user's back. #}
|
||||
{% set col_labels = (col_def.get('x-options') or col_def.get('x_options') or {}).get('labels') or {} %}
|
||||
<select name="{{ full_key }}.{{ item_index }}.{{ col_name }}"
|
||||
class="block w-full px-2 py-1 border border-gray-300 rounded text-sm bg-white">
|
||||
{% for opt in col_enum %}{% if opt is not none %}
|
||||
<option value="{{ opt }}" {% if col_value == opt or (col_value is none and col_def.get('default') == opt) %}selected{% endif %}>{{ opt }}</option>
|
||||
<option value="{{ opt }}" {% if col_value == opt or (col_value is none and col_def.get('default') == opt) %}selected{% endif %}>{{ col_labels.get(opt, opt) }}</option>
|
||||
{% endif %}{% endfor %}
|
||||
</select>
|
||||
{% elif col_xwidget == 'date-picker' %}
|
||||
|
||||