test: add drift guards for cross-file contracts

Three guard suites that pin contracts spanning multiple files, where one
side changing unilaterally breaks the other silently:

- test_version_comparison_consistency.py: the repo's four version
  comparators (compatibility.parse_semver, api_v3's packaging-based
  _is_plugin_update_available, store_manager update_plugin's raw string
  equality, skin_runtime._major) answer differently on the same inputs.
  A table pins each one's verdict; update_plugin is driven through its
  real code path to show the SUSPECTED BUGs: 'v1.2.0' vs '1.2.0'
  triggers a full reinstall the UI calls unnecessary, and a locally-ahead
  plugin gets downgraded. A pairwise-ordering check keeps parse_semver
  agreeing with packaging on plain X.Y.Z.
- test/web_interface/test_secret_separation_parity.py: api_v3.py carries
  three inline copies of find_secret_fields/separate_secrets that lack
  the canonical module's array-item support. The copy count is asserted
  exact (it may only go down; new copies must import
  src/web_interface/secret_helpers), the missing-array-support gap is
  asserted so it can't grow silently, and the canonical behavior that
  migration will adopt is documented executably.
- test_discovery_path_contract.py: the three 'where is plugin X'
  resolvers (PluginManager discovery, StoreManager._find_plugin_path,
  SchemaManager.get_schema_path) agree on the configured directory, and
  their divergent fallback chains are characterized. Also pins the
  .standalone-backup- naming contract shared by store rollback and
  discovery, and _resolve_skin_target's path-traversal rejection.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh
This commit is contained in:
Claude
2026-08-06 22:31:03 +00:00
parent dfba88b599
commit b8b0c8e2e2
3 changed files with 433 additions and 0 deletions
+167
View File
@@ -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
+148
View File
@@ -0,0 +1,148 @@
"""
Drift guard: the repo has FOUR version-comparison implementations, and they
do not agree. This file pins each one's answer on the same inputs so any
future change to one of them (or a fifth copy appearing) surfaces here.
The four:
1. src/plugin_system/compatibility.py — parse_semver / tuple comparison
(install gate).
2. web_interface/blueprints/api_v3.py — _is_plugin_update_available, uses
packaging.version (update badge in the UI).
3. src/plugin_system/store_manager.py — update_plugin's raw STRING EQUALITY
for monorepo plugins ("local_version == remote_version").
4. src/skin_system/skin_runtime.py — _major, int(major) gate for the
skin API.
SUSPECTED BUG (characterized here, not fixed): #3 disagrees with #2. For
"v1.2.0" vs "1.2.0" the UI says "no update available" while update_plugin
performs a full reinstall; for a locally-ahead plugin ("2.0.0" installed,
registry "1.9.0") the UI says no update but update_plugin DOWNGRADES via
reinstall. Unifying on one comparator is tracked follow-up work; when that
lands, the expectations in TestStoreManagerStringEquality flip and this
file is the reminder to update them deliberately.
"""
import json
from unittest.mock import patch
import pytest
from packaging.version import parse as pkg_parse
from src.plugin_system.compatibility import 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) pairs and what each comparator concludes.
CASES = [
# pair parse_semver equal? api_v3 update? store equal-string?
(("1.2.0", "1.2.0"), True, False, True),
(("v1.2.0", "1.2.0"), True, False, False),
(("1.2", "1.2.0"), True, False, False),
(("1.2.0", "1.2.0-rc1"), True, False, False),
(("1.2.0", "1.3.0"), False, True, False),
(("2.0.0", "1.9.0"), False, False, False),
]
class TestComparatorMatrix:
@pytest.mark.parametrize("pair,semver_equal,api_update,store_equal", CASES)
def test_parse_semver_equality(self, pair, semver_equal, api_update, store_equal):
a, b = pair
assert (parse_semver(a) == parse_semver(b)) is semver_equal
@pytest.mark.parametrize("pair,semver_equal,api_update,store_equal", CASES)
def test_api_v3_update_available(self, pair, semver_equal, api_update, store_equal):
installed, latest = pair
assert _is_plugin_update_available(installed, latest) is api_update
@pytest.mark.parametrize("pair,semver_equal,api_update,store_equal", CASES)
def test_store_manager_string_equality(self, pair, semver_equal, api_update, store_equal):
# The literal comparison update_plugin performs at its
# "already at latest version" check.
a, b = pair
assert (a == b) is store_equal
class TestStoreManagerStringEquality:
"""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_triggers_reinstall_despite_semantic_equality(self, tmp_path):
# SUSPECTED BUG: packaging (and api_v3) treat these as equal; the
# string comparison does not, so the user gets a full reinstall.
store, info = self._store(tmp_path, "v1.2.0", "1.2.0")
result, reinstall = self._run_update(store, info)
reinstall.assert_called_once()
assert result is True
def test_locally_ahead_version_triggers_downgrade_reinstall(self, tmp_path):
# SUSPECTED BUG: a plugin ahead of the registry (local dev build) is
# "updated" — i.e. downgraded — because inequality is the only test.
store, info = self._store(tmp_path, "2.0.0", "1.9.0")
result, reinstall = self._run_update(store, info)
reinstall.assert_called_once()
assert result is True
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})"
@@ -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"}]}