Compare commits

..
Author SHA1 Message Date
Claude 06766b408f fix(backup): address CodeRabbit and CodeQL findings on PR #439
- first_time_install.sh: verify chown/chmod succeed and the final
  owner/group/mode on config_secrets.json before reporting success;
  exit with a clear error otherwise instead of swallowing failures.
- api_v3.py: replace the predictable .writetest probe with an
  exclusive NamedTemporaryFile to avoid a race with concurrent
  resolvers; log the preferred/fallback export path and OSError when
  falling back to the reinstall-deleted directory.
- api_v3.py: mark a restore as failed when plugin reinstalls fail,
  even if file restoration itself succeeded, so the endpoint no longer
  reports HTTP 200 success on a partial restore.
- api_v3.py: stringify plugin IDs before joining them into the error
  message so a malformed backup's non-string plugin_id can't raise a
  TypeError and mask the detailed response.
- backup_manager.py / api_v3.py: stop putting raw exception text (originating
  from a user-controlled backup file) into restore results returned to
  the client; log full details server-side instead. Addresses the
  CodeQL "stack trace information exposure" alert.
- test coverage: add a test for get_plugin_info() resolving a
  manifest id, and assert the disabled restore_wifi path also skips
  and omits ytm_auth.json.
2026-08-06 22:25:37 +00:00
ChuckBuildsandClaude Opus 5 7d83ca742a fix(backup): restore must not depend on owning the file it replaces
Follow-up from testing the previous commit on real hardware, where the
secrets fix turned out to be both too narrow and slightly wrong.

Too narrow: config.json, wifi_config.json and ytm_auth.json are installed
root-owned and group-readable exactly like the secrets file, so all four
were unrestorable by the web service, not just one. `shutil.copy2` opens
the destination for writing, which needs permission on the *existing
file*; the web user could create files in that directory all day and
still not replace them.

Slightly wrong: the previous commit loosened the secrets file to
group-writable. That was treating the symptom. The real error was
deciding ownership from `ledmatrix.service` -- the display service, which
runs as root and only ever *reads* secrets -- when the account that
*writes* them is the web interface, which deliberately does not run as
root. Ownership now follows the web service's user and the mode stays
640.

`_copy_file` writes a temporary file alongside the target and renames
over it. That needs only directory permission, so a restore no longer
cares who owns the destination, and it is atomic: a crash mid-restore can
no longer leave a half-written config. The destination's mode is carried
across so restoring secrets does not widen them to the umask, and its
owner is carried across too when the OS allows it -- only root can hand a
file to another user, so a restore run by the web service keeps its own
ownership rather than pretending to preserve root's.

Verified on a device with all four config files set root-owned 640 and
unwritable by the web user: before, every one failed with EACCES; after,
the restore reports success with no errors and all four sections
restored, mode still 640, root still able to read them and the web
service still able to write them.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Udr6MfaFLUPhX5Fgo67Jf5
2026-08-06 15:33:13 -04:00
ChuckBuildsandClaude Opus 5 d44e31ac9d ci: run the backup suites
test_backup_manager.py existed but was never enrolled, so the tests that
should have guarded backup and restore have not run on a pull request.
That is part of why the restore bugs in the previous commit reached a
device: the suite was there, it just was not watching. Adds it alongside
the new registry-resolution tests.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Udr6MfaFLUPhX5Fgo67Jf5
2026-08-06 15:33:13 -04:00
ChuckBuildsandClaude Opus 5 5cf92f29b1 fix(backup): make a restored device match the one that was backed up
Found by wiping a working device and reinstalling from scratch. Every
problem here is invisible until you actually do that, which is why a
green test suite and eleven hours of uptime had not surfaced any of them.

**Four enabled plugins vanished on restore.** Weather, stocks, music and
leaderboard have a registry `id` that differs from the `id` in their own
manifest: the registry calls them `weather`, everything else calls them
`ledmatrix-weather`. Installation already prefers the manifest id for the
directory name and warns when the two disagree, so on disk, in
config.json and in a backup they are `ledmatrix-weather` -- but nothing
resolved that in reverse. Restore asked the store for `ledmatrix-weather`
and got "Plugin not found in registry", four times, and the device came
back missing four plugins the user had enabled.

Registry lookup now falls back to matching `plugin_path`, which already
records `plugins/ledmatrix-weather`. Renaming the published ids would
have orphaned `plugin_state.json` entries keyed on the old ones. Exact id
still wins, so a path that collides with another entry's id cannot
shadow it. Against the live registry and a real 28-plugin install this
takes unresolvable directories from five to one -- the one being
starlark-apps, which is genuinely not in the registry.

**Secrets could not be restored at all.** A fresh install left
config_secrets.json group-readable but not group-writable, and the web
interface -- which is what performs a restore -- does not necessarily run
as the owner. Every other file in the backup restored; secrets failed
with EACCES. Now group-writable, so the account running the web UI can
put them back.

**A partial restore reported "Restore had errors" and nothing else.**
That is the same message whether the whole thing failed or it quietly
dropped your API keys. It now names what was restored, what failed, and
which plugins were not reinstalled.

**ytm_auth.json was never in the backup.** It sits in config/ beside the
three files that are, and is pure device-local auth: losing it silently
signs the user out of YouTube Music. Backed up and restored with the
wifi config, which it resembles.

**Backups were written inside the directory a reinstall deletes.**
config/backups/exports is destroyed by the reinstall the user was told to
make it before. Exports now go beside the install, falling back to the
old path when that is not writable.

**The installer reboots without asking in non-interactive mode**, which
the README did not mention -- easy to hit when piping the install, and
alarming when a device you are installing onto disappears. Documented,
with --no-reboot-prompt. Its log also claimed root:ledmatrix while
printing a hardcoded group name rather than the one it used.

Tests: registry resolution gets its own suite, including the collision
case and third-party entries with an empty plugin_path. The existing
round-trip test passed throughout this because its fixture plugin has a
directory name equal to its id -- the one shape that cannot fail -- so it
now carries ytm_auth too.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Udr6MfaFLUPhX5Fgo67Jf5
2026-08-06 15:30:01 -04:00
48 changed files with 590 additions and 2178 deletions
+37 -21
View File
@@ -4,9 +4,6 @@ on:
pull_request:
push:
branches: [main]
# Manual runs against any branch — useful when a PR's automatic run
# needs a re-run or didn't get created.
workflow_dispatch:
# Both jobs only check out the repo and run pytest.
permissions:
@@ -16,12 +13,6 @@ jobs:
plugin-safety:
name: Plugin safety harness + unit tests
runs-on: ubuntu-latest
env:
# The bundled fixture plugin gives the harness at least one real plugin
# to render, and REQUIRE_PLUGINS turns "discovered zero plugins" into a
# hard failure instead of a silent all-skip green run.
LEDMATRIX_PLUGINS_DIR: test/fixtures/plugins
LEDMATRIX_REQUIRE_PLUGINS: "1"
steps:
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
with:
@@ -38,9 +29,12 @@ jobs:
pip install -r requirements.txt -r requirements-test.txt
pip install RGBMatrixEmulator
- name: Run plugin safety harness
- name: Run harness + visual rendering tests
run: |
pytest --no-cov test/plugins/
pytest --no-cov \
test/plugins/test_harness.py \
test/plugins/test_visual_rendering.py \
test/plugins/test_plugin_matrix.py
unit-tests:
name: Core unit tests
@@ -61,15 +55,37 @@ jobs:
pip install -r requirements.txt -r requirements-test.txt
pip install RGBMatrixEmulator
# Run the ENTIRE test tree (except test/plugins, which the
# plugin-safety job owns). New test files are enrolled automatically;
# excluding anything requires a visible, commented --ignore here.
# Coverage is measured and enforced only in this step — pytest.ini
# deliberately carries no coverage flags so local runs stay fast.
# Safety net for the shared sports/scroll/style infrastructure. These
# suites existed but were not enrolled in CI, so a refactor of
# src/base_classes or src/common could regress them silently. Enrolled
# explicitly (not `pytest test/`) so known hardware-only suites don't
# break CI; grow this list as more suites are made headless.
- name: Run core unit suites
run: |
pytest -m "not hardware" test/ \
--ignore=test/plugins \
--cov=src --cov=web_interface \
--cov-report=term \
--cov-fail-under=45
pytest --no-cov \
test/test_skin_system.py \
test/test_font_manager.py \
test/test_data_sources.py \
test/test_api_extractors.py \
test/test_scroll_helper.py \
test/test_scroll_helper_continuous.py \
test/test_adaptive_layout.py \
test/test_loader_compat_warning.py \
test/test_sports_base_characterization.py \
test/test_element_style.py \
test/test_sports_core_promotions.py \
test/test_sports_modes_promotions.py \
test/test_sports_capabilities.py \
test/test_sports_scroll.py \
test/test_version_consistency.py \
test/test_plugin_compatibility_gate.py \
test/test_install_preserves_existing.py \
test/test_core_owned_config_keys.py \
test/test_async_plugin_updates.py \
test/test_plugin_update_reservation.py \
test/test_template_targets.py \
test/test_widget_scripts.py \
test/test_doc_links.py \
test/test_registry_id_resolution.py \
test/test_backup_manager.py \
test/web_interface/test_cache.py
+6
View File
@@ -365,6 +365,12 @@ sudo bash ./first_time_install.sh
This single script installs services, dependencies, configures permissions and sudoers, and validates the setup.
It finishes by asking whether to reboot. If you run it non-interactively — piped, over a script, or with `-y` — there is no one to ask, so **it reboots immediately without prompting**. Pass `--no-reboot-prompt` to install without rebooting:
```bash
sudo bash ./first_time_install.sh -y --no-reboot-prompt
```
</details>
</details>
+45 -18
View File
@@ -1480,27 +1480,54 @@ if [ -f "$PROJECT_ROOT_DIR/config/config.json" ]; then
fi
# Set proper permissions for secrets file (restrictive: owner rw, group r)
# If service runs as root, set ownership to root so it can read as owner
# Otherwise, use ACTUAL_USER and rely on group membership
# Owned by whoever WRITES the file, which is the web interface.
#
# This used to read the User= of ledmatrix.service — the display service —
# and, finding root, hand the file to root:ledmatrix 640. But the display
# service only ever reads secrets, and root can read any file regardless of
# mode. The account that *writes* them is the web interface: it saves config
# edits and performs backup restores, and it deliberately does not run as root
# (a web server should not). So a root-owned, group-read-only file left the web
# UI unable to write its own secrets, and restoring a backup failed with
# "Permission denied: config_secrets.json" while every other file in the same
# backup restored fine.
#
# Owning by the writer keeps the tighter 640 rather than loosening to
# group-writable, and root still reads it as superuser.
if [ -f "$PROJECT_ROOT_DIR/config/config_secrets.json" ]; then
# Check if service runs as root (from service file or template)
SERVICE_USER="root"
if [ -f "/etc/systemd/system/ledmatrix.service" ]; then
SERVICE_USER=$(grep "^User=" /etc/systemd/system/ledmatrix.service | cut -d'=' -f2 || echo "root")
elif [ -f "$PROJECT_ROOT_DIR/systemd/ledmatrix.service" ]; then
SERVICE_USER=$(grep "^User=" "$PROJECT_ROOT_DIR/systemd/ledmatrix.service" | cut -d'=' -f2 || echo "root")
# The web service is the writer; fall back to the display service, then to
# the installing user, so an unusual layout still lands somewhere sensible.
SECRETS_OWNER=""
for unit in "/etc/systemd/system/ledmatrix-web.service" \
"$PROJECT_ROOT_DIR/systemd/ledmatrix-web.service"; do
if [ -f "$unit" ]; then
SECRETS_OWNER=$(grep -m1 "^User=" "$unit" | cut -d'=' -f2)
[ -n "$SECRETS_OWNER" ] && break
fi
done
if [ -z "$SECRETS_OWNER" ]; then
SECRETS_OWNER="$ACTUAL_USER"
fi
if [ "$SERVICE_USER" = "root" ]; then
# Service runs as root - set ownership to root so it can read as owner
chown "root:$LEDMATRIX_GROUP" "$PROJECT_ROOT_DIR/config/config_secrets.json" || true
echo "✓ Secrets file permissions set (root:ledmatrix for root service)"
else
# Service runs as regular user - use ACTUAL_USER and rely on group membership
chown "$ACTUAL_USER:$LEDMATRIX_GROUP" "$PROJECT_ROOT_DIR/config/config_secrets.json" || true
echo "✓ Secrets file permissions set ($ACTUAL_USER:ledmatrix)"
SECRETS_FILE="$PROJECT_ROOT_DIR/config/config_secrets.json"
# A root-owned file is only correct when the writer really is root.
if ! chown "$SECRETS_OWNER:$LEDMATRIX_GROUP" "$SECRETS_FILE"; then
echo "✗ ERROR: Failed to set ownership on $SECRETS_FILE to $SECRETS_OWNER:$LEDMATRIX_GROUP" >&2
echo " Try: sudo chown $SECRETS_OWNER:$LEDMATRIX_GROUP $SECRETS_FILE" >&2
exit 1
fi
chmod 640 "$PROJECT_ROOT_DIR/config/config_secrets.json"
if ! chmod 640 "$SECRETS_FILE"; then
echo "✗ ERROR: Failed to set permissions on $SECRETS_FILE to 640" >&2
echo " Try: sudo chmod 640 $SECRETS_FILE" >&2
exit 1
fi
ACTUAL_OWNERSHIP=$(stat -c '%U:%G' "$SECRETS_FILE" 2>/dev/null || echo "unknown")
ACTUAL_MODE=$(stat -c '%a' "$SECRETS_FILE" 2>/dev/null || echo "unknown")
if [ "$ACTUAL_OWNERSHIP" != "$SECRETS_OWNER:$LEDMATRIX_GROUP" ] || [ "$ACTUAL_MODE" != "640" ]; then
echo "✗ ERROR: $SECRETS_FILE ended up as $ACTUAL_OWNERSHIP mode $ACTUAL_MODE, expected $SECRETS_OWNER:$LEDMATRIX_GROUP mode 640" >&2
echo " The web interface may be unable to read or write config_secrets.json." >&2
exit 1
fi
echo "✓ Secrets file owned by the web service user ($SECRETS_OWNER:$LEDMATRIX_GROUP, mode 640)"
fi
# Set proper permissions for YTM auth file (readable by all users including root service)
+6 -3
View File
@@ -10,13 +10,16 @@ python_functions = test_*
testpaths = test
# Output options
# Coverage is deliberately NOT configured here: a bare local `pytest` should
# be fast and dependency-light. Coverage is measured and enforced in exactly
# one place — the unit-tests job in .github/workflows/test.yml.
# Note: Coverage options require pytest-cov to be installed
# Run: pip install pytest-cov
addopts =
-v
--strict-markers
--tb=short
--cov=src
--cov-report=term-missing
--cov-report=html
--cov-fail-under=30
# Markers
markers =
-2
View File
@@ -4,6 +4,4 @@ pytest>=9.0.3,<10.0.0
pytest-cov>=4.1.0,<5.0.0
pytest-mock>=3.11.0,<4.0.0
freezegun>=1.2,<2 # deterministic time for golden-image tests
psutil>=6.0.0,<8.0.0 # optional at runtime; installed for tests so the
# /system/status endpoint's real path is exercised
mypy>=1.5.0,<2.0.0 # static type checking (also pinned in .pre-commit-config.yaml)
+98 -8
View File
@@ -16,6 +16,7 @@ import json
import logging
import os
import shutil
import stat
import socket
import tempfile
import zipfile
@@ -82,6 +83,10 @@ BUNDLED_FONTS: frozenset[str] = frozenset({
_CONFIG_REL = Path("config/config.json")
_SECRETS_REL = Path("config/config_secrets.json")
_WIFI_REL = Path("config/wifi_config.json")
# Sits in config/ next to the three above and is pure user state — a
# YouTube Music session that has to be re-authenticated by hand if lost.
# It was omitted from backups, so a restore silently signed the user out.
_YTM_REL = Path("config/ytm_auth.json")
_FONTS_REL = Path("assets/fonts")
_PLUGIN_UPLOADS_REL = Path("assets/plugins")
_STATE_REL = Path("data/plugin_state.json")
@@ -303,6 +308,9 @@ def create_backup(
if (project_root / _WIFI_REL).exists():
zf.write(project_root / _WIFI_REL, _WIFI_REL.as_posix())
contents.append("wifi")
if (project_root / _YTM_REL).exists():
zf.write(project_root / _YTM_REL, _YTM_REL.as_posix())
contents.append("ytm_auth")
# User-uploaded fonts.
user_fonts = iter_user_fonts(project_root)
@@ -348,6 +356,7 @@ def preview_backup_contents(project_root: Path) -> Dict[str, Any]:
"has_config": (project_root / _CONFIG_REL).exists(),
"has_secrets": (project_root / _SECRETS_REL).exists(),
"has_wifi": (project_root / _WIFI_REL).exists(),
"has_ytm_auth": (project_root / _YTM_REL).exists(),
"user_fonts": [p.name for p in iter_user_fonts(project_root)],
"plugin_uploads": len(iter_plugin_uploads(project_root)),
"plugins": list_installed_plugins(project_root),
@@ -429,6 +438,8 @@ def validate_backup(zip_path: Path) -> Tuple[bool, str, Dict[str, Any]]:
detected.append("secrets")
if _WIFI_REL.as_posix() in names:
detected.append("wifi")
if _YTM_REL.as_posix() in names:
detected.append("ytm_auth")
if any(n.startswith(_FONTS_REL.as_posix() + "/") for n in names):
detected.append("fonts")
if any(
@@ -481,8 +492,61 @@ def _extract_zip_safe(zip_path: Path, dest_dir: Path) -> None:
def _copy_file(src: Path, dst: Path) -> None:
"""Replace ``dst`` with ``src``, atomically, without needing to own ``dst``.
``shutil.copy2`` opens the destination for writing, so it needs write
permission on the *existing file*. Several config files are installed
root-owned and group-readable while the web interface which is what runs
a restore deliberately runs as a non-root user. Restoring those failed
with EACCES even though the account could create files in the same
directory perfectly well.
Writing a temporary file alongside and renaming over the target needs only
directory permission, which the web user has. It is also atomic: a crash
mid-restore can no longer leave a half-written config behind.
The destination's existing mode is preserved when there is one, so
restoring secrets does not silently widen them to the umask default.
"""
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, dst)
existing_mode: Optional[int] = None
existing_owner: Optional[Tuple[int, int]] = None
if dst.exists():
try:
info = dst.stat()
existing_mode = stat.S_IMODE(info.st_mode)
existing_owner = (info.st_uid, info.st_gid)
except OSError:
existing_mode = None
existing_owner = None
fd, tmp_name = tempfile.mkstemp(dir=str(dst.parent), prefix=f".{dst.name}.", suffix=".tmp")
os.close(fd)
tmp_path = Path(tmp_name)
try:
shutil.copyfile(src, tmp_path)
if existing_mode is not None:
os.chmod(tmp_path, existing_mode)
else:
shutil.copymode(src, tmp_path)
if existing_owner is not None:
# Replacing a file creates a new inode owned by whoever is running,
# which would silently move a root-owned config to the web user.
# Carry the previous owner across when the OS permits it — only
# root can hand a file to another user, so this is best-effort and
# a plain restore as the web user simply keeps its own ownership.
try:
os.chown(tmp_path, existing_owner[0], existing_owner[1])
except (OSError, PermissionError):
pass
os.replace(tmp_path, dst)
except BaseException:
try:
tmp_path.unlink()
except OSError:
pass
raise
def restore_backup(
@@ -513,7 +577,8 @@ def restore_backup(
try:
_extract_zip_safe(Path(zip_path), tmp_dir)
except (ValueError, zipfile.BadZipFile, OSError) as e:
result.errors.append(f"Failed to extract backup: {e}")
logger.error("[Backup] Failed to extract backup: %s", e, exc_info=True)
result.errors.append("Failed to extract backup")
return result
# Main config.
@@ -522,7 +587,8 @@ def restore_backup(
_copy_file(tmp_dir / _CONFIG_REL, project_root / _CONFIG_REL)
result.restored.append("config")
except OSError as e:
result.errors.append(f"Failed to restore config.json: {e}")
logger.error("[Backup] Failed to restore config.json: %s", e, exc_info=True)
result.errors.append("Failed to restore config.json")
elif (tmp_dir / _CONFIG_REL).exists():
result.skipped.append("config")
@@ -532,7 +598,10 @@ def restore_backup(
_copy_file(tmp_dir / _SECRETS_REL, project_root / _SECRETS_REL)
result.restored.append("secrets")
except OSError as e:
result.errors.append(f"Failed to restore config_secrets.json: {e}")
logger.error(
"[Backup] Failed to restore config_secrets.json: %s", e, exc_info=True
)
result.errors.append("Failed to restore config_secrets.json")
elif (tmp_dir / _SECRETS_REL).exists():
result.skipped.append("secrets")
@@ -542,10 +611,26 @@ def restore_backup(
_copy_file(tmp_dir / _WIFI_REL, project_root / _WIFI_REL)
result.restored.append("wifi")
except OSError as e:
result.errors.append(f"Failed to restore wifi_config.json: {e}")
logger.error(
"[Backup] Failed to restore wifi_config.json: %s", e, exc_info=True
)
result.errors.append("Failed to restore wifi_config.json")
elif (tmp_dir / _WIFI_REL).exists():
result.skipped.append("wifi")
# YouTube Music session. Follows restore_wifi rather than getting its
# own flag: it is device-local auth in the same sense, and a separate
# toggle for one file would be noise in the restore dialog.
if options.restore_wifi and (tmp_dir / _YTM_REL).exists():
try:
_copy_file(tmp_dir / _YTM_REL, project_root / _YTM_REL)
result.restored.append("ytm_auth")
except OSError as e:
logger.error("[Backup] Failed to restore ytm_auth.json: %s", e, exc_info=True)
result.errors.append("Failed to restore ytm_auth.json")
elif (tmp_dir / _YTM_REL).exists():
result.skipped.append("ytm_auth")
# User fonts — skip anything that collides with a bundled font.
tmp_fonts = tmp_dir / _FONTS_REL
if options.restore_fonts and tmp_fonts.exists():
@@ -560,7 +645,10 @@ def restore_backup(
_copy_file(font, project_root / _FONTS_REL / font.name)
restored_count += 1
except OSError as e:
result.errors.append(f"Failed to restore font {font.name}: {e}")
logger.error(
"[Backup] Failed to restore font %s: %s", font.name, e, exc_info=True
)
result.errors.append(f"Failed to restore font {font.name}")
if restored_count:
result.restored.append(f"fonts ({restored_count})")
elif tmp_fonts.exists():
@@ -581,7 +669,8 @@ def restore_backup(
_copy_file(src, project_root / rel)
count += 1
except OSError as e:
result.errors.append(f"Failed to restore {rel}: {e}")
logger.error("[Backup] Failed to restore %s: %s", rel, e, exc_info=True)
result.errors.append(f"Failed to restore {rel}")
if count:
result.restored.append(f"plugin_uploads ({count})")
elif tmp_uploads.exists():
@@ -599,7 +688,8 @@ def restore_backup(
if isinstance(p, dict) and p.get("plugin_id")
]
except (OSError, json.JSONDecodeError) as e:
result.errors.append(f"Could not read plugins.json: {e}")
logger.error("[Backup] Could not read plugins.json: %s", e, exc_info=True)
result.errors.append("Could not read plugins.json")
result.success = not result.errors
return result
-5
View File
@@ -383,15 +383,10 @@ class SportsCore(ABC):
ctx = skin_runtime.build_context(self, game, size=size)
card = skin.render_vegas_card(ctx, dict(game))
if card is not None:
# A successful render clears accumulated strikes, mirroring
# _render_game — transient failures must not add up across
# the session and disable a working skin.
self._skin_failures = 0
return card
ctx = skin_runtime.build_context(self, game, size=size)
render = getattr(skin, f"render_{self.SKIN_MODE}")
if render(ctx, dict(game)):
self._skin_failures = 0
return ctx.canvas
except Exception:
# Card failures count toward the same 3-strike session disable
+20 -39
View File
@@ -106,9 +106,14 @@ class ConfigManager:
Returns:
SaveResult with status and details
"""
# Load current secrets to preserve them (raises if unreadable — see
# _load_secrets_for_save)
secrets_content = self._load_secrets_for_save()
# Load current secrets to preserve them
secrets_content = {}
if os.path.exists(self.secrets_path):
try:
with open(self.secrets_path, 'r') as f_secrets:
secrets_content = json.load(f_secrets)
except Exception as e:
self.logger.warning(f"Could not load secrets file {self.secrets_path} during save: {e}")
# Strip secrets from main config before saving
config_to_write = self._strip_secrets_recursive(new_config_data, secrets_content)
@@ -285,43 +290,19 @@ class ConfigManager:
result[key] = value
return result
def _load_secrets_for_save(self) -> Dict[str, Any]:
"""Load config_secrets.json for stripping before a save.
A missing secrets file is fine (nothing to strip). But a file that
EXISTS and cannot be read or parsed means stripping is impossible
and the in-memory config being saved has secrets deep-merged into it,
so proceeding would write them into config.json in plaintext. That
was the historical behavior; it is now a hard refusal. The save
raises so the caller (and user) fixes the secrets file instead of
silently leaking its contents into the world-readable main config.
"""
if not os.path.exists(self.secrets_path):
return {}
try:
with open(self.secrets_path, 'r') as f_secrets:
return json.load(f_secrets)
# Only the expected read/parse failures — an unexpected implementation
# error should propagate as itself, not masquerade as a secrets-file
# problem. (JSONDecodeError and UnicodeDecodeError are ValueErrors.)
except (OSError, ValueError, RecursionError) as e:
error_msg = (
f"Refusing to save config: secrets file {self.secrets_path} exists "
f"but could not be loaded ({e}). Saving without it would write "
f"merged secret values into config.json in plaintext. Fix or "
f"remove the secrets file, then retry."
)
self.logger.error("[Config] %s", error_msg, exc_info=True)
raise ConfigError(error_msg, config_path=self.secrets_path) from e
def save_config(self, new_config_data: Dict[str, Any]) -> None:
"""Save configuration to the main JSON file, stripping out secrets.
Raises ConfigError when the secrets file exists but cannot be loaded,
because stripping would be impossible and secrets would leak into
config.json.
"""
secrets_content = self._load_secrets_for_save()
"""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.
config_to_write = self._strip_secrets_recursive(new_config_data, secrets_content)
-39
View File
@@ -180,45 +180,6 @@ def declared_min_version(manifest: Dict[str, Any]) -> Optional[str]:
return None
def is_update_available(installed_version: str, latest_version: str) -> bool:
"""Return True when the registry's ``latest_version`` is strictly newer
than the installed version.
THE shared comparator for "should this plugin be updated?" used by both
the web UI's update badge (`api_v3._is_plugin_update_available`) and the
store's `update_plugin` reinstall decision, so the two can never disagree.
Uses PEP 440-aware comparison (``packaging``), which also normalizes
equivalent spellings: ``v1.2.0`` == ``1.2.0`` and ``1.2`` == ``1.2.0``, so
cosmetic differences never trigger a reinstall and a locally modified
plugin whose version is *ahead* of the registry is never "updated"
(downgraded). If either version string can't be parsed the mismatch is
surfaced (True) so the user can reconcile, rather than silently hiding a
potential update.
"""
if not installed_version or not latest_version:
return False
if not isinstance(installed_version, str) or not isinstance(latest_version, str):
# A malformed manifest/registry can carry a number (1.2) or worse;
# packaging would raise TypeError. Surface the mismatch instead.
return True
if installed_version == latest_version:
return False
try:
from packaging.version import parse as _parse_version, InvalidVersion
except ImportError:
# packaging is a core dependency, but if it's somehow unavailable we
# can't compare semantically — surface the mismatch we already know
# exists (the two strings differ).
return True
try:
return _parse_version(latest_version) > _parse_version(installed_version)
except InvalidVersion:
# Unparseable version string: we can't tell direction, so surface the
# mismatch rather than silently hiding a potential update.
return True
def check(manifest: Dict[str, Any], core_version: str) -> Tuple[bool, Optional[str]]:
"""Return ``(compatible, reason)``.
+36 -16
View File
@@ -1143,7 +1143,7 @@ class PluginStoreManager:
"""
registry = self.fetch_registry()
plugins = registry.get('plugins', []) or []
plugin_info = next((p for p in plugins if p['id'] == plugin_id), None)
plugin_info = self._match_registry_entry(plugins, plugin_id)
if not plugin_info:
return None
@@ -1183,6 +1183,37 @@ class PluginStoreManager:
return plugin_info
@staticmethod
def _match_registry_entry(plugins: List[Dict], plugin_id: str) -> Optional[Dict]:
"""Find a registry entry by its id, or by the directory it installs to.
Four shipped plugins have a registry ``id`` that differs from the ``id``
in their own manifest: ``weather`` installs to ``plugins/ledmatrix-weather``,
and likewise stocks, music and leaderboard. Installation already prefers
the manifest id for the directory name, so on disk, in ``config.json``
and in a backup manifest those plugins are called ``ledmatrix-weather``.
Only the registry calls them ``weather``, and nothing resolved that in
reverse: restoring a backup asked the store for ``ledmatrix-weather``
and got "Plugin not found in registry", silently dropping four enabled
plugins from a restored device.
Matching ``plugin_path`` fixes it without renaming any published id,
which would orphan ``plugin_state.json`` entries keyed on the old ones.
Exact id always wins, so an entry whose *path* happens to collide with
another entry's id cannot shadow it.
"""
if not plugin_id:
return None
exact = next((p for p in plugins if p.get('id') == plugin_id), None)
if exact is not None:
return exact
for entry in plugins:
path = (entry.get('plugin_path') or '').rstrip('/')
if path and path.rsplit('/', 1)[-1] == plugin_id:
return entry
return None
def get_registry_info(self, plugin_id: str) -> Optional[Dict]:
"""
Get plugin information from the registry cache only (no GitHub API calls).
@@ -1198,7 +1229,7 @@ class PluginStoreManager:
"""
registry = self.fetch_registry()
plugins = registry.get('plugins', []) or []
return next((p for p in plugins if p.get('id') == plugin_id), None)
return self._match_registry_entry(plugins, plugin_id)
def install_plugin(self, plugin_id: str, branch: Optional[str] = None) -> bool:
"""Install a plugin, keeping any existing install until the new one is
@@ -2969,10 +3000,7 @@ class PluginStoreManager:
remote_branch = plugin_info_remote.get('branch') or plugin_info_remote.get('default_branch')
# Compare local manifest version against registry latest_version
# to avoid unnecessary reinstalls for monorepo plugins. Uses the
# same semantic comparator as the web UI's update badge, so
# equivalent spellings ("v1.2.0" vs "1.2.0") never trigger a
# reinstall and a locally-ahead version is never downgraded.
# to avoid unnecessary reinstalls for monorepo plugins
try:
local_manifest_path = plugin_path / "manifest.json"
if local_manifest_path.exists():
@@ -2980,16 +3008,8 @@ class PluginStoreManager:
local_manifest = json.load(f)
local_version = local_manifest.get('version', '')
remote_version = plugin_info_remote.get('latest_version', '')
from src.plugin_system.compatibility import is_update_available
# No truthiness gate: the shared comparator already treats
# a missing version on either side as "no update", and the
# store must agree with the UI badge in that case too. A
# missing manifest (not just a missing version field)
# still falls through to the reinstall recovery path.
if not is_update_available(local_version, remote_version):
self.logger.info(
f"Plugin {plugin_id} already at latest version "
f"(installed {local_version}, registry {remote_version})")
if local_version and remote_version and local_version == remote_version:
self.logger.info(f"Plugin {plugin_id} already at latest version {local_version}")
return True
except Exception as e:
self.logger.debug(f"Could not compare versions for {plugin_id}: {e}")
@@ -1,31 +0,0 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "CI Fixture Plugin",
"type": "object",
"properties": {
"enabled": {
"type": "boolean",
"default": true
},
"display_duration": {
"type": "number",
"default": 5
},
"border_color": {
"type": "array",
"items": {"type": "integer", "minimum": 0, "maximum": 255},
"minItems": 3,
"maxItems": 3,
"default": [0, 255, 0],
"description": "RGB color of the border rectangle."
},
"diagonal_color": {
"type": "array",
"items": {"type": "integer", "minimum": 0, "maximum": 255},
"minItems": 3,
"maxItems": 3,
"default": [255, 0, 0],
"description": "RGB color of the diagonals."
}
}
}
-44
View File
@@ -1,44 +0,0 @@
"""
CI fixture plugin.
Exists so the plugin safety harness (test/plugins/test_plugin_matrix.py and
the plugin-safety CI job) always has at least one real plugin to load and
render without it, an empty plugins/ directory turns the whole job into a
green no-op. The render is deliberately trivial and fully deterministic:
a border rectangle plus both diagonals, sized from the display manager's
declared dimensions. No fonts, no network, no time dependence, so golden
images are stable across platforms.
"""
from PIL import ImageDraw
from src.plugin_system.base_plugin import BasePlugin
class CIFixturePlugin(BasePlugin):
"""Deterministic CI-only fixture plugin: renders a border + diagonals
pattern sized from the display's declared dimensions. Never shipped to
devices; exists solely so the plugin safety harness has a real plugin
to exercise in CI."""
def update(self) -> None:
"""Nothing to fetch — the render is self-contained."""
def display(self, force_clear: bool = False) -> None:
self.display_manager.clear()
width = self.display_manager.matrix.width
height = self.display_manager.matrix.height
border = tuple(self.config.get("border_color", [0, 255, 0]))
diagonal = tuple(self.config.get("diagonal_color", [255, 0, 0]))
image = self.display_manager.image
draw = ImageDraw.Draw(image)
# Blank only the declared panel area, then draw edge-to-edge content:
# the border proves the plugin reads dynamic dimensions (any overflow
# or underfill at any size is a harness bug or a dimensions bug), the
# diagonals make golden comparisons sensitive to size/offset drift.
draw.rectangle([0, 0, width - 1, height - 1], fill=(0, 0, 0))
draw.rectangle([0, 0, width - 1, height - 1], outline=border)
draw.line([0, 0, width - 1, height - 1], fill=diagonal)
draw.line([0, height - 1, width - 1, 0], fill=diagonal)
self.display_manager.update_display()
-13
View File
@@ -1,13 +0,0 @@
{
"id": "ci-fixture-plugin",
"name": "CI Fixture Plugin",
"version": "1.0.0",
"description": "Bundled test fixture so the plugin safety harness always has at least one real plugin to render in CI. Draws a deterministic border + diagonals pattern at any panel size. Not installable from the store and never shipped to devices.",
"author": "LEDMatrix",
"entry_point": "manager.py",
"class_name": "CIFixturePlugin",
"display_modes": ["ci-fixture"],
"update_interval": 3600,
"min_ledmatrix_version": "2.0.0",
"compatible_versions": [">=2.0.0"]
}
@@ -1,7 +0,0 @@
# No dependencies — the fixture must load in any environment.
#
# Pillow is deliberately NOT pinned here even though manager.py imports
# PIL: it is a core LEDMatrix dependency (see the repo-root
# requirements.txt), so it is always present wherever the harness runs,
# and the harness loads plugins with install_deps=False anyway. Pinning
# it here would only invite a needless pip install during test runs.
Binary file not shown.

Before

Width:  |  Height:  |  Size: 359 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 586 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 849 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 1.2 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 395 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 319 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 466 B

Binary file not shown.

Before

Width:  |  Height:  |  Size: 454 B

+2 -10
View File
@@ -23,17 +23,9 @@ os.environ['EMULATOR'] = 'true'
def plugins_dir() -> Path:
"""Get the plugins directory path.
Honors LEDMATRIX_PLUGINS_DIR (first entry) when set the same override
test_plugin_matrix.py uses, so CI can point every plugin suite at the
bundled fixture plugins. Otherwise checks plugins/ first, then falls
back to plugin-repos/ for monorepo development environments.
Checks plugins/ first, then falls back to plugin-repos/
for monorepo development environments.
"""
env = os.environ.get('LEDMATRIX_PLUGINS_DIR')
if env:
first = env.split(os.pathsep)[0]
if first:
return Path(first)
plugins_path = project_root / 'plugins'
plugin_repos_path = project_root / 'plugin-repos'
@@ -1,10 +1,5 @@
"""
Integration tests for basketball-scoreboard plugin.
Requires the real plugin to be installed (plugins/ or plugin-repos/,
or the dir named by LEDMATRIX_PLUGINS_DIR) on machines without it,
every test here skips by design. CI covers plugin safety with the
bundled fixture plugin via test_plugin_matrix.py instead.
"""
import pytest
-5
View File
@@ -1,10 +1,5 @@
"""
Integration tests for calendar plugin.
Requires the real plugin to be installed (plugins/ or plugin-repos/,
or the dir named by LEDMATRIX_PLUGINS_DIR) on machines without it,
every test here skips by design. CI covers plugin safety with the
bundled fixture plugin via test_plugin_matrix.py instead.
"""
import pytest
-5
View File
@@ -1,10 +1,5 @@
"""
Integration tests for clock-simple plugin.
Requires the real plugin to be installed (plugins/ or plugin-repos/,
or the dir named by LEDMATRIX_PLUGINS_DIR) on machines without it,
every test here skips by design. CI covers plugin safety with the
bundled fixture plugin via test_plugin_matrix.py instead.
"""
import pytest
-5
View File
@@ -1,10 +1,5 @@
"""
Integration tests for odds-ticker plugin.
Requires the real plugin to be installed (plugins/ or plugin-repos/,
or the dir named by LEDMATRIX_PLUGINS_DIR) on machines without it,
every test here skips by design. CI covers plugin safety with the
bundled fixture plugin via test_plugin_matrix.py instead.
"""
import pytest
-5
View File
@@ -1,10 +1,5 @@
"""
Integration tests for soccer-scoreboard plugin.
Requires the real plugin to be installed (plugins/ or plugin-repos/,
or the dir named by LEDMATRIX_PLUGINS_DIR) on machines without it,
every test here skips by design. CI covers plugin safety with the
bundled fixture plugin via test_plugin_matrix.py instead.
"""
import pytest
-5
View File
@@ -1,10 +1,5 @@
"""
Integration tests for text-display plugin.
Requires the real plugin to be installed (plugins/ or plugin-repos/,
or the dir named by LEDMATRIX_PLUGINS_DIR) on machines without it,
every test here skips by design. CI covers plugin safety with the
bundled fixture plugin via test_plugin_matrix.py instead.
"""
import pytest
+52
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import json
import stat
import zipfile
from pathlib import Path
@@ -41,6 +42,13 @@ def _make_project(root: Path) -> Path:
json.dumps({"ap_mode": {"ssid": "LEDMatrix"}}),
encoding="utf-8",
)
# Device-local auth that lives in config/ like the three above. It was
# omitted from backups, so a restore silently signed the user out of
# YouTube Music and they had to re-authenticate by hand.
(root / "config" / "ytm_auth.json").write_text(
json.dumps({"token": "YTM-TOKEN"}),
encoding="utf-8",
)
fonts = root / "assets" / "fonts"
fonts.mkdir(parents=True)
@@ -240,6 +248,10 @@ def test_restore_roundtrip(project: Path, empty_project: Path, tmp_path: Path) -
restored_secrets = json.loads((empty_project / "config" / "config_secrets.json").read_text())
assert restored_secrets["ledmatrix-weather"]["api_key"] == "SECRET"
assert "ytm_auth" in result.restored
restored_ytm = json.loads((empty_project / "config" / "ytm_auth.json").read_text())
assert restored_ytm["token"] == "YTM-TOKEN"
# User font restored, bundled font untouched.
assert (empty_project / "assets" / "fonts" / "my-custom-font.ttf").read_bytes() == b"\x00\x01USER"
assert (empty_project / "assets" / "fonts" / "5x7.bdf").read_text() == "BUNDLED"
@@ -271,6 +283,10 @@ def test_restore_honors_options(project: Path, empty_project: Path, tmp_path: Pa
assert result.plugins_to_install == []
assert "secrets" in result.skipped
assert "wifi" in result.skipped
# ytm_auth rides on restore_wifi rather than its own flag -- disabling
# wifi restore must not leave a stale session token behind.
assert "ytm_auth" in result.skipped
assert not (empty_project / "config" / "ytm_auth.json").exists()
def test_restore_rejects_malicious_zip(empty_project: Path, tmp_path: Path) -> None:
@@ -282,3 +298,39 @@ def test_restore_rejects_malicious_zip(empty_project: Path, tmp_path: Path) -> N
# validate_backup catches it before extraction.
assert not result.success
assert any("unsafe" in e.lower() for e in result.errors)
def test_restore_over_a_file_the_user_cannot_write(
project: Path, empty_project: Path, tmp_path: Path
) -> None:
"""Restore must not need write permission on the destination *file*.
Reproduces what a fresh install leaves behind: config files owned by root
and only group-readable, while the web interface that performs the restore
runs as a non-root user. shutil.copy2 opens the destination for writing and
failed with EACCES; writing alongside and renaming needs only directory
permission, which that account has.
Simulated here by making the destination read-only the owner cannot
open it for writing either, but can still replace it within its directory.
"""
zip_path = create_backup(project, output_dir=tmp_path / "exports")
# Pre-existing, read-only destinations.
(empty_project / "config").mkdir(parents=True, exist_ok=True)
for name in ("config.json", "config_secrets.json", "wifi_config.json", "ytm_auth.json"):
target = empty_project / "config" / name
target.write_text("{}", encoding="utf-8")
target.chmod(0o444)
result = restore_backup(zip_path, empty_project, RestoreOptions())
assert result.success, result.errors
for section in ("config", "secrets", "wifi", "ytm_auth"):
assert section in result.restored, f"{section} not restored: {result.errors}"
restored = json.loads((empty_project / "config" / "config.json").read_text())
assert restored["my-plugin"]["favorites"] == ["A", "B"]
# The destination's mode is preserved rather than widened to the umask.
assert stat.S_IMODE((empty_project / "config" / "config_secrets.json").stat().st_mode) == 0o444
-110
View File
@@ -1,110 +0,0 @@
"""
Tests for BasePlugin.get_display_duration ~100 lines of type coercion that
every plugin's rotation slot depends on, previously untested.
The contract: a positive number wins wherever it comes from; everything else
falls through instance attr config the 15.0 default, logging on the way.
"""
from unittest.mock import MagicMock
import pytest
from src.plugin_system.base_plugin import BasePlugin
class _MinimalPlugin(BasePlugin):
def update(self):
pass
def display(self, force_clear=False):
pass
def make_plugin(config=None, instance_duration="__unset__"):
plugin = _MinimalPlugin(
plugin_id="duration-test",
config=config or {},
display_manager=MagicMock(),
cache_manager=MagicMock(),
plugin_manager=MagicMock(),
)
if instance_duration != "__unset__":
plugin.display_duration = instance_duration
return plugin
class TestInstanceVariable:
def test_positive_int_wins(self):
assert make_plugin(instance_duration=30).get_display_duration() == 30.0
def test_positive_float_wins(self):
assert make_plugin(instance_duration=12.5).get_display_duration() == 12.5
def test_returns_float_type(self):
result = make_plugin(instance_duration=30).get_display_duration()
assert isinstance(result, float)
def test_numeric_string_wins(self):
assert make_plugin(instance_duration="25").get_display_duration() == 25.0
def test_zero_falls_through_to_config(self):
plugin = make_plugin(config={"display_duration": 20},
instance_duration=0)
assert plugin.get_display_duration() == 20.0
def test_negative_falls_through_to_config(self):
plugin = make_plugin(config={"display_duration": 20},
instance_duration=-5)
assert plugin.get_display_duration() == 20.0
def test_none_falls_through_to_config(self):
plugin = make_plugin(config={"display_duration": 20},
instance_duration=None)
assert plugin.get_display_duration() == 20.0
def test_garbage_string_falls_through(self):
plugin = make_plugin(config={"display_duration": 20},
instance_duration="abc")
assert plugin.get_display_duration() == 20.0
def test_non_positive_string_falls_through(self):
plugin = make_plugin(config={"display_duration": 20},
instance_duration="0")
assert plugin.get_display_duration() == 20.0
def test_unexpected_type_falls_through(self):
plugin = make_plugin(config={"display_duration": 20},
instance_duration=[30])
assert plugin.get_display_duration() == 20.0
def test_bool_true_is_one_second(self):
# Characterized quirk: bool is an int subclass, so display_duration =
# True passes the isinstance((int, float)) branch and returns 1.0.
assert make_plugin(instance_duration=True).get_display_duration() == 1.0
class TestConfigFallback:
def test_config_number(self):
assert make_plugin({"display_duration": 20}).get_display_duration() == 20.0
def test_config_numeric_string(self):
assert make_plugin({"display_duration": "12.5"}).get_display_duration() == 12.5
def test_missing_config_uses_default(self):
assert make_plugin({}).get_display_duration() == 15.0
def test_config_zero_uses_default(self):
assert make_plugin({"display_duration": 0}).get_display_duration() == 15.0
def test_config_negative_uses_default(self):
assert make_plugin({"display_duration": -10}).get_display_duration() == 15.0
def test_config_garbage_string_uses_default(self):
assert make_plugin({"display_duration": "soon"}).get_display_duration() == 15.0
def test_config_unexpected_type_uses_default(self):
assert make_plugin({"display_duration": {"s": 5}}).get_display_duration() == 15.0
def test_config_none_uses_default(self):
assert make_plugin({"display_duration": None}).get_display_duration() == 15.0
-271
View File
@@ -1,271 +0,0 @@
"""
Tests for src/plugin_system/compatibility.py the "can this plugin run on
this core?" gate used by both the plugin loader (advisory) and the store
manager (blocking at install/update time).
This module had zero direct test coverage despite guarding every install.
These tests pin the documented contract: refuse only on evidence, resolve
every uncertain case (unparseable versions, missing fields, untrustworthy
core) to compatible.
"""
import pytest
from src.plugin_system.compatibility import (
TRUSTWORTHY_FLOOR,
parse_semver,
_parse_strict,
_satisfies_range,
satisfies_compatible_versions,
declared_min_version,
check,
)
class TestParseSemver:
def test_plain_triplet(self):
assert parse_semver("1.2.3") == (1, 2, 3)
def test_leading_v_tolerated(self):
assert parse_semver("v3.2.1") == (3, 2, 1)
def test_prerelease_suffix_stripped(self):
# "3.2.0-rc1" must NOT parse as (3, 2, 1) — a release candidate must
# not rank above its own release.
assert parse_semver("3.2.0-rc1") == (3, 2, 0)
def test_build_suffix_stripped(self):
# "3.2.0+build42" must NOT parse as (3, 2, 42).
assert parse_semver("3.2.0+build42") == (3, 2, 0)
def test_two_part_version_pads_zero(self):
assert parse_semver("1.2") == (1, 2, 0)
def test_one_part_version_pads_zeros(self):
assert parse_semver("2") == (2, 0, 0)
def test_extra_parts_ignored(self):
assert parse_semver("1.2.3.4") == (1, 2, 3)
def test_non_string_returns_none(self):
assert parse_semver(None) is None
assert parse_semver(123) is None
assert parse_semver((1, 2, 3)) is None
def test_garbage_with_no_digits_is_lenient_zero(self):
# Documented leniency: digit-scraping yields (0, 0, 0) for pure
# garbage. Fine for a floor (0.0.0 never blocks), wrong for ranges —
# which is why ranges go through _parse_strict instead.
assert parse_semver("garbage") == (0, 0, 0)
def test_whitespace_stripped(self):
assert parse_semver(" 1.2.3 ") == (1, 2, 3)
class TestParseStrict:
def test_accepts_real_versions(self):
assert _parse_strict("1.2.3") == (1, 2, 3)
assert _parse_strict("v1.2.3-rc1") == (1, 2, 3)
assert _parse_strict("2.0") == (2, 0, 0)
def test_rejects_garbage(self):
assert _parse_strict("not-a-version") is None
assert _parse_strict("") is None
def test_rejects_non_string(self):
assert _parse_strict(None) is None
class TestSatisfiesRange:
CORE = (3, 1, 0)
@pytest.mark.parametrize("spec,expected", [
(">=3.0.0", True),
(">=3.1.0", True),
(">=3.2.0", False),
("<=3.1.0", True),
("<=3.0.9", False),
(">3.0.9", True),
(">3.1.0", False),
("<3.2.0", True),
("<3.1.0", False),
])
def test_comparison_operators(self, spec, expected):
assert _satisfies_range(self.CORE, spec) is expected
def test_tilde_allows_patch_only(self):
# ~3.1.0 means >=3.1.0, <3.2.0
assert _satisfies_range((3, 1, 5), "~3.1.0") is True
assert _satisfies_range((3, 2, 0), "~3.1.0") is False
assert _satisfies_range((3, 0, 9), "~3.1.0") is False
def test_caret_allows_minor_and_patch(self):
# ^3.1.0 means >=3.1.0, <4.0.0
assert _satisfies_range((3, 9, 9), "^3.1.0") is True
assert _satisfies_range((4, 0, 0), "^3.1.0") is False
assert _satisfies_range((3, 0, 0), "^3.1.0") is False
def test_bare_exact_version(self):
assert _satisfies_range((3, 1, 0), "3.1.0") is True
assert _satisfies_range((3, 1, 1), "3.1.0") is False
def test_inclusive_dash_range(self):
assert _satisfies_range((2, 5, 0), "2.0.0 - 3.1.0") is True
assert _satisfies_range((2, 0, 0), "2.0.0 - 3.1.0") is True
assert _satisfies_range((3, 1, 0), "2.0.0 - 3.1.0") is True
assert _satisfies_range((3, 1, 1), "2.0.0 - 3.1.0") is False
def test_unparseable_spec_returns_none_not_false(self):
# Garbage must read as "no evidence", never as a refusal — an
# unrecognised spelling must not cost a user a working install.
assert _satisfies_range(self.CORE, "banana") is None
assert _satisfies_range(self.CORE, ">=banana") is None
assert _satisfies_range(self.CORE, "") is None
assert _satisfies_range(self.CORE, "banana - 3.0.0") is None
class TestSatisfiesCompatibleVersions:
def test_any_entry_satisfying_wins(self):
manifest = {"compatible_versions": ["<1.0.0", ">=3.0.0"]}
assert satisfies_compatible_versions(manifest, (3, 1, 0)) is True
def test_all_entries_failing_is_false(self):
manifest = {"compatible_versions": ["<1.0.0", "2.0.0 - 2.9.9"]}
assert satisfies_compatible_versions(manifest, (3, 1, 0)) is False
def test_absent_field_returns_none(self):
assert satisfies_compatible_versions({}, (3, 1, 0)) is None
def test_empty_list_returns_none(self):
assert satisfies_compatible_versions(
{"compatible_versions": []}, (3, 1, 0)) is None
def test_non_list_returns_none(self):
assert satisfies_compatible_versions(
{"compatible_versions": ">=2.0.0"}, (3, 1, 0)) is None
def test_all_unparseable_entries_returns_none(self):
manifest = {"compatible_versions": ["banana", 42, None]}
assert satisfies_compatible_versions(manifest, (3, 1, 0)) is None
def test_mixed_parseable_and_garbage_uses_parseable(self):
manifest = {"compatible_versions": ["banana", ">=3.0.0"]}
assert satisfies_compatible_versions(manifest, (3, 1, 0)) is True
class TestDeclaredMinVersion:
def test_top_level_field(self):
assert declared_min_version({"min_ledmatrix_version": "2.1.0"}) == "2.1.0"
def test_requires_dict_fallback(self):
manifest = {"requires": {"min_ledmatrix_version": "2.2.0"}}
assert declared_min_version(manifest) == "2.2.0"
def test_versions_array_fallback(self):
manifest = {"versions": [{"ledmatrix_min_version": "2.3.0"}]}
assert declared_min_version(manifest) == "2.3.0"
def test_versions_array_deprecated_spelling(self):
manifest = {"versions": [{"ledmatrix_min": "2.4.0"}]}
assert declared_min_version(manifest) == "2.4.0"
def test_top_level_wins_over_versions_array(self):
manifest = {
"min_ledmatrix_version": "2.1.0",
"versions": [{"ledmatrix_min_version": "9.9.9"}],
}
assert declared_min_version(manifest) == "2.1.0"
def test_requires_as_list_does_not_raise(self):
# A hand-edited manifest can carry `requires` as a list; this used to
# raise AttributeError and one malformed manifest would take down the
# whole install path.
assert declared_min_version({"requires": ["something"]}) is None
def test_versions_as_dict_does_not_raise(self):
# Same for `versions` as a mapping (used to raise KeyError).
assert declared_min_version({"versions": {"0": {}}}) is None
def test_nothing_declared_returns_none(self):
assert declared_min_version({}) is None
class TestCheck:
def test_compatible_when_nothing_declared(self):
assert check({}, "3.1.0") == (True, None)
def test_min_version_blocks_older_core(self):
manifest = {"name": "Test Plugin", "min_ledmatrix_version": "3.2.0"}
ok, reason = check(manifest, "3.1.0")
assert ok is False
assert "3.2.0" in reason and "3.1.0" in reason
def test_min_version_allows_equal_core(self):
manifest = {"min_ledmatrix_version": "3.1.0"}
assert check(manifest, "3.1.0") == (True, None)
def test_compatible_versions_upper_bound_blocks(self):
# A range is the only field that can express "not compatible with
# newer cores" — it must win even when the floor passes.
manifest = {
"name": "Old Plugin",
"min_ledmatrix_version": "2.0.0",
"compatible_versions": ["2.0.0 - 2.9.9"],
}
ok, reason = check(manifest, "3.1.0")
assert ok is False
assert "2.0.0 - 2.9.9" in reason
def test_unparseable_core_with_high_floor_is_blocked(self):
manifest = {"min_ledmatrix_version": "3.2.0",
"compatible_versions": [">=3.2.0"]}
# An unparseable core version is "unknown", not "old"... but note
# parse_semver("garbage") == (0,0,0) which is below TRUSTWORTHY_FLOOR,
# so this rides the untrustworthy-core branch: floor > 2.0.0 blocks.
ok, reason = check(manifest, "garbage")
assert ok is False
assert "too old to identify reliably" in reason
def test_untrustworthy_core_allows_ecosystem_baseline_floor(self):
# A core reporting 1.0.0 may really be v3.1.0 (which shipped with a
# wrong __version__). Floors at or below TRUSTWORTHY_FLOOR must not
# block, or that population could install nothing.
manifest = {"min_ledmatrix_version": "2.0.0",
"compatible_versions": [">=2.0.0"]}
assert check(manifest, "1.0.0") == (True, None)
def test_untrustworthy_core_blocks_floor_above_baseline(self):
# But a floor above 2.0.0 needs modules that no core reporting below
# the floor can have — the one refusal on that branch.
manifest = {"name": "New Plugin", "min_ledmatrix_version": "3.2.0"}
ok, reason = check(manifest, "1.0.0")
assert ok is False
assert "too old to identify reliably" in reason
def test_untrustworthy_core_ignores_compatible_versions(self):
# On the untrustworthy branch only the declared floor is consulted;
# ranges cannot be evaluated against a version that isn't evidence.
manifest = {"compatible_versions": ["2.0.0 - 2.9.9"]}
assert check(manifest, "1.0.0") == (True, None)
def test_floor_exactly_at_trustworthy_floor_is_allowed(self):
floor = ".".join(str(n) for n in TRUSTWORTHY_FLOOR)
manifest = {"min_ledmatrix_version": floor}
assert check(manifest, "1.0.0") == (True, None)
def test_reason_uses_manifest_name(self):
manifest = {"name": "Fancy Clock", "min_ledmatrix_version": "9.0.0"}
ok, reason = check(manifest, "3.1.0")
assert ok is False
assert reason.startswith("Fancy Clock")
def test_reason_falls_back_to_id(self):
manifest = {"id": "fancy-clock", "min_ledmatrix_version": "9.0.0"}
ok, reason = check(manifest, "3.1.0")
assert ok is False
assert reason.startswith("fancy-clock")
def test_prerelease_core_compares_equal_to_release(self):
# Documented: prereleases compare equal to their release.
manifest = {"min_ledmatrix_version": "3.2.0"}
assert check(manifest, "3.2.0-rc1") == (True, None)
-179
View File
@@ -1,179 +0,0 @@
"""
Tests for the ConfigManager secrets round-trip and the load_config fast path.
The contract under test: config_secrets.json values are deep-merged INTO the
in-memory config at load time, and stripped back OUT before anything is
written to config.json so secrets live in exactly one file on disk. This
suite pins that round-trip plus its sharp edges, including the guard that a
save REFUSES (ConfigError) when the secrets file exists but can't be loaded,
rather than leaking merged secrets into config.json in plaintext.
Complements test_config_manager.py, which covers loading/migration/validation.
"""
import json
import os
import pytest
from src.config_manager import ConfigManager
from src.exceptions import ConfigError
def make_manager(tmp_path, config=None, secrets=None):
"""A ConfigManager over tmp_path files, template migration neutralized."""
config_file = tmp_path / "config.json"
secrets_file = tmp_path / "config_secrets.json"
config_file.write_text(json.dumps(config if config is not None else {}))
if secrets is not None:
secrets_file.write_text(json.dumps(secrets))
manager = ConfigManager(config_path=str(config_file),
secrets_path=str(secrets_file))
# Point the (CWD-relative) template at nothing so migration never runs —
# these tests assert exact on-disk contents.
manager.template_path = str(tmp_path / "no-template.json")
return manager
class TestLoadMergesSecrets:
def test_secrets_deep_merged_into_config(self, tmp_path):
manager = make_manager(
tmp_path,
config={"weather": {"city": "Austin"}, "timezone": "UTC"},
secrets={"weather": {"api_key": "s3cret"}},
)
loaded = manager.load_config()
assert loaded["weather"] == {"city": "Austin", "api_key": "s3cret"}
assert loaded["timezone"] == "UTC"
def test_secret_scalar_overrides_config_value(self, tmp_path):
manager = make_manager(
tmp_path,
config={"weather": {"api_key": "YOUR_API_KEY"}},
secrets={"weather": {"api_key": "real-key"}},
)
assert manager.load_config()["weather"]["api_key"] == "real-key"
def test_missing_secrets_file_loads_config_fine(self, tmp_path):
manager = make_manager(tmp_path, config={"timezone": "UTC"})
assert manager.load_config() == {"timezone": "UTC"}
def test_corrupt_secrets_file_loads_config_without_secrets(self, tmp_path):
manager = make_manager(tmp_path, config={"timezone": "UTC"})
(tmp_path / "config_secrets.json").write_text("{not json")
loaded = manager.load_config()
assert loaded["timezone"] == "UTC"
class TestSaveStripsSecrets:
def test_round_trip_keeps_secrets_out_of_config_json(self, tmp_path):
manager = make_manager(
tmp_path,
config={"weather": {"city": "Austin"}},
secrets={"weather": {"api_key": "s3cret"}},
)
loaded = manager.load_config()
assert loaded["weather"]["api_key"] == "s3cret" # merged in memory
manager.save_config(loaded)
on_disk = json.loads((tmp_path / "config.json").read_text())
assert "api_key" not in on_disk.get("weather", {})
assert on_disk["weather"]["city"] == "Austin"
# In-memory config still carries the secret for runtime use.
assert manager.config["weather"]["api_key"] == "s3cret"
def test_group_dropped_when_only_secrets_remain(self, tmp_path):
# _strip_secrets_recursive drops a group entirely when nothing
# non-secret is left in it.
manager = make_manager(
tmp_path,
config={},
secrets={"weather": {"api_key": "s3cret"}},
)
manager.save_config({"weather": {"api_key": "s3cret"}, "timezone": "UTC"})
on_disk = json.loads((tmp_path / "config.json").read_text())
assert on_disk == {"timezone": "UTC"}
def test_scalar_secret_key_stripped_at_top_level(self, tmp_path):
manager = make_manager(tmp_path, config={}, secrets={"token": "t"})
manager.save_config({"token": "t", "timezone": "UTC"})
on_disk = json.loads((tmp_path / "config.json").read_text())
assert on_disk == {"timezone": "UTC"}
def test_corrupt_secrets_file_refuses_save_no_plaintext_leak(self, tmp_path):
# Regression guard: when the secrets file exists but is corrupt at
# save time, stripping is impossible — the save must raise instead of
# writing the merged secrets into config.json in plaintext (the
# historical behavior).
manager = make_manager(
tmp_path,
config={"weather": {"city": "Austin"}},
secrets={"weather": {"api_key": "s3cret"}},
)
loaded = manager.load_config()
(tmp_path / "config_secrets.json").write_text("{corrupt")
with pytest.raises(ConfigError):
manager.save_config(loaded)
# On-disk config untouched: no secret leaked.
on_disk = json.loads((tmp_path / "config.json").read_text())
assert "api_key" not in on_disk.get("weather", {})
def test_corrupt_secrets_file_refuses_atomic_save_too(self, tmp_path):
# Same refusal on the atomic save path, which shared the leak.
manager = make_manager(
tmp_path,
config={"weather": {"city": "Austin"}},
secrets={"weather": {"api_key": "s3cret"}},
)
loaded = manager.load_config()
(tmp_path / "config_secrets.json").write_text("{corrupt")
with pytest.raises(ConfigError):
manager.save_config_atomic(loaded)
on_disk = json.loads((tmp_path / "config.json").read_text())
assert "api_key" not in on_disk.get("weather", {})
class TestLoadFastPath:
def test_unchanged_files_return_cached_dict(self, tmp_path):
manager = make_manager(tmp_path, config={"timezone": "UTC"})
first = manager.load_config()
second = manager.load_config()
assert second is first # same aliased dict, no re-read
def test_touching_secrets_file_invalidates_cache(self, tmp_path):
manager = make_manager(
tmp_path,
config={"weather": {}},
secrets={"weather": {"api_key": "old"}},
)
assert manager.load_config()["weather"]["api_key"] == "old"
secrets_file = tmp_path / "config_secrets.json"
secrets_file.write_text(json.dumps({"weather": {"api_key": "new"}}))
# Force a different mtime_ns in case the write landed within the
# filesystem's timestamp granularity.
os.utime(secrets_file, ns=(1, 1))
assert manager.load_config()["weather"]["api_key"] == "new"
def test_same_mtime_same_size_change_served_stale(self, tmp_path):
# Characterized fast-path blind spot: the signature is (mtime_ns,
# size) only, so a same-length content swap with a forged identical
# mtime is not detected. Real writes bump mtime_ns, so this is
# acceptable — but it is a contract worth pinning.
manager = make_manager(tmp_path, config={"timezone": "AAA"})
config_file = tmp_path / "config.json"
os.utime(config_file, ns=(1_000_000_000, 1_000_000_000))
manager._loaded_sig = None
first = manager.load_config()
assert first["timezone"] == "AAA"
config_file.write_text(json.dumps({"timezone": "BBB"})) # same length
os.utime(config_file, ns=(1_000_000_000, 1_000_000_000))
assert manager.load_config()["timezone"] == "AAA" # stale, by design
-167
View File
@@ -1,167 +0,0 @@
"""
Drift guard: three components independently answer "where is plugin X?" and
their answers must stay coherent plus the `.standalone-backup-` naming
contract that install/rollback shares with discovery.
The three resolvers:
1. PluginManager._scan_directory_for_plugins scans ONLY the configured dir.
2. PluginStoreManager._find_plugin_path configured dir, then a sibling
`plugins/` fallback derived from the configured dir's parent.
3. SchemaManager.get_schema_path configured dir, then project-root
`plugins/`, then `plugin-repos/`, then case-insensitive scans.
The divergence is characterized (a plugin visible to the store/schema
fallbacks but invisible to discovery is a real support-issue shape) so any
change to the fallback chains is a deliberate one.
The `.standalone-backup-` contract: store_manager renames a plugin dir aside
with that substring during install/rollback; discovery MUST skip such dirs
or a half-finished install would surface a ghost plugin. The substring is
duplicated as a literal in both files this test breaks if either side
changes it unilaterally.
"""
import json
import logging
import threading
from pathlib import Path
import pytest
from src.plugin_system.plugin_manager import PluginManager
from src.plugin_system.schema_manager import SchemaManager
from src.plugin_system.store_manager import PluginStoreManager
def _write_plugin(base: Path, plugin_id: str, dir_name: str = None):
plugin_dir = base / (dir_name or plugin_id)
plugin_dir.mkdir(parents=True)
(plugin_dir / "manifest.json").write_text(json.dumps({
"id": plugin_id, "name": plugin_id, "version": "1.0.0",
}))
(plugin_dir / "config_schema.json").write_text(json.dumps({
"type": "object", "properties": {"enabled": {"type": "boolean"}},
}))
return plugin_dir
def _scanner():
"""A PluginManager stripped to just its discovery machinery — the full
constructor wires config/schema/health managers this test doesn't need."""
pm = object.__new__(PluginManager)
pm.logger = logging.getLogger("test_discovery_path_contract")
pm._discovery_lock = threading.Lock()
pm.plugin_manifests = {}
pm.plugin_directories = {}
return pm
class TestResolversAgreeOnConfiguredDir:
def test_all_three_find_a_plugin_in_the_configured_dir(self, tmp_path):
plugins_dir = tmp_path / "plugin-repos"
plugin_dir = _write_plugin(plugins_dir, "demo-plugin")
found = _scanner()._scan_directory_for_plugins(plugins_dir)
assert found == ["demo-plugin"]
store = PluginStoreManager(
plugins_dir=str(plugins_dir),
uninstalled_registry_path=str(tmp_path / "uninstalled.json"))
assert store._find_plugin_path("demo-plugin") == plugin_dir
schema = SchemaManager(plugins_dir=plugins_dir, project_root=tmp_path)
assert schema.get_schema_path("demo-plugin") == \
plugin_dir / "config_schema.json"
class TestFallbackDivergence:
def test_plugin_only_in_plugins_dir_fallback(self, tmp_path):
"""Characterized divergence: configured dir is plugin-repos/, but the
plugin sits in a sibling plugins/. The store and schema fallbacks
find it; discovery does NOT so the plugin is installable/
configurable but never loads. Pinned so a change to any fallback
chain shows up here."""
configured = tmp_path / "plugin-repos"
configured.mkdir()
legacy_dir = _write_plugin(tmp_path / "plugins", "legacy-plugin")
# Discovery: invisible.
assert _scanner()._scan_directory_for_plugins(configured) == []
# Store fallback: visible (parent-of-configured / 'plugins').
store = PluginStoreManager(
plugins_dir=str(configured),
uninstalled_registry_path=str(tmp_path / "uninstalled.json"))
assert store._find_plugin_path("legacy-plugin") == legacy_dir
# Schema fallback: visible (project_root / 'plugins').
schema = SchemaManager(plugins_dir=configured, project_root=tmp_path)
assert schema.get_schema_path("legacy-plugin") == \
legacy_dir / "config_schema.json"
def test_schema_manager_probes_plugins_before_plugin_repos(self, tmp_path):
# Documented order (also in CLAUDE.md): plugins/ wins over
# plugin-repos/ when the same id exists in both.
in_plugins = _write_plugin(tmp_path / "plugins", "dupe")
_write_plugin(tmp_path / "plugin-repos", "dupe")
schema = SchemaManager(plugins_dir=None, project_root=tmp_path)
assert schema.get_schema_path("dupe") == \
in_plugins / "config_schema.json"
def test_schema_manager_case_insensitive_fallback(self, tmp_path):
plugin_dir = _write_plugin(tmp_path / "plugins", "MyPlugin",
dir_name="MyPlugin")
schema = SchemaManager(plugins_dir=None, project_root=tmp_path)
assert schema.get_schema_path("myplugin") == \
plugin_dir / "config_schema.json"
class TestStandaloneBackupContract:
def test_discovery_skips_backup_dirs(self, tmp_path):
plugins_dir = tmp_path / "plugins"
_write_plugin(plugins_dir, "real-plugin")
# A rollback-in-progress dir with a valid manifest must NOT surface.
_write_plugin(plugins_dir, "real-plugin",
dir_name="real-plugin.standalone-backup-migrating")
found = _scanner()._scan_directory_for_plugins(plugins_dir)
assert found == ["real-plugin"]
def test_backup_substring_literal_matches_across_files(self):
"""The substring is duplicated in plugin_manager (skip check) and
store_manager (rename-aside names). If either side changes it, the
other silently stops honoring the contract this test is the
tripwire."""
root = Path(__file__).resolve().parents[1]
pm_text = (root / "src/plugin_system/plugin_manager.py").read_text()
sm_text = (root / "src/plugin_system/store_manager.py").read_text()
assert "'.standalone-backup-'" in pm_text.replace('"', "'")
assert ".standalone-backup-" in sm_text
class TestSkinTargetResolution:
def _store(self, tmp_path):
return PluginStoreManager(
plugins_dir=str(tmp_path / "plugins"),
uninstalled_registry_path=str(tmp_path / "uninstalled.json"))
def test_valid_skin_id_resolves_inside_skins_dir(self, tmp_path):
from src.skin_system import skin_runtime
store = self._store(tmp_path)
target = store._resolve_skin_target("my-skin")
assert target is not None
assert target.parent == skin_runtime.get_skins_directory().resolve()
@pytest.mark.parametrize("bad_id", [
"../evil",
"..",
"a/../../etc",
"/etc/passwd",
"skin/../../outside",
"",
None,
123,
])
def test_traversal_and_malformed_ids_rejected(self, tmp_path, bad_id):
store = self._store(tmp_path)
assert store._resolve_skin_target(bad_id) is None
+7
View File
@@ -15,6 +15,13 @@ class TestDisplayControllerInitialization:
assert test_display_controller.plugin_manager is not None
assert test_display_controller.available_modes == []
@pytest.mark.skip(reason="No assertions; init logic is covered by test_init_success and fixture setup")
def test_plugin_discovery_and_loading(self, test_display_controller):
"""Test plugin discovery and loading during initialization."""
pm = test_display_controller.plugin_manager
pm.discover_plugins.return_value = ["plugin1", "plugin2"]
pm.get_plugin.return_value = MagicMock()
class TestDisplayControllerModeRotation:
"""Test display mode rotation logic."""
@@ -11,18 +11,9 @@ orphaning VegasModeCoordinator.mark_plugin_updated() -- it has had zero
callers since.
"""
import os
from typing import Dict, List, Optional
from unittest.mock import MagicMock
# display_controller imports display_manager, which imports the hardware
# rgbmatrix module unless EMULATOR=true was set before import. Use the
# emulator (same convention as test_display_dirty_tracking.py and
# test/plugins/conftest.py) so this file collects on machines without the
# hardware library — and so display_manager gets the emulator binding no
# matter which test module imports it first.
os.environ.setdefault("EMULATOR", "true")
from src.display_controller import DisplayController
+9 -26
View File
@@ -1,15 +1,6 @@
import os
import pytest
from unittest.mock import MagicMock, patch
from PIL import ImageDraw
# display_manager imports the hardware rgbmatrix module at import time unless
# EMULATOR=true. Use the emulator (same convention as
# test_display_dirty_tracking.py) so this file collects standalone instead of
# relying on collection order — the tests below patch RGBMatrix/
# RGBMatrixOptions explicitly, so the underlying binding doesn't matter here.
os.environ.setdefault("EMULATOR", "true")
from src.display_manager import DisplayManager
@pytest.fixture
@@ -86,26 +77,18 @@ class TestDisplayManagerDrawing:
assert dm.matrix.Clear.called
def test_draw_text(self, test_config, mock_rgb_matrix):
"""Text drawn through draw_text must actually light pixels."""
from PIL import Image, ImageDraw, ImageFont
import src.display_manager as dm_mod
"""Test text drawing."""
with patch.dict('os.environ', {'EMULATOR': 'false'}):
DisplayManager._instance = None
dm = DisplayManager(test_config, suppress_test_pattern=True)
# The fixture replaces the module's freetype with a MagicMock,
# which breaks draw_text's isinstance(font, freetype.Face) check
# (and silently swallows the draw). Give the mock a real class so
# isinstance works and the PIL path is taken.
dm_mod.freetype.Face = type("_FakeFace", (), {})
# Start from a known-black canvas so the assertion below can only
# pass if draw_text itself lit something.
dm.image = Image.new('RGB', (dm.width, dm.height))
dm.draw = ImageDraw.Draw(dm.image)
dm = DisplayManager(test_config)
dm.draw_text("Test", 0, 0, font=ImageFont.load_default())
# Mock font
font = MagicMock()
assert dm.image.convert("L").getbbox() is not None, \
"draw_text lit no pixels"
dm.draw_text("Test", 0, 0, font)
# Verify draw_text was called (DisplayManager uses freetype/PIL)
# The actual implementation uses freetype or PIL, not graphics module
assert True # draw_text should execute without error
def test_draw_image(self, test_config, mock_rgb_matrix):
"""Test image drawing."""
+66 -110
View File
@@ -1,126 +1,82 @@
"""
Tests for src/font_manager.py FontManager loading, caching, fallback,
and BDF handling, exercised against the real bundled fonts in assets/fonts.
This file replaces an earlier version whose tests were try/except blocks
ending in `assert True` they executed the code but could not fail. Every
test here asserts observable behavior: returned font types, cache identity,
fallback selection, and BDF native-size reading.
"""
import freetype
import pytest
from PIL import ImageFont
from unittest.mock import patch
from src.font_manager import FontManager
@pytest.fixture
def fm():
"""A FontManager over the real assets/fonts catalog."""
return FontManager({})
def mock_freetype():
"""Mock freetype module."""
with patch('src.font_manager.freetype') as mock_freetype:
yield mock_freetype
class TestFontManager:
"""Test FontManager functionality."""
class TestCatalog:
def test_bundled_common_fonts_are_registered(self, fm):
# These aliases are hardcoded in FontManager.common_fonts and the
# files ship in assets/fonts — all three must resolve.
for family in ("press_start", "four_by_six", "five_by_seven"):
assert family in fm.font_catalog, f"{family} missing from catalog"
def test_init(self, test_config, mock_freetype):
"""Test FontManager initialization."""
# Ensure BDF files exist check passes
with patch('os.path.exists', return_value=True):
fm = FontManager(test_config)
assert fm.config == test_config
assert hasattr(fm, 'font_cache') # FontManager uses font_cache, not fonts
def test_catalog_families_are_lowercase_filenames(self, fm):
# _scan_fonts_directory lowercases the filename stem.
assert all(name == name.lower() for name in fm.font_catalog)
def test_get_font_success(self, test_config, mock_freetype):
"""Test successful font loading."""
with patch('os.path.exists', return_value=True), \
patch('os.path.join', side_effect=lambda *args: "/".join(args)):
fm = FontManager(test_config)
class TestGetFont:
def test_ttf_family_returns_usable_pil_font(self, fm):
font = fm.get_font("press_start", 8)
assert isinstance(font, ImageFont.FreeTypeFont)
# Usable: it can measure text.
bbox = font.getbbox("Hi")
assert bbox[2] > bbox[0]
# Request a font (get_font requires family and size_px)
# Font may be None if font file doesn't exist in test, that's ok
try:
font = fm.get_font("small", 12) # family and size_px required
# Just verify the method can be called
assert True # FontManager.get_font() executed
except (TypeError, AttributeError):
# If method signature doesn't match, that's ok for now
assert True
def test_bdf_family_returns_freetype_face(self, fm):
font = fm.get_font("five_by_seven", 7)
assert isinstance(font, freetype.Face)
def test_get_font_missing_file(self, test_config, mock_freetype):
"""Test handling of missing font file."""
with patch('os.path.exists', return_value=False):
fm = FontManager(test_config)
def test_repeat_call_returns_cached_identity(self, fm):
first = fm.get_font("press_start", 8)
hits_before = fm.performance_stats["cache_hits"]
second = fm.get_font("press_start", 8)
assert second is first
assert fm.performance_stats["cache_hits"] == hits_before + 1
# Request a font where file doesn't exist
# get_font requires family and size_px
try:
font = fm.get_font("small", 12) # family and size_px required
# Font may be None if file doesn't exist, that's ok
assert True # Method executed
except (TypeError, AttributeError):
assert True # Method signature may differ
def test_different_sizes_get_distinct_cache_entries(self, fm):
small = fm.get_font("press_start", 8)
large = fm.get_font("press_start", 16)
assert small is not large
assert "press_start_8" in fm.font_cache
assert "press_start_16" in fm.font_cache
def test_get_font_invalid_name(self, test_config, mock_freetype):
"""Test requesting invalid font name."""
with patch('os.path.exists', return_value=True):
fm = FontManager(test_config)
def test_unknown_family_falls_back_to_default_without_raising(self, fm):
failed_before = fm.performance_stats["failed_loads"]
font = fm.get_font("no-such-family", 10)
# The documented fallback is PIL's default font (whose concrete type
# varies across Pillow versions), recorded as a failed load. It must
# still be usable for measurement.
assert type(font) is type(ImageFont.load_default())
assert font.getbbox("Hi")[2] > 0
assert fm.performance_stats["failed_loads"] == failed_before + 1
# Request unknown font (get_font requires family and size_px)
try:
font = fm.get_font("nonexistent_font", 12) # family and size_px required
# Font may be None for unknown font, that's ok
assert True # Method executed
except (TypeError, AttributeError):
assert True # Method signature may differ
def test_corrupt_font_file_falls_back_to_default(self, fm, tmp_path):
bad = tmp_path / "broken.ttf"
bad.write_text("this is not a font file")
fm.font_catalog["broken"] = str(bad)
failed_before = fm.performance_stats["failed_loads"]
font = fm.get_font("broken", 10)
assert type(font) is type(ImageFont.load_default())
assert font.getbbox("Hi")[2] > 0
assert fm.performance_stats["failed_loads"] == failed_before + 1
def test_get_font_with_fallback(self, test_config, mock_freetype):
"""Test font loading with fallback."""
# FontManager.get_font() requires family and size_px
# This test verifies the method exists and can be called
fm = FontManager(test_config)
assert hasattr(fm, 'get_font')
assert True # Method exists, implementation may vary
def test_load_custom_font(self, test_config, mock_freetype):
"""Test loading a custom font file directly."""
with patch('os.path.exists', return_value=True):
fm = FontManager(test_config)
class TestBdfNativeSize:
def test_five_by_seven_reports_native_height(self, fm):
# 5x7.bdf declares a 7px strike; requesting other sizes still renders
# the native size, so callers need this to know the truth.
assert fm.get_native_bdf_size("five_by_seven") == 7
def test_ttf_family_has_no_native_size(self, fm):
assert fm.get_native_bdf_size("press_start") is None
def test_unknown_family_has_no_native_size(self, fm):
assert fm.get_native_bdf_size("no-such-family") is None
class TestMeasureText:
def test_ttf_measurement_is_positive_and_cached(self, fm):
font = fm.get_font("press_start", 8)
width, height, baseline = fm.measure_text("SCORE", font)
assert width > 0 and height > 0
# Cached: same result object path on second call.
assert fm.measure_text("SCORE", font) == (width, height, baseline)
assert ("SCORE", id(font)) in fm.metrics_cache
def test_longer_text_measures_wider(self, fm):
font = fm.get_font("press_start", 8)
short, _, _ = fm.measure_text("AB", font)
long, _, _ = fm.measure_text("ABCD", font)
assert long > short
class TestCacheLifecycle:
def test_clear_cache_empties_both_caches(self, fm):
font = fm.get_font("press_start", 8)
fm.measure_text("X", font)
assert fm.font_cache and fm.metrics_cache
fm.clear_cache()
assert not fm.font_cache
assert not fm.metrics_cache
def test_reload_config_bumps_generation_and_clears(self, fm):
fm.get_font("press_start", 8)
gen_before = fm.cache_generation
fm.reload_config({})
assert fm.cache_generation == gen_before + 1
assert not fm.font_cache
# FontManager uses add_font or get_font, not load_font
# Just verify the manager can handle font operations
# The actual method depends on implementation
assert hasattr(fm, 'get_font') or hasattr(fm, 'add_font')
+114
View File
@@ -0,0 +1,114 @@
"""A plugin must be findable in the registry by the id it calls itself.
Four shipped plugins have a registry ``id`` that differs from the ``id`` in
their own ``manifest.json``:
directory / manifest.json id registry id
ledmatrix-weather weather
ledmatrix-stocks stocks
ledmatrix-music music
ledmatrix-leaderboard leaderboard
The installer already knows about this: it deliberately names the install
directory after the *manifest* id (store_manager, "Use manifest ID for
directory name"), and warns when the two disagree. So on disk, in
``config.json`` and in a backup manifest, these plugins are called
``ledmatrix-weather``. Only the registry calls them ``weather``.
Nothing resolved that in reverse. Asking the store to install
``ledmatrix-weather`` -- which is exactly what restoring a backup does --
failed with "Plugin not found in registry", and four enabled plugins went
missing from a restored device with no error surfaced to the user.
Renaming the registry ids would orphan existing ``plugin_state.json`` entries
keyed on the old ones, so the lookup resolves ``plugin_path`` instead: the
registry already records ``plugins/ledmatrix-weather``, which is unambiguous
and needs no published identity to change.
"""
import os
import sys
from typing import Any, Dict, List, Optional
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from src.plugin_system.store_manager import PluginStoreManager # noqa: E402
# Shaped like the real registry: id and plugin_path basename disagree for the
# first entry, agree for the second.
REGISTRY: Dict[str, List[Dict[str, Any]]] = {
"plugins": [
{
"id": "weather",
"name": "Weather",
"plugin_path": "plugins/ledmatrix-weather",
"repo": "https://github.com/ChuckBuilds/ledmatrix-plugins",
},
{
"id": "ledmatrix-flights",
"name": "Flights",
"plugin_path": "plugins/ledmatrix-flights",
"repo": "https://github.com/ChuckBuilds/ledmatrix-plugins",
},
{
"id": "third-party",
"name": "Third Party",
"plugin_path": "",
"repo": "https://github.com/someone/thing",
},
]
}
@pytest.fixture
def store(monkeypatch: pytest.MonkeyPatch) -> PluginStoreManager:
manager = PluginStoreManager.__new__(PluginStoreManager)
monkeypatch.setattr(manager, "fetch_registry", lambda *a, **k: REGISTRY, raising=False)
return manager
def _ids(entry: Optional[Dict[str, Any]]) -> Optional[str]:
return entry.get("id") if entry else None
class TestRegistryLookupByManifestId:
def test_get_plugin_info_resolves_manifest_id(self, store: PluginStoreManager) -> None:
"""get_plugin_info() delegates to the same lookup as get_registry_info()."""
assert (
_ids(store.get_plugin_info("ledmatrix-weather", fetch_latest_from_github=False))
== "weather"
)
def test_exact_registry_id_still_resolves(self, store: PluginStoreManager) -> None:
assert _ids(store.get_registry_info("weather")) == "weather"
def test_manifest_id_resolves_via_plugin_path(self, store: PluginStoreManager) -> None:
"""The case that broke restore: asked by the name on disk."""
assert _ids(store.get_registry_info("ledmatrix-weather")) == "weather", (
"a plugin installed as 'ledmatrix-weather' could not be found in a "
"registry that lists it under plugin_path plugins/ledmatrix-weather")
def test_matching_id_and_path_unaffected(self, store: PluginStoreManager) -> None:
assert _ids(store.get_registry_info("ledmatrix-flights")) == "ledmatrix-flights"
def test_unknown_plugin_still_returns_none(self, store: PluginStoreManager) -> None:
assert store.get_registry_info("no-such-plugin") is None
def test_empty_plugin_path_is_not_a_wildcard(self, store: PluginStoreManager) -> None:
"""Third-party entries carry plugin_path "" — that must not match ""."""
assert store.get_registry_info("") is None
def test_exact_id_wins_over_a_path_match(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""If some other entry's path collides with a real id, id wins."""
registry = {
"plugins": [
{"id": "decoy", "plugin_path": "plugins/weather"},
{"id": "weather", "plugin_path": "plugins/ledmatrix-weather"},
]
}
manager = PluginStoreManager.__new__(PluginStoreManager)
monkeypatch.setattr(manager, "fetch_registry", lambda *a, **k: registry, raising=False)
assert _ids(manager.get_registry_info("weather")) == "weather"
-104
View File
@@ -1,104 +0,0 @@
"""
Tests for SchemaManager.merge_with_defaults the merge every plugin config
passes through at load time (schema defaults + user config, with None
replacement). A regression here silently changes every plugin's effective
config, so the exact branch behavior is pinned, including the
characterized type-mismatch cases.
"""
import pytest
from src.plugin_system.schema_manager import SchemaManager
@pytest.fixture
def sm(tmp_path):
return SchemaManager(plugins_dir=str(tmp_path))
class TestBasicMerge:
def test_missing_keys_filled_from_defaults(self, sm):
merged = sm.merge_with_defaults(
{"city": "Austin"}, {"city": "NYC", "units": "metric"})
assert merged == {"city": "Austin", "units": "metric"}
def test_present_keys_preserved(self, sm):
merged = sm.merge_with_defaults({"enabled": False}, {"enabled": True})
assert merged["enabled"] is False
def test_nested_three_level_merge(self, sm):
config = {"a": {"b": {"c": 1}}}
defaults = {"a": {"b": {"c": 0, "d": 2}, "e": 3}}
merged = sm.merge_with_defaults(config, defaults)
assert merged == {"a": {"b": {"c": 1, "d": 2}, "e": 3}}
def test_inputs_not_mutated(self, sm):
config = {"a": {"b": 1}}
defaults = {"a": {"b": 0, "c": 2}, "d": 3}
sm.merge_with_defaults(config, defaults)
assert config == {"a": {"b": 1}}
assert defaults == {"a": {"b": 0, "c": 2}, "d": 3}
def test_merged_values_are_copies_not_aliases(self, sm):
config = {"teams": ["DAL"]}
merged = sm.merge_with_defaults(config, {"teams": []})
merged["teams"].append("HOU")
assert config["teams"] == ["DAL"] # user's list untouched
class TestNoneReplacement:
def test_none_replaced_by_default(self, sm):
merged = sm.merge_with_defaults({"units": None}, {"units": "metric"})
assert merged["units"] == "metric"
def test_falsey_non_none_values_kept(self, sm):
merged = sm.merge_with_defaults(
{"enabled": False, "count": 0, "label": ""},
{"enabled": True, "count": 5, "label": "x"},
)
assert merged == {"enabled": False, "count": 0, "label": ""}
def test_nested_none_replaced(self, sm):
merged = sm.merge_with_defaults(
{"style": {"color": None}}, {"style": {"color": "red"}})
assert merged["style"]["color"] == "red"
def test_none_with_no_default_stays_none(self, sm):
merged = sm.merge_with_defaults({"extra": None}, {})
assert merged["extra"] is None
def test_none_replaced_by_dict_default_is_a_copy(self, sm):
defaults = {"style": {"color": "red"}}
merged = sm.merge_with_defaults({"style": None}, defaults)
assert merged["style"] == {"color": "red"}
merged["style"]["color"] = "blue"
assert defaults["style"]["color"] == "red"
class TestTypeMismatches:
def test_user_scalar_over_dict_default_wins(self, sm):
# Characterized: a scalar user value replaces a dict default outright.
merged = sm.merge_with_defaults(
{"style": "compact"}, {"style": {"color": "red"}})
assert merged["style"] == "compact"
def test_user_dict_over_scalar_default_wins(self, sm):
merged = sm.merge_with_defaults(
{"style": {"color": "red"}}, {"style": "compact"})
assert merged["style"] == {"color": "red"}
def test_arrays_replaced_wholesale_not_merged(self, sm):
# Pinned contract: arrays never element-merge — the user's array is
# the whole answer, even when shorter than the default.
merged = sm.merge_with_defaults(
{"teams": ["DAL"]}, {"teams": ["NYG", "PHI", "WAS"]})
assert merged["teams"] == ["DAL"]
def test_empty_user_array_beats_default(self, sm):
merged = sm.merge_with_defaults({"teams": []}, {"teams": ["NYG"]})
assert merged["teams"] == []
def test_extra_user_keys_survive(self, sm):
# Keys with no schema default pass through untouched.
merged = sm.merge_with_defaults({"custom_flag": 7}, {"known": 1})
assert merged == {"known": 1, "custom_flag": 7}
-119
View File
@@ -450,122 +450,3 @@ class TestExampleSkin:
ctx = skin_runtime.build_context(host, game, size=size)
assert getattr(skin, f"render_{mode}")(ctx, game) is True
assert ctx.canvas.convert("L").getbbox() is not None
class TestRenderSkinCard:
"""render_skin_card (vegas cards) shares _render_game's 3-strike counter.
Both paths reset the counter on success transient failures must not
accumulate across a session and disable a working skin.
"""
def _probe(self, skin):
from src.base_classes.sports import SportsCore
probe = _FallbackProbe(skin)
probe.render_skin_card = (
lambda game, size: SportsCore.render_skin_card(probe, game, size))
return probe
def test_vegas_card_returned_when_skin_provides_one(self):
card_img = Image.new("RGB", (96, 32), (0, 0, 255))
class CardSkin(ScoreboardSkin):
def render_vegas_card(self, ctx, game):
return card_img
probe = self._probe(CardSkin({}, {}))
assert probe.render_skin_card({}, (96, 32)) is card_img
def test_vegas_card_none_falls_through_to_mode_renderer(self):
class ModeOnlySkin(ScoreboardSkin):
def render_live(self, ctx, game):
ctx.draw.rectangle([0, 0, 5, 5], fill=(255, 0, 0))
return True
probe = self._probe(ModeOnlySkin({}, {}))
card = probe.render_skin_card({}, (96, 32))
assert card is not None
assert card.size == (96, 32)
assert card.convert("L").getbbox() is not None
def test_skin_declining_returns_none(self):
probe = self._probe(ScoreboardSkin({}, {})) # all renders -> False
assert probe.render_skin_card({}, (96, 32)) is None
assert probe._skin_failures == 0 # declining is not a failure
def test_no_skin_returns_none(self):
probe = self._probe(None)
assert probe.render_skin_card({}, (96, 32)) is None
def test_card_failures_count_toward_shared_disable(self):
class BrokenCardSkin(ScoreboardSkin):
calls = 0
def render_vegas_card(self, ctx, game):
BrokenCardSkin.calls += 1
raise ValueError("kaboom")
probe = self._probe(BrokenCardSkin({}, {}))
for _ in range(5):
assert probe.render_skin_card({}, (96, 32)) is None
# Skin stopped being consulted after the 3rd failure...
assert BrokenCardSkin.calls == 3
assert probe._skin_failures == 3
# ...and the shared counter also disables _render_game's skin path.
probe._render_game({"status_text": "Q1"})
assert probe.builtin_calls == 1
assert BrokenCardSkin.calls == 3 # not consulted again
def test_card_success_resets_strikes(self):
"""A successful card render clears accumulated strikes (mirroring
_render_game) 2 failures + a success + 1 failure leaves the skin
enabled with a single strike, instead of disabling it."""
card_img = Image.new("RGB", (96, 32), (0, 0, 255))
class FlakyCardSkin(ScoreboardSkin):
fail = True
def render_vegas_card(self, ctx, game):
if FlakyCardSkin.fail:
raise ValueError("kaboom")
return card_img
probe = self._probe(FlakyCardSkin({}, {}))
FlakyCardSkin.fail = True
probe.render_skin_card({}, (96, 32))
probe.render_skin_card({}, (96, 32))
assert probe._skin_failures == 2
FlakyCardSkin.fail = False
assert probe.render_skin_card({}, (96, 32)) is card_img
assert probe._skin_failures == 0 # success cleared the strikes
FlakyCardSkin.fail = True
probe.render_skin_card({}, (96, 32))
assert probe._skin_failures == 1 # counting from the reset state
FlakyCardSkin.fail = False
assert probe.render_skin_card({}, (96, 32)) is card_img # still enabled
def test_card_success_via_mode_renderer_also_resets_strikes(self):
"""The fallthrough path (render_vegas_card None -> mode renderer
True) resets the counter as well."""
class ModeOnlySkin(ScoreboardSkin):
def render_live(self, ctx, game):
ctx.draw.rectangle([0, 0, 5, 5], fill=(255, 0, 0))
return True
probe = self._probe(ModeOnlySkin({}, {}))
probe._skin_failures = 2
assert probe.render_skin_card({}, (96, 32)) is not None
assert probe._skin_failures == 0
def test_render_game_success_also_resets_strikes(self):
"""Same reset contract on the display path, for symmetry."""
class GoodSkin(ScoreboardSkin):
def render_live(self, ctx, game):
return True
probe = _FallbackProbe(GoodSkin({}, {}))
probe._skin_failures = 2
probe._render_game({"status_text": "Q1"})
assert probe._skin_failures == 0
-175
View File
@@ -1,175 +0,0 @@
"""
Drift guard for version comparison.
There is now ONE shared "should this plugin update?" comparator
`src.plugin_system.compatibility.is_update_available` used by both the web
UI's update badge (`api_v3._is_plugin_update_available`) and the store's
`update_plugin` reinstall decision, so the badge and the actual reinstall
can never disagree. (Historically the store used raw string equality, which
reinstalled over cosmetic differences like "v1.2.0" vs "1.2.0" and even
DOWNGRADED locally-ahead plugins; this file's tests killed that.)
Two other version parsers legitimately remain and are pinned here so they
don't drift: `compatibility.parse_semver` (the install-compatibility gate,
range-spec oriented) and `skin_runtime._major` (skin API major gate).
"""
import json
from unittest.mock import patch
import pytest
from packaging.version import parse as pkg_parse
from src.plugin_system.compatibility import is_update_available, parse_semver
from src.skin_system.skin_runtime import _major
from src.plugin_system.store_manager import PluginStoreManager
from web_interface.blueprints.api_v3 import _is_plugin_update_available
# (installed, registry) -> update available?
CASES = [
(("1.2.0", "1.2.0"), False), # identical
(("v1.2.0", "1.2.0"), False), # cosmetic v-prefix, semantically equal
(("1.2", "1.2.0"), False), # short form, semantically equal
(("1.2.0", "1.2.0-rc1"), False), # rc of same release is not newer
(("1.2.0", "1.3.0"), True), # registry genuinely newer
(("2.0.0", "1.9.0"), False), # locally ahead — never downgrade
(("abc.def", "1.0.0"), True), # unparseable — surface the mismatch
(("", "1.0.0"), False), # missing either side — nothing to do
(("1.0.0", ""), False),
]
class TestSharedComparatorMalformedInputs:
def test_truthy_non_string_surfaces_mismatch(self):
# A malformed manifest can carry version as a number; packaging would
# raise TypeError on it. The comparator must not raise.
assert is_update_available(1.2, "1.2.0") is True
assert is_update_available("1.2.0", 1.3) is True
def test_falsy_non_string_means_nothing_to_do(self):
assert is_update_available(None, "1.0.0") is False
assert is_update_available("1.0.0", None) is False
assert is_update_available(0, "1.0.0") is False
class TestSharedComparator:
@pytest.mark.parametrize("pair,expected", CASES)
def test_is_update_available(self, pair, expected):
installed, latest = pair
assert is_update_available(installed, latest) is expected
@pytest.mark.parametrize("pair,expected", CASES)
def test_api_v3_helper_agrees(self, pair, expected):
# The UI badge helper must be a pure alias of the shared comparator.
installed, latest = pair
assert _is_plugin_update_available(installed, latest) is expected
class TestStoreManagerUsesSharedComparator:
"""Drive update_plugin's real code path to its version check."""
def _store(self, tmp_path, local_version, registry_version):
plugin_dir = tmp_path / "plugins" / "demo-plugin"
plugin_dir.mkdir(parents=True)
(plugin_dir / "manifest.json").write_text(json.dumps({
"id": "demo-plugin", "version": local_version,
}))
store = PluginStoreManager(
plugins_dir=str(tmp_path / "plugins"),
uninstalled_registry_path=str(tmp_path / "uninstalled.json"),
)
registry_info = {
"id": "demo-plugin",
"repo": "https://github.com/example/ledmatrix-plugins",
"latest_version": registry_version,
}
return store, registry_info
def _run_update(self, store, registry_info):
with patch.object(store, "fetch_registry", return_value={"plugins": [registry_info]}), \
patch.object(store, "get_plugin_info", return_value=registry_info), \
patch.object(store, "_reinstall_with_rollback", return_value=True) as reinstall:
result = store.update_plugin("demo-plugin")
return result, reinstall
def test_equal_strings_skip_reinstall(self, tmp_path):
store, info = self._store(tmp_path, "1.2.0", "1.2.0")
result, reinstall = self._run_update(store, info)
assert result is True
reinstall.assert_not_called()
def test_v_prefix_equivalent_skips_reinstall(self, tmp_path):
# "v1.2.0" == "1.2.0" semantically — no pointless reinstall.
store, info = self._store(tmp_path, "v1.2.0", "1.2.0")
result, reinstall = self._run_update(store, info)
assert result is True
reinstall.assert_not_called()
def test_locally_ahead_version_is_never_downgraded(self, tmp_path):
# A plugin ahead of the registry (local dev build) must not be
# "updated" — that would be a downgrade.
store, info = self._store(tmp_path, "2.0.0", "1.9.0")
result, reinstall = self._run_update(store, info)
assert result is True
reinstall.assert_not_called()
def test_registry_newer_triggers_reinstall(self, tmp_path):
store, info = self._store(tmp_path, "1.2.0", "1.3.0")
result, reinstall = self._run_update(store, info)
reinstall.assert_called_once()
assert result is True
def test_unparseable_version_surfaces_via_reinstall(self, tmp_path):
# Direction unknowable → reconcile by reinstalling from the registry.
store, info = self._store(tmp_path, "abc.def", "1.0.0")
result, reinstall = self._run_update(store, info)
reinstall.assert_called_once()
assert result is True
def test_empty_local_version_follows_comparator_no_reinstall(self, tmp_path):
# The comparator says "nothing to do" for a missing version, and the
# store must agree with the UI badge — no reinstall.
store, info = self._store(tmp_path, "", "1.0.0")
result, reinstall = self._run_update(store, info)
assert result is True
reinstall.assert_not_called()
def test_empty_registry_version_follows_comparator_no_reinstall(self, tmp_path):
store, info = self._store(tmp_path, "1.0.0", "")
result, reinstall = self._run_update(store, info)
assert result is True
reinstall.assert_not_called()
class TestSkinRuntimeMajor:
def test_plain_versions(self):
assert _major("1.0.0") == 1
assert _major("2.1") == 2
def test_int_input_tolerated(self):
assert _major(2) == 2
def test_garbage_returns_none(self):
assert _major("garbage") is None
assert _major(None) is None
def test_v_prefix_not_tolerated(self):
# Unlike parse_semver, _major does NOT strip a leading 'v' —
# a skin.json declaring "v1.0.0" fails the API gate. Characterized
# so a manifest-format loosening elsewhere doesn't silently diverge.
assert _major("v1.0.0") is None
class TestParseSemverAgreesWithPackaging:
"""parse_semver and packaging must agree on ordering for plain X.Y.Z —
the region where the two ecosystems overlap and must never diverge."""
PLAIN = ["0.1.0", "1.0.0", "1.2.0", "1.2.3", "1.10.0", "2.0.0", "10.0.1"]
def test_pairwise_ordering_matches(self):
for a in self.PLAIN:
for b in self.PLAIN:
ours = parse_semver(a) < parse_semver(b)
theirs = pkg_parse(a) < pkg_parse(b)
assert ours == theirs, f"ordering diverges on ({a}, {b})"
-3
View File
@@ -393,9 +393,6 @@ class TestSystemAPI:
@patch('web_interface.blueprints.api_v3.subprocess')
def test_get_system_status(self, mock_subprocess, client):
"""Test getting system status."""
# The endpoint returns 503 without psutil, which is an optional
# runtime dependency (requirements-test.txt installs it for CI).
pytest.importorskip("psutil")
mock_result = MagicMock()
mock_result.stdout = 'active\n'
mock_result.returncode = 0
-232
View File
@@ -1,232 +0,0 @@
"""
Unit tests for the module-level helper functions in
web_interface/blueprints/api_v3.py.
These helpers back the plugin config save endpoint (the largest function in
the repo) and the store's update-available detection, but were previously
exercised only indirectly through full Flask route tests. Testing them
directly pins behavior that the routes rely on including a few
characterized quirks marked below.
"""
import sys
from pathlib import Path
from typing import Any, ClassVar, Dict
import pytest
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))
from web_interface.blueprints.api_v3 import ( # noqa: E402
_is_plugin_update_available,
_coerce_to_bool,
deep_merge,
_parse_form_value,
_get_schema_property,
_set_nested_value,
)
class TestIsPluginUpdateAvailable:
def test_equal_versions_no_update(self):
assert _is_plugin_update_available("1.2.0", "1.2.0") is False
def test_newer_registry_version_needs_update(self):
assert _is_plugin_update_available("1.2.0", "1.3.0") is True
def test_installed_ahead_of_registry_no_update(self):
# A locally modified plugin ahead of the registry must not be
# flagged — this is the whole point of semantic comparison here.
assert _is_plugin_update_available("2.0.0", "1.9.0") is False
def test_empty_versions_no_update(self):
assert _is_plugin_update_available("", "1.0.0") is False
assert _is_plugin_update_available("1.0.0", "") is False
assert _is_plugin_update_available("", "") is False
def test_v_prefix_parses_as_equal(self):
# packaging.version treats "v1.2.0" == "1.2.0" (PEP 440 tolerates the
# prefix), so no update is flagged. Contrast with store_manager's
# string-equality check — see test_version_comparison_consistency.py.
assert _is_plugin_update_available("v1.2.0", "1.2.0") is False
def test_two_part_version_parses_as_equal(self):
assert _is_plugin_update_available("1.2", "1.2.0") is False
def test_unparseable_version_surfaces_mismatch(self):
# Direction unknowable → surface the difference rather than hide a
# potential update.
assert _is_plugin_update_available("abc.def", "1.0.0") is True
def test_prerelease_below_release(self):
assert _is_plugin_update_available("1.2.0-rc1", "1.2.0") is True
class TestCoerceToBool:
@pytest.mark.parametrize("value", ["true", "TRUE", "on", "1", "yes", "YES"])
def test_truthy_strings(self, value):
assert _coerce_to_bool(value) is True
@pytest.mark.parametrize("value", ["false", "off", "0", "no", "", "banana"])
def test_falsey_strings(self, value):
assert _coerce_to_bool(value) is False
def test_none_is_false(self):
assert _coerce_to_bool(None) is False
def test_bools_pass_through(self):
assert _coerce_to_bool(True) is True
assert _coerce_to_bool(False) is False
def test_int_only_one_is_true(self):
# Characterized quirk: ints coerce via `value == 1`, so 2 (truthy in
# Python) is False here.
assert _coerce_to_bool(1) is True
assert _coerce_to_bool(2) is False
assert _coerce_to_bool(0) is False
def test_other_types_false(self):
assert _coerce_to_bool([1]) is False
assert _coerce_to_bool({"a": 1}) is False
class TestDeepMerge:
def test_nested_dicts_merge_recursively(self):
base = {"a": {"x": 1, "y": 2}, "b": 1}
update = {"a": {"y": 3, "z": 4}}
assert deep_merge(base, update) == {"a": {"x": 1, "y": 3, "z": 4}, "b": 1}
def test_scalar_over_dict_replaces(self):
assert deep_merge({"a": {"x": 1}}, {"a": 5}) == {"a": 5}
def test_dict_over_scalar_replaces(self):
assert deep_merge({"a": 5}, {"a": {"x": 1}}) == {"a": {"x": 1}}
def test_lists_replaced_wholesale(self):
assert deep_merge({"a": [1, 2]}, {"a": [3]}) == {"a": [3]}
def test_top_level_not_mutated_but_shallow_copy(self):
# Characterized: result = base.copy() protects base's top level, but
# nested dicts NOT touched by the update are shared by reference.
base = {"a": {"x": 1}, "keep": {"y": 2}}
result = deep_merge(base, {"a": {"x": 9}})
assert base == {"a": {"x": 1}, "keep": {"y": 2}} # base unchanged
assert result["keep"] is base["keep"] # untouched subtree is shared
class TestParseFormValue:
def test_boolean_strings(self):
assert _parse_form_value("true") is True
assert _parse_form_value("False") is False
def test_null_like_strings(self):
assert _parse_form_value("null") is None
assert _parse_form_value("none") is None
assert _parse_form_value("") is None
def test_none_passthrough(self):
assert _parse_form_value(None) is None
def test_numbers(self):
assert _parse_form_value("42") == 42
assert isinstance(_parse_form_value("42"), int)
assert _parse_form_value("3.5") == 3.5
assert isinstance(_parse_form_value("3.5"), float)
def test_json_array_parsed_before_numbers(self):
# RGB arrays like "[255, 0, 0]" must come back as lists.
assert _parse_form_value("[255, 0, 0]") == [255, 0, 0]
def test_json_object(self):
assert _parse_form_value('{"a": 1}') == {"a": 1}
def test_malformed_json_falls_back_to_string(self):
assert _parse_form_value("[not json") == "[not json"
def test_plain_string_returned_unstripped(self):
# The original value (not the stripped copy) is returned.
assert _parse_form_value(" hello ") == " hello "
def test_non_string_passthrough(self):
assert _parse_form_value(7) == 7
assert _parse_form_value([1, 2]) == [1, 2]
class TestGetSchemaProperty:
SCHEMA: ClassVar[Dict[str, Any]] = {
"properties": {
"brightness": {"type": "integer"},
"customization": {
"type": "object",
"properties": {
"time_text": {
"type": "object",
"properties": {"font": {"type": "string"}},
},
},
},
"fifa.world": {"type": "object",
"properties": {"enabled": {"type": "boolean"}}},
}
}
def test_top_level_lookup(self):
assert _get_schema_property(self.SCHEMA, "brightness") == {"type": "integer"}
def test_nested_dot_path(self):
prop = _get_schema_property(self.SCHEMA, "customization.time_text.font")
assert prop == {"type": "string"}
def test_dotted_schema_key_matched_longest_first(self):
# League keys like "fifa.world" contain a literal dot and must match
# as a single key, not be split into nested fifa -> world lookups.
prop = _get_schema_property(self.SCHEMA, "fifa.world.enabled")
assert prop == {"type": "boolean"}
def test_missing_path_returns_none(self):
assert _get_schema_property(self.SCHEMA, "nope.nope") is None
def test_no_properties_returns_none(self):
assert _get_schema_property({}, "a") is None
assert _get_schema_property(None, "a") is None
class TestSetNestedValue:
def test_sets_top_level(self):
config = {}
_set_nested_value(config, "brightness", 80)
assert config == {"brightness": 80}
def test_creates_intermediate_dicts(self):
config = {}
_set_nested_value(config, "customization.time_text.font", "5x7")
assert config == {"customization": {"time_text": {"font": "5x7"}}}
def test_merges_into_existing_nested_dict(self):
config = {"customization": {"color": "red"}}
_set_nested_value(config, "customization.font", "5x7")
assert config == {"customization": {"color": "red", "font": "5x7"}}
def test_scalar_intermediate_replaced_with_dict(self):
# Characterized: a non-dict intermediate is silently replaced.
config = {"customization": "oops"}
_set_nested_value(config, "customization.font", "5x7")
assert config == {"customization": {"font": "5x7"}}
def test_existing_dotted_key_preserved(self):
# An existing literal "fifa.world" key must be updated in place, not
# exploded into nested {"fifa": {"world": ...}}.
config = {"fifa.world": {"enabled": False}}
_set_nested_value(config, "fifa.world.enabled", True)
assert config == {"fifa.world": {"enabled": True}}
def test_none_does_not_overwrite_existing(self):
config = {"a": 1}
_set_nested_value(config, "a", None)
assert config == {"a": 1}
def test_none_sets_missing_key(self):
config = {}
_set_nested_value(config, "a", None)
assert config == {"a": None}
-241
View File
@@ -1,241 +0,0 @@
"""
Tests for src/web_interface/secret_helpers.py the canonical secret
identification / separation / masking helpers.
This module is the extracted single source of truth for x-secret handling,
but until now had zero test coverage (only ``mask_secret_fields`` is even
imported by production code, from pages_v3). api_v3.py still carries three
inline re-implementations of ``find_secret_fields``/``separate_secrets``
see test_secret_separation_parity.py so pinning the canonical behavior
here is a precondition for ever migrating those copies.
"""
import copy
from src.web_interface.secret_helpers import (
find_secret_fields,
separate_secrets,
mask_secret_fields,
mask_all_secret_values,
remove_empty_secrets,
)
SCHEMA_PROPS = {
"api_key": {"type": "string", "x-secret": True},
"city": {"type": "string"},
"auth": {
"type": "object",
"properties": {
"token": {"type": "string", "x-secret": True},
"username": {"type": "string"},
},
},
"accounts": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": {"type": "string"},
"token": {"type": "string", "x-secret": True},
},
},
},
"recovery_codes": {
"type": "array",
"items": {"type": "string", "x-secret": True},
},
}
class TestFindSecretFields:
def test_top_level_secret(self):
assert "api_key" in find_secret_fields(SCHEMA_PROPS)
def test_non_secret_not_included(self):
assert "city" not in find_secret_fields(SCHEMA_PROPS)
def test_nested_object_secret_uses_dot_path(self):
assert "auth.token" in find_secret_fields(SCHEMA_PROPS)
assert "auth.username" not in find_secret_fields(SCHEMA_PROPS)
def test_array_item_object_secret_uses_bracket_path(self):
assert "accounts[].token" in find_secret_fields(SCHEMA_PROPS)
def test_array_of_secrets_uses_bracket_path(self):
assert "recovery_codes[]" in find_secret_fields(SCHEMA_PROPS)
def test_full_set(self):
assert find_secret_fields(SCHEMA_PROPS) == {
"api_key", "auth.token", "accounts[].token", "recovery_codes[]",
}
def test_non_dict_properties_tolerated(self):
assert find_secret_fields({"weird": "not-a-dict"}) == set()
def test_non_dict_input_returns_empty(self):
assert find_secret_fields(None) == set()
assert find_secret_fields([]) == set()
class TestSeparateSecrets:
def test_flat_partition(self):
regular, secrets = separate_secrets(
{"api_key": "s3cret", "city": "Austin"}, {"api_key"})
assert regular == {"city": "Austin"}
assert secrets == {"api_key": "s3cret"}
def test_nested_partition(self):
config = {"auth": {"token": "t0k", "username": "chuck"}}
regular, secrets = separate_secrets(config, {"auth.token"})
assert regular == {"auth": {"username": "chuck"}}
assert secrets == {"auth": {"token": "t0k"}}
def test_empty_nested_dicts_pruned_from_regular(self):
# A dict that is all secrets leaves nothing behind on the regular
# side — the key must be dropped, not kept as {}.
config = {"auth": {"token": "t0k"}}
regular, secrets = separate_secrets(config, {"auth.token"})
assert regular == {}
assert secrets == {"auth": {"token": "t0k"}}
def test_whole_array_secret(self):
config = {"recovery_codes": ["a", "b"], "city": "Austin"}
regular, secrets = separate_secrets(config, {"recovery_codes[]"})
assert regular == {"city": "Austin"}
assert secrets == {"recovery_codes": ["a", "b"]}
def test_array_item_secrets_produce_parallel_lists(self):
# Per-item secrets keep the arrays index-aligned so they can be
# recombined: regular gets the stripped items, secrets a parallel
# list of the extracted values.
config = {"accounts": [
{"name": "a", "token": "ta"},
{"name": "b", "token": "tb"},
]}
regular, secrets = separate_secrets(config, {"accounts[].token"})
assert regular == {"accounts": [{"name": "a"}, {"name": "b"}]}
assert secrets == {"accounts": [{"token": "ta"}, {"token": "tb"}]}
def test_array_item_non_dict_items_get_placeholder(self):
config = {"accounts": [{"name": "a", "token": "ta"}, "oddball"]}
regular, secrets = separate_secrets(config, {"accounts[].token"})
assert regular == {"accounts": [{"name": "a"}, "oddball"]}
assert secrets == {"accounts": [{"token": "ta"}, {}]}
def test_array_without_secret_paths_stays_regular(self):
config = {"teams": ["DAL", "HOU"]}
regular, secrets = separate_secrets(config, {"api_key"})
assert regular == {"teams": ["DAL", "HOU"]}
assert secrets == {}
def test_round_trip_loses_nothing(self):
# separate + naive recombine must reconstruct the original config.
config = {
"api_key": "k",
"city": "Austin",
"auth": {"token": "t", "username": "chuck"},
"recovery_codes": ["a", "b"],
}
paths = find_secret_fields(SCHEMA_PROPS)
regular, secrets = separate_secrets(copy.deepcopy(config), paths)
def recombine(reg, sec):
out = copy.deepcopy(reg)
for k, v in sec.items():
if isinstance(v, dict) and isinstance(out.get(k), dict):
out[k] = recombine(out[k], v)
else:
out[k] = v
return out
assert recombine(regular, secrets) == config
class TestMaskSecretFields:
def test_masks_present_secret_to_empty_string(self):
result = mask_secret_fields({"api_key": "s3cret"}, SCHEMA_PROPS)
assert result["api_key"] == ""
def test_leaves_non_secret_untouched(self):
result = mask_secret_fields({"city": "Austin"}, SCHEMA_PROPS)
assert result["city"] == "Austin"
def test_none_and_empty_left_alone(self):
result = mask_secret_fields({"api_key": None}, SCHEMA_PROPS)
assert result["api_key"] is None
result = mask_secret_fields({"api_key": ""}, SCHEMA_PROPS)
assert result["api_key"] == ""
def test_falsey_but_set_values_are_masked(self):
# 0 and False are real values; the check is `is not None and != ''`.
# Note False == '' is False in Python, so False IS masked; 0 == '' is
# also False, so 0 is masked too.
result = mask_secret_fields({"api_key": 0}, SCHEMA_PROPS)
assert result["api_key"] == ""
result = mask_secret_fields({"api_key": False}, SCHEMA_PROPS)
assert result["api_key"] == ""
def test_nested_object_masked_without_mutating_input(self):
config = {"auth": {"token": "t0k", "username": "chuck"}}
original = copy.deepcopy(config)
result = mask_secret_fields(config, SCHEMA_PROPS)
assert result["auth"]["token"] == ""
assert result["auth"]["username"] == "chuck"
assert config == original # input not mutated
def test_array_of_secrets_masked_elementwise(self):
result = mask_secret_fields(
{"recovery_codes": ["a", "b"]}, SCHEMA_PROPS)
assert result["recovery_codes"] == ["", ""]
def test_array_of_objects_masked_per_item(self):
config = {"accounts": [{"name": "a", "token": "ta"}, "oddball"]}
result = mask_secret_fields(config, SCHEMA_PROPS)
assert result["accounts"][0] == {"name": "a", "token": ""}
assert result["accounts"][1] == "oddball"
def test_non_dict_schema_property_tolerated(self):
assert mask_secret_fields({"x": 1}, {"x": "bogus"}) == {"x": 1}
class TestMaskAllSecretValues:
def test_real_values_replaced_with_bullets(self):
assert mask_all_secret_values({"key": "abc"}) == {"key": "••••••••"}
def test_placeholders_preserved(self):
# YOUR_* placeholders must survive so the UI can show "not set".
result = mask_all_secret_values({"key": "YOUR_API_KEY_HERE"})
assert result == {"key": "YOUR_API_KEY_HERE"}
def test_empty_and_none_preserved(self):
assert mask_all_secret_values({"a": "", "b": None}) == {"a": "", "b": None}
def test_recurses_into_nested_dicts(self):
result = mask_all_secret_values({"plugin": {"token": "t", "empty": ""}})
assert result == {"plugin": {"token": "••••••••", "empty": ""}}
def test_non_string_real_values_masked(self):
assert mask_all_secret_values({"port": 8080}) == {"port": "••••••••"}
class TestRemoveEmptySecrets:
def test_strips_empty_string(self):
assert remove_empty_secrets({"a": "", "b": "real"}) == {"b": "real"}
def test_strips_whitespace_only(self):
assert remove_empty_secrets({"a": " "}) == {}
def test_strips_none(self):
assert remove_empty_secrets({"a": None}) == {}
def test_prunes_empty_nested_dicts(self):
assert remove_empty_secrets({"plugin": {"token": ""}}) == {}
def test_keeps_nested_real_values(self):
result = remove_empty_secrets({"plugin": {"token": "t", "empty": ""}})
assert result == {"plugin": {"token": "t"}}
def test_keeps_falsey_non_string_values(self):
# 0 and False are neither None nor blank strings — they are kept.
assert remove_empty_secrets({"a": 0, "b": False}) == {"a": 0, "b": False}
@@ -1,118 +0,0 @@
"""
Drift guard for the duplicated secret-separation logic.
src/web_interface/secret_helpers.py is the canonical implementation of
find_secret_fields / separate_secrets, but web_interface/blueprints/api_v3.py
still carries THREE inline nested-function copies of each (in the plugin
config GET, POST, and reset endpoints). The copies lack the canonical
module's array-item support (`accounts[].token`), so migrating an endpoint
onto the module is a behavior change that must be made deliberately.
This file guards two things:
1. The copy count can only go DOWN. A fourth copy appearing means someone
re-implemented the logic again instead of importing secret_helpers.
2. The known behavioral gap is documented as an executable fact, so whoever
migrates the endpoints knows exactly what changes.
"""
import re
from pathlib import Path
from src.web_interface.secret_helpers import find_secret_fields, separate_secrets
API_V3_PATH = (Path(__file__).resolve().parents[2]
/ "web_interface" / "blueprints" / "api_v3.py")
# Update DOWNWARD as endpoints migrate onto src/web_interface/secret_helpers.
EXPECTED_INLINE_COPIES = 3
class TestInlineCopyCount:
def _count(self, name: str) -> int:
source = API_V3_PATH.read_text(encoding="utf-8")
return len(re.findall(rf"^\s*def {name}\(", source, flags=re.MULTILINE))
def test_find_secret_fields_copy_count(self):
count = self._count("find_secret_fields")
assert count == EXPECTED_INLINE_COPIES, (
f"api_v3.py has {count} inline find_secret_fields definitions, "
f"expected {EXPECTED_INLINE_COPIES}. New code must import it from "
f"src/web_interface/secret_helpers instead of re-implementing it; "
f"if you migrated an endpoint, lower EXPECTED_INLINE_COPIES."
)
def test_separate_secrets_copy_count(self):
count = self._count("separate_secrets")
assert count == EXPECTED_INLINE_COPIES, (
f"api_v3.py has {count} inline separate_secrets definitions, "
f"expected {EXPECTED_INLINE_COPIES}. New code must import it from "
f"src/web_interface/secret_helpers instead of re-implementing it; "
f"if you migrated an endpoint, lower EXPECTED_INLINE_COPIES."
)
def test_inline_copies_lack_array_item_support(self):
"""The documented gap: no inline copy recurses into array `items`
schemas, so array-item secrets (accounts[].token) are NOT routed to
config_secrets.json by these endpoints. The canonical module handles
them. When an endpoint migrates onto the module that behavior
changes (a fix, but a deliberate one).
If this fails, an inline copy has grown array support duplicating
the canonical module even harder. Migrate the endpoint onto
src/web_interface/secret_helpers instead.
"""
for body in self._inline_bodies("find_secret_fields"):
# Array handling requires checking type == 'array'; no inline
# copy does. (Can't grep bare "items" — properties.items() the
# dict method appears legitimately.)
assert "'array'" not in body and '"array"' not in body
@staticmethod
def _inline_bodies(name: str):
"""Extract each inline def's body from api_v3.py by indentation."""
lines = API_V3_PATH.read_text(encoding="utf-8").splitlines()
bodies = []
i = 0
while i < len(lines):
match = re.match(rf"^(\s+)def {name}\(", lines[i])
if not match:
i += 1
continue
indent = len(match.group(1))
body = [lines[i]]
i += 1
while i < len(lines):
line = lines[i]
if line.strip() and (len(line) - len(line.lstrip())) <= indent:
break
body.append(line)
i += 1
bodies.append("\n".join(body))
assert bodies, f"no inline {name} definitions found"
return bodies
class TestCanonicalArrayItemBehavior:
"""Executable documentation of what migrating endpoints will change."""
SCHEMA = {
"accounts": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": {"type": "string"},
"token": {"type": "string", "x-secret": True},
},
},
},
}
def test_canonical_module_routes_array_item_secrets(self):
paths = find_secret_fields(self.SCHEMA)
assert "accounts[].token" in paths
config = {"accounts": [{"name": "a", "token": "s3cret"}]}
regular, secrets = separate_secrets(config, paths)
assert regular == {"accounts": [{"name": "a"}]}
assert secrets == {"accounts": [{"token": "s3cret"}]}
@@ -367,10 +367,6 @@ class TestStateReconciliationUnrecoverable(unittest.TestCase):
self.store_manager.fetch_registry.return_value = {"plugins": []}
self.store_manager.install_plugin.return_value = False
self.store_manager.was_recently_uninstalled.return_value = False
# A bare Mock() returns a truthy Mock for is_plugin_uninstalled(),
# which reads as "persistently uninstalled" and skips auto-repair
# entirely — these tests need the repair path to run.
self.store_manager.is_plugin_uninstalled.return_value = False
self.reconciler = StateReconciliation(
state_manager=self.state_manager,
+77 -9
View File
@@ -124,13 +124,28 @@ def _is_plugin_update_available(installed_version: str, latest_version: str) ->
"""Return True when the registry's ``latest_version`` is strictly newer
than the installed version.
Thin alias for the shared comparator in
`src.plugin_system.compatibility.is_update_available` the store's
`update_plugin` uses the same function, so the UI badge and the actual
reinstall decision can never disagree.
Uses PEP 440 / semver-aware comparison so a locally modified plugin whose
version is *ahead* of the published registry is not flagged as needing an
update. If either version string can't be parsed, falls back to a plain
inequality check (any difference is surfaced so the user can reconcile).
"""
from src.plugin_system.compatibility import is_update_available
return is_update_available(installed_version, latest_version)
if not installed_version or not latest_version:
return False
if installed_version == latest_version:
return False
try:
from packaging.version import parse as _parse_version, InvalidVersion
except ImportError:
# packaging is a core dependency, but if it's somehow unavailable we
# can't compare semantically — surface the mismatch we already know
# exists (the two strings differ).
return True
try:
return _parse_version(latest_version) > _parse_version(installed_version)
except InvalidVersion:
# Unparseable version string: we can't tell direction, so surface the
# mismatch rather than silently hiding a potential update.
return True
def _ensure_cache_manager():
"""Ensure cache manager is initialized."""
@@ -7833,7 +7848,35 @@ def clear_old_errors():
# Backup / Restore
# ---------------------------------------------------------------------------
_BACKUP_EXPORT_DIR = PROJECT_ROOT / "config" / "backups" / "exports"
def _resolve_backup_export_dir() -> Path:
"""Where exported backups live: beside the install, not inside it.
They used to be written to ``<project>/config/backups/exports``. That is
inside the directory a reinstall deletes, so the documented recovery path
-- export a backup, then reinstall -- destroyed the backup it had just
told the user to make. Anyone who downloaded the ZIP was fine; anyone
relying on the on-device copy was not.
Falls back to the old location when the parent directory is not writable,
so an unusual layout degrades to previous behaviour instead of failing to
export at all.
"""
preferred = PROJECT_ROOT.parent / "ledmatrix-backups"
fallback = PROJECT_ROOT / "config" / "backups" / "exports"
try:
preferred.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(dir=preferred, prefix=".writetest-"):
pass
return preferred
except OSError as e:
logger.warning(
f"[Backup] Export dir {preferred} is not writable ({e}); "
f"falling back to {fallback}, which a reinstall will delete"
)
return fallback
_BACKUP_EXPORT_DIR = _resolve_backup_export_dir()
def _safe_backup_path(filename: str) -> Path:
@@ -7983,11 +8026,36 @@ def backup_restore():
else:
result.plugins_failed.append({'plugin_id': pid, 'error': 'Store manager unavailable'})
except Exception as pe:
result.plugins_failed.append({'plugin_id': pid, 'error': str(pe)})
logger.error(
"[Backup] Failed to reinstall plugin %r: %s", pid, pe, exc_info=True
)
result.plugins_failed.append({'plugin_id': pid, 'error': 'Installation failed; see server logs'})
# A restore that dropped files can still report success if the only
# failures were plugin reinstalls, since those don't touch result.errors.
if result.plugins_failed:
result.success = False
data = result.to_dict()
if not result.success:
return jsonify({'status': 'error', 'message': 'Restore had errors', 'data': data}), 500
# Name what failed, and what nonetheless landed. A restore is
# partial far more often than it is total -- a fresh install can
# leave config_secrets.json unwritable by the web service, so
# config restores and secrets do not. "Restore had errors" alone
# left the user unable to tell a wholly failed restore from one
# that quietly dropped their API keys.
failed_plugins = [
str(p.get('plugin_id')) for p in (result.plugins_failed or []) if p.get('plugin_id')
]
parts = []
if result.restored:
parts.append(f"restored: {', '.join(result.restored)}")
if result.errors:
parts.append(f"failed: {'; '.join(result.errors)}")
if failed_plugins:
parts.append(f"plugins not reinstalled: {', '.join(failed_plugins)}")
message = 'Restore incomplete — ' + ('. '.join(parts) if parts else 'see logs')
return jsonify({'status': 'error', 'message': message, 'data': data}), 500
return jsonify({'status': 'success', 'data': data})
except Exception as e:
logger.error("backup_restore failed: %s", e, exc_info=True)