Test suite overhaul + fixes for the three bugs it uncovered (#441)
* ci: run the whole test tree and make the plugin-safety job assert something real The unit-tests CI job ran an explicit 24-file allowlist that had rotted: 63 of 90 test files (display, vegas, store manager, web API, web_interface) never ran on a PR. The job now runs all of test/ (minus test/plugins, which the plugin-safety job owns) so new test files are enrolled by default and any exclusion needs a visible, commented --ignore. The plugin-safety job was a green no-op: plugins/ is empty in CI, so every test skipped with 'Manifest not found'. It now renders a bundled deterministic fixture plugin (test/fixtures/plugins/ci-fixture-plugin, golden images included for all 8 default sizes) via LEDMATRIX_PLUGINS_DIR, and sets LEDMATRIX_REQUIRE_PLUGINS=1 so discovering zero plugins fails loudly instead of skipping green. The per-plugin suites document that they target dev machines with real plugins installed. Coverage is now measured and enforced in exactly one place — the CI unit-tests step (--cov=src --cov=web_interface --cov-fail-under=45, from a measured 47% baseline). pytest.ini previously declared --cov-fail-under=30 but CI always passed --no-cov, so the gate had never run anywhere; local pytest is now coverage-free and fast. Enabling the 63 unenrolled files surfaced three cases of test rot, fixed here: test_display_controller_vegas_tick.py could not collect without the hardware rgbmatrix module (now uses the emulator convention), the state-reconciliation unrecoverable-cache tests broke when production added the is_plugin_uninstalled tombstone check (bare Mock returned truthy), and test_get_system_status assumed the optional psutil dependency (now installed via requirements-test.txt and guarded by importorskip). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh * test: replace can't-fail tests with real assertions test_font_manager.py was 5 of 6 tests shaped as 'try: call(); assert True / except: assert True' — running in CI while unable to fail on any regression. Rewritten against the real FontManager API and the bundled assets/fonts: returned font types, cache-hit identity, distinct entries per size, default-font fallback for unknown families and corrupt files (recorded in failed_loads), BDF native-size reading, text measurement, and cache lifecycle. test_display_manager.py's test_draw_text ended in 'assert True'; it now renders onto a known-black canvas and asserts pixels were actually lit — which required un-breaking the fixture's freetype MagicMock so draw_text's isinstance check doesn't silently swallow the draw. test_display_controller.py carried a permanently-skipped test whose skip reason already declared it redundant; deleted. Both display test files now set EMULATOR=true before importing display_manager (the same convention as test_display_dirty_tracking.py) so they collect standalone instead of depending on which test module imports display_manager first. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh * test: cover the untested fragile logic (compatibility gate, secrets, config merges, durations, skin cards) New unit tests for pure or filesystem-only logic that previously had zero direct coverage: - test_compatibility.py: the semver install gate (parse_semver suffix handling, every range operator, TRUSTWORTHY_FLOOR behavior for cores reporting untrustworthy versions, 'more restrictive wins', and the malformed-manifest shapes that used to raise). - test/web_interface/test_secret_helpers.py: the canonical x-secret helpers — find/separate/mask/remove, array-item secrets, no input mutation, and a separate->recombine round-trip. - test/web_interface/test_api_v3_helpers.py: the module-level helpers behind the plugin config save endpoint (_is_plugin_update_available, _coerce_to_bool including the int==1 quirk, deep_merge including its shared-subtree shallowness, _parse_form_value, dotted-key-aware _get_schema_property/_set_nested_value). - test_base_plugin_duration.py: get_display_duration's full coercion ladder (instance attr -> config -> 15.0), including the bool-is-int quirk where display_duration=True means one second. - test_config_manager_secrets.py: the secrets round-trip — deep-merge on load, strip on save, group pruning, the load fast path — and two characterized sharp edges marked SUSPECTED BUG: an unreadable secrets file at save time writes secrets into config.json in plaintext, and a same-mtime-same-size content swap is served stale. - test_schema_manager_merge.py: merge_with_defaults branch behavior (None replacement vs falsey preservation, dict-vs-scalar mismatches, arrays replaced wholesale, defaults never mutated). - test_skin_system.py (extended): render_skin_card shares _render_game's 3-strike counter but never resets it on success — the asymmetry is pinned in both directions, along with card fallthrough and the disable interaction between the two paths. Suspected bugs are characterized, not fixed — each carries a comment so a future behavior change is deliberate rather than accidental. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh * test: add drift guards for cross-file contracts Three guard suites that pin contracts spanning multiple files, where one side changing unilaterally breaks the other silently: - test_version_comparison_consistency.py: the repo's four version comparators (compatibility.parse_semver, api_v3's packaging-based _is_plugin_update_available, store_manager update_plugin's raw string equality, skin_runtime._major) answer differently on the same inputs. A table pins each one's verdict; update_plugin is driven through its real code path to show the SUSPECTED BUGs: 'v1.2.0' vs '1.2.0' triggers a full reinstall the UI calls unnecessary, and a locally-ahead plugin gets downgraded. A pairwise-ordering check keeps parse_semver agreeing with packaging on plain X.Y.Z. - test/web_interface/test_secret_separation_parity.py: api_v3.py carries three inline copies of find_secret_fields/separate_secrets that lack the canonical module's array-item support. The copy count is asserted exact (it may only go down; new copies must import src/web_interface/secret_helpers), the missing-array-support gap is asserted so it can't grow silently, and the canonical behavior that migration will adopt is documented executably. - test_discovery_path_contract.py: the three 'where is plugin X' resolvers (PluginManager discovery, StoreManager._find_plugin_path, SchemaManager.get_schema_path) agree on the configured directory, and their divergent fallback chains are characterized. Also pins the .standalone-backup- naming contract shared by store rollback and discovery, and _resolve_skin_target's path-traversal rejection. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh * test: address review feedback — fixture lifecycle, test names, ClassVar - ci-fixture-plugin: call display_manager.clear() before rendering (per plugin guidelines — the fixture should model a well-behaved plugin), add a class docstring, and document why Pillow is deliberately not pinned in its requirements.txt (core dependency; harness installs nothing). - Rename two tests whose names contradicted their assertions: test_unparseable_core_version_is_compatible -> test_unparseable_core_with_high_floor_is_blocked, and test_unreadable_secrets_file... -> test_corrupt_secrets_file... - Annotate TestGetSchemaProperty.SCHEMA as ClassVar (RUF012). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh * ci: allow manual test.yml runs via workflow_dispatch Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh * fix: unify version comparison, refuse secret-leaking saves, reset skin strikes on card success Fixes the three suspected bugs this PR's characterization tests pinned, flipping those tests to assert the corrected behavior: - plugins/store: ONE shared update comparator. New compatibility.is_update_available() (PEP 440 via packaging) is now used by both the web UI's update badge (api_v3._is_plugin_update_available is a thin alias) and store_manager.update_plugin's reinstall decision. Previously update_plugin used raw string equality: 'v1.2.0' vs '1.2.0' triggered a full reinstall the UI called unnecessary, and a locally- ahead plugin (2.0.0 installed, registry 1.9.0) was silently DOWNGRADED. Now equivalent spellings skip the reinstall and locally-ahead versions are never downgraded; unparseable versions still reconcile by reinstalling from the registry. - config: save_config and save_config_atomic now refuse (ConfigError) when config_secrets.json exists but cannot be loaded. Both previously proceeded without stripping, writing the merged secrets into config.json in plaintext. The shared _load_secrets_for_save() helper raises with an actionable message instead; a missing secrets file is still fine (nothing to strip), and _migrate_config's catch-all keeps boot resilient. - skins: render_skin_card resets _skin_failures on both success paths (vegas card returned, or mode renderer handled), mirroring _render_game. Transient card failures no longer accumulate across a session until they permanently disable a working skin. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh * fix: harden shared comparator edges from review - is_update_available: reject truthy non-string versions (a malformed manifest can carry a number; packaging raises TypeError on those) by surfacing the mismatch instead of raising. - store_manager.update_plugin: drop the truthiness gate around the comparator so a missing version on either side follows the shared 'no update' verdict, keeping the store consistent with the UI badge; a missing manifest still uses the reinstall recovery path. - config_manager._load_secrets_for_save: catch only expected read/parse failures (OSError/ValueError/RecursionError) so implementation bugs propagate as themselves, and log with traceback. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh --------- Co-authored-by: Claude <noreply@anthropic.com>
@@ -0,0 +1,31 @@
|
||||
{
|
||||
"$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."
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
"""
|
||||
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()
|
||||
@@ -0,0 +1,13 @@
|
||||
{
|
||||
"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"]
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
# 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.
|
||||
|
After Width: | Height: | Size: 359 B |
|
After Width: | Height: | Size: 586 B |
|
After Width: | Height: | Size: 849 B |
|
After Width: | Height: | Size: 1.2 KiB |
|
After Width: | Height: | Size: 395 B |
|
After Width: | Height: | Size: 319 B |
|
After Width: | Height: | Size: 466 B |
|
After Width: | Height: | Size: 454 B |
@@ -23,9 +23,17 @@ os.environ['EMULATOR'] = 'true'
|
||||
def plugins_dir() -> Path:
|
||||
"""Get the plugins directory path.
|
||||
|
||||
Checks plugins/ first, then falls back to plugin-repos/
|
||||
for monorepo development environments.
|
||||
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.
|
||||
"""
|
||||
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,5 +1,10 @@
|
||||
"""
|
||||
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,5 +1,10 @@
|
||||
"""
|
||||
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,5 +1,10 @@
|
||||
"""
|
||||
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,5 +1,10 @@
|
||||
"""
|
||||
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,5 +1,10 @@
|
||||
"""
|
||||
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,5 +1,10 @@
|
||||
"""
|
||||
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
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
"""
|
||||
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
|
||||
@@ -0,0 +1,271 @@
|
||||
"""
|
||||
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)
|
||||
@@ -0,0 +1,179 @@
|
||||
"""
|
||||
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
|
||||
@@ -0,0 +1,167 @@
|
||||
"""
|
||||
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,13 +15,6 @@ 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,9 +11,18 @@ 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,6 +1,15 @@
|
||||
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
|
||||
@@ -77,18 +86,26 @@ class TestDisplayManagerDrawing:
|
||||
assert dm.matrix.Clear.called
|
||||
|
||||
def test_draw_text(self, test_config, mock_rgb_matrix):
|
||||
"""Test text drawing."""
|
||||
"""Text drawn through draw_text must actually light pixels."""
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
import src.display_manager as dm_mod
|
||||
with patch.dict('os.environ', {'EMULATOR': 'false'}):
|
||||
dm = DisplayManager(test_config)
|
||||
|
||||
# Mock font
|
||||
font = MagicMock()
|
||||
|
||||
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
|
||||
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.draw_text("Test", 0, 0, font=ImageFont.load_default())
|
||||
|
||||
assert dm.image.convert("L").getbbox() is not None, \
|
||||
"draw_text lit no pixels"
|
||||
|
||||
def test_draw_image(self, test_config, mock_rgb_matrix):
|
||||
"""Test image drawing."""
|
||||
|
||||
@@ -1,82 +1,126 @@
|
||||
"""
|
||||
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 unittest.mock import patch
|
||||
from PIL import ImageFont
|
||||
|
||||
from src.font_manager import FontManager
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_freetype():
|
||||
"""Mock freetype module."""
|
||||
with patch('src.font_manager.freetype') as mock_freetype:
|
||||
yield mock_freetype
|
||||
def fm():
|
||||
"""A FontManager over the real assets/fonts catalog."""
|
||||
return FontManager({})
|
||||
|
||||
class TestFontManager:
|
||||
"""Test FontManager functionality."""
|
||||
|
||||
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_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)
|
||||
|
||||
# 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_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)
|
||||
|
||||
# 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_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)
|
||||
|
||||
# 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_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)
|
||||
|
||||
# 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')
|
||||
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_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)
|
||||
|
||||
|
||||
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]
|
||||
|
||||
def test_bdf_family_returns_freetype_face(self, fm):
|
||||
font = fm.get_font("five_by_seven", 7)
|
||||
assert isinstance(font, freetype.Face)
|
||||
|
||||
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
|
||||
|
||||
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_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
|
||||
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
"""
|
||||
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,3 +450,122 @@ 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
|
||||
|
||||
@@ -0,0 +1,175 @@
|
||||
"""
|
||||
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,6 +393,9 @@ 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
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
"""
|
||||
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}
|
||||
@@ -0,0 +1,241 @@
|
||||
"""
|
||||
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}
|
||||
@@ -0,0 +1,118 @@
|
||||
"""
|
||||
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,6 +367,10 @@ 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,
|
||||
|
||||