diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index aadebc4d..ecc49eb0 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -4,6 +4,9 @@ on: pull_request: push: branches: [main] + # Manual runs against any branch — useful when a PR's automatic run + # needs a re-run or didn't get created. + workflow_dispatch: # Both jobs only check out the repo and run pytest. permissions: @@ -13,6 +16,12 @@ jobs: plugin-safety: name: Plugin safety harness + unit tests runs-on: ubuntu-latest + env: + # The bundled fixture plugin gives the harness at least one real plugin + # to render, and REQUIRE_PLUGINS turns "discovered zero plugins" into a + # hard failure instead of a silent all-skip green run. + LEDMATRIX_PLUGINS_DIR: test/fixtures/plugins + LEDMATRIX_REQUIRE_PLUGINS: "1" steps: - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 with: @@ -29,12 +38,9 @@ jobs: pip install -r requirements.txt -r requirements-test.txt pip install RGBMatrixEmulator - - name: Run harness + visual rendering tests + - name: Run plugin safety harness run: | - pytest --no-cov \ - test/plugins/test_harness.py \ - test/plugins/test_visual_rendering.py \ - test/plugins/test_plugin_matrix.py + pytest --no-cov test/plugins/ unit-tests: name: Core unit tests @@ -55,35 +61,15 @@ jobs: pip install -r requirements.txt -r requirements-test.txt pip install RGBMatrixEmulator - # Safety net for the shared sports/scroll/style infrastructure. These - # suites existed but were not enrolled in CI, so a refactor of - # src/base_classes or src/common could regress them silently. Enrolled - # explicitly (not `pytest test/`) so known hardware-only suites don't - # break CI; grow this list as more suites are made headless. + # Run the ENTIRE test tree (except test/plugins, which the + # plugin-safety job owns). New test files are enrolled automatically; + # excluding anything requires a visible, commented --ignore here. + # Coverage is measured and enforced only in this step — pytest.ini + # deliberately carries no coverage flags so local runs stay fast. - name: Run core unit suites run: | - pytest --no-cov \ - test/test_skin_system.py \ - test/test_font_manager.py \ - test/test_data_sources.py \ - test/test_api_extractors.py \ - test/test_scroll_helper.py \ - test/test_scroll_helper_continuous.py \ - test/test_adaptive_layout.py \ - test/test_loader_compat_warning.py \ - test/test_sports_base_characterization.py \ - test/test_element_style.py \ - test/test_sports_core_promotions.py \ - test/test_sports_modes_promotions.py \ - test/test_sports_capabilities.py \ - test/test_sports_scroll.py \ - test/test_version_consistency.py \ - test/test_plugin_compatibility_gate.py \ - test/test_install_preserves_existing.py \ - test/test_core_owned_config_keys.py \ - test/test_async_plugin_updates.py \ - test/test_plugin_update_reservation.py \ - test/test_template_targets.py \ - test/test_widget_scripts.py \ - test/test_doc_links.py \ - test/web_interface/test_cache.py + pytest -m "not hardware" test/ \ + --ignore=test/plugins \ + --cov=src --cov=web_interface \ + --cov-report=term \ + --cov-fail-under=45 diff --git a/pytest.ini b/pytest.ini index a13cc6cb..aaffa360 100644 --- a/pytest.ini +++ b/pytest.ini @@ -10,16 +10,13 @@ python_functions = test_* testpaths = test # Output options -# Note: Coverage options require pytest-cov to be installed -# Run: pip install pytest-cov -addopts = +# Coverage is deliberately NOT configured here: a bare local `pytest` should +# be fast and dependency-light. Coverage is measured and enforced in exactly +# one place — the unit-tests job in .github/workflows/test.yml. +addopts = -v --strict-markers --tb=short - --cov=src - --cov-report=term-missing - --cov-report=html - --cov-fail-under=30 # Markers markers = diff --git a/requirements-test.txt b/requirements-test.txt index 2faccf94..60e26951 100644 --- a/requirements-test.txt +++ b/requirements-test.txt @@ -4,4 +4,6 @@ pytest>=9.0.3,<10.0.0 pytest-cov>=4.1.0,<5.0.0 pytest-mock>=3.11.0,<4.0.0 freezegun>=1.2,<2 # deterministic time for golden-image tests +psutil>=6.0.0,<8.0.0 # optional at runtime; installed for tests so the + # /system/status endpoint's real path is exercised mypy>=1.5.0,<2.0.0 # static type checking (also pinned in .pre-commit-config.yaml) diff --git a/src/base_classes/sports/core.py b/src/base_classes/sports/core.py index e43c6e45..3291a1bf 100644 --- a/src/base_classes/sports/core.py +++ b/src/base_classes/sports/core.py @@ -383,10 +383,15 @@ class SportsCore(ABC): ctx = skin_runtime.build_context(self, game, size=size) card = skin.render_vegas_card(ctx, dict(game)) if card is not None: + # A successful render clears accumulated strikes, mirroring + # _render_game — transient failures must not add up across + # the session and disable a working skin. + self._skin_failures = 0 return card ctx = skin_runtime.build_context(self, game, size=size) render = getattr(skin, f"render_{self.SKIN_MODE}") if render(ctx, dict(game)): + self._skin_failures = 0 return ctx.canvas except Exception: # Card failures count toward the same 3-strike session disable diff --git a/src/config_manager.py b/src/config_manager.py index 8a7edbb1..569b295e 100644 --- a/src/config_manager.py +++ b/src/config_manager.py @@ -106,18 +106,13 @@ class ConfigManager: Returns: SaveResult with status and details """ - # Load current secrets to preserve them - secrets_content = {} - if os.path.exists(self.secrets_path): - try: - with open(self.secrets_path, 'r') as f_secrets: - secrets_content = json.load(f_secrets) - except Exception as e: - self.logger.warning(f"Could not load secrets file {self.secrets_path} during save: {e}") - + # Load current secrets to preserve them (raises if unreadable — see + # _load_secrets_for_save) + secrets_content = self._load_secrets_for_save() + # Strip secrets from main config before saving config_to_write = self._strip_secrets_recursive(new_config_data, secrets_content) - + # Use atomic manager to save atomic_mgr = self._get_atomic_manager() result = atomic_mgr.save_config_atomic( @@ -290,19 +285,43 @@ class ConfigManager: result[key] = value return result + def _load_secrets_for_save(self) -> Dict[str, Any]: + """Load config_secrets.json for stripping before a save. + + A missing secrets file is fine (nothing to strip). But a file that + EXISTS and cannot be read or parsed means stripping is impossible — + and the in-memory config being saved has secrets deep-merged into it, + so proceeding would write them into config.json in plaintext. That + was the historical behavior; it is now a hard refusal. The save + raises so the caller (and user) fixes the secrets file instead of + silently leaking its contents into the world-readable main config. + """ + if not os.path.exists(self.secrets_path): + return {} + try: + with open(self.secrets_path, 'r') as f_secrets: + return json.load(f_secrets) + # Only the expected read/parse failures — an unexpected implementation + # error should propagate as itself, not masquerade as a secrets-file + # problem. (JSONDecodeError and UnicodeDecodeError are ValueErrors.) + except (OSError, ValueError, RecursionError) as e: + error_msg = ( + f"Refusing to save config: secrets file {self.secrets_path} exists " + f"but could not be loaded ({e}). Saving without it would write " + f"merged secret values into config.json in plaintext. Fix or " + f"remove the secrets file, then retry." + ) + self.logger.error("[Config] %s", error_msg, exc_info=True) + raise ConfigError(error_msg, config_path=self.secrets_path) from e + def save_config(self, new_config_data: Dict[str, Any]) -> None: - """Save configuration to the main JSON file, stripping out secrets.""" - secrets_content = {} - if os.path.exists(self.secrets_path): - try: - with open(self.secrets_path, 'r') as f_secrets: - secrets_content = json.load(f_secrets) - except Exception as e: - self.logger.warning(f"Could not load secrets file {self.secrets_path} during save: {e}") - # Continue without stripping if secrets can't be loaded, or handle as critical error - # For now, we'll proceed cautiously and save the full new_config_data if secrets are unreadable - # to prevent accidental data loss if the secrets file is temporarily corrupt. - # A more robust approach might be to fail the save or use a cached version of secrets. + """Save configuration to the main JSON file, stripping out secrets. + + Raises ConfigError when the secrets file exists but cannot be loaded, + because stripping would be impossible and secrets would leak into + config.json. + """ + secrets_content = self._load_secrets_for_save() config_to_write = self._strip_secrets_recursive(new_config_data, secrets_content) diff --git a/src/plugin_system/compatibility.py b/src/plugin_system/compatibility.py index 784a6b6a..9b676cd2 100644 --- a/src/plugin_system/compatibility.py +++ b/src/plugin_system/compatibility.py @@ -180,6 +180,45 @@ def declared_min_version(manifest: Dict[str, Any]) -> Optional[str]: return None +def is_update_available(installed_version: str, latest_version: str) -> bool: + """Return True when the registry's ``latest_version`` is strictly newer + than the installed version. + + THE shared comparator for "should this plugin be updated?" — used by both + the web UI's update badge (`api_v3._is_plugin_update_available`) and the + store's `update_plugin` reinstall decision, so the two can never disagree. + + Uses PEP 440-aware comparison (``packaging``), which also normalizes + equivalent spellings: ``v1.2.0`` == ``1.2.0`` and ``1.2`` == ``1.2.0``, so + cosmetic differences never trigger a reinstall — and a locally modified + plugin whose version is *ahead* of the registry is never "updated" + (downgraded). If either version string can't be parsed the mismatch is + surfaced (True) so the user can reconcile, rather than silently hiding a + potential update. + """ + if not installed_version or not latest_version: + return False + if not isinstance(installed_version, str) or not isinstance(latest_version, str): + # A malformed manifest/registry can carry a number (1.2) or worse; + # packaging would raise TypeError. Surface the mismatch instead. + return True + if installed_version == latest_version: + return False + try: + from packaging.version import parse as _parse_version, InvalidVersion + except ImportError: + # packaging is a core dependency, but if it's somehow unavailable we + # can't compare semantically — surface the mismatch we already know + # exists (the two strings differ). + return True + try: + return _parse_version(latest_version) > _parse_version(installed_version) + except InvalidVersion: + # Unparseable version string: we can't tell direction, so surface the + # mismatch rather than silently hiding a potential update. + return True + + def check(manifest: Dict[str, Any], core_version: str) -> Tuple[bool, Optional[str]]: """Return ``(compatible, reason)``. diff --git a/src/plugin_system/store_manager.py b/src/plugin_system/store_manager.py index e13b3e5b..8e27ab8f 100644 --- a/src/plugin_system/store_manager.py +++ b/src/plugin_system/store_manager.py @@ -2969,7 +2969,10 @@ class PluginStoreManager: remote_branch = plugin_info_remote.get('branch') or plugin_info_remote.get('default_branch') # Compare local manifest version against registry latest_version - # to avoid unnecessary reinstalls for monorepo plugins + # to avoid unnecessary reinstalls for monorepo plugins. Uses the + # same semantic comparator as the web UI's update badge, so + # equivalent spellings ("v1.2.0" vs "1.2.0") never trigger a + # reinstall and a locally-ahead version is never downgraded. try: local_manifest_path = plugin_path / "manifest.json" if local_manifest_path.exists(): @@ -2977,8 +2980,16 @@ class PluginStoreManager: local_manifest = json.load(f) local_version = local_manifest.get('version', '') remote_version = plugin_info_remote.get('latest_version', '') - if local_version and remote_version and local_version == remote_version: - self.logger.info(f"Plugin {plugin_id} already at latest version {local_version}") + from src.plugin_system.compatibility import is_update_available + # No truthiness gate: the shared comparator already treats + # a missing version on either side as "no update", and the + # store must agree with the UI badge in that case too. A + # missing manifest (not just a missing version field) + # still falls through to the reinstall recovery path. + if not is_update_available(local_version, remote_version): + self.logger.info( + f"Plugin {plugin_id} already at latest version " + f"(installed {local_version}, registry {remote_version})") return True except Exception as e: self.logger.debug(f"Could not compare versions for {plugin_id}: {e}") diff --git a/test/fixtures/plugins/ci-fixture-plugin/config_schema.json b/test/fixtures/plugins/ci-fixture-plugin/config_schema.json new file mode 100644 index 00000000..9410d90a --- /dev/null +++ b/test/fixtures/plugins/ci-fixture-plugin/config_schema.json @@ -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." + } + } +} diff --git a/test/fixtures/plugins/ci-fixture-plugin/manager.py b/test/fixtures/plugins/ci-fixture-plugin/manager.py new file mode 100644 index 00000000..02805f60 --- /dev/null +++ b/test/fixtures/plugins/ci-fixture-plugin/manager.py @@ -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() diff --git a/test/fixtures/plugins/ci-fixture-plugin/manifest.json b/test/fixtures/plugins/ci-fixture-plugin/manifest.json new file mode 100644 index 00000000..40d188cb --- /dev/null +++ b/test/fixtures/plugins/ci-fixture-plugin/manifest.json @@ -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"] +} diff --git a/test/fixtures/plugins/ci-fixture-plugin/requirements.txt b/test/fixtures/plugins/ci-fixture-plugin/requirements.txt new file mode 100644 index 00000000..7fefb0f5 --- /dev/null +++ b/test/fixtures/plugins/ci-fixture-plugin/requirements.txt @@ -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. diff --git a/test/fixtures/plugins/ci-fixture-plugin/test/golden/128x32/ci-fixture.png b/test/fixtures/plugins/ci-fixture-plugin/test/golden/128x32/ci-fixture.png new file mode 100644 index 00000000..32608a8c Binary files /dev/null and b/test/fixtures/plugins/ci-fixture-plugin/test/golden/128x32/ci-fixture.png differ diff --git a/test/fixtures/plugins/ci-fixture-plugin/test/golden/128x64/ci-fixture.png b/test/fixtures/plugins/ci-fixture-plugin/test/golden/128x64/ci-fixture.png new file mode 100644 index 00000000..de738059 Binary files /dev/null and b/test/fixtures/plugins/ci-fixture-plugin/test/golden/128x64/ci-fixture.png differ diff --git a/test/fixtures/plugins/ci-fixture-plugin/test/golden/128x96/ci-fixture.png b/test/fixtures/plugins/ci-fixture-plugin/test/golden/128x96/ci-fixture.png new file mode 100644 index 00000000..3107e6da Binary files /dev/null and b/test/fixtures/plugins/ci-fixture-plugin/test/golden/128x96/ci-fixture.png differ diff --git a/test/fixtures/plugins/ci-fixture-plugin/test/golden/256x128/ci-fixture.png b/test/fixtures/plugins/ci-fixture-plugin/test/golden/256x128/ci-fixture.png new file mode 100644 index 00000000..dd43ab76 Binary files /dev/null and b/test/fixtures/plugins/ci-fixture-plugin/test/golden/256x128/ci-fixture.png differ diff --git a/test/fixtures/plugins/ci-fixture-plugin/test/golden/256x32/ci-fixture.png b/test/fixtures/plugins/ci-fixture-plugin/test/golden/256x32/ci-fixture.png new file mode 100644 index 00000000..01638b37 Binary files /dev/null and b/test/fixtures/plugins/ci-fixture-plugin/test/golden/256x32/ci-fixture.png differ diff --git a/test/fixtures/plugins/ci-fixture-plugin/test/golden/64x32/ci-fixture.png b/test/fixtures/plugins/ci-fixture-plugin/test/golden/64x32/ci-fixture.png new file mode 100644 index 00000000..9312505b Binary files /dev/null and b/test/fixtures/plugins/ci-fixture-plugin/test/golden/64x32/ci-fixture.png differ diff --git a/test/fixtures/plugins/ci-fixture-plugin/test/golden/64x64/ci-fixture.png b/test/fixtures/plugins/ci-fixture-plugin/test/golden/64x64/ci-fixture.png new file mode 100644 index 00000000..7095ed8e Binary files /dev/null and b/test/fixtures/plugins/ci-fixture-plugin/test/golden/64x64/ci-fixture.png differ diff --git a/test/fixtures/plugins/ci-fixture-plugin/test/golden/96x48/ci-fixture.png b/test/fixtures/plugins/ci-fixture-plugin/test/golden/96x48/ci-fixture.png new file mode 100644 index 00000000..239561e4 Binary files /dev/null and b/test/fixtures/plugins/ci-fixture-plugin/test/golden/96x48/ci-fixture.png differ diff --git a/test/plugins/conftest.py b/test/plugins/conftest.py index 27bb6718..cff6dd42 100644 --- a/test/plugins/conftest.py +++ b/test/plugins/conftest.py @@ -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' diff --git a/test/plugins/test_basketball_scoreboard.py b/test/plugins/test_basketball_scoreboard.py index 0fbd8b79..b93fb89f 100644 --- a/test/plugins/test_basketball_scoreboard.py +++ b/test/plugins/test_basketball_scoreboard.py @@ -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 diff --git a/test/plugins/test_calendar.py b/test/plugins/test_calendar.py index 18528fdb..4d874940 100644 --- a/test/plugins/test_calendar.py +++ b/test/plugins/test_calendar.py @@ -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 diff --git a/test/plugins/test_clock_simple.py b/test/plugins/test_clock_simple.py index 507feec9..9b25a59e 100644 --- a/test/plugins/test_clock_simple.py +++ b/test/plugins/test_clock_simple.py @@ -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 diff --git a/test/plugins/test_odds_ticker.py b/test/plugins/test_odds_ticker.py index 90209548..231ed7de 100644 --- a/test/plugins/test_odds_ticker.py +++ b/test/plugins/test_odds_ticker.py @@ -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 diff --git a/test/plugins/test_soccer_scoreboard.py b/test/plugins/test_soccer_scoreboard.py index 36212bfd..d5526fb8 100644 --- a/test/plugins/test_soccer_scoreboard.py +++ b/test/plugins/test_soccer_scoreboard.py @@ -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 diff --git a/test/plugins/test_text_display.py b/test/plugins/test_text_display.py index a43815ea..34adfa49 100644 --- a/test/plugins/test_text_display.py +++ b/test/plugins/test_text_display.py @@ -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 diff --git a/test/test_base_plugin_duration.py b/test/test_base_plugin_duration.py new file mode 100644 index 00000000..4039e1fc --- /dev/null +++ b/test/test_base_plugin_duration.py @@ -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 diff --git a/test/test_compatibility.py b/test/test_compatibility.py new file mode 100644 index 00000000..34f88869 --- /dev/null +++ b/test/test_compatibility.py @@ -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) diff --git a/test/test_config_manager_secrets.py b/test/test_config_manager_secrets.py new file mode 100644 index 00000000..94c7a6a2 --- /dev/null +++ b/test/test_config_manager_secrets.py @@ -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 diff --git a/test/test_discovery_path_contract.py b/test/test_discovery_path_contract.py new file mode 100644 index 00000000..7c60b88c --- /dev/null +++ b/test/test_discovery_path_contract.py @@ -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 diff --git a/test/test_display_controller.py b/test/test_display_controller.py index 782ce69e..8f89c31f 100644 --- a/test/test_display_controller.py +++ b/test/test_display_controller.py @@ -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.""" diff --git a/test/test_display_controller_vegas_tick.py b/test/test_display_controller_vegas_tick.py index 643359f3..538e9272 100644 --- a/test/test_display_controller_vegas_tick.py +++ b/test/test_display_controller_vegas_tick.py @@ -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 diff --git a/test/test_display_manager.py b/test/test_display_manager.py index 63c49114..9c8b1468 100644 --- a/test/test_display_manager.py +++ b/test/test_display_manager.py @@ -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.""" diff --git a/test/test_font_manager.py b/test/test_font_manager.py index 11240724..16b093a3 100644 --- a/test/test_font_manager.py +++ b/test/test_font_manager.py @@ -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 diff --git a/test/test_schema_manager_merge.py b/test/test_schema_manager_merge.py new file mode 100644 index 00000000..64a82c9e --- /dev/null +++ b/test/test_schema_manager_merge.py @@ -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} diff --git a/test/test_skin_system.py b/test/test_skin_system.py index eb94b8fb..1a7138c2 100644 --- a/test/test_skin_system.py +++ b/test/test_skin_system.py @@ -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 diff --git a/test/test_version_comparison_consistency.py b/test/test_version_comparison_consistency.py new file mode 100644 index 00000000..9a27b59c --- /dev/null +++ b/test/test_version_comparison_consistency.py @@ -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})" diff --git a/test/test_web_api.py b/test/test_web_api.py index 418bb68c..8cadf922 100644 --- a/test/test_web_api.py +++ b/test/test_web_api.py @@ -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 diff --git a/test/web_interface/test_api_v3_helpers.py b/test/web_interface/test_api_v3_helpers.py new file mode 100644 index 00000000..dc21864b --- /dev/null +++ b/test/web_interface/test_api_v3_helpers.py @@ -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} diff --git a/test/web_interface/test_secret_helpers.py b/test/web_interface/test_secret_helpers.py new file mode 100644 index 00000000..5f33e899 --- /dev/null +++ b/test/web_interface/test_secret_helpers.py @@ -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} diff --git a/test/web_interface/test_secret_separation_parity.py b/test/web_interface/test_secret_separation_parity.py new file mode 100644 index 00000000..65c5576e --- /dev/null +++ b/test/web_interface/test_secret_separation_parity.py @@ -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"}]} diff --git a/test/web_interface/test_state_reconciliation.py b/test/web_interface/test_state_reconciliation.py index 19f23833..7d01b886 100644 --- a/test/web_interface/test_state_reconciliation.py +++ b/test/web_interface/test_state_reconciliation.py @@ -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, diff --git a/web_interface/blueprints/api_v3.py b/web_interface/blueprints/api_v3.py index aea7d592..95f7ddb1 100644 --- a/web_interface/blueprints/api_v3.py +++ b/web_interface/blueprints/api_v3.py @@ -124,28 +124,13 @@ def _is_plugin_update_available(installed_version: str, latest_version: str) -> """Return True when the registry's ``latest_version`` is strictly newer than the installed version. - Uses PEP 440 / semver-aware comparison so a locally modified plugin whose - version is *ahead* of the published registry is not flagged as needing an - update. If either version string can't be parsed, falls back to a plain - inequality check (any difference is surfaced so the user can reconcile). + Thin alias for the shared comparator in + `src.plugin_system.compatibility.is_update_available` — the store's + `update_plugin` uses the same function, so the UI badge and the actual + reinstall decision can never disagree. """ - if not installed_version or not latest_version: - return False - if installed_version == latest_version: - return False - try: - from packaging.version import parse as _parse_version, InvalidVersion - except ImportError: - # packaging is a core dependency, but if it's somehow unavailable we - # can't compare semantically — surface the mismatch we already know - # exists (the two strings differ). - return True - try: - return _parse_version(latest_version) > _parse_version(installed_version) - except InvalidVersion: - # Unparseable version string: we can't tell direction, so surface the - # mismatch rather than silently hiding a potential update. - return True + from src.plugin_system.compatibility import is_update_available + return is_update_available(installed_version, latest_version) def _ensure_cache_manager(): """Ensure cache manager is initialized."""