test(sync): cover the announce loop, and reject non-finite scroll positions

Three review findings from the follower receive path.

Non-finite scroll x reached follower rendering. json.loads accepts the
bare NaN/Infinity literals and float() accepts them as strings, so
"x": NaN arrived as a real float and was stored verbatim. NaN loses
every comparison the scroll code makes, so a follower given one sits on
a position it can never advance past. It now raises through the existing
malformed-control-message guard, which logs and drops the packet and
leaves the last good position in place.

_broadcast_available() only proves the host accepts sendto() for a
broadcast; a network that accepts the send and drops the packet would
let TestRealSocketHandshake run to its five-second deadline and fail on
assertions the code did not break. The deadline now distinguishes the
two: if not one packet crossed in either direction, that is the
environment, and the test skips rather than reporting a protocol
failure.

That skip could hide a real regression in the announcing side, so
TestFollowerAnnounceLoop covers it on mock sockets, where no network is
involved and nothing can skip: hello carries this display's hardware
config and goes to the broadcast address, heartbeats follow, an empty
hardware config falls back to 32x64x1, hello is not resent before its
interval, and a send failure is swallowed rather than killing the loop.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

.github/workflows/test.yml

@@ -72,4 +72,4 @@ jobs:
             --ignore=test/plugins \
             --cov=src --cov=web_interface \
             --cov-report=term \
-            --cov-fail-under=48
+            --cov-fail-under=52

src/common/logo_helper.py

@@ -6,6 +6,8 @@ Extracted from LEDMatrix core to provide reusable functionality for plugins.
 """
 
 import logging
+import os
+import tempfile
 from pathlib import Path
 from typing import Dict, List, Optional, Union
 
@@ -19,6 +21,10 @@ from src.common.permission_utils import (
 )
 
 
+# Well above any real team logo; bounds what a remote URL can write to disk.
+MAX_LOGO_BYTES = 10 * 1024 * 1024
+
+
 class LogoHelper:
     """
     Helper class for logo loading, caching, and resizing.
@@ -226,7 +232,10 @@ class LogoHelper:
         return {
             'cached_logos': len(self._logo_cache),
             'cache_size_limit': self.cache_size,
-            'cache_usage_percent': (len(self._logo_cache) / self.cache_size) * 100
+            'cache_usage_percent': (
+                (len(self._logo_cache) / self.cache_size) * 100
+                if self.cache_size else 0
+            ),
         }
     
     def _resize_logo(self, logo: Image.Image, max_width: Optional[int] = None, 
@@ -258,17 +267,60 @@ class LogoHelper:
         self._cache_order.append(cache_key)
     
     def _download_logo(self, url: str, file_path: Path) -> None:
-        """Download logo from URL."""
+        """Download logo from URL.
+
+        The response size is capped and the saved file is verified as a
+        decodable image before it is left on disk: a logo URL is remote
+        input, and without this an oversized or malformed response would
+        be cached for every later load_logo() call to trip over.
+
+        The body is streamed and counted as it arrives rather than read
+        through response.content, which buffers the whole thing first —
+        a server that omits Content-Length and never stops sending would
+        exhaust memory before any size check could run. Nothing lands at
+        file_path until the download completes and decodes, so a failed
+        download cannot leave a truncated logo behind either.
+        """
         # Ensure directory exists with proper permissions
         ensure_directory_permissions(file_path.parent, get_assets_dir_mode())
 
-        # Download with timeout
-        response = self.session.get(url, timeout=30)
+        # A unique temp name, not a fixed "<name>.part": two plugins can
+        # ask for the same logo at once, and a shared name would let them
+        # interleave writes into one file, publish the mixture, or delete
+        # each other's partial. Same directory, so os.replace stays atomic.
+        fd, tmp_name = tempfile.mkstemp(
+            dir=str(file_path.parent), prefix=file_path.name + '.', suffix='.part')
+        tmp_path = Path(tmp_name)
+        try:
+            # fdopen outermost so the descriptor mkstemp handed back is
+            # always adopted and closed, including when the request itself
+            # raises — load_logo_with_download swallows that, so a leak
+            # here would accumulate quietly on a URL that keeps failing.
+            with os.fdopen(fd, 'wb') as f:
+                with self.session.get(url, timeout=30, stream=True) as response:
                     response.raise_for_status()
+                    downloaded = 0
+                    for chunk in response.iter_content(chunk_size=64 * 1024):
+                        if not chunk:
+                            continue
+                        downloaded += len(chunk)
+                        if downloaded > MAX_LOGO_BYTES:
+                            raise ValueError(
+                                f"Logo at {url} exceeds the "
+                                f"{MAX_LOGO_BYTES}-byte limit; not saved")
+                        f.write(chunk)
 
-        # Save to file
-        with open(file_path, 'wb') as f:
-            f.write(response.content)
+            # Verify it decodes before it becomes the cached logo. PIL
+            # raises DecompressionBombError past its own pixel limit; a
+            # partial or non-image response raises UnidentifiedImageError
+            # (an OSError subclass).
+            with Image.open(tmp_path) as probe:
+                probe.load()
+
+            os.replace(tmp_path, file_path)
+        except BaseException:
+            tmp_path.unlink(missing_ok=True)
+            raise
 
         # Set proper file permissions after saving
         ensure_file_permissions(file_path, get_assets_file_mode())

src/common/sync_manager.py

@@ -19,6 +19,7 @@ Port default: 5765 (UDP). Open this port on both Pis if ufw is active:
 
 import io
 import json
+import math
 import os
 import socket
 import struct
@@ -37,6 +38,13 @@ _RAW_MAGIC = b'SYNC_RAW'
 _RAW_HEADER = struct.Struct('<HH')  # width, height (uint16 LE)
 
 
+# Upper bound on a decoded frame/scroll image. Generous for any real scroll
+# image (a leader's full cycle is long but only panel-height tall), and low
+# enough that a crafted image from any host on the LAN cannot force a large
+# allocation on the render thread. Applied on both receive paths — the TCP
+# image server and the follower's legacy-PNG UDP fallback.
+_MAX_FRAME_W, _MAX_FRAME_H = 100_000, 256
+
 SYNC_PORT = 5765
 HELLO_INTERVAL = 5.0       # follower broadcasts hello every 5 s
 HEARTBEAT_INTERVAL = 2.0   # follower sends heartbeat every 2 s
@@ -101,6 +109,7 @@ class DisplaySyncManager:
         self._peer_chain: int = 0
         self._last_heartbeat_time: float = 0.0
         self._leader_width: int = 0  # set by display_controller after init
+        self._oversized_frame_warned: bool = False
 
         # Follower state
         self._follower_state = FollowerState.STANDALONE
@@ -174,6 +183,10 @@ class DisplaySyncManager:
                 continue
             except Exception as exc:
                 self.logger.debug("Sync leader recv error: %s", exc)
+                # Brief backoff: a socket left in a bad state raises
+                # immediately, which would otherwise spin this thread at
+                # 100% CPU logging the same error.
+                time.sleep(0.1)
 
     def _handle_hello(self, msg: dict, sender_ip: str) -> None:
         hw = self._hw_config
@@ -273,11 +286,10 @@ class DisplaySyncManager:
                             break
                         data.extend(chunk)
                     img = Image.open(io.BytesIO(data))
-                    _MAX_W, _MAX_H = 100_000, 256  # generous for any real scroll image
-                    if img.width > _MAX_W or img.height > _MAX_H:
+                    if img.width > _MAX_FRAME_W or img.height > _MAX_FRAME_H:
                         self.logger.warning(
                             "Sync: rejected oversized scroll image %dx%d (max %dx%d) from %s",
-                            img.width, img.height, _MAX_W, _MAX_H, addr,
+                            img.width, img.height, _MAX_FRAME_W, _MAX_FRAME_H, addr,
                         )
                         continue
                     try:
@@ -396,7 +408,7 @@ class DisplaySyncManager:
             data = header + arr.tobytes()
             if len(data) <= 65000:
                 self._send_sock.sendto(data, (self._peer_ip, self.port))
-            elif not getattr(self, '_oversized_frame_warned', False):
+            elif not self._oversized_frame_warned:
                 self._oversized_frame_warned = True
                 self.logger.warning(
                     "Sync: frame too large for UDP (%d bytes, max 65000) — "
@@ -451,25 +463,8 @@ class DisplaySyncManager:
         )
         self.write_status_file()
 
-    def _follower_recv_loop(self) -> None:
-        while self._running:
-            try:
-                data, addr = self._recv_sock.recvfrom(65535)
-                sender_ip = addr[0]
-
-                if data[:8] == _RAW_MAGIC or len(data) > 512:
-                    # Frame data: prefer magic-tagged raw RGB; fall back to legacy PNG
-                    try:
-                        if data[:8] == _RAW_MAGIC:
-                            w, h = _RAW_HEADER.unpack(data[8:12])
-                            raw = data[12:]
-                            img = Image.frombuffer(
-                                "RGB", (w, h), raw, "raw", "RGB", 0, 1
-                            )
-                        else:
-                            # Fallback: try legacy PNG
-                            img = Image.open(io.BytesIO(data))
-                            img.load()
+    def _handle_received_frame(self, img: Image.Image, sender_ip: str) -> None:
+        """Record a decoded leader frame and enter follower mode if needed."""
         with self._frame_lock:
             self._latest_frame = img
         self._last_leader_frame_time = time.time()
@@ -482,12 +477,62 @@ class DisplaySyncManager:
                 sender_ip,
             )
             self.write_status_file()
+
+    def _follower_recv_loop(self) -> None:
+        while self._running:
+            try:
+                data, addr = self._recv_sock.recvfrom(65535)
+                sender_ip = addr[0]
+
+                if data[:8] == _RAW_MAGIC:
+                    # Magic-tagged raw RGB frame — self-describing, no guessing.
+                    try:
+                        w, h = _RAW_HEADER.unpack(data[8:12])
+                        raw = data[12:]
+                        img = Image.frombuffer(
+                            "RGB", (w, h), raw, "raw", "RGB", 0, 1
+                        )
+                        self._handle_received_frame(img, sender_ip)
                     except Exception as exc:
                         self.logger.debug("Sync: frame decode error: %s", exc)
                 else:
-                    # Control message
+                    # No magic prefix. Whether the payload parses as JSON
+                    # decides between a control message and a legacy
+                    # (pre-magic) PNG frame — both wire formats are
+                    # self-describing, so no size heuristic is needed. A
+                    # >512-byte control message used to be misrouted into
+                    # image decode and silently dropped.
                     try:
                         msg = json.loads(data.decode("utf-8"))
+                    except (json.JSONDecodeError, UnicodeDecodeError):
+                        # Not JSON — try a legacy PNG frame.
+                        try:
+                            img = Image.open(io.BytesIO(data))
+                            if img.width > _MAX_FRAME_W or img.height > _MAX_FRAME_H:
+                                # Same cap the TCP image path applies: decode
+                                # is deferred until load(), so check first.
+                                self.logger.debug(
+                                    "Sync: rejected oversized legacy frame %dx%d from %s",
+                                    img.width, img.height, sender_ip,
+                                )
+                                continue
+                            img.load()
+                            self._handle_received_frame(img, sender_ip)
+                        except Exception as exc:
+                            self.logger.debug("Sync: frame decode error: %s", exc)
+                        continue
+
+                    # It parsed, so it is a control message and never a
+                    # frame. Read and validate its fields under a guard —
+                    # a UDP payload is attacker-shaped, so a non-object
+                    # body makes .get() raise AttributeError and an "sx"
+                    # carrying a non-numeric x raises ValueError/TypeError
+                    # — but dispatch the callback *outside* it. Running
+                    # the callback in here would let a fault in someone
+                    # else's code read as a malformed packet and be
+                    # logged as one.
+                    fire_new_cycle = False
+                    try:
                         t = msg.get("t")
                         if t == "hello_ack":
                             self._leader_ip = sender_ip
@@ -501,7 +546,17 @@ class DisplaySyncManager:
                             self.write_status_file()
                         elif t == "sx":
                             # Vegas scroll-position sync — tiny message, renders locally
-                            self._latest_scroll_x = float(msg["x"])
+                            scroll_x = float(msg["x"])
+                            if not math.isfinite(scroll_x):
+                                # json.loads accepts the NaN/Infinity literals,
+                                # and float("nan") accepts the strings, so a
+                                # non-finite x reaches here intact. Left alone
+                                # it poisons every offset computed from it —
+                                # NaN comparisons are all false, so the
+                                # follower renders a frame it can never scroll
+                                # back from. Treat it as malformed.
+                                raise ValueError(f"non-finite scroll x: {msg['x']!r}")
+                            self._latest_scroll_x = scroll_x
                             self._last_leader_frame_time = time.time()
                             self._leader_ip = sender_ip
                             if self._follower_state == FollowerState.STANDALONE:
@@ -511,19 +566,22 @@ class DisplaySyncManager:
                                     sender_ip,
                                 )
                                 self.write_status_file()
-                                if self._on_new_cycle:
-                                    self._on_new_cycle()  # build initial scroll image
+                                fire_new_cycle = True  # build initial scroll image
                         elif t == "nc":
                             # Leader started a new scroll cycle — rebuild local image
-                            if self._on_new_cycle:
+                            fire_new_cycle = True
+                    except (KeyError, AttributeError, TypeError, ValueError) as exc:
+                        self.logger.debug("Sync: malformed control message: %s", exc)
+                        continue
+
+                    if fire_new_cycle and self._on_new_cycle:
                         self._on_new_cycle()
-                    except (json.JSONDecodeError, UnicodeDecodeError, KeyError):
-                        pass
 
             except socket.timeout:
                 continue
             except Exception as exc:
                 self.logger.debug("Sync follower recv error: %s", exc)
+                time.sleep(0.1)
 
     def _follower_announce_loop(self) -> None:
         hw = self._hw_config

src/web_interface/api_helpers.py

@@ -30,16 +30,14 @@ def success_response(
     """
     response_data = create_success_response(data, message, metadata)
 
-    # Add request metadata if available
-    if metadata is None:
-        metadata = {}
-    
-    # Add timing if request start time is available
+    # Timing is merged into whatever the caller passed, without inventing a
+    # metadata block for responses that have neither.
+    enriched = dict(metadata) if metadata is not None else {}
     if hasattr(request, 'start_time'):
-        metadata['response_time_ms'] = int((time.time() - request.start_time) * 1000)
+        enriched['response_time_ms'] = int((time.time() - request.start_time) * 1000)
 
-    if metadata:
-        response_data['metadata'] = metadata
+    if metadata is not None or enriched:
+        response_data['metadata'] = enriched
 
     return jsonify(response_data)
 

src/web_interface/error_handler.py

@@ -142,13 +142,16 @@ def create_success_response(
         "status": "success"
     }
     
+    # All three use `is not None` rather than truthiness: "" and {} are
+    # values a caller chose to send, and dropping them silently would make
+    # the response shape depend on the data.
     if data is not None:
         response["data"] = data
 
-    if message:
+    if message is not None:
         response["message"] = message
 
-    if metadata:
+    if metadata is not None:
         response["metadata"] = metadata
 
     return response

src/web_interface/errors.py

@@ -89,7 +89,11 @@ class WebInterfaceError:
         self.category = category or self._infer_category(error_code)
         self.details = details
         self.context = context or {}
-        self.suggested_fixes = suggested_fixes or self._get_default_suggestions(error_code)
+        # `is None`, not truthiness: an explicit [] means "this caller has
+        # no suggestions to offer", which the default list would override.
+        self.suggested_fixes = (
+            suggested_fixes if suggested_fixes is not None
+            else self._get_default_suggestions(error_code))
         self.original_error = original_error
     
     def _infer_category(self, error_code: ErrorCode) -> ErrorCategory:

src/web_interface/validators.py

@@ -43,10 +43,15 @@ def validate_image_url(url: str) -> Tuple[bool, Optional[str]]:
     if any(handler in url_lower for handler in ['onerror=', 'onload=', 'onclick=']):
         return False, "Event handlers not allowed in URLs"
     
+    # Reject directory traversal anywhere, not only in relative paths:
+    # http://host/../secret is as much a traversal attempt as /../secret.
+    if '..' in url:
+        return False, "Invalid path: directory traversal not allowed"
+
     # Allow relative paths starting with /
     if url.startswith('/'):
-        # Validate it's a safe relative path (no directory traversal)
-        if '..' in url or url.startswith('//'):
+        # // would be a protocol-relative URL, not a local path
+        if url.startswith('//'):
             return False, "Invalid relative path"
         return True, None
     
@@ -104,10 +109,11 @@ def validate_file_upload(filename: str, max_size_mb: int = 10,
     if '..' in filename or '/' in filename or '\\' in filename:
         return False, "Filename contains invalid characters"
     
-    # Check extension if specified
+    # Check extension if specified. Both sides are lowercased: the caller's
+    # list is as likely to hold '.TTF' as the filename is.
     if allowed_extensions:
         file_ext = Path(filename).suffix.lower()
-        if file_ext not in allowed_extensions:
+        if file_ext not in [ext.lower() for ext in allowed_extensions]:
             return False, f"File extension must be one of: {', '.join(allowed_extensions)}"
     
     return True, None
@@ -147,7 +153,8 @@ def validate_numeric_range(value: float, min_val: Optional[float] = None,
     Returns:
         Tuple of (is_valid, error_message)
     """
-    if not isinstance(value, (int, float)):
+    # bool is an int subclass, so True would otherwise validate as 1.
+    if not isinstance(value, (int, float)) or isinstance(value, bool):
         return False, "Value must be a number"
     
     if min_val is not None and value < min_val:
@@ -183,7 +190,15 @@ def validate_string_length(text: str, min_length: Optional[int] = None,
 
 def sanitize_plugin_config(config: dict) -> dict:
     """
-    Sanitize plugin configuration input to prevent injection.
+    Restrict a plugin config to safe key names and value types.
+
+    Drops keys that are not plain identifiers and values that are not
+    JSON-ish scalars, lists, or dicts, recursing into the latter two.
+
+    String values are returned **unescaped**: output escaping is the
+    template layer's job, and escaping here would store the escaped form
+    in config.json. Do not read this function as XSS protection for
+    rendered output.
 
     Args:
         config: Configuration dictionary

test/_api_v3_test_helpers.py

@@ -0,0 +1,75 @@
+"""
+Shared scaffolding for api_v3 blueprint tests.
+
+Not a test module (the leading underscore keeps pytest from collecting
+it). It is the pytest-fixture equivalent of ``_make_client()`` in
+test_uninstall_and_reconcile_endpoint.py, which is unittest-style and
+requires ``self.addCleanup``.
+
+The api_v3 blueprint keeps its managers as attributes on a module-level
+singleton, not in Flask app state, so replacing them with mocks leaks
+into every later test that imports api_v3 unless the originals are put
+back. ``api_v3_client`` snapshots and restores them around each test.
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+from flask import Flask
+
+
+# Every manager attribute the blueprint reads. Anything missing here keeps
+# whatever a previously-run test left on the singleton.
+API_V3_MANAGER_ATTRS = (
+    'config_manager', 'plugin_manager', 'plugin_store_manager',
+    'plugin_state_manager', 'saved_repositories_manager', 'schema_manager',
+    'operation_queue', 'operation_history', 'cache_manager',
+)
+
+_SENTINEL = object()
+
+
+def build_app(blueprint):
+    app = Flask(__name__)
+    app.config['TESTING'] = True
+    app.config['SECRET_KEY'] = 'test'
+    app.register_blueprint(blueprint, url_prefix='/api/v3')
+    return app
+
+
+@pytest.fixture
+def api_v3_module():
+    """The api_v3 module with every manager replaced by a MagicMock.
+
+    Restores the original attributes afterwards. Tests point individual
+    managers at real objects (a ConfigManager over tmp_path, say) or set
+    them to None to exercise the not-initialized branches.
+    """
+    from web_interface.blueprints import api_v3 as module
+
+    originals = {
+        name: getattr(module.api_v3, name, _SENTINEL)
+        for name in API_V3_MANAGER_ATTRS
+    }
+    for name in API_V3_MANAGER_ATTRS:
+        setattr(module.api_v3, name, MagicMock())
+    # Default to the direct path; queue tests opt in explicitly.
+    module.api_v3.operation_queue = None
+
+    yield module
+
+    for name, original in originals.items():
+        if original is _SENTINEL:
+            if hasattr(module.api_v3, name):
+                try:
+                    delattr(module.api_v3, name)
+                except AttributeError:
+                    pass
+        else:
+            setattr(module.api_v3, name, original)
+
+
+@pytest.fixture
+def api_v3_client(api_v3_module):
+    """Flask test client wired to the mocked blueprint."""
+    return build_app(api_v3_module.api_v3).test_client()

test/test_api_v3_calendar_credentials.py

@@ -0,0 +1,226 @@
+"""
+Endpoint tests for POST /plugins/calendar/upload-credentials.
+
+The endpoint takes an uploaded Google OAuth credentials file, writes it
+into the calendar plugin's directory as credentials.json at mode 0600, and
+copies any previous file aside first. It had no tests.
+
+Regression coverage for two fixed bugs:
+- The OAuth-shape check sat inside `except Exception: pass`, so a valid
+  JSON document that is not an object — a bare `42`, a list, a string —
+  raised TypeError on the membership test, was swallowed, and got saved
+  as credentials.json anyway.
+- Each overwrite created a timestamped backup and nothing ever removed
+  them, so every re-upload left another complete copy of the user's OAuth
+  client credentials in the plugin directory, indefinitely.
+"""
+
+import io
+import json
+import os
+import stat
+import sys
+import time
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
+
+URL = "/api/v3/plugins/calendar/upload-credentials"
+
+VALID_CREDENTIALS = {
+    "installed": {
+        "client_id": "abc.apps.googleusercontent.com",
+        "client_secret": "shh",
+        "redirect_uris": ["http://localhost"],
+    }
+}
+
+
+@pytest.fixture
+def plugin_dir(tmp_path, api_v3_module):
+    directory = tmp_path / "plugins" / "calendar"
+    directory.mkdir(parents=True)
+    api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(directory)
+    return directory
+
+
+def upload(client, content, filename="credentials.json"):
+    # bytes are sent verbatim (to exercise malformed input); anything else
+    # is serialized, so None becomes the JSON literal null rather than an
+    # empty body.
+    payload = content if isinstance(content, bytes) else json.dumps(content).encode()
+    return client.post(
+        URL,
+        data={"file": (io.BytesIO(payload), filename)},
+        content_type="multipart/form-data",
+    )
+
+
+def backups(plugin_dir):
+    return sorted(plugin_dir.glob("credentials.json.backup.*"))
+
+
+class TestRequestValidation:
+    def test_no_file_part_is_a_400(self, api_v3_client, plugin_dir):
+        response = api_v3_client.post(URL, data={}, content_type="multipart/form-data")
+        assert response.status_code == 400
+        assert "No file provided" in response.get_json()["message"]
+
+    def test_empty_filename_is_a_400(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, VALID_CREDENTIALS, filename="")
+        assert response.status_code == 400
+
+    @pytest.mark.parametrize("filename", ["creds.txt", "creds.pem", "creds"])
+    def test_non_json_extension_is_a_400(self, api_v3_client, plugin_dir, filename):
+        response = upload(api_v3_client, VALID_CREDENTIALS, filename=filename)
+        assert response.status_code == 400
+        assert "JSON file" in response.get_json()["message"]
+
+    def test_uppercase_json_extension_accepted(self, api_v3_client, plugin_dir):
+        assert upload(api_v3_client, VALID_CREDENTIALS,
+                      filename="CREDENTIALS.JSON").status_code == 200
+
+    def test_oversized_file_is_a_400(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, b"x" * (1024 * 1024 + 1))
+        assert response.status_code == 400
+        assert "1MB" in response.get_json()["message"]
+        assert not (plugin_dir / "credentials.json").exists()
+
+    def test_invalid_json_is_a_400(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, b"{not json")
+        assert response.status_code == 400
+        assert "not valid JSON" in response.get_json()["message"]
+        assert not (plugin_dir / "credentials.json").exists()
+
+    def test_missing_plugin_directory_is_a_404(self, api_v3_client, api_v3_module, tmp_path):
+        api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(
+            tmp_path / "not-installed")
+        assert upload(api_v3_client, VALID_CREDENTIALS).status_code == 404
+
+
+class TestOAuthShapeValidation:
+    def test_installed_key_accepted(self, api_v3_client, plugin_dir):
+        assert upload(api_v3_client, VALID_CREDENTIALS).status_code == 200
+
+    def test_web_key_accepted(self, api_v3_client, plugin_dir):
+        assert upload(api_v3_client, {"web": {"client_id": "x"}}).status_code == 200
+
+    def test_object_without_oauth_keys_is_a_400(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, {"something": "else"})
+        assert response.status_code == 400
+        assert "valid Google OAuth" in response.get_json()["message"]
+        assert not (plugin_dir / "credentials.json").exists()
+
+    @pytest.mark.parametrize("content", [42, "a string", [1, 2, 3], True, None])
+    def test_valid_json_that_is_not_an_object_is_rejected(
+            self, api_v3_client, plugin_dir, content):
+        # Regression: `'installed' not in 42` raises TypeError, which the
+        # bare `except Exception: pass` swallowed — the file was then saved
+        # as credentials.json despite being unusable as credentials.
+        response = upload(api_v3_client, content)
+        assert response.status_code == 400
+        assert "valid Google OAuth" in response.get_json()["message"]
+        assert not (plugin_dir / "credentials.json").exists()
+
+
+class TestSaving:
+    def test_file_written_with_contents_intact(self, api_v3_client, plugin_dir):
+        response = upload(api_v3_client, VALID_CREDENTIALS)
+        assert response.status_code == 200
+        saved = json.loads((plugin_dir / "credentials.json").read_text())
+        assert saved == VALID_CREDENTIALS
+
+    def test_response_reports_the_path(self, api_v3_client, plugin_dir):
+        body = upload(api_v3_client, VALID_CREDENTIALS).get_json()
+        assert body["path"].endswith("credentials.json")
+
+    def test_permissions_are_owner_only(self, api_v3_client, plugin_dir):
+        upload(api_v3_client, VALID_CREDENTIALS)
+        mode = stat.S_IMODE((plugin_dir / "credentials.json").stat().st_mode)
+        assert mode == 0o600
+
+    def test_first_upload_creates_no_backup(self, api_v3_client, plugin_dir):
+        upload(api_v3_client, VALID_CREDENTIALS)
+        assert backups(plugin_dir) == []
+
+    def test_overwrite_backs_up_the_previous_file(self, api_v3_client, plugin_dir):
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"old": 1}}))
+        upload(api_v3_client, VALID_CREDENTIALS)
+        assert len(backups(plugin_dir)) == 1
+        assert json.loads(backups(plugin_dir)[0].read_text()) == {"installed": {"old": 1}}
+        assert json.loads((plugin_dir / "credentials.json").read_text()) == VALID_CREDENTIALS
+
+
+class TestBackupPruning:
+    def _seed(self, plugin_dir, count):
+        """Create `count` backups with distinct, increasing mtimes."""
+        now = int(time.time())
+        for i in range(count):
+            path = plugin_dir / f"credentials.json.backup.{now - (count - i) * 10}"
+            path.write_text(json.dumps({"installed": {"gen": i}}))
+            os.utime(path, (now - (count - i) * 10, now - (count - i) * 10))
+
+    def test_old_backups_are_pruned(self, api_v3_client, plugin_dir):
+        # Regression: nothing ever removed these, so a plugin directory
+        # accumulated one full copy of the user's OAuth credentials per
+        # re-upload, forever.
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
+        self._seed(plugin_dir, 7)
+        assert len(backups(plugin_dir)) == 7
+
+        upload(api_v3_client, VALID_CREDENTIALS)
+        assert len(backups(plugin_dir)) == 5
+
+    def test_the_newest_backups_are_the_ones_kept(self, api_v3_client, plugin_dir):
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
+        self._seed(plugin_dir, 7)
+
+        upload(api_v3_client, VALID_CREDENTIALS)
+        remaining = backups(plugin_dir)
+        # The just-created backup (of "cur") plus the four newest seeds.
+        contents = [json.loads(p.read_text()) for p in remaining]
+        assert {"installed": {"cur": 1}} in contents
+        assert {"installed": {"gen": 0}} not in contents  # oldest seed gone
+
+    def test_under_the_limit_nothing_is_removed(self, api_v3_client, plugin_dir):
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
+        self._seed(plugin_dir, 2)
+        upload(api_v3_client, VALID_CREDENTIALS)
+        assert len(backups(plugin_dir)) == 3  # 2 seeded + 1 new
+
+    def test_repeated_uploads_stay_bounded(
+            self, api_v3_client, plugin_dir, api_v3_module, monkeypatch):
+        # The backup filename carries int(time.time()), so uploads inside
+        # the same second all write the same name and overwrite each other.
+        # Advance a fake clock a second per round — otherwise this never
+        # reaches six backups and the bound holds for the wrong reason.
+        clock = {"now": int(time.time())}
+        monkeypatch.setattr(
+            api_v3_module, "time", SimpleNamespace(time=lambda: clock["now"]))
+        for i in range(10):
+            clock["now"] += 1
+            upload(api_v3_client, {"installed": {"round": i}})
+            os.utime(plugin_dir / "credentials.json",
+                     (clock["now"], clock["now"]))
+        remaining = backups(plugin_dir)
+        assert len(remaining) == 5
+        # And they are the five most recent rounds, not an arbitrary five.
+        kept = sorted(int(p.name.rsplit(".", 1)[1]) for p in remaining)
+        assert kept == [clock["now"] - 4 + i for i in range(5)]
+
+    def test_unremovable_backup_does_not_fail_the_upload(
+            self, api_v3_client, plugin_dir, monkeypatch):
+        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
+        self._seed(plugin_dir, 7)
+
+        def refuse(self):
+            raise OSError("read-only filesystem")
+        monkeypatch.setattr(Path, "unlink", refuse)
+
+        # Pruning is housekeeping; failing it must not lose the upload.
+        assert upload(api_v3_client, VALID_CREDENTIALS).status_code == 200

test/test_api_v3_music_auth_endpoints.py

@@ -0,0 +1,302 @@
+"""
+Endpoint tests for /plugins/authenticate/spotify and .../ytm.
+
+The Spotify step-2 handler writes a Python wrapper script to a temp file
+with the user's redirect URL embedded in it, then runs that file through
+subprocess. That is the most dangerous shape in the blueprint and had no
+tests: the URL is user input reaching generated source code.
+
+The two endpoints are NOT symmetrical, despite the matching names. Only
+Spotify has a two-step flow, a wrapper script, and a redirect_url; YTM
+just runs its script directly.
+
+Regression coverage for one fixed bug: the wrapper file was unlinked in
+the success/failure branch and again in the TimeoutExpired handler, so
+any other failure from subprocess.run — the interpreter missing, a fork
+failure, an interrupted call — left a temp file containing the user's
+redirect URL behind.
+"""
+
+import ast
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
+
+
+@pytest.fixture
+def plugin_dir(tmp_path, api_v3_module):
+    """A plugin directory containing both auth scripts."""
+    directory = tmp_path / "plugins" / "ledmatrix-music"
+    directory.mkdir(parents=True)
+    (directory / "authenticate_spotify.py").write_text("print('spotify')\n")
+    (directory / "authenticate_ytm.py").write_text("print('ytm')\n")
+    api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(directory)
+    return directory
+
+
+def completed(returncode=0, stdout="ok", stderr=""):
+    return subprocess.CompletedProcess(
+        args=["python3"], returncode=returncode, stdout=stdout, stderr=stderr)
+
+
+class TestSpotifyPreconditions:
+    URL = "/api/v3/plugins/authenticate/spotify"
+
+    def test_missing_plugin_directory_is_404(self, api_v3_client, api_v3_module, tmp_path):
+        api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(
+            tmp_path / "not-installed")
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 404
+        assert response.get_json()["message"] == "Plugin not found"
+
+    def test_none_plugin_directory_is_404(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = None
+        assert api_v3_client.post(self.URL, json={}).status_code == 404
+
+    def test_missing_auth_script_is_404(self, api_v3_client, plugin_dir):
+        (plugin_dir / "authenticate_spotify.py").unlink()
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 404
+        assert "script not found" in response.get_json()["message"]
+
+
+class TestSpotifyStepTwo:
+    """redirect_url present — the wrapper-script path."""
+
+    URL = "/api/v3/plugins/authenticate/spotify"
+
+    def test_success(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed(0, "done")):
+            response = api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
+        assert response.status_code == 200
+        body = response.get_json()
+        assert body["status"] == "success"
+        assert body["output"] == "done"
+
+    def test_script_failure_is_a_400_with_combined_output(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed(1, "out", "err")):
+            response = api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
+        assert response.status_code == 400
+        assert response.get_json()["output"] == "outerr"
+
+    def test_timeout_is_a_408(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run",
+                          side_effect=subprocess.TimeoutExpired("python3", 120)):
+            response = api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
+        assert response.status_code == 408
+        assert "timed out" in response.get_json()["message"]
+
+    def test_runs_a_list_argv_never_a_shell(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed()) as run:
+            api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
+        args, kwargs = run.call_args
+        assert isinstance(args[0], list)
+        assert args[0][0] == "python3"
+        assert kwargs.get("shell") in (None, False)
+
+    def test_timeout_is_bounded(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed()) as run:
+            api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
+        assert run.call_args.kwargs["timeout"] == 120
+
+
+class TestSpotifyWrapperCleanup:
+    URL = "/api/v3/plugins/authenticate/spotify"
+
+    def _wrapper_paths_after(self, api_v3_client, run_mock):
+        """Run the endpoint and return the wrapper path subprocess saw."""
+        seen = {}
+
+        def capture(args, **kwargs):
+            seen["path"] = args[1]
+            return run_mock(args, **kwargs)
+
+        with patch.object(subprocess, "run", side_effect=capture):
+            api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
+        return seen["path"]
+
+    def test_removed_after_success(self, api_v3_client, plugin_dir):
+        path = self._wrapper_paths_after(api_v3_client, lambda *a, **kw: completed())
+        assert not os.path.exists(path)
+
+    def test_removed_after_script_failure(self, api_v3_client, plugin_dir):
+        path = self._wrapper_paths_after(
+            api_v3_client, lambda *a, **kw: completed(1, "out", "err"))
+        assert not os.path.exists(path)
+
+    def test_removed_after_timeout(self, api_v3_client, plugin_dir):
+        def raise_timeout(*a, **kw):
+            raise subprocess.TimeoutExpired("python3", 120)
+        path = self._wrapper_paths_after(api_v3_client, raise_timeout)
+        assert not os.path.exists(path)
+
+    def test_removed_when_subprocess_cannot_start(self, api_v3_client, plugin_dir):
+        # Regression: cleanup lived in the success/failure branch and in the
+        # TimeoutExpired handler only. An OSError from subprocess.run itself
+        # — no interpreter, fork failure — skipped both and left the wrapper,
+        # which contains the user's redirect URL, on disk.
+        def raise_oserror(*a, **kw):
+            raise OSError("[Errno 12] Cannot allocate memory")
+        path = self._wrapper_paths_after(api_v3_client, raise_oserror)
+        assert not os.path.exists(path)
+
+
+class TestSpotifyRedirectUrlIsNotInjectable:
+    """The wrapper embeds redirect_url into generated Python source."""
+
+    URL = "/api/v3/plugins/authenticate/spotify"
+
+    ADVERSARIAL = [
+        '''http://cb/?code=x"''',
+        """http://cb/?code=x'""",
+        'http://cb/?code=x\\',
+        'http://cb/?code=x\nimport os; os.system("id")',
+        'http://cb/?code=x"""\nimport os\n"""',
+        "http://cb/?code=x'''",
+        'http://cb/?code=x\\"\\n',
+        '"; import os; os.system("id"); "',
+    ]
+
+    def _wrapper_source(self, api_v3_client, redirect_url):
+        captured = {}
+
+        def capture(args, **kwargs):
+            captured["source"] = Path(args[1]).read_text()
+            return completed()
+
+        with patch.object(subprocess, "run", side_effect=capture):
+            api_v3_client.post(self.URL, json={"redirect_url": redirect_url})
+        return captured["source"]
+
+    @pytest.mark.parametrize("redirect_url", ADVERSARIAL)
+    def test_wrapper_is_still_valid_python(self, api_v3_client, plugin_dir, redirect_url):
+        # If escaping failed, the generated file would not parse at all.
+        source = self._wrapper_source(api_v3_client, redirect_url)
+        ast.parse(source)
+
+    @pytest.mark.parametrize("redirect_url", ADVERSARIAL)
+    def test_url_survives_as_one_string_literal(
+            self, api_v3_client, plugin_dir, redirect_url):
+        # Stronger than "it parses": the URL must still be a single string
+        # assigned to redirect_url, not code that escaped into statements.
+        source = self._wrapper_source(api_v3_client, redirect_url)
+        tree = ast.parse(source)
+        assigned = [
+            node.value.value for node in ast.walk(tree)
+            if isinstance(node, ast.Assign)
+            and isinstance(node.value, ast.Constant)
+            and any(getattr(t, "id", None) == "redirect_url" for t in node.targets)
+        ]
+        assert assigned == [redirect_url.strip()]
+
+    def test_injected_call_does_not_become_a_statement(self, api_v3_client, plugin_dir):
+        source = self._wrapper_source(
+            api_v3_client, 'http://cb/\nimport os; os.system("id")')
+        tree = ast.parse(source)
+        imported = {
+            alias.name for node in ast.walk(tree)
+            if isinstance(node, ast.Import) for alias in node.names
+        }
+        # The wrapper legitimately imports sys, subprocess and os; what it
+        # must not gain is a *call* smuggled in through the URL.
+        calls = [
+            node for node in ast.walk(tree)
+            if isinstance(node, ast.Call)
+            and isinstance(node.func, ast.Attribute)
+            and node.func.attr == "system"
+        ]
+        assert calls == []
+
+
+class TestSpotifyStepOne:
+    """No redirect_url — the OAuth-URL path, which imports the script."""
+
+    URL = "/api/v3/plugins/authenticate/spotify"
+
+    def test_script_without_credentials_helper_is_an_error(
+            self, api_v3_client, plugin_dir):
+        # The stub script defines neither get_auth_url nor
+        # load_spotify_credentials, so no URL can be produced.
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code in (400, 500)
+        assert response.get_json()["status"] == "error"
+
+    def test_unusable_credentials_do_not_leak_into_the_response(
+            self, api_v3_client, plugin_dir):
+        (plugin_dir / "authenticate_spotify.py").write_text(
+            "def load_spotify_credentials():\n"
+            "    return ('id-abc', 'super-secret-value', None)\n"
+        )
+        response = api_v3_client.post(self.URL, json={})
+        assert "super-secret-value" not in response.get_data(as_text=True)
+
+    def test_script_raising_on_import_is_handled(self, api_v3_client, plugin_dir):
+        (plugin_dir / "authenticate_spotify.py").write_text("raise RuntimeError('boom')\n")
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 500
+        assert response.get_json()["status"] == "error"
+
+    def test_bodyless_post_reaches_step_one(self, api_v3_client, plugin_dir):
+        # Covered by the silent=True fix: previously a 500 from body parsing.
+        response = api_v3_client.post(self.URL)
+        assert response.status_code in (400, 500)
+        assert response.get_json()["status"] == "error"
+
+    def test_whitespace_redirect_url_is_treated_as_absent(
+            self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed()) as run:
+            api_v3_client.post(self.URL, json={"redirect_url": "   "})
+        # Step 2 never runs, so no wrapper is executed.
+        run.assert_not_called()
+
+
+class TestYouTubeMusic:
+    """No wrapper script and no redirect_url — deliberately not symmetric."""
+
+    URL = "/api/v3/plugins/authenticate/ytm"
+
+    def test_missing_plugin_directory_is_404(self, api_v3_client, api_v3_module, tmp_path):
+        api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(
+            tmp_path / "not-installed")
+        assert api_v3_client.post(self.URL).status_code == 404
+
+    def test_missing_script_is_404(self, api_v3_client, plugin_dir):
+        (plugin_dir / "authenticate_ytm.py").unlink()
+        response = api_v3_client.post(self.URL)
+        assert response.status_code == 404
+        assert "script not found" in response.get_json()["message"]
+
+    def test_success(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed(0, "authorized")):
+            response = api_v3_client.post(self.URL)
+        assert response.status_code == 200
+        assert response.get_json()["output"] == "authorized"
+
+    def test_failure_is_a_400_with_combined_output(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed(1, "out", "err")):
+            response = api_v3_client.post(self.URL)
+        assert response.status_code == 400
+        assert response.get_json()["output"] == "outerr"
+
+    def test_timeout_is_a_408(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run",
+                          side_effect=subprocess.TimeoutExpired("python3", 60)):
+            assert api_v3_client.post(self.URL).status_code == 408
+
+    def test_runs_the_script_directly_without_a_shell(self, api_v3_client, plugin_dir):
+        with patch.object(subprocess, "run", return_value=completed()) as run:
+            api_v3_client.post(self.URL)
+        args, kwargs = run.call_args
+        assert args[0][0] == "python3"
+        assert args[0][1].endswith("authenticate_ytm.py")
+        assert kwargs.get("shell") in (None, False)
+        assert kwargs["timeout"] == 60

test/test_api_v3_optional_body.py

@@ -0,0 +1,136 @@
+"""
+Regression tests: POST endpoints whose body is optional must accept a
+request that has no body at all.
+
+Six handlers in api_v3 read their body as ``request.get_json() or {}``.
+The ``or {}`` states the intent plainly — every field is optional, so a
+bodyless POST should fall back to defaults. But ``get_json()`` without
+``silent=True`` raises ``UnsupportedMediaType`` when the request carries
+no JSON Content-Type, and it raises *before* ``or {}`` is evaluated. Each
+handler's catch-all then turned that into a 500.
+
+So the natural way to call these endpoints — a POST with no body, which
+is what curl, a fetch() without options, and most HTTP clients send by
+default — failed on every one of them. The shipped UI always sends a JSON
+object, which is why this went unnoticed.
+
+This file covers the endpoints whose bodyless behaviour is not already
+tested in their own suite.
+"""
+
+import re
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
+
+
+class TestOnDemandStart:
+    URL = "/api/v3/display/on-demand/start"
+
+    def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module):
+        response = api_v3_client.post(self.URL)
+        # The endpoint may still reject the request on its own terms (no
+        # plugin_id, nothing to display); what it must not do is fail with
+        # a 500 raised out of body parsing.
+        assert response.status_code != 500
+
+    def test_json_body_still_works(self, api_v3_client, api_v3_module):
+        assert api_v3_client.post(self.URL, json={}).status_code != 500
+
+
+class TestResetPluginConfig:
+    URL = "/api/v3/plugins/config/reset"
+
+    def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module):
+        assert api_v3_client.post(self.URL).status_code != 500
+
+    def test_json_body_still_works(self, api_v3_client, api_v3_module):
+        assert api_v3_client.post(self.URL, json={}).status_code != 500
+
+
+class TestDeleteOfTheDayJson:
+    URL = "/api/v3/plugins/of-the-day/json/delete"
+
+    def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module):
+        assert api_v3_client.post(self.URL).status_code != 500
+
+    def test_json_body_still_works(self, api_v3_client, api_v3_module):
+        assert api_v3_client.post(self.URL, json={}).status_code != 500
+
+
+class TestPluginLimits:
+    URL = "/api/v3/plugins/clock/limits"
+
+    def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module):
+        assert api_v3_client.post(self.URL).status_code != 500
+
+
+class TestMissingBodyGivesTheDeclaredError:
+    """Handlers that answer "No data provided" must actually be able to.
+
+    A second group of handlers reads `data = request.get_json()` and then
+    guards with `if not data: return 400`. That guard is unreachable for a
+    request with no JSON body, because get_json() raises first — so the
+    caller got a 500 "an error occurred; see logs for details" instead of
+    the 400 the handler plainly intends to send.
+    """
+
+    @pytest.mark.parametrize("url", [
+        "/api/v3/plugins/install",
+        "/api/v3/plugins/install-from-url",
+        "/api/v3/plugins/registry-from-url",
+        "/api/v3/config/raw/main",
+        "/api/v3/config/raw/secrets",
+        "/api/v3/cache/delete",
+    ])
+    def test_bodyless_post_gets_a_400_not_a_500(self, api_v3_client, api_v3_module, url):
+        response = api_v3_client.post(url)
+        assert response.status_code == 400, (
+            f"{url} answered {response.status_code}: "
+            f"{response.get_data(as_text=True)[:200]}")
+
+    @pytest.mark.parametrize("url", [
+        "/api/v3/plugins/install",
+        "/api/v3/config/raw/main",
+    ])
+    def test_malformed_json_gets_a_400_not_a_500(self, api_v3_client, api_v3_module, url):
+        response = api_v3_client.post(
+            url, data="{not json", content_type="application/json")
+        assert response.status_code == 400
+
+
+class TestNoBodyReadContradictsItsOwnGuard:
+    SOURCE = Path(__file__).parent.parent / "web_interface/blueprints/api_v3.py"
+
+    def test_no_or_default_read_is_unguarded(self):
+        """`get_json() or <default>` is a contradiction without silent=True.
+
+        Writing `or {}` declares the body optional; omitting silent=True
+        means the call raises before the default can apply.
+        """
+        offenders = [
+            line.strip() for line in self.SOURCE.read_text().splitlines()
+            if "request.get_json()" in line and " or " in line
+        ]
+        assert offenders == [], (
+            "these reads declare a default but raise before reaching it; "
+            f"use get_json(silent=True): {offenders}")
+
+    def test_no_not_data_guard_is_unreachable(self):
+        """A `if not data:` guard needs a read that can actually return None."""
+        lines = self.SOURCE.read_text().splitlines()
+        offenders = []
+        for i, line in enumerate(lines):
+            if re.search(r"=\s*request\.get_json\(\)\s*$", line):
+                window = "\n".join(lines[i + 1:i + 3])
+                if re.search(r"if\s+(not\s+data\b|data\s+is\s+None)", window):
+                    offenders.append(f"line {i + 1}: {line.strip()}")
+        assert offenders == [], (
+            "these handlers guard on a missing body but raise before the "
+            f"guard runs; use get_json(silent=True): {offenders}")

test/test_api_v3_plugin_install_endpoints.py

@@ -0,0 +1,302 @@
+"""
+Endpoint tests for POST /plugins/install and POST /plugins/install-from-url.
+
+Both were only ever tested at the PluginStoreManager layer, so the route
+logic — the queue-vs-direct branch, schema invalidation, plugin discovery,
+state and history recording — was unexercised.
+
+/plugins/install carries the same install logic twice: once inside the
+operation-queue callback and once in the direct fallback. The paired
+tests below assert both branches produce the same side effects, so the
+duplication cannot quietly drift.
+"""
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
+
+INSTALL = "/api/v3/plugins/install"
+FROM_URL = "/api/v3/plugins/install-from-url"
+
+
+@pytest.fixture
+def queued(api_v3_module):
+    """Enable the operation queue and run its callback synchronously."""
+    queue = MagicMock()
+
+    def enqueue(operation_type, plugin_id, operation_callback=None):
+        queue.callback_result = operation_callback(MagicMock())
+        return "op-123"
+
+    queue.enqueue_operation.side_effect = enqueue
+    api_v3_module.api_v3.operation_queue = queue
+    return queue
+
+
+def side_effects(module):
+    """The manager calls a successful install is expected to make."""
+    api = module.api_v3
+    return {
+        "schema_invalidated": api.schema_manager.invalidate_cache.call_args_list,
+        "discovered": api.plugin_manager.discover_plugins.call_count,
+        "loaded": api.plugin_manager.load_plugin.call_args_list,
+        "state_set": api.plugin_state_manager.set_plugin_installed.call_args_list,
+        "history": api.operation_history.record_operation.call_args_list,
+    }
+
+
+class TestInstallValidation:
+    def test_uninitialized_store_manager_is_a_500(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager = None
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert response.status_code == 500
+        assert "not initialized" in response.get_json()["message"]
+
+    def test_missing_plugin_id_is_a_400(self, api_v3_client, api_v3_module):
+        response = api_v3_client.post(INSTALL, json={})
+        assert response.status_code == 400
+        assert "plugin_id required" in response.get_json()["message"]
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.assert_not_called()
+
+    def test_empty_body_is_a_400(self, api_v3_client, api_v3_module):
+        assert api_v3_client.post(INSTALL, json=None).status_code == 400
+
+
+class TestInstallDirectPath:
+    """operation_queue is None — the fallback branch."""
+
+    def test_success(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert response.status_code == 200
+        assert response.get_json()["status"] == "success"
+
+    def test_success_side_effects(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        effects = side_effects(api_v3_module)
+        assert effects["schema_invalidated"] == [(("clock",), {})]
+        assert effects["discovered"] == 1
+        assert effects["loaded"] == [(("clock",), {})]
+        assert effects["state_set"] == [(("clock",), {})]
+        assert effects["history"][0].kwargs["status"] == "success"
+
+    def test_branch_forwarded_to_the_manager(self, api_v3_client, api_v3_module):
+        manager = api_v3_module.api_v3.plugin_store_manager
+        manager.install_plugin.return_value = True
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
+        manager.install_plugin.assert_called_once_with("clock", branch="dev")
+
+    def test_branch_named_in_the_message(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
+        assert "(branch: dev)" in response.get_json()["message"]
+
+    def test_failure_is_a_500(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert response.status_code == 500
+        assert "Failed to install" in response.get_json()["message"]
+
+    def test_failure_mentions_missing_registry_entry(self, api_v3_client, api_v3_module):
+        manager = api_v3_module.api_v3.plugin_store_manager
+        manager.install_plugin.return_value = False
+        manager.get_plugin_info.return_value = None
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "ghost"})
+        assert "not found in registry" in response.get_json()["message"]
+
+    def test_failure_omits_registry_note_when_plugin_is_known(
+            self, api_v3_client, api_v3_module):
+        manager = api_v3_module.api_v3.plugin_store_manager
+        manager.install_plugin.return_value = False
+        manager.get_plugin_info.return_value = {"id": "clock"}
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert "not found in registry" not in response.get_json()["message"]
+
+    def test_failure_recorded_in_history(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        record = api_v3_module.api_v3.operation_history.record_operation.call_args
+        assert record.kwargs["status"] == "failed"
+
+    def test_no_side_effects_on_failure(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        effects = side_effects(api_v3_module)
+        assert effects["schema_invalidated"] == []
+        assert effects["loaded"] == []
+        assert effects["state_set"] == []
+
+
+class TestInstallQueuedPath:
+    """operation_queue present — the callback branch."""
+
+    def test_returns_an_operation_id(self, api_v3_client, api_v3_module, queued):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert response.status_code == 200
+        assert response.get_json()["data"]["operation_id"] == "op-123"
+
+    def test_message_says_queued(self, api_v3_client, api_v3_module, queued):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert "queued" in response.get_json()["message"]
+
+    def test_callback_success_side_effects(self, api_v3_client, api_v3_module, queued):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        effects = side_effects(api_v3_module)
+        assert effects["schema_invalidated"] == [(("clock",), {})]
+        assert effects["discovered"] == 1
+        assert effects["loaded"] == [(("clock",), {})]
+        assert effects["state_set"] == [(("clock",), {})]
+        assert effects["history"][0].kwargs["status"] == "success"
+
+    def test_callback_reports_success(self, api_v3_client, api_v3_module, queued):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert queued.callback_result["success"] is True
+
+    def test_callback_failure_raises_for_the_queue(self, api_v3_client, api_v3_module, queued):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
+        # The callback signals failure by raising, so the queue can mark the
+        # operation failed; the route's catch-all turns it into a 500.
+        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        assert response.status_code == 500
+
+    def test_callback_failure_recorded_in_history(self, api_v3_client, api_v3_module, queued):
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
+        record = api_v3_module.api_v3.operation_history.record_operation.call_args
+        assert record.kwargs["status"] == "failed"
+
+    def test_branch_forwarded_from_the_callback(self, api_v3_client, api_v3_module, queued):
+        manager = api_v3_module.api_v3.plugin_store_manager
+        manager.install_plugin.return_value = True
+        api_v3_client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
+        manager.install_plugin.assert_called_once_with("clock", branch="dev")
+
+
+class TestInstallPathsAgree:
+    """The queue callback and the direct fallback duplicate the same logic."""
+
+    def _run(self, client, module, install_ok, queue):
+        module.api_v3.plugin_store_manager.install_plugin.return_value = install_ok
+        client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
+        return side_effects(module)
+
+    def test_success_side_effects_match(self, api_v3_client, api_v3_module):
+        direct = self._run(api_v3_client, api_v3_module, True, None)
+
+        # Reset and re-run through the queue.
+        for mock in (api_v3_module.api_v3.schema_manager,
+                     api_v3_module.api_v3.plugin_manager,
+                     api_v3_module.api_v3.plugin_state_manager,
+                     api_v3_module.api_v3.operation_history):
+            mock.reset_mock()
+        queue = MagicMock()
+        queue.enqueue_operation.side_effect = (
+            lambda t, p, operation_callback=None: operation_callback(MagicMock()) and "op")
+        api_v3_module.api_v3.operation_queue = queue
+        queued = self._run(api_v3_client, api_v3_module, True, queue)
+
+        assert direct["schema_invalidated"] == queued["schema_invalidated"]
+        assert direct["discovered"] == queued["discovered"]
+        assert direct["loaded"] == queued["loaded"]
+        assert direct["state_set"] == queued["state_set"]
+        assert (direct["history"][0].kwargs["status"]
+                == queued["history"][0].kwargs["status"])
+        assert (direct["history"][0].kwargs["details"]
+                == queued["history"][0].kwargs["details"])
+
+    def test_only_the_message_wording_differs(self, api_v3_client, api_v3_module):
+        # Characterized: the direct path says "Plugin installed
+        # successfully" while the queue callback says "Plugin clock
+        # installed successfully". Cosmetic, and the queue's text is
+        # internal to the operation record rather than the HTTP response.
+        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
+        direct = api_v3_client.post(INSTALL, json={"plugin_id": "clock"}).get_json()
+        assert direct["message"] == "Plugin installed successfully"
+
+
+class TestInstallFromUrl:
+    def test_uninitialized_store_manager_is_a_500(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager = None
+        assert api_v3_client.post(FROM_URL, json={"repo_url": "http://x"}).status_code == 500
+
+    def test_missing_repo_url_is_a_400(self, api_v3_client, api_v3_module):
+        response = api_v3_client.post(FROM_URL, json={})
+        assert response.status_code == 400
+        assert "repo_url required" in response.get_json()["message"]
+
+    def test_success(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
+            "success": True, "plugin_id": "clock", "name": "Clock"}
+        response = api_v3_client.post(FROM_URL, json={"repo_url": "https://github.com/o/r"})
+        assert response.status_code == 200
+        body = response.get_json()
+        assert body["plugin_id"] == "clock"
+        assert body["name"] == "Clock"
+
+    def test_all_optional_arguments_forwarded(self, api_v3_client, api_v3_module):
+        manager = api_v3_module.api_v3.plugin_store_manager
+        manager.install_from_url.return_value = {"success": True, "plugin_id": "clock"}
+        api_v3_client.post(FROM_URL, json={
+            "repo_url": "  https://github.com/o/r  ",
+            "plugin_id": "clock",
+            "plugin_path": "plugins/clock",
+            "branch": "dev",
+        })
+        manager.install_from_url.assert_called_once_with(
+            repo_url="https://github.com/o/r",
+            plugin_id="clock",
+            plugin_path="plugins/clock",
+            branch="dev",
+        )
+
+    def test_success_invalidates_schema_and_loads_plugin(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
+            "success": True, "plugin_id": "clock"}
+        api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
+        api_v3_module.api_v3.schema_manager.invalidate_cache.assert_called_once_with("clock")
+        api_v3_module.api_v3.plugin_manager.load_plugin.assert_called_once_with("clock")
+
+    def test_success_without_plugin_id_skips_discovery(self, api_v3_client, api_v3_module):
+        # install_from_url can succeed without naming the plugin; there is
+        # then nothing to invalidate or load.
+        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
+            "success": True, "plugin_id": None}
+        api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
+        api_v3_module.api_v3.schema_manager.invalidate_cache.assert_not_called()
+        api_v3_module.api_v3.plugin_manager.load_plugin.assert_not_called()
+
+    def test_branch_from_result_included(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
+            "success": True, "plugin_id": "clock", "branch": "dev"}
+        body = api_v3_client.post(FROM_URL, json={"repo_url": "http://x"}).get_json()
+        assert body["branch"] == "dev"
+        assert "(branch: dev)" in body["message"]
+
+    def test_failure_reports_the_managers_error(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
+            "success": False, "error": "repo not found"}
+        response = api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
+        assert response.status_code == 500
+        assert response.get_json()["message"] == "repo not found"
+
+    def test_failure_without_error_uses_fallback_text(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
+            "success": False}
+        response = api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
+        assert "Failed to install plugin from URL" in response.get_json()["message"]
+
+    def test_manager_exception_is_a_500(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.install_from_url.side_effect = (
+            RuntimeError("boom"))
+        assert api_v3_client.post(FROM_URL, json={"repo_url": "http://x"}).status_code == 500

test/test_api_v3_registry_endpoints.py

@@ -0,0 +1,179 @@
+"""
+Endpoint tests for the plugin-registry routes in api_v3:
+POST /plugins/store/refresh and POST /plugins/registry-from-url.
+
+Both reach out to the network through PluginStoreManager (mocked here) and
+had no endpoint-level coverage; registry-from-url in particular takes a
+user-supplied URL and hands it straight to the manager.
+"""
+
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
+
+
+class TestRefreshPluginStore:
+    URL = "/api/v3/plugins/store/refresh"
+
+    def test_uninitialized_manager_is_a_500(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager = None
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 500
+        assert "not initialized" in response.get_json()["message"]
+
+    def test_success_reports_plugin_count(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {
+            "plugins": [{"id": "a"}, {"id": "b"}, {"id": "c"}]}
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 200
+        assert response.get_json()["plugin_count"] == 3
+
+    def test_forces_a_refresh_rather_than_using_cache(self, api_v3_client, api_v3_module):
+        manager = api_v3_module.api_v3.plugin_store_manager
+        manager.fetch_registry.return_value = {"plugins": []}
+        api_v3_client.post(self.URL, json={})
+        manager.fetch_registry.assert_called_once_with(force_refresh=True)
+
+    def test_empty_registry_reports_zero(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {}
+        response = api_v3_client.post(self.URL, json={})
+        assert response.get_json()["plugin_count"] == 0
+
+    def test_no_body_is_accepted(self, api_v3_client, api_v3_module):
+        # Regression: `request.get_json() or {}` says a missing body is
+        # fine, but get_json() raises UnsupportedMediaType before `or {}`
+        # is reached, so a bodyless POST — the natural way to call a
+        # refresh endpoint — came back 500.
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
+        assert api_v3_client.post(self.URL).status_code == 200
+
+    def test_body_without_json_content_type_is_accepted(
+            self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
+        response = api_v3_client.post(self.URL, data="", content_type="text/plain")
+        assert response.status_code == 200
+
+    def test_malformed_json_body_falls_back_to_defaults(
+            self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
+        response = api_v3_client.post(
+            self.URL, data="{not json", content_type="application/json")
+        assert response.status_code == 200
+
+    @pytest.mark.parametrize("key", ["fetch_commit_info", "fetch_latest_versions"])
+    def test_either_commit_info_key_extends_the_message(
+            self, api_v3_client, api_v3_module, key):
+        # fetch_latest_versions is the older spelling; both must work.
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
+        response = api_v3_client.post(self.URL, json={key: True})
+        assert "commit metadata" in response.get_json()["message"]
+
+    def test_message_stays_plain_without_the_flag(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
+        response = api_v3_client.post(self.URL, json={})
+        assert response.get_json()["message"] == "Plugin store refreshed"
+
+    def test_network_failure_is_a_500(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.side_effect = (
+            ConnectionError("github unreachable"))
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 500
+        assert response.get_json()["message"] == "An error occurred; see logs for details"
+
+    def test_failure_body_carries_no_traceback_or_paths(
+            self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry.side_effect = (
+            RuntimeError("failed at /home/user/LEDMatrix/src/secret.py line 42"))
+        body = api_v3_client.post(self.URL, json={}).get_json()
+        assert "Traceback" not in str(body)
+        # `details` is describe_exception output: one line, type-named,
+        # credential-redacted. It may quote the message, but never a stack.
+        assert body["details"].startswith("RuntimeError:")
+        assert "\n" not in body["details"]
+
+
+class TestRegistryFromUrl:
+    URL = "/api/v3/plugins/registry-from-url"
+
+    def test_uninitialized_manager_is_a_500(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager = None
+        response = api_v3_client.post(self.URL, json={"repo_url": "http://x"})
+        assert response.status_code == 500
+
+    def test_missing_repo_url_is_a_400(self, api_v3_client, api_v3_module):
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 400
+        assert "repo_url required" in response.get_json()["message"]
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.assert_not_called()
+
+    def test_success_returns_the_plugin_list(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = {
+            "plugins": [{"id": "clock"}]}
+        response = api_v3_client.post(
+            self.URL, json={"repo_url": "https://github.com/o/r"})
+        assert response.status_code == 200
+        body = response.get_json()
+        assert body["plugins"] == [{"id": "clock"}]
+        assert body["registry_url"] == "https://github.com/o/r"
+
+    def test_url_is_trimmed_before_use(self, api_v3_client, api_v3_module):
+        manager = api_v3_module.api_v3.plugin_store_manager
+        manager.fetch_registry_from_url.return_value = {"plugins": []}
+        api_v3_client.post(self.URL, json={"repo_url": "  https://github.com/o/r  "})
+        manager.fetch_registry_from_url.assert_called_once_with("https://github.com/o/r")
+
+    def test_registry_without_plugins_key_returns_empty_list(
+            self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = {
+            "other": 1}
+        response = api_v3_client.post(self.URL, json={"repo_url": "http://x"})
+        assert response.get_json()["plugins"] == []
+
+    def test_no_registry_found_is_a_400(self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = None
+        response = api_v3_client.post(self.URL, json={"repo_url": "http://x/not-a-registry"})
+        assert response.status_code == 400
+        assert "Failed to fetch registry" in response.get_json()["message"]
+
+    @pytest.mark.parametrize("url", [
+        "not a url",
+        "javascript:alert(1)",
+        "file:///etc/passwd",
+        "http://localhost:8080/admin",
+    ])
+    def test_unusable_urls_fail_cleanly(self, api_v3_client, api_v3_module, url):
+        # Characterization: the handler performs no URL validation of its
+        # own — whatever the manager makes of the URL decides the outcome.
+        # What is pinned here is that a rejected URL produces a clean 400
+        # rather than a traceback or a 500.
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = None
+        response = api_v3_client.post(self.URL, json={"repo_url": url})
+        assert response.status_code == 400
+        assert "Traceback" not in str(response.get_json())
+
+    def test_fetch_exception_is_a_500_without_internals(
+            self, api_v3_client, api_v3_module):
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.side_effect = (
+            ValueError("parse failed in /srv/app/internal.py"))
+        response = api_v3_client.post(self.URL, json={"repo_url": "http://x"})
+        assert response.status_code == 500
+        body = response.get_json()
+        assert body["message"] == "An error occurred; see logs for details"
+        assert "Traceback" not in str(body)
+
+    def test_non_string_repo_url_is_rejected(self, api_v3_client, api_v3_module):
+        # Regression: .strip() on a non-string raised, and the catch-all
+        # reported the caller's own mistake as a server fault.
+        response = api_v3_client.post(self.URL, json={"repo_url": 12345})
+        assert response.status_code == 400
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.assert_not_called()
+
+    def test_blank_repo_url_is_rejected(self, api_v3_client, api_v3_module):
+        response = api_v3_client.post(self.URL, json={"repo_url": "   "})
+        assert response.status_code == 400
+        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.assert_not_called()

test/test_api_v3_wifi_endpoints.py

@@ -0,0 +1,240 @@
+"""
+Endpoint tests for the /wifi/* routes in api_v3.
+
+These routes drive the host's actual networking — connecting, dropping a
+connection, switching the radio off — and had no endpoint-level tests at
+all. WiFiManager is mocked throughout; nothing here may touch real
+networking.
+
+Each handler does `from src.wifi_manager import WiFiManager` inside the
+function body, so the patch target is the class at its definition site.
+"""
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
+
+
+@pytest.fixture
+def wifi_manager():
+    """Patch WiFiManager where it is defined; yield the instance mock."""
+    with patch("src.wifi_manager.WiFiManager") as cls:
+        instance = MagicMock()
+        cls.return_value = instance
+        yield instance
+
+
+class TestConnect:
+    URL = "/api/v3/wifi/connect"
+
+    def test_success(self, api_v3_client, wifi_manager):
+        wifi_manager.connect_to_network.return_value = (True, "Connected to HomeNet")
+        response = api_v3_client.post(self.URL, json={"ssid": "HomeNet", "password": "pw"})
+        assert response.status_code == 200
+        assert response.get_json()["message"] == "Connected to HomeNet"
+        wifi_manager.connect_to_network.assert_called_once_with("HomeNet", "pw")
+
+    def test_missing_body_rejected(self, api_v3_client, wifi_manager):
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 400
+        wifi_manager.connect_to_network.assert_not_called()
+
+    def test_missing_ssid_rejected(self, api_v3_client, wifi_manager):
+        response = api_v3_client.post(self.URL, json={"password": "pw"})
+        assert response.status_code == 400
+        assert "SSID is required" in response.get_json()["message"]
+        wifi_manager.connect_to_network.assert_not_called()
+
+    @pytest.mark.parametrize("ssid", ["", "   ", "\t"])
+    def test_blank_ssid_rejected(self, api_v3_client, wifi_manager, ssid):
+        response = api_v3_client.post(self.URL, json={"ssid": ssid})
+        assert response.status_code == 400
+        wifi_manager.connect_to_network.assert_not_called()
+
+    def test_ssid_is_trimmed(self, api_v3_client, wifi_manager):
+        wifi_manager.connect_to_network.return_value = (True, "ok")
+        api_v3_client.post(self.URL, json={"ssid": "  HomeNet  "})
+        wifi_manager.connect_to_network.assert_called_once_with("HomeNet", "")
+
+    def test_missing_password_becomes_empty_string(self, api_v3_client, wifi_manager):
+        wifi_manager.connect_to_network.return_value = (True, "ok")
+        api_v3_client.post(self.URL, json={"ssid": "OpenNet"})
+        wifi_manager.connect_to_network.assert_called_once_with("OpenNet", "")
+
+    def test_null_password_becomes_empty_string(self, api_v3_client, wifi_manager):
+        wifi_manager.connect_to_network.return_value = (True, "ok")
+        api_v3_client.post(self.URL, json={"ssid": "OpenNet", "password": None})
+        wifi_manager.connect_to_network.assert_called_once_with("OpenNet", "")
+
+    def test_failure_reports_the_managers_reason(self, api_v3_client, wifi_manager):
+        wifi_manager.connect_to_network.return_value = (False, "Bad password")
+        response = api_v3_client.post(self.URL, json={"ssid": "HomeNet"})
+        assert response.status_code == 400
+        assert response.get_json()["message"] == "Bad password"
+
+    def test_failure_without_reason_uses_fallback_text(self, api_v3_client, wifi_manager):
+        wifi_manager.connect_to_network.return_value = (False, None)
+        response = api_v3_client.post(self.URL, json={"ssid": "HomeNet"})
+        assert response.status_code == 400
+        assert response.get_json()["message"] == "Failed to connect to network"
+
+    def test_manager_exception_is_a_500_without_leaking_internals(
+            self, api_v3_client, wifi_manager):
+        wifi_manager.connect_to_network.side_effect = RuntimeError(
+            "/usr/lib/secret/path blew up")
+        response = api_v3_client.post(self.URL, json={"ssid": "HomeNet"})
+        assert response.status_code == 500
+        body = response.get_json()
+        assert body["message"] == "An error occurred; see logs for details"
+        # `details` comes from describe_exception, which is deliberately
+        # safe to return (redacted, capped) — it names the type.
+        assert "RuntimeError" in body["details"]
+
+
+class TestDisconnect:
+    URL = "/api/v3/wifi/disconnect"
+
+    def test_success(self, api_v3_client, wifi_manager):
+        wifi_manager.disconnect_from_network.return_value = (True, "Disconnected")
+        response = api_v3_client.post(self.URL)
+        assert response.status_code == 200
+        assert response.get_json()["message"] == "Disconnected"
+
+    def test_failure(self, api_v3_client, wifi_manager):
+        wifi_manager.disconnect_from_network.return_value = (False, "Not connected")
+        response = api_v3_client.post(self.URL)
+        assert response.status_code == 400
+        assert response.get_json()["message"] == "Not connected"
+
+    def test_failure_without_reason_uses_fallback(self, api_v3_client, wifi_manager):
+        wifi_manager.disconnect_from_network.return_value = (False, "")
+        response = api_v3_client.post(self.URL)
+        assert response.get_json()["message"] == "Failed to disconnect from network"
+
+    def test_exception_is_a_500(self, api_v3_client, wifi_manager):
+        wifi_manager.disconnect_from_network.side_effect = OSError("nmcli missing")
+        assert api_v3_client.post(self.URL).status_code == 500
+
+
+class TestApMode:
+    ENABLE = "/api/v3/wifi/ap/enable"
+    DISABLE = "/api/v3/wifi/ap/disable"
+
+    def test_enable_success(self, api_v3_client, wifi_manager):
+        wifi_manager.enable_ap_mode.return_value = (True, "AP enabled")
+        response = api_v3_client.post(self.ENABLE, json={})
+        assert response.status_code == 200
+        wifi_manager.enable_ap_mode.assert_called_once_with(force=False)
+
+    @pytest.mark.parametrize("raw,expected", [
+        (True, True), (False, False),
+        ("true", True), ("TRUE", True), ("1", True),
+        ("false", False), ("no", False), ("yes", False),
+        (1, False),  # only real True or the listed strings count
+    ])
+    def test_force_coercion(self, api_v3_client, wifi_manager, raw, expected):
+        wifi_manager.enable_ap_mode.return_value = (True, "ok")
+        api_v3_client.post(self.ENABLE, json={"force": raw})
+        wifi_manager.enable_ap_mode.assert_called_once_with(force=expected)
+
+    def test_enable_without_body(self, api_v3_client, wifi_manager):
+        wifi_manager.enable_ap_mode.return_value = (True, "ok")
+        assert api_v3_client.post(self.ENABLE).status_code == 200
+
+    def test_enable_failure(self, api_v3_client, wifi_manager):
+        wifi_manager.enable_ap_mode.return_value = (False, "hostapd missing")
+        response = api_v3_client.post(self.ENABLE, json={})
+        assert response.status_code == 400
+        assert response.get_json()["message"] == "hostapd missing"
+
+    def test_disable_success(self, api_v3_client, wifi_manager):
+        wifi_manager.disable_ap_mode.return_value = (True, "AP disabled")
+        assert api_v3_client.post(self.DISABLE).status_code == 200
+
+    def test_disable_failure(self, api_v3_client, wifi_manager):
+        wifi_manager.disable_ap_mode.return_value = (False, "not running")
+        assert api_v3_client.post(self.DISABLE).status_code == 400
+
+    def test_enable_exception_is_a_500(self, api_v3_client, wifi_manager):
+        wifi_manager.enable_ap_mode.side_effect = RuntimeError("boom")
+        assert api_v3_client.post(self.ENABLE, json={}).status_code == 500
+
+
+class TestRadio:
+    URL = "/api/v3/wifi/radio"
+
+    def test_get_state(self, api_v3_client, wifi_manager):
+        wifi_manager.get_wifi_radio_state.return_value = {
+            "enabled": True, "ethernet_connected": False}
+        response = api_v3_client.get(self.URL)
+        assert response.status_code == 200
+        assert response.get_json()["data"]["enabled"] is True
+
+    def test_get_state_exception_is_a_500(self, api_v3_client, wifi_manager):
+        wifi_manager.get_wifi_radio_state.side_effect = OSError("rfkill missing")
+        assert api_v3_client.get(self.URL).status_code == 500
+
+    def test_enabled_is_required(self, api_v3_client, wifi_manager):
+        response = api_v3_client.post(self.URL, json={})
+        assert response.status_code == 400
+        assert "enabled is required" in response.get_json()["message"]
+        wifi_manager.set_wifi_radio.assert_not_called()
+
+    def test_enable_success(self, api_v3_client, wifi_manager):
+        wifi_manager.set_wifi_radio.return_value = (True, "Radio on", None)
+        wifi_manager.get_wifi_radio_state.return_value = {"enabled": True}
+        response = api_v3_client.post(self.URL, json={"enabled": True})
+        assert response.status_code == 200
+        wifi_manager.set_wifi_radio.assert_called_once_with(True, force=False)
+
+    @pytest.mark.parametrize("raw,expected", [
+        (True, True), ("true", True), ("1", True), ("yes", True),
+        (False, False), ("false", False), ("off", False), (0, False),
+    ])
+    def test_enabled_coercion_is_string_aware(
+            self, api_v3_client, wifi_manager, raw, expected):
+        # bool("false") is True, so the endpoint parses strings explicitly
+        # rather than trusting truthiness — it is a public contract, not
+        # only the shipped UI which always sends real JSON booleans.
+        wifi_manager.set_wifi_radio.return_value = (True, "ok", None)
+        wifi_manager.get_wifi_radio_state.return_value = {}
+        api_v3_client.post(self.URL, json={"enabled": raw})
+        wifi_manager.set_wifi_radio.assert_called_once_with(expected, force=False)
+
+    def test_force_passed_through(self, api_v3_client, wifi_manager):
+        wifi_manager.set_wifi_radio.return_value = (True, "ok", None)
+        wifi_manager.get_wifi_radio_state.return_value = {}
+        api_v3_client.post(self.URL, json={"enabled": False, "force": "true"})
+        wifi_manager.set_wifi_radio.assert_called_once_with(False, force=True)
+
+    def test_refusal_reports_reason(self, api_v3_client, wifi_manager):
+        # Disabling the radio without Ethernet would lock the user out of
+        # this very interface, so the manager can refuse with a reason.
+        wifi_manager.set_wifi_radio.return_value = (
+            False, "Refusing: no wired fallback", "no_ethernet")
+        response = api_v3_client.post(self.URL, json={"enabled": False})
+        assert response.status_code == 400
+        body = response.get_json()
+        assert body["reason"] == "no_ethernet"
+        assert "Refusing" in body["message"]
+
+    def test_exception_is_a_500(self, api_v3_client, wifi_manager):
+        wifi_manager.set_wifi_radio.side_effect = RuntimeError("boom")
+        assert api_v3_client.post(self.URL, json={"enabled": True}).status_code == 500
+
+
+class TestNoRealNetworking:
+    def test_wifi_manager_is_never_constructed_for_real(self, api_v3_client):
+        # Guard against a future refactor moving the import to module level,
+        # where the fixture's patch of the definition site would stop
+        # applying and the tests would start driving real networking.
+        with patch("src.wifi_manager.WiFiManager") as cls:
+            cls.return_value.disconnect_from_network.return_value = (True, "ok")
+            api_v3_client.post("/api/v3/wifi/disconnect")
+            assert cls.called

test/test_logo_helper.py

@@ -0,0 +1,423 @@
+"""
+Tests for src/common/logo_helper.py — logo loading, LRU caching, resizing,
+and download-with-fallback. Previously untested: nothing in test/ referenced
+this module at all.
+
+Real PIL images under tmp_path are used rather than mocked ones, since
+load_logo() does real Path.exists() and Image.open() calls; only the HTTP
+session and the permission helpers are patched.
+
+Regression coverage for two fixed bugs:
+- _download_logo wrote response.content to disk with no size cap and no
+  check that the bytes decoded as an image, so a hostile or broken URL
+  could leave arbitrary/oversized content cached in the assets directory.
+- get_cache_stats() divided by self.cache_size unguarded, raising
+  ZeroDivisionError for a helper constructed with cache_size=0.
+"""
+
+import logging
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+import requests
+from PIL import Image, UnidentifiedImageError
+
+from src.common.logo_helper import MAX_LOGO_BYTES, LogoHelper
+
+
+@pytest.fixture(autouse=True)
+def _no_real_chmod(monkeypatch):
+    # Keep the permission helpers out of the way: their own env detection
+    # is not what these tests are about.
+    monkeypatch.setattr("src.common.logo_helper.ensure_directory_permissions", MagicMock())
+    monkeypatch.setattr("src.common.logo_helper.ensure_file_permissions", MagicMock())
+
+
+@pytest.fixture
+def helper():
+    return LogoHelper(display_width=64, display_height=32,
+                      logger=logging.getLogger("test.logo_helper"))
+
+
+def write_logo(path: Path, size=(20, 20), color=(255, 0, 0), fmt="PNG") -> Path:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    Image.new("RGB", size, color).save(path, format=fmt)
+    return path
+
+
+def fake_response(content: bytes, chunk_size: int = 64 * 1024):
+    """Stand-in for a streamed requests.Response.
+
+    _download_logo opens `with session.get(..., stream=True)` and reads
+    through iter_content(), so the fake has to be a context manager that
+    yields the body in pieces rather than exposing it as .content.
+    Chunking is the fake's own, not the caller's, so a test can dribble a
+    body out in small pieces.
+    """
+    response = MagicMock()
+    response.__enter__.return_value = response
+    response.__exit__.return_value = False
+    response.raise_for_status = MagicMock()
+
+    def _iter_content(*_args, **_kwargs):
+        for i in range(0, len(content), chunk_size):
+            yield content[i:i + chunk_size]
+
+    response.iter_content = _iter_content
+    return response
+
+
+def endless_response(chunk: bytes = b"\x00" * 65536):
+    """A server that declares no length and never stops sending.
+
+    This is the case response.content could not survive: it buffers to
+    completion, so the size check never got a chance to run.
+    """
+    response = MagicMock()
+    response.__enter__.return_value = response
+    response.__exit__.return_value = False
+    response.raise_for_status = MagicMock()
+
+    def _iter_content(*_args, **_kwargs):
+        while True:
+            yield chunk
+
+    response.iter_content = _iter_content
+    return response
+
+
+def png_bytes(size=(20, 20), color=(0, 128, 0)) -> bytes:
+    import io
+    buf = io.BytesIO()
+    Image.new("RGB", size, color).save(buf, format="PNG")
+    return buf.getvalue()
+
+
+class TestLoadLogo:
+    def test_loads_and_converts_to_rgba(self, helper, tmp_path):
+        path = write_logo(tmp_path / "PHI.png")
+        logo = helper.load_logo("PHI", path)
+        assert logo is not None
+        assert logo.mode == "RGBA"
+
+    def test_missing_file_returns_none(self, helper, tmp_path, caplog):
+        with caplog.at_level(logging.WARNING):
+            assert helper.load_logo("NOPE", tmp_path / "missing.png") is None
+        assert "Logo not found" in caplog.text
+
+    def test_second_load_is_served_from_cache(self, helper, tmp_path):
+        path = write_logo(tmp_path / "PHI.png")
+        first = helper.load_logo("PHI", path)
+        path.unlink()  # cache hit must not touch the filesystem
+        assert helper.load_logo("PHI", path) is first
+
+    def test_cache_key_includes_requested_size(self, helper, tmp_path):
+        # A panel-size change must not hand back a logo sized for the old
+        # dimensions, so the two sizes get separate cache entries.
+        path = write_logo(tmp_path / "PHI.png", size=(100, 100))
+        small = helper.load_logo("PHI", path, max_width=10, max_height=10)
+        large = helper.load_logo("PHI", path, max_width=50, max_height=50)
+        assert small is not large
+        assert small.size != large.size
+        assert len(helper._logo_cache) == 2
+
+    def test_default_size_is_one_and_a_half_display(self, helper, tmp_path):
+        path = write_logo(tmp_path / "PHI.png", size=(500, 500))
+        logo = helper.load_logo("PHI", path)
+        assert logo.width <= int(64 * 1.5)
+        assert logo.height <= int(32 * 1.5)
+
+    def test_smaller_image_is_not_upscaled(self, helper, tmp_path):
+        path = write_logo(tmp_path / "PHI.png", size=(8, 8))
+        assert helper.load_logo("PHI", path, max_width=64, max_height=64).size == (8, 8)
+
+    def test_larger_image_is_downscaled_preserving_aspect(self, helper, tmp_path):
+        path = write_logo(tmp_path / "PHI.png", size=(200, 100))
+        logo = helper.load_logo("PHI", path, max_width=50, max_height=50)
+        assert logo.width <= 50 and logo.height <= 50
+        assert logo.width == 50 and logo.height == 25  # 2:1 preserved
+
+    def test_string_path_accepted(self, helper, tmp_path):
+        path = write_logo(tmp_path / "PHI.png")
+        assert helper.load_logo("PHI", str(path)) is not None
+
+    def test_corrupt_file_returns_none(self, helper, tmp_path, caplog):
+        bad = tmp_path / "bad.png"
+        bad.write_bytes(b"not an image")
+        with caplog.at_level(logging.ERROR):
+            assert helper.load_logo("BAD", bad) is None
+        assert "Error loading logo" in caplog.text
+
+
+class TestCacheManagement:
+    def test_lru_evicts_oldest(self, tmp_path):
+        helper = LogoHelper(64, 32, cache_size=2, logger=MagicMock())
+        paths = [write_logo(tmp_path / f"T{i}.png") for i in range(3)]
+        for i, path in enumerate(paths):
+            helper.load_logo(f"T{i}", path)
+        assert len(helper._logo_cache) == 2
+        assert not any(k.startswith("T0_") for k in helper._logo_cache)
+
+    def test_cache_hit_refreshes_lru_position(self, tmp_path):
+        helper = LogoHelper(64, 32, cache_size=2, logger=MagicMock())
+        a, b, c = [write_logo(tmp_path / f"{n}.png") for n in ("A", "B", "C")]
+        helper.load_logo("A", a)
+        helper.load_logo("B", b)
+        helper.load_logo("A", a)   # A is now most-recently used
+        helper.load_logo("C", c)   # evicts B, not A
+        assert any(k.startswith("A_") for k in helper._logo_cache)
+        assert not any(k.startswith("B_") for k in helper._logo_cache)
+
+    def test_clear_cache_empties_both_structures(self, helper, tmp_path):
+        helper.load_logo("PHI", write_logo(tmp_path / "PHI.png"))
+        helper.clear_cache()
+        assert helper._logo_cache == {}
+        assert helper._cache_order == []
+
+    def test_cache_stats(self, tmp_path):
+        helper = LogoHelper(64, 32, cache_size=4, logger=MagicMock())
+        helper.load_logo("PHI", write_logo(tmp_path / "PHI.png"))
+        stats = helper.get_cache_stats()
+        assert stats["cached_logos"] == 1
+        assert stats["cache_size_limit"] == 4
+        assert stats["cache_usage_percent"] == 25
+
+    def test_zero_cache_size_does_not_divide_by_zero(self):
+        # Regression: this raised ZeroDivisionError.
+        stats = LogoHelper(64, 32, cache_size=0, logger=MagicMock()).get_cache_stats()
+        assert stats["cache_usage_percent"] == 0
+        assert stats["cache_size_limit"] == 0
+
+
+class TestLoadLogoWithDownload:
+    def test_existing_file_skips_download(self, helper, tmp_path):
+        path = write_logo(tmp_path / "PHI.png")
+        helper.session.get = MagicMock()
+        assert helper.load_logo_with_download("PHI", path, "http://x/logo.png") is not None
+        helper.session.get.assert_not_called()
+
+    def test_downloads_then_loads(self, helper, tmp_path):
+        path = tmp_path / "PHI.png"
+        helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
+        logo = helper.load_logo_with_download("PHI", path, "http://x/logo.png")
+        assert logo is not None
+        assert path.exists()
+        # stream=True is load-bearing: it is what lets the size cap apply
+        # before the body is buffered.
+        helper.session.get.assert_called_once_with(
+            "http://x/logo.png", timeout=30, stream=True)
+
+    def test_download_failure_falls_back_to_placeholder(self, helper, tmp_path):
+        helper.session.get = MagicMock(
+            side_effect=requests.RequestException("connection reset"))
+        logo = helper.load_logo_with_download(
+            "PHI", tmp_path / "PHI.png", "http://x/logo.png",
+            max_width=20, max_height=20)
+        assert logo is not None and logo.size == (20, 20)  # placeholder
+
+    def test_http_error_falls_back_to_placeholder(self, helper, tmp_path):
+        response = fake_response(b"")
+        response.raise_for_status.side_effect = requests.HTTPError("404")
+        helper.session.get = MagicMock(return_value=response)
+        logo = helper.load_logo_with_download(
+            "PHI", tmp_path / "PHI.png", "http://x/logo.png",
+            max_width=20, max_height=20)
+        assert logo is not None and logo.size == (20, 20)
+
+    def test_no_url_and_no_file_gives_placeholder(self, helper, tmp_path):
+        logo = helper.load_logo_with_download(
+            "PHI", tmp_path / "missing.png", None, max_width=16, max_height=16)
+        assert logo is not None and logo.size == (16, 16)
+
+
+class TestDownloadLogo:
+    def test_writes_file_and_sets_permissions(self, helper, tmp_path):
+        path = tmp_path / "assets" / "PHI.png"
+        # Directory creation is ensure_directory_permissions' job, and the
+        # autouse fixture stubs it out — so make the directory here.
+        path.parent.mkdir()
+        helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
+        with patch("src.common.logo_helper.ensure_directory_permissions") as dirs, \
+             patch("src.common.logo_helper.ensure_file_permissions") as files:
+            helper._download_logo("http://x/logo.png", path)
+        assert path.exists()
+        dirs.assert_called_once()
+        files.assert_called_once()
+        assert dirs.call_args[0][0] == path.parent
+
+    def test_oversized_response_is_rejected_without_writing(self, helper, tmp_path):
+        # Regression: an unbounded response.content was written straight to
+        # disk, so a hostile URL chose how many bytes landed in assets/.
+        path = tmp_path / "huge.png"
+        helper.session.get = MagicMock(
+            return_value=fake_response(b"\x00" * (MAX_LOGO_BYTES + 1)))
+        with pytest.raises(ValueError, match="exceeds the"):
+            helper._download_logo("http://x/huge.png", path)
+        assert not path.exists()
+
+    def test_unbounded_response_is_aborted_at_the_cap(self, helper, tmp_path):
+        # Regression: the cap used to be checked against response.content,
+        # which buffers the whole body first — so a server that omits
+        # Content-Length and never stops sending exhausted memory before
+        # the check could run. Streaming counts bytes as they arrive, so
+        # this terminates instead of hanging.
+        path = tmp_path / "endless.png"
+        helper.session.get = MagicMock(return_value=endless_response())
+        with pytest.raises(ValueError, match="exceeds the"):
+            helper._download_logo("http://x/endless.png", path)
+        assert not path.exists()
+
+    def test_no_partial_file_is_left_when_the_stream_dies(self, helper, tmp_path):
+        # A transfer that fails midway must not leave a truncated logo
+        # where the real one belongs — load_logo() would cache it.
+        path = tmp_path / "cut.png"
+        real = png_bytes()
+
+        def _dies_midway(*_args, **_kwargs):
+            yield real[:20]
+            raise OSError("connection reset")
+
+        response = MagicMock()
+        response.__enter__.return_value = response
+        response.__exit__.return_value = False
+        response.raise_for_status = MagicMock()
+        response.iter_content = _dies_midway
+        helper.session.get = MagicMock(return_value=response)
+
+        with pytest.raises(OSError):
+            helper._download_logo("http://x/cut.png", path)
+        assert not path.exists()
+        assert list(tmp_path.glob("*.part")) == []
+
+    def test_concurrent_downloads_do_not_share_a_temp_file(self, helper, tmp_path):
+        # Two plugins can ask for the same logo at once. A fixed
+        # "<name>.part" would let them interleave writes into one file and
+        # publish the mixture; each download gets its own temp name.
+        path = tmp_path / "PHI.png"
+        seen = []
+        real_mkstemp = tempfile.mkstemp
+
+        def record(*args, **kwargs):
+            fd, name = real_mkstemp(*args, **kwargs)
+            seen.append(name)
+            return fd, name
+
+        with patch("src.common.logo_helper.tempfile.mkstemp", side_effect=record):
+            helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
+            helper._download_logo("http://x/logo.png", path)
+            helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
+            helper._download_logo("http://x/logo.png", path)
+
+        assert len(seen) == 2 and seen[0] != seen[1]
+        assert path.exists()
+        assert list(tmp_path.glob("*.part")) == []  # both cleaned up
+
+    def test_request_failure_leaves_no_temp_file(self, helper, tmp_path):
+        # mkstemp creates the file up front, so an error before any bytes
+        # arrive still has something to clean up.
+        helper.session.get = MagicMock(
+            side_effect=requests.RequestException("connection reset"))
+        with pytest.raises(requests.RequestException):
+            helper._download_logo("http://x/logo.png", tmp_path / "PHI.png")
+        assert list(tmp_path.glob("*")) == []
+
+    def test_non_image_response_is_deleted_and_raises(self, helper, tmp_path):
+        # Regression: undecodable bytes stayed on disk, so every later
+        # load_logo() call hit the corrupt file instead of re-downloading.
+        path = tmp_path / "bad.png"
+        helper.session.get = MagicMock(return_value=fake_response(b"<html>404</html>"))
+        # Specifically Pillow's identify failure, not any OSError: the
+        # point is that the bytes did not decode, and OSError alone would
+        # also admit unrelated filesystem faults.
+        with pytest.raises(UnidentifiedImageError):
+            helper._download_logo("http://x/bad.png", path)
+        assert not path.exists()
+        assert list(tmp_path.glob("*.part")) == []
+
+    def test_decompression_bomb_is_deleted_and_raises(self, helper, tmp_path, monkeypatch):
+        path = tmp_path / "bomb.png"
+        helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
+
+        class Bomb:
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *a):
+                return False
+
+            def load(self):
+                raise Image.DecompressionBombError("too many pixels")
+
+        monkeypatch.setattr("src.common.logo_helper.Image.open", lambda *a, **kw: Bomb())
+        with pytest.raises(Image.DecompressionBombError):
+            helper._download_logo("http://x/bomb.png", path)
+        assert not path.exists()
+
+    def test_bad_download_surfaces_as_placeholder_not_crash(self, helper, tmp_path):
+        # The new guards raise, and load_logo_with_download's existing
+        # broad except turns that into the placeholder path.
+        helper.session.get = MagicMock(return_value=fake_response(b"garbage"))
+        logo = helper.load_logo_with_download(
+            "PHI", tmp_path / "PHI.png", "http://x/bad.png",
+            max_width=12, max_height=12)
+        assert logo is not None and logo.size == (12, 12)
+
+
+class TestLogoVariations:
+    def test_plain_abbreviation_returns_itself(self, helper):
+        assert helper.get_logo_variations("PHI") == ["PHI"]
+
+    def test_ampersand_expanded(self, helper):
+        assert "TAAND M" in helper.get_logo_variations("TA& M")
+
+    def test_and_contracted(self, helper):
+        assert "T&M" in helper.get_logo_variations("TANDM")
+
+    def test_special_case_appends_known_aliases(self, helper):
+        variations = helper.get_logo_variations("TA&M")
+        assert "TAMU" in variations and "TEXASAM" in variations
+        assert "TAANDM" in variations  # the generic & rule still applies
+
+
+class TestNormalizeAbbreviation:
+    def test_uppercases_and_strips(self, helper):
+        assert helper.normalize_abbreviation("  phi  ") == "PHI"
+
+    def test_ampersand_becomes_and(self, helper):
+        assert helper.normalize_abbreviation("TA&M") == "TAANDM"
+
+    def test_internal_spaces_removed(self, helper):
+        assert helper.normalize_abbreviation("New York") == "NEWYORK"
+
+    def test_deliberately_differs_from_logo_downloader(self, helper):
+        # Pinned, not a bug: LogoDownloader.normalize_abbreviation replaces
+        # filesystem-unsafe characters but keeps spaces, and plugins call
+        # that one. Changing either changes which logo filenames resolve on
+        # existing installs. Both docstrings say so explicitly.
+        from src.logo_downloader import LogoDownloader
+        assert helper.normalize_abbreviation("New York") == "NEWYORK"
+        assert LogoDownloader.normalize_abbreviation("New York") == "NEW YORK"
+
+
+class TestPlaceholderLogo:
+    def test_uses_requested_dimensions(self, helper):
+        assert helper._create_placeholder_logo("PHI", 30, 20).size == (30, 20)
+
+    def test_defaults_to_one_and_a_half_display(self, helper):
+        assert helper._create_placeholder_logo("PHI").size == (96, 48)
+
+    def test_is_rgba(self, helper):
+        assert helper._create_placeholder_logo("PHI", 10, 10).mode == "RGBA"
+
+    def test_invalid_dimensions_return_none(self, helper, caplog):
+        with caplog.at_level(logging.ERROR):
+            assert helper._create_placeholder_logo("PHI", -5, -5) is None
+        assert "Error creating placeholder" in caplog.text
+
+
+class TestSessionConfiguration:
+    def test_user_agent_and_accept_headers(self, helper):
+        assert helper.session.headers["User-Agent"] == "LEDMatrix-Common/1.0"
+        assert helper.session.headers["Accept"] == "image/*"

test/web_interface/test_api_v3_backup_paths.py

@@ -0,0 +1,220 @@
+"""
+Path-containment tests for the backup file routes:
+GET /backup/download/<filename>, DELETE /backup/<filename>, and the
+listing/validation routes alongside them.
+
+Both filename routes take user input straight from the URL and turn it
+into a filesystem path, one to read and one to unlink. `_safe_backup_path`
+is what stops that from reaching outside the export directory, and it had
+no tests.
+
+This is verification of existing containment, not a fix: no bypass was
+found. The tests exist so that a later "just let dots through" change has
+to argue with something.
+"""
+
+import io
+import json
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+from flask import Flask
+
+project_root = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(project_root))
+
+from web_interface.blueprints import api_v3 as api_v3_module  # noqa: E402
+from web_interface.blueprints.api_v3 import api_v3  # noqa: E402
+
+_MANAGER_ATTRS = (
+    'config_manager', 'plugin_manager', 'plugin_store_manager',
+    'plugin_state_manager', 'saved_repositories_manager', 'schema_manager',
+    'operation_queue', 'operation_history', 'cache_manager',
+)
+_SENTINEL = object()
+
+# Anything that tries to name a file outside the export directory, or that
+# is not a plain <name>.zip.
+TRAVERSAL_ATTEMPTS = [
+    "../../etc/passwd",
+    "../config.json",
+    "..%2f..%2fetc%2fpasswd",
+    "....//....//etc/passwd",
+    "/etc/passwd",
+    "..\\..\\config.json",
+    "backup.zip/../../../etc/passwd",
+    ".hidden.zip",
+    "backup.txt",
+    "backup.zip.exe",
+    "",
+    ".",
+    "..",
+]
+
+
+@pytest.fixture
+def env(tmp_path, monkeypatch):
+    export_dir = tmp_path / "backups"
+    export_dir.mkdir()
+    monkeypatch.setattr(api_v3_module, "_BACKUP_EXPORT_DIR", export_dir)
+
+    # A file outside the export dir that a traversal would be reaching for.
+    secret = tmp_path / "config.json"
+    secret.write_text(json.dumps({"secret": "do not touch"}))
+
+    originals = {name: getattr(api_v3, name, _SENTINEL) for name in _MANAGER_ATTRS}
+    for name in _MANAGER_ATTRS:
+        setattr(api_v3, name, MagicMock())
+
+    app = Flask(__name__)
+    app.config["TESTING"] = True
+    app.register_blueprint(api_v3, url_prefix="/api/v3")
+
+    class Env:
+        pass
+
+    e = Env()
+    e.client = app.test_client()
+    e.export_dir = export_dir
+    e.secret = secret
+    yield e
+
+    for name, original in originals.items():
+        if original is _SENTINEL:
+            if hasattr(api_v3, name):
+                delattr(api_v3, name)
+        else:
+            setattr(api_v3, name, original)
+
+
+def make_backup(export_dir, name="backup-2026-01-01.zip"):
+    path = export_dir / name
+    path.write_bytes(b"PK\x03\x04fake zip")
+    return path
+
+
+class TestSafeBackupPath:
+    """The containment helper itself."""
+
+    @pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS)
+    def test_rejects_unsafe_names(self, env, filename):
+        assert api_v3_module._safe_backup_path(filename) is None
+
+    def test_rejects_none(self, env):
+        assert api_v3_module._safe_backup_path(None) is None
+
+    @pytest.mark.parametrize("filename", [
+        "backup.zip",
+        "backup-2026-01-01.zip",
+        "backup_2026.01.01-v2.zip",
+        "a.zip",
+    ])
+    def test_accepts_plain_zip_names(self, env, filename):
+        resolved = api_v3_module._safe_backup_path(filename)
+        assert resolved is not None
+        assert resolved.parent == env.export_dir.resolve()
+
+    def test_result_is_always_inside_the_export_dir(self, env):
+        resolved = api_v3_module._safe_backup_path("backup.zip")
+        resolved.relative_to(env.export_dir.resolve())  # raises if outside
+
+    def test_overlong_name_rejected(self, env):
+        assert api_v3_module._safe_backup_path("a" * 250 + ".zip") is None
+
+
+class TestDownload:
+    def test_downloads_an_existing_backup(self, env):
+        make_backup(env.export_dir)
+        response = env.client.get("/api/v3/backup/download/backup-2026-01-01.zip")
+        assert response.status_code == 200
+        assert response.data == b"PK\x03\x04fake zip"
+
+    def test_missing_file_is_a_404(self, env):
+        response = env.client.get("/api/v3/backup/download/never-made.zip")
+        assert response.status_code == 404
+
+    @pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS)
+    def test_traversal_attempts_are_refused(self, env, filename):
+        response = env.client.get(f"/api/v3/backup/download/{filename}")
+        # However the request is turned away — 404 from the containment
+        # check, or 308/405 from routing never matching at all — what
+        # matters is that no file outside the export directory is served.
+        assert response.status_code != 200
+        assert b"do not touch" not in response.data
+
+
+class TestDelete:
+    def test_deletes_an_existing_backup(self, env):
+        path = make_backup(env.export_dir)
+        response = env.client.delete("/api/v3/backup/backup-2026-01-01.zip")
+        assert response.status_code == 200
+        assert not path.exists()
+
+    def test_missing_file_is_a_404(self, env):
+        response = env.client.delete("/api/v3/backup/never-made.zip")
+        assert response.status_code == 404
+
+    @pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS)
+    def test_traversal_attempts_delete_nothing(self, env, filename):
+        response = env.client.delete(f"/api/v3/backup/{filename}")
+        assert response.status_code != 200
+        assert env.secret.exists()  # the file a traversal was aiming at
+
+    def test_only_the_named_backup_is_removed(self, env):
+        keep = make_backup(env.export_dir, "keep.zip")
+        drop = make_backup(env.export_dir, "drop.zip")
+        env.client.delete("/api/v3/backup/drop.zip")
+        assert keep.exists()
+        assert not drop.exists()
+
+    def test_directory_with_a_matching_name_is_not_removed(self, env):
+        # The delete loop matches by name but requires a regular file.
+        (env.export_dir / "sneaky.zip").mkdir()
+        response = env.client.delete("/api/v3/backup/sneaky.zip")
+        assert response.status_code == 404
+        assert (env.export_dir / "sneaky.zip").is_dir()
+
+
+class TestList:
+    def test_lists_only_zip_files(self, env):
+        make_backup(env.export_dir, "one.zip")
+        (env.export_dir / "notes.txt").write_text("ignore me")
+        response = env.client.get("/api/v3/backup/list")
+        assert response.status_code == 200
+        names = [entry["filename"] for entry in response.get_json()["data"]]
+        assert names == ["one.zip"]
+
+    def test_empty_directory_lists_nothing(self, env):
+        response = env.client.get("/api/v3/backup/list")
+        assert response.get_json()["data"] == []
+
+    def test_entries_carry_size_and_timestamp(self, env):
+        make_backup(env.export_dir, "one.zip")
+        entry = env.client.get("/api/v3/backup/list").get_json()["data"][0]
+        assert entry["size"] == len(b"PK\x03\x04fake zip")
+        assert entry["created_at"]
+
+
+class TestValidate:
+    def test_missing_file_is_a_400(self, env):
+        response = env.client.post("/api/v3/backup/validate", data={},
+                                   content_type="multipart/form-data")
+        assert response.status_code == 400
+        assert "No backup_file" in response.get_json()["message"]
+
+    def test_invalid_archive_is_a_400(self, env):
+        response = env.client.post(
+            "/api/v3/backup/validate",
+            data={"backup_file": (io.BytesIO(b"not a zip"), "bad.zip")},
+            content_type="multipart/form-data")
+        assert response.status_code == 400
+        assert "Invalid or corrupted" in response.get_json()["message"]
+
+    def test_validation_does_not_leave_temp_files_in_the_export_dir(self, env):
+        env.client.post(
+            "/api/v3/backup/validate",
+            data={"backup_file": (io.BytesIO(b"not a zip"), "bad.zip")},
+            content_type="multipart/form-data")
+        assert list(env.export_dir.iterdir()) == []

test/web_interface/test_api_v3_backup_restore.py

@@ -0,0 +1,262 @@
+"""
+Endpoint tests for POST /backup/restore.
+
+Restore is the most destructive operation the web interface exposes: it
+overwrites config, secrets, WiFi settings and fonts, and reinstalls
+plugins. It had no tests.
+
+restore_backup itself is mocked — this file is about what the route does
+with the request and with the result, not about ZIP handling, which
+belongs to backup_manager's own tests.
+
+Regression coverage for one fixed bug: a malformed `options` field fell
+back to {}, and since every RestoreOptions flag defaults to True, that
+turned a mis-serialized narrow restore into a full one — secrets
+included — with no indication anything had been ignored.
+"""
+
+import io
+import json
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+from flask import Flask
+
+project_root = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(project_root))
+
+from web_interface.blueprints.api_v3 import api_v3  # noqa: E402
+
+URL = "/api/v3/backup/restore"
+
+_MANAGER_ATTRS = (
+    'config_manager', 'plugin_manager', 'plugin_store_manager',
+    'plugin_state_manager', 'saved_repositories_manager', 'schema_manager',
+    'operation_queue', 'operation_history', 'cache_manager',
+)
+_SENTINEL = object()
+
+
+class FakeResult:
+    """Stand-in for backup_manager.RestoreResult."""
+
+    def __init__(self, success=True, restored=None, errors=None,
+                 plugins_to_install=None):
+        self.success = success
+        self.restored = restored if restored is not None else ["config"]
+        self.errors = errors or []
+        self.plugins_to_install = plugins_to_install or []
+        self.plugins_installed = []
+        self.plugins_failed = []
+
+    def to_dict(self):
+        return {
+            "success": self.success,
+            "restored": self.restored,
+            "errors": self.errors,
+            "plugins_installed": self.plugins_installed,
+            "plugins_failed": self.plugins_failed,
+        }
+
+
+@pytest.fixture
+def client():
+    originals = {name: getattr(api_v3, name, _SENTINEL) for name in _MANAGER_ATTRS}
+    for name in _MANAGER_ATTRS:
+        setattr(api_v3, name, MagicMock())
+
+    app = Flask(__name__)
+    app.config["TESTING"] = True
+    app.register_blueprint(api_v3, url_prefix="/api/v3")
+    yield app.test_client()
+
+    for name, original in originals.items():
+        if original is _SENTINEL:
+            if hasattr(api_v3, name):
+                delattr(api_v3, name)
+        else:
+            setattr(api_v3, name, original)
+
+
+@pytest.fixture
+def restore():
+    """Patch backup_manager.restore_backup (imported inside the handler)."""
+    with patch("src.backup_manager.restore_backup") as mock:
+        mock.return_value = FakeResult()
+        yield mock
+
+
+def post(client, options=None, filename="backup.zip", content=b"PK\x03\x04fake"):
+    data = {"backup_file": (io.BytesIO(content), filename)}
+    if options is not None:
+        data["options"] = options
+    return client.post(URL, data=data, content_type="multipart/form-data")
+
+
+class TestRequestValidation:
+    def test_missing_file_is_a_400(self, client, restore):
+        response = client.post(URL, data={}, content_type="multipart/form-data")
+        assert response.status_code == 400
+        assert "No backup_file" in response.get_json()["message"]
+        restore.assert_not_called()
+
+    def test_absent_options_defaults_to_a_full_restore(self, client, restore):
+        # Documented default, not the bug: omitting options entirely means
+        # "restore everything".
+        post(client)
+        options = restore.call_args[0][2]
+        assert options.restore_config is True
+        assert options.restore_secrets is True
+        assert options.reinstall_plugins is True
+
+    def test_partial_options_are_honoured(self, client, restore):
+        post(client, options=json.dumps({
+            "restore_secrets": False, "reinstall_plugins": False}))
+        options = restore.call_args[0][2]
+        assert options.restore_secrets is False
+        assert options.reinstall_plugins is False
+        assert options.restore_config is True  # unspecified stays default
+
+    @pytest.mark.parametrize("raw", ["{not json", "", "{'single': 'quotes'}"])
+    def test_malformed_options_are_refused(self, client, restore, raw):
+        # Regression: this fell back to {}, and every flag defaults to
+        # True, so a caller asking for a narrow restore and mis-serializing
+        # it got a full one — secrets overwritten — and no warning.
+        response = post(client, options=raw)
+        assert response.status_code == 400
+        assert "Invalid options" in response.get_json()["message"]
+        restore.assert_not_called()
+
+    @pytest.mark.parametrize("raw", ["[1,2,3]", '"a string"', "42", "true", "null"])
+    def test_options_that_are_not_an_object_are_refused(self, client, restore, raw):
+        response = post(client, options=raw)
+        assert response.status_code == 400
+        restore.assert_not_called()
+
+    def test_empty_object_is_accepted_as_all_defaults(self, client, restore):
+        assert post(client, options="{}").status_code == 200
+        assert restore.call_args[0][2].restore_config is True
+
+
+class TestSuccess:
+    def test_success_returns_the_result(self, client, restore):
+        restore.return_value = FakeResult(success=True, restored=["config", "secrets"])
+        response = post(client)
+        assert response.status_code == 200
+        body = response.get_json()
+        assert body["status"] == "success"
+        assert body["data"]["restored"] == ["config", "secrets"]
+
+    def test_temp_file_is_cleaned_up(self, client, restore):
+        seen = {}
+
+        def capture(path, project_root, options):
+            seen["path"] = Path(path)
+            assert seen["path"].exists()  # present while restoring
+            return FakeResult()
+
+        restore.side_effect = capture
+        post(client)
+        assert not seen["path"].exists()
+
+    def test_temp_file_cleaned_up_even_when_restore_raises(self, client, restore):
+        seen = {}
+
+        def blow_up(path, project_root, options):
+            seen["path"] = Path(path)
+            raise RuntimeError("corrupt archive")
+
+        restore.side_effect = blow_up
+        response = post(client)
+        assert response.status_code == 500
+        assert not seen["path"].exists()
+
+
+class TestPluginReinstall:
+    def test_plugins_are_reinstalled_when_requested(self, client, restore):
+        restore.return_value = FakeResult(
+            plugins_to_install=[{"plugin_id": "clock"}, {"plugin_id": "weather"}])
+        api_v3.plugin_store_manager.install_plugin.return_value = True
+        response = post(client)
+        assert response.status_code == 200
+        assert response.get_json()["data"]["plugins_installed"] == ["clock", "weather"]
+
+    def test_reinstall_skipped_when_not_requested(self, client, restore):
+        restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}])
+        post(client, options=json.dumps({"reinstall_plugins": False}))
+        api_v3.plugin_store_manager.install_plugin.assert_not_called()
+
+    def test_entries_without_a_plugin_id_are_skipped(self, client, restore):
+        restore.return_value = FakeResult(plugins_to_install=[{}, {"plugin_id": "clock"}])
+        api_v3.plugin_store_manager.install_plugin.return_value = True
+        post(client)
+        assert api_v3.plugin_store_manager.install_plugin.call_count == 1
+
+    def test_failed_reinstall_turns_the_whole_restore_into_an_error(
+            self, client, restore):
+        # Pinned as intentional: file restoration succeeded and does not
+        # touch result.errors, but a user whose plugins did not come back
+        # should not be told the restore was a success.
+        restore.return_value = FakeResult(
+            success=True, plugins_to_install=[{"plugin_id": "clock"}])
+        api_v3.plugin_store_manager.install_plugin.return_value = False
+        response = post(client)
+        assert response.status_code == 500
+        body = response.get_json()
+        assert body["status"] == "error"
+        assert "clock" in body["message"]
+
+    def test_message_names_what_landed_and_what_did_not(self, client, restore):
+        restore.return_value = FakeResult(
+            success=True, restored=["config", "fonts"],
+            plugins_to_install=[{"plugin_id": "clock"}])
+        api_v3.plugin_store_manager.install_plugin.return_value = False
+        message = post(client).get_json()["message"]
+        assert "restored: config, fonts" in message
+        assert "plugins not reinstalled: clock" in message
+
+    def test_install_exception_is_recorded_without_leaking_details(
+            self, client, restore):
+        restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}])
+        api_v3.plugin_store_manager.install_plugin.side_effect = RuntimeError(
+            "/srv/internal/path exploded")
+        body = post(client).get_json()
+        failures = body["data"]["plugins_failed"]
+        assert failures[0]["plugin_id"] == "clock"
+        assert "/srv/internal/path" not in json.dumps(body)
+
+    def test_missing_store_manager_is_reported_per_plugin(self, client, restore):
+        restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}])
+        api_v3.plugin_store_manager = None
+        with patch("web_interface.blueprints.api_v3.plugin_store_manager", None):
+            body = post(client).get_json()
+        assert body["data"]["plugins_failed"][0]["error"] == "Store manager unavailable"
+
+
+class TestFailureReporting:
+    def test_restore_errors_produce_a_500(self, client, restore):
+        restore.return_value = FakeResult(
+            success=False, restored=[], errors=["config: permission denied"])
+        response = post(client)
+        assert response.status_code == 500
+        assert "permission denied" in response.get_json()["message"]
+
+    def test_partial_restore_names_both_sides(self, client, restore):
+        restore.return_value = FakeResult(
+            success=False, restored=["config"], errors=["secrets: unwritable"])
+        message = post(client).get_json()["message"]
+        assert "restored: config" in message
+        assert "failed: secrets: unwritable" in message
+
+    def test_failure_without_detail_still_says_something(self, client, restore):
+        restore.return_value = FakeResult(success=False, restored=[], errors=[])
+        message = post(client).get_json()["message"]
+        assert "Restore incomplete" in message
+
+    def test_unexpected_exception_is_a_500(self, client, restore):
+        restore.side_effect = RuntimeError("boom")
+        response = post(client)
+        assert response.status_code == 500
+        assert response.get_json()["status"] == "error"

test/web_interface/test_api_v3_config_raw.py

@@ -0,0 +1,204 @@
+"""
+Endpoint tests for POST /config/raw/main and POST /config/raw/secrets.
+
+These write whatever JSON they are given straight to config.json and
+config_secrets.json, bypassing the secret-separation path that
+/config/main and the plugin-config endpoints go through. Given how much
+care the rest of the config surface takes to keep secrets out of
+config.json, an untested pair of endpoints that writes it verbatim is
+worth pinning precisely.
+
+Like test_api_v3_secret_roundtrip.py, these run a REAL ConfigManager over
+tmp_path so the assertions are against files on disk rather than mock
+calls.
+"""
+
+import json
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+from flask import Flask
+
+project_root = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(project_root))
+
+from src.config_manager import ConfigManager  # noqa: E402
+from src.exceptions import ConfigError  # noqa: E402
+from web_interface.blueprints.api_v3 import api_v3  # noqa: E402
+
+MAIN = "/api/v3/config/raw/main"
+SECRETS = "/api/v3/config/raw/secrets"
+
+
+@pytest.fixture
+def env(tmp_path):
+    config_file = tmp_path / "config.json"
+    config_file.write_text(json.dumps({"timezone": "UTC"}))
+    secrets_file = tmp_path / "config_secrets.json"
+
+    config_manager = ConfigManager(
+        config_path=str(config_file), secrets_path=str(secrets_file))
+    config_manager.template_path = str(tmp_path / "no-template.json")
+
+    _SENTINEL = object()
+    attrs = ('config_manager', 'plugin_manager', 'plugin_store_manager',
+             'plugin_state_manager', 'saved_repositories_manager',
+             'schema_manager', 'operation_queue', 'operation_history',
+             'cache_manager')
+    originals = {name: getattr(api_v3, name, _SENTINEL) for name in attrs}
+
+    for name in attrs:
+        setattr(api_v3, name, MagicMock())
+    api_v3.config_manager = config_manager
+
+    app = Flask(__name__)
+    app.config["TESTING"] = True
+    app.register_blueprint(api_v3, url_prefix="/api/v3")
+
+    class Env:
+        pass
+
+    e = Env()
+    e.client = app.test_client()
+    e.config_manager = config_manager
+    e.config_file = config_file
+    e.secrets_file = secrets_file
+    yield e
+
+    for name, original in originals.items():
+        if original is _SENTINEL:
+            if hasattr(api_v3, name):
+                delattr(api_v3, name)
+        else:
+            setattr(api_v3, name, original)
+
+
+class TestSaveRawMain:
+    def test_writes_the_body_to_config_json(self, env):
+        response = env.client.post(MAIN, json={"timezone": "America/Chicago"})
+        assert response.status_code == 200
+        assert json.loads(env.config_file.read_text()) == {"timezone": "America/Chicago"}
+
+    def test_replaces_rather_than_merges(self, env):
+        env.client.post(MAIN, json={"only": "this"})
+        assert json.loads(env.config_file.read_text()) == {"only": "this"}
+
+    def test_does_not_touch_the_secrets_file(self, env):
+        env.secrets_file.write_text(json.dumps({"weather": {"api_key": "k"}}))
+        env.client.post(MAIN, json={"timezone": "UTC"})
+        assert json.loads(env.secrets_file.read_text()) == {"weather": {"api_key": "k"}}
+
+    def test_uninitialized_manager_is_a_500(self, env):
+        api_v3.config_manager = None
+        response = env.client.post(MAIN, json={"timezone": "UTC"})
+        assert response.status_code == 500
+        assert "not initialized" in response.get_json()["message"]
+
+    def test_empty_object_is_a_400(self, env):
+        response = env.client.post(MAIN, json={})
+        assert response.status_code == 400
+        assert "No data provided" in response.get_json()["message"]
+
+    def test_bodyless_post_is_a_400(self, env):
+        response = env.client.post(MAIN)
+        assert response.status_code == 400
+        assert "No data provided" in response.get_json()["message"]
+
+    def test_malformed_json_is_a_400_in_the_app_shape(self, env):
+        response = env.client.post(MAIN, data="{not json",
+                                   content_type="application/json")
+        assert response.status_code == 400
+        body = response.get_json()
+        assert body["status"] == "error"
+        # A body that was sent but does not parse is a distinct mistake
+        # from sending none, and says so. Previously the handler's own
+        # json.JSONDecodeError arm was unreachable — Werkzeug raised
+        # first — so this collapsed into "No data provided".
+        assert "Invalid JSON in request body" in body["message"]
+
+    def test_config_error_is_a_500_with_context(self, env, monkeypatch):
+        def refuse(kind, data):
+            raise ConfigError("cannot write", config_path="/etc/x.json")
+        monkeypatch.setattr(env.config_manager, "save_raw_file_content", refuse)
+        response = env.client.post(MAIN, json={"timezone": "UTC"})
+        assert response.status_code == 500
+        assert "/etc/x.json" in json.dumps(response.get_json())
+
+    def test_unexpected_error_is_a_500(self, env, monkeypatch):
+        def boom(kind, data):
+            raise RuntimeError("disk on fire")
+        monkeypatch.setattr(env.config_manager, "save_raw_file_content", boom)
+        response = env.client.post(MAIN, json={"timezone": "UTC"})
+        assert response.status_code == 500
+        assert response.get_json()["status"] == "error"
+
+
+class TestSaveRawSecrets:
+    def test_writes_only_to_the_secrets_file(self, env):
+        response = env.client.post(SECRETS, json={"weather": {"api_key": "s3cret"}})
+        assert response.status_code == 200
+        assert json.loads(env.secrets_file.read_text()) == {"weather": {"api_key": "s3cret"}}
+
+    def test_secret_values_never_reach_config_json(self, env):
+        env.client.post(SECRETS, json={"weather": {"api_key": "s3cret"}})
+        assert "s3cret" not in env.config_file.read_text()
+
+    def test_existing_main_config_is_untouched(self, env):
+        before = env.config_file.read_text()
+        env.client.post(SECRETS, json={"weather": {"api_key": "k"}})
+        assert env.config_file.read_text() == before
+
+    def test_github_token_is_reloaded_for_the_store_manager(self, env):
+        store = MagicMock()
+        store._load_github_token.return_value = "ghp_new"
+        api_v3.plugin_store_manager = store
+        env.client.post(SECRETS, json={"github": {"token": "ghp_new"}})
+        store._load_github_token.assert_called_once()
+        assert store.github_token == "ghp_new"
+
+    def test_absent_store_manager_is_fine(self, env):
+        api_v3.plugin_store_manager = None
+        assert env.client.post(SECRETS, json={"a": 1}).status_code == 200
+
+    def test_uninitialized_manager_is_a_500(self, env):
+        api_v3.config_manager = None
+        assert env.client.post(SECRETS, json={"a": 1}).status_code == 500
+
+    def test_empty_object_is_a_400(self, env):
+        assert env.client.post(SECRETS, json={}).status_code == 400
+
+    def test_bodyless_post_is_a_400(self, env):
+        assert env.client.post(SECRETS).status_code == 400
+
+    def test_error_is_a_500(self, env, monkeypatch):
+        def boom(kind, data):
+            raise RuntimeError("nope")
+        monkeypatch.setattr(env.config_manager, "save_raw_file_content", boom)
+        assert env.client.post(SECRETS, json={"a": 1}).status_code == 500
+
+
+class TestRawEndpointsBypassSecretSeparation:
+    """Pinned behaviour, deliberately not "fixed".
+
+    These endpoints are the escape hatch for editing the config files
+    directly from the web UI's raw JSON editor. They write what they are
+    given, so a secret typed into the main-config editor lands in
+    config.json in plain text — unlike /config/main and the plugin-config
+    endpoints, which route x-secret fields into config_secrets.json.
+
+    That is the point of a raw editor, but it is a sharp edge worth
+    stating out loud: anyone adding a "convenience" that posts plugin
+    config through this endpoint would silently lose secret separation.
+    """
+
+    def test_secret_shaped_keys_are_written_verbatim_to_main(self, env):
+        env.client.post(MAIN, json={"weather": {"api_key": "PLAINTEXT-KEY"}})
+        on_disk = json.loads(env.config_file.read_text())
+        assert on_disk["weather"]["api_key"] == "PLAINTEXT-KEY"
+
+    def test_no_separation_happens_on_the_raw_path(self, env):
+        env.client.post(MAIN, json={"weather": {"api_key": "PLAINTEXT-KEY"}})
+        # Nothing was moved aside into the secrets file.
+        assert not env.secrets_file.exists() or "PLAINTEXT-KEY" not in env.secrets_file.read_text()

test/web_interface/test_error_handler.py

@@ -0,0 +1,149 @@
+"""
+Tests for the response builders in src/web_interface/error_handler.py and
+the success path in src/web_interface/api_helpers.py.
+
+describe_exception() in the same module is already covered by
+test/test_web_error_detail.py and is not duplicated here.
+
+Regression coverage for one fixed bug: create_success_response used
+truthiness for `message` and `metadata` while using `is not None` for
+`data`, so an explicitly-passed "" or {} was silently dropped —
+api_helpers.success_response() repeated the same gate, which is the path
+every api_v3 endpoint actually calls.
+"""
+
+import pytest
+from flask import Flask
+
+from src.web_interface.api_helpers import success_response
+from src.web_interface.error_handler import (
+    create_error_response,
+    create_success_response,
+)
+from src.web_interface.errors import ErrorCode, WebInterfaceError
+
+
+@pytest.fixture
+def app():
+    return Flask(__name__)
+
+
+class TestCreateErrorResponse:
+    def test_returns_response_and_status_tuple(self, app):
+        with app.test_request_context():
+            response, status = create_error_response(
+                ErrorCode.CONFIG_SAVE_FAILED, "could not save")
+        assert status == 500
+        assert response.get_json()["message"] == "could not save"
+
+    def test_status_code_passthrough(self, app):
+        with app.test_request_context():
+            _, status = create_error_response(
+                ErrorCode.INVALID_INPUT, "bad", status_code=400)
+        assert status == 400
+
+    def test_body_matches_the_error_dataclass(self, app):
+        with app.test_request_context():
+            response, _ = create_error_response(
+                ErrorCode.NETWORK_ERROR, "offline",
+                details="connection refused", context={"url": "http://x"})
+        expected = WebInterfaceError(
+            error_code=ErrorCode.NETWORK_ERROR, message="offline",
+            details="connection refused", context={"url": "http://x"}).to_dict()
+        assert response.get_json() == expected
+
+    def test_none_context_produces_no_context_key(self, app):
+        with app.test_request_context():
+            response, _ = create_error_response(ErrorCode.SYSTEM_ERROR, "boom")
+        assert "context" not in response.get_json()
+
+    def test_suggested_fixes_passed_through(self, app):
+        with app.test_request_context():
+            response, _ = create_error_response(
+                ErrorCode.SYSTEM_ERROR, "boom", suggested_fixes=["Try again"])
+        assert response.get_json()["suggested_fixes"] == ["Try again"]
+
+
+class TestCreateSuccessResponse:
+    def test_bare_success(self):
+        assert create_success_response() == {"status": "success"}
+
+    def test_data_included(self):
+        assert create_success_response(data={"a": 1})["data"] == {"a": 1}
+
+    @pytest.mark.parametrize("falsy", [0, "", False, {}, []])
+    def test_falsy_data_is_still_included(self, falsy):
+        assert create_success_response(data=falsy)["data"] == falsy
+
+    def test_none_data_omitted(self):
+        assert "data" not in create_success_response(data=None)
+
+    def test_message_included(self):
+        assert create_success_response(message="done")["message"] == "done"
+
+    def test_empty_message_is_still_included(self):
+        # Regression: `if message:` dropped an explicitly-passed "".
+        assert create_success_response(message="")["message"] == ""
+
+    def test_none_message_omitted(self):
+        assert "message" not in create_success_response(message=None)
+
+    def test_metadata_included(self):
+        assert create_success_response(metadata={"v": 1})["metadata"] == {"v": 1}
+
+    def test_empty_metadata_is_still_included(self):
+        # Regression: `if metadata:` dropped an explicitly-passed {}.
+        assert create_success_response(metadata={})["metadata"] == {}
+
+    def test_none_metadata_omitted(self):
+        assert "metadata" not in create_success_response(metadata=None)
+
+
+class TestSuccessResponseHelper:
+    """api_helpers.success_response — the wrapper every endpoint calls."""
+
+    def test_plain_response_has_no_metadata_block(self, app):
+        with app.test_request_context():
+            body = success_response(data={"a": 1}).get_json()
+        assert body == {"status": "success", "data": {"a": 1}}
+
+    def test_explicit_empty_metadata_survives_the_wrapper(self, app):
+        # Regression: the wrapper re-gated metadata on truthiness after
+        # create_success_response had already included it, so {} was
+        # dropped again on the way out.
+        with app.test_request_context():
+            body = success_response(data=None, metadata={}).get_json()
+        assert body["metadata"] == {}
+
+    def test_caller_metadata_preserved(self, app):
+        with app.test_request_context():
+            body = success_response(metadata={"version": "1.2"}).get_json()
+        assert body["metadata"]["version"] == "1.2"
+
+    def test_timing_added_when_request_has_start_time(self, app):
+        with app.test_request_context() as ctx:
+            ctx.request.start_time = 0.0
+            body = success_response(data={"a": 1}).get_json()
+        assert "response_time_ms" in body["metadata"]
+
+    def test_timing_merges_with_caller_metadata(self, app):
+        with app.test_request_context() as ctx:
+            ctx.request.start_time = 0.0
+            body = success_response(metadata={"version": "1.2"}).get_json()
+        assert body["metadata"]["version"] == "1.2"
+        assert "response_time_ms" in body["metadata"]
+
+    def test_caller_metadata_dict_is_not_mutated(self, app):
+        # The helper used to add response_time_ms straight into the dict the
+        # caller passed, so a module-level or reused metadata dict would
+        # accumulate timings from previous requests.
+        caller_metadata = {"version": "1.2"}
+        with app.test_request_context() as ctx:
+            ctx.request.start_time = 0.0
+            success_response(metadata=caller_metadata)
+        assert caller_metadata == {"version": "1.2"}
+
+    def test_message_passed_through(self, app):
+        with app.test_request_context():
+            body = success_response(message="saved").get_json()
+        assert body["message"] == "saved"

test/web_interface/test_errors.py

@@ -0,0 +1,208 @@
+"""
+Tests for src/web_interface/errors.py — the structured error type behind
+every API error response (category inference, default suggestions, the
+JSON shape, and exception conversion).
+
+Pure logic; no Flask context needed.
+
+Regression coverage for one fixed bug: suggested_fixes used `or`, so a
+caller passing [] to mean "no suggestions" silently got the default list.
+"""
+
+import pytest
+
+from src.web_interface.errors import ErrorCategory, ErrorCode, WebInterfaceError
+
+
+class TestCategoryInference:
+    @pytest.mark.parametrize("code,expected", [
+        (ErrorCode.CONFIG_SAVE_FAILED, ErrorCategory.CONFIGURATION),
+        (ErrorCode.CONFIG_ROLLBACK_FAILED, ErrorCategory.CONFIGURATION),
+        (ErrorCode.PLUGIN_NOT_FOUND, ErrorCategory.PLUGIN),
+        (ErrorCode.PLUGIN_OPERATION_CONFLICT, ErrorCategory.PLUGIN),
+        (ErrorCode.VALIDATION_ERROR, ErrorCategory.VALIDATION),
+        (ErrorCode.SCHEMA_VALIDATION_FAILED, ErrorCategory.VALIDATION),
+        (ErrorCode.INVALID_INPUT, ErrorCategory.VALIDATION),
+        (ErrorCode.NETWORK_ERROR, ErrorCategory.NETWORK),
+        (ErrorCode.API_ERROR, ErrorCategory.NETWORK),
+        (ErrorCode.TIMEOUT, ErrorCategory.NETWORK),
+        (ErrorCode.PERMISSION_DENIED, ErrorCategory.PERMISSION),
+        (ErrorCode.FILE_PERMISSION_ERROR, ErrorCategory.PERMISSION),
+        (ErrorCode.SYSTEM_ERROR, ErrorCategory.SYSTEM),
+        (ErrorCode.SERVICE_UNAVAILABLE, ErrorCategory.SYSTEM),
+        (ErrorCode.UNKNOWN_ERROR, ErrorCategory.UNKNOWN),
+    ])
+    def test_every_code_prefix_maps_to_its_category(self, code, expected):
+        assert WebInterfaceError(code, "msg").category is expected
+
+    def test_explicit_category_overrides_inference(self):
+        error = WebInterfaceError(
+            ErrorCode.CONFIG_SAVE_FAILED, "msg", category=ErrorCategory.SYSTEM)
+        assert error.category is ErrorCategory.SYSTEM
+
+    def test_every_error_code_gets_a_category(self):
+        # No code may fall through uncategorized as the enum grows.
+        for code in ErrorCode:
+            assert isinstance(WebInterfaceError(code, "msg").category, ErrorCategory)
+
+
+class TestDefaultSuggestions:
+    def test_mapped_code_gets_specific_suggestions(self):
+        fixes = WebInterfaceError(ErrorCode.CONFIG_SAVE_FAILED, "msg").suggested_fixes
+        assert "Check available disk space" in fixes
+
+    def test_unmapped_code_gets_generic_fallback(self):
+        # PLUGIN_UPDATE_FAILED has no entry in suggestions_map.
+        fixes = WebInterfaceError(ErrorCode.PLUGIN_UPDATE_FAILED, "msg").suggested_fixes
+        assert fixes == ["Review error details and try again"]
+
+    def test_explicit_suggestions_win(self):
+        error = WebInterfaceError(
+            ErrorCode.CONFIG_SAVE_FAILED, "msg", suggested_fixes=["Do the thing"])
+        assert error.suggested_fixes == ["Do the thing"]
+
+    def test_explicit_empty_list_is_respected(self):
+        # Regression: `suggested_fixes or default` treated [] as "unset",
+        # so a caller could not express "I have no suggestions".
+        error = WebInterfaceError(
+            ErrorCode.CONFIG_SAVE_FAILED, "msg", suggested_fixes=[])
+        assert error.suggested_fixes == []
+
+    def test_none_still_gets_defaults(self):
+        error = WebInterfaceError(
+            ErrorCode.CONFIG_SAVE_FAILED, "msg", suggested_fixes=None)
+        assert len(error.suggested_fixes) > 0
+
+
+class TestToDict:
+    def test_base_keys_always_present(self):
+        result = WebInterfaceError(ErrorCode.SYSTEM_ERROR, "boom").to_dict()
+        assert result["status"] == "error"
+        assert result["error_code"] == "SYSTEM_ERROR"
+        assert result["error_category"] == "system"
+        assert result["message"] == "boom"
+
+    def test_details_included_when_set(self):
+        result = WebInterfaceError(
+            ErrorCode.SYSTEM_ERROR, "boom", details="disk full").to_dict()
+        assert result["details"] == "disk full"
+
+    def test_details_omitted_when_absent(self):
+        assert "details" not in WebInterfaceError(ErrorCode.SYSTEM_ERROR, "boom").to_dict()
+
+    def test_context_included_when_non_empty(self):
+        result = WebInterfaceError(
+            ErrorCode.SYSTEM_ERROR, "boom", context={"path": "/tmp/x"}).to_dict()
+        assert result["context"] == {"path": "/tmp/x"}
+
+    def test_empty_context_is_omitted(self):
+        # Pinned as intentional, not a bug: __init__ normalizes context to
+        # {}, and an empty context carries no information, so it is left out
+        # rather than padding every error body with "context": {}.
+        result = WebInterfaceError(ErrorCode.SYSTEM_ERROR, "boom", context={}).to_dict()
+        assert "context" not in result
+
+    def test_empty_suggestions_omitted(self):
+        result = WebInterfaceError(
+            ErrorCode.SYSTEM_ERROR, "boom", suggested_fixes=[]).to_dict()
+        assert "suggested_fixes" not in result
+
+    def test_is_json_serializable(self):
+        import json
+        error = WebInterfaceError(
+            ErrorCode.NETWORK_ERROR, "boom",
+            details="timeout", context={"url": "http://x"})
+        assert json.loads(json.dumps(error.to_dict()))["error_code"] == "NETWORK_ERROR"
+
+
+class TestFromException:
+    @pytest.mark.parametrize("exc_name,expected", [
+        ("ConfigError", ErrorCode.CONFIG_LOAD_FAILED),
+        ("PluginError", ErrorCode.PLUGIN_LOAD_FAILED),
+        ("PermissionError", ErrorCode.PERMISSION_DENIED),
+        ("AccessDenied", ErrorCode.PERMISSION_DENIED),
+        ("ValidationError", ErrorCode.VALIDATION_ERROR),
+        ("SchemaError", ErrorCode.VALIDATION_ERROR),
+        ("NetworkError", ErrorCode.NETWORK_ERROR),
+        ("ConnectionError", ErrorCode.NETWORK_ERROR),
+        ("TimeoutError", ErrorCode.TIMEOUT),
+        ("SomethingElse", ErrorCode.UNKNOWN_ERROR),
+    ])
+    def test_code_inferred_from_exception_class_name(self, exc_name, expected):
+        exc = type(exc_name, (Exception,), {})("boom")
+        assert WebInterfaceError.from_exception(exc).error_code is expected
+
+    def test_explicit_code_skips_inference(self):
+        error = WebInterfaceError.from_exception(
+            ValueError("boom"), error_code=ErrorCode.PLUGIN_NOT_FOUND)
+        assert error.error_code is ErrorCode.PLUGIN_NOT_FOUND
+
+    def test_message_is_the_safe_one_not_the_exception_text(self):
+        # The raw exception text is not echoed into `message`; that field is
+        # a fixed, user-facing string per code.
+        error = WebInterfaceError.from_exception(ValueError("secret-ish detail"))
+        assert error.message == "An unexpected error occurred"
+        assert "secret-ish" not in error.message
+
+    def test_exception_type_recorded_in_context(self):
+        error = WebInterfaceError.from_exception(ValueError("boom"))
+        assert error.context["exception_type"] == "ValueError"
+
+    def test_caller_context_is_preserved_alongside_type(self):
+        error = WebInterfaceError.from_exception(
+            ValueError("boom"), context={"plugin_id": "clock"})
+        assert error.context["plugin_id"] == "clock"
+        assert error.context["exception_type"] == "ValueError"
+
+    def test_caller_supplied_exception_type_is_overwritten(self):
+        error = WebInterfaceError.from_exception(
+            ValueError("boom"), context={"exception_type": "Fake"})
+        assert error.context["exception_type"] == "ValueError"
+
+    def test_original_error_retained(self):
+        exc = ValueError("boom")
+        assert WebInterfaceError.from_exception(exc).original_error is exc
+
+    def test_every_code_has_a_safe_message(self):
+        for code in ErrorCode:
+            assert WebInterfaceError._safe_message(code)
+
+
+class TestExceptionDetails:
+    def test_context_dict_is_flattened(self):
+        exc = ValueError("boom")
+        exc.context = {"config_path": "/etc/x.json", "line": 4}
+        details = WebInterfaceError._get_exception_details(exc)
+        assert "config_path: /etc/x.json" in details
+        assert "line: 4" in details
+        assert "; " in details
+
+    def test_exception_type_key_excluded(self):
+        exc = ValueError("boom")
+        exc.context = {"exception_type": "ValueError", "path": "/tmp/x"}
+        details = WebInterfaceError._get_exception_details(exc)
+        assert "exception_type" not in details
+        assert details == "path: /tmp/x"
+
+    def test_context_with_only_exception_type_gives_none(self):
+        exc = ValueError("boom")
+        exc.context = {"exception_type": "ValueError"}
+        assert WebInterfaceError._get_exception_details(exc) is None
+
+    def test_no_context_attribute_gives_none(self):
+        assert WebInterfaceError._get_exception_details(ValueError("boom")) is None
+
+    def test_non_dict_context_gives_none(self):
+        exc = ValueError("boom")
+        exc.context = "not a dict"
+        assert WebInterfaceError._get_exception_details(exc) is None
+
+    def test_empty_context_gives_none(self):
+        exc = ValueError("boom")
+        exc.context = {}
+        assert WebInterfaceError._get_exception_details(exc) is None
+
+    def test_details_flow_into_from_exception(self):
+        exc = ValueError("boom")
+        exc.context = {"config_path": "/etc/x.json"}
+        assert "config_path" in WebInterfaceError.from_exception(exc).details

test/web_interface/test_validators.py

@@ -0,0 +1,284 @@
+"""
+Tests for src/web_interface/validators.py.
+
+dedup_unique_arrays is already covered by test_dedup_unique_arrays.py and
+is not repeated here; this file covers the other eight functions, none of
+which had any tests.
+
+Regression coverage for three fixed bugs:
+- validate_numeric_range accepted True/False, since bool subclasses int.
+- validate_file_upload lowercased the filename's extension but not the
+  caller's allowed_extensions list, so ['.TTF'] rejected 'font.ttf'.
+- validate_image_url only checked for '..' inside the relative-path
+  branch, so http://host/../secret passed validation untouched.
+"""
+
+import pytest
+
+from src.web_interface.validators import (
+    escape_html,
+    sanitize_plugin_config,
+    validate_file_upload,
+    validate_font_awesome_class,
+    validate_image_url,
+    validate_mime_type,
+    validate_numeric_range,
+    validate_string_length,
+)
+
+
+class TestEscapeHtml:
+    def test_escapes_all_five_entities(self):
+        assert escape_html("""<a href="x">O'Neill & co</a>""") == (
+            "&lt;a href=&quot;x&quot;&gt;O&#x27;Neill &amp; co&lt;/a&gt;")
+
+    def test_ampersand_is_escaped_first_so_nothing_double_escapes(self):
+        # If '<' were replaced before '&', the '&' of '&lt;' would be
+        # escaped again into '&amp;lt;'.
+        assert escape_html("<") == "&lt;"
+        assert escape_html("&") == "&amp;"
+        assert escape_html("&<") == "&amp;&lt;"
+
+    def test_plain_text_unchanged(self):
+        assert escape_html("hello world") == "hello world"
+
+    def test_non_string_is_coerced(self):
+        assert escape_html(42) == "42"
+        assert escape_html(None) == "None"
+
+    def test_script_tag_neutralized(self):
+        assert "<script>" not in escape_html("<script>alert(1)</script>")
+
+
+class TestValidateImageUrl:
+    @pytest.mark.parametrize("url", [
+        "javascript:alert(1)",
+        "JavaScript:alert(1)",
+        "JAVASCRIPT:alert(1)",
+        "data:text/html;base64,PHNjcmlwdD4=",
+        "vbscript:msgbox(1)",
+        "file:///etc/passwd",
+    ])
+    def test_dangerous_protocols_rejected(self, url):
+        valid, error = validate_image_url(url)
+        assert valid is False and "protocol" in error.lower()
+
+    @pytest.mark.parametrize("url", [
+        "http://x/a.png?onerror=alert(1)",
+        "http://x/a.png#onload=alert(1)",
+        "http://x/onclick=alert(1).png",
+    ])
+    def test_event_handlers_rejected(self, url):
+        valid, error = validate_image_url(url)
+        assert valid is False and "Event handlers" in error
+
+    @pytest.mark.parametrize("url", ["", None, 123, []])
+    def test_empty_or_non_string_rejected(self, url):
+        assert validate_image_url(url)[0] is False
+
+    def test_http_and_https_allowed(self):
+        assert validate_image_url("http://example.com/logo.png") == (True, None)
+        assert validate_image_url("https://example.com/logo.png") == (True, None)
+
+    def test_other_schemes_rejected(self):
+        valid, error = validate_image_url("ftp://example.com/logo.png")
+        assert valid is False and "http://" in error
+
+    def test_relative_path_allowed(self):
+        assert validate_image_url("/static/logo.png") == (True, None)
+
+    def test_protocol_relative_url_rejected(self):
+        assert validate_image_url("//evil.com/logo.png")[0] is False
+
+    def test_relative_traversal_rejected(self):
+        assert validate_image_url("/static/../../etc/passwd")[0] is False
+
+    def test_absolute_url_traversal_rejected(self):
+        # Regression: the '..' check used to sit inside the leading-slash
+        # branch, so an absolute URL skipped it entirely.
+        valid, error = validate_image_url("http://example.com/../secret")
+        assert valid is False and "traversal" in error.lower()
+
+    def test_bare_traversal_rejected(self):
+        assert validate_image_url("../../etc/passwd")[0] is False
+
+
+class TestValidateFontAwesomeClass:
+    @pytest.mark.parametrize("cls", ["fa-star", "fas fa-star", "fa-solid fa-house"])
+    def test_valid_classes_accepted(self, cls):
+        assert validate_font_awesome_class(cls) == (True, None)
+
+    @pytest.mark.parametrize("cls", ["star", "glyphicon-star", ""])
+    def test_classes_without_fa_prefix_rejected(self, cls):
+        assert validate_font_awesome_class(cls)[0] is False
+
+    def test_injection_attempt_rejected(self):
+        assert validate_font_awesome_class('fa-star" onload="alert(1)')[0] is False
+
+    def test_angle_brackets_rejected(self):
+        assert validate_font_awesome_class("<script>fa-star</script>")[0] is False
+
+    def test_non_string_rejected(self):
+        valid, error = validate_font_awesome_class(None)
+        assert valid is False and "string" in error
+
+    def test_explicit_fa_check_is_unreachable_but_harmless(self):
+        # Characterized, not fixed: the regex already requires 'fa-', so the
+        # follow-up `if 'fa-' not in class_name` can never fire. Anything
+        # lacking 'fa-' is rejected by the pattern first, with the pattern's
+        # own message.
+        valid, error = validate_font_awesome_class("star")
+        assert valid is False
+        assert error == "Invalid Font Awesome class name format"
+
+
+class TestValidateFileUpload:
+    def test_plain_filename_accepted(self):
+        assert validate_file_upload("logo.png") == (True, None)
+
+    @pytest.mark.parametrize("filename", [
+        "../etc/passwd", "dir/file.png", "dir\\file.png", "..\\..\\secrets",
+    ])
+    def test_traversal_characters_rejected(self, filename):
+        valid, error = validate_file_upload(filename)
+        assert valid is False and "invalid characters" in error
+
+    @pytest.mark.parametrize("filename", ["", None, 123])
+    def test_empty_or_non_string_rejected(self, filename):
+        assert validate_file_upload(filename)[0] is False
+
+    def test_allowed_extension_accepted(self):
+        assert validate_file_upload("font.ttf", allowed_extensions=[".ttf", ".otf"]) == (True, None)
+
+    def test_disallowed_extension_rejected(self):
+        valid, error = validate_file_upload("evil.exe", allowed_extensions=[".ttf"])
+        assert valid is False and "extension" in error
+
+    def test_uppercase_filename_extension_matches(self):
+        assert validate_file_upload("FONT.TTF", allowed_extensions=[".ttf"]) == (True, None)
+
+    def test_uppercase_allowed_list_matches(self):
+        # Regression: only the filename side was lowercased, so a caller
+        # passing ['.TTF'] rejected every valid .ttf upload.
+        assert validate_file_upload("font.ttf", allowed_extensions=[".TTF"]) == (True, None)
+
+    def test_no_extension_list_skips_the_check(self):
+        assert validate_file_upload("anything.xyz") == (True, None)
+
+
+class TestValidateMimeType:
+    def test_known_type_accepted(self):
+        assert validate_mime_type("logo.png", ["image/png"]) == (True, None)
+
+    def test_mismatched_type_rejected(self):
+        valid, error = validate_mime_type("logo.png", ["image/jpeg"])
+        assert valid is False and "not allowed" in error
+
+    def test_undeterminable_type_rejected(self):
+        valid, error = validate_mime_type("mystery.zzz", ["image/png"])
+        assert valid is False and "Could not determine" in error
+
+    def test_guess_type_failure_is_caught(self, monkeypatch):
+        import mimetypes
+        monkeypatch.setattr(mimetypes, "guess_type",
+                            lambda *a, **kw: (_ for _ in ()).throw(RuntimeError("boom")))
+        valid, error = validate_mime_type("logo.png", ["image/png"])
+        assert valid is False and "Error validating MIME type" in error
+
+
+class TestValidateNumericRange:
+    def test_value_in_range(self):
+        assert validate_numeric_range(5, min_val=0, max_val=10) == (True, None)
+
+    def test_boundaries_are_inclusive(self):
+        assert validate_numeric_range(0, min_val=0, max_val=10) == (True, None)
+        assert validate_numeric_range(10, min_val=0, max_val=10) == (True, None)
+
+    def test_below_minimum_rejected(self):
+        valid, error = validate_numeric_range(-1, min_val=0)
+        assert valid is False and "at least" in error
+
+    def test_above_maximum_rejected(self):
+        valid, error = validate_numeric_range(11, max_val=10)
+        assert valid is False and "at most" in error
+
+    def test_floats_accepted(self):
+        assert validate_numeric_range(2.5, min_val=0, max_val=10) == (True, None)
+
+    def test_no_bounds_accepts_any_number(self):
+        assert validate_numeric_range(-9999) == (True, None)
+
+    @pytest.mark.parametrize("value", ["5", None, [], {}])
+    def test_non_numeric_rejected(self, value):
+        valid, error = validate_numeric_range(value, min_val=0, max_val=10)
+        assert valid is False and error == "Value must be a number"
+
+    @pytest.mark.parametrize("value", [True, False])
+    def test_booleans_rejected(self, value):
+        # Regression: bool subclasses int, so True passed the isinstance
+        # check and then compared as 1 against the range.
+        valid, error = validate_numeric_range(value, min_val=0, max_val=10)
+        assert valid is False and error == "Value must be a number"
+
+
+class TestValidateStringLength:
+    def test_within_range(self):
+        assert validate_string_length("hello", min_length=1, max_length=10) == (True, None)
+
+    def test_boundaries_are_inclusive(self):
+        assert validate_string_length("abc", min_length=3, max_length=3) == (True, None)
+
+    def test_too_short_rejected(self):
+        valid, error = validate_string_length("", min_length=1)
+        assert valid is False and "at least" in error
+
+    def test_too_long_rejected(self):
+        valid, error = validate_string_length("abcdef", max_length=3)
+        assert valid is False and "at most" in error
+
+    def test_non_string_rejected(self):
+        valid, error = validate_string_length(123, max_length=10)
+        assert valid is False and "must be a string" in error
+
+    def test_no_bounds_accepts_anything(self):
+        assert validate_string_length("") == (True, None)
+
+
+class TestSanitizePluginConfig:
+    def test_valid_keys_and_scalars_kept(self):
+        config = {"enabled": True, "count": 3, "ratio": 1.5, "name": "clock"}
+        assert sanitize_plugin_config(config) == config
+
+    @pytest.mark.parametrize("key", ["has space", "has-dash", "has.dot", "has/slash", ""])
+    def test_invalid_key_names_dropped(self, key):
+        assert sanitize_plugin_config({key: "value", "good": 1}) == {"good": 1}
+
+    def test_non_string_keys_dropped(self):
+        assert sanitize_plugin_config({1: "a", "good": 2}) == {"good": 2}
+
+    def test_nested_dicts_recursed(self):
+        result = sanitize_plugin_config({"outer": {"inner": 1, "bad key": 2}})
+        assert result == {"outer": {"inner": 1}}
+
+    def test_list_of_scalars_preserved(self):
+        assert sanitize_plugin_config({"teams": ["PHI", "NYG"]})["teams"] == ["PHI", "NYG"]
+
+    def test_list_of_dicts_recursed(self):
+        result = sanitize_plugin_config({"items": [{"ok": 1, "bad key": 2}]})
+        assert result["items"] == [{"ok": 1}]
+
+    def test_unknown_value_types_dropped(self):
+        assert sanitize_plugin_config({"weird": {1, 2, 3}, "good": 1}) == {"good": 1}
+
+    def test_none_values_dropped(self):
+        assert sanitize_plugin_config({"nothing": None, "good": 1}) == {"good": 1}
+
+    def test_strings_are_not_html_escaped(self):
+        # Pinned, not a bug: escaping here would persist the escaped form in
+        # config.json. Output escaping belongs to the template layer, which
+        # the function's docstring now says explicitly.
+        payload = "<script>alert(1)</script>"
+        assert sanitize_plugin_config({"title": payload})["title"] == payload
+
+    def test_empty_config(self):
+        assert sanitize_plugin_config({}) == {}

web_interface/blueprints/api_v3.py

@@ -328,7 +328,7 @@ def save_schedule_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
@@ -536,7 +536,7 @@ def save_dim_schedule_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
@@ -1345,18 +1345,20 @@ def save_raw_main_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        data = request.get_json()
+        # silent=True so a malformed body returns None instead of raising
+        # Werkzeug's own BadRequest, which would answer in a different
+        # shape than this API's. Distinguish the two causes: a body that
+        # was sent but does not parse is a different mistake from no body.
+        data = request.get_json(silent=True)
+        if data is None and request.get_data():
+            return jsonify({'status': 'error', 'message': 'Invalid JSON in request body'}), 400
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
-        # Validate that it's valid JSON (already parsed by request.get_json())
         # Save the raw config file
         api_v3.config_manager.save_raw_file_content('main', data)
 
         return jsonify({'status': 'success', 'message': 'Main configuration saved successfully'})
-    except json.JSONDecodeError as e:
-        logger.error('Invalid JSON', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'Invalid JSON in request body'}), 400
     except Exception as e:
         from src.exceptions import ConfigError
         logger.error("Error saving raw main config", exc_info=True)
@@ -1391,7 +1393,11 @@ def save_raw_secrets_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        data = request.get_json()
+        # See save_raw_main_config: silent parsing, with a sent-but-broken
+        # body reported separately from a missing one.
+        data = request.get_json(silent=True)
+        if data is None and request.get_data():
+            return jsonify({'status': 'error', 'message': 'Invalid JSON in request body'}), 400
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
@@ -1403,9 +1409,6 @@ def save_raw_secrets_config():
             api_v3.plugin_store_manager.github_token = api_v3.plugin_store_manager._load_github_token()
 
         return jsonify({'status': 'success', 'message': 'Secrets configuration saved successfully'})
-    except json.JSONDecodeError as e:
-        logger.error('Invalid JSON', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'Invalid JSON in request body'}), 400
     except Exception as e:
         from src.exceptions import ConfigError
         logger.error("Error saving raw secrets config", exc_info=True)
@@ -2411,7 +2414,7 @@ def get_on_demand_status():
 def start_on_demand_display():
     """Request the display controller to run a specific plugin on-demand."""
     try:
-        data = request.get_json() or {}
+        data = request.get_json(silent=True) or {}
         plugin_id = data.get('plugin_id')
         mode = data.get('mode')
         duration = data.get('duration')
@@ -2935,7 +2938,7 @@ def manage_plugin_limits(plugin_id):
                 })
         else:
             # POST - Set limits
-            data = request.get_json() or {}
+            data = request.get_json(silent=True) or {}
             from src.plugin_system.resource_monitor import ResourceLimits
 
             limits = ResourceLimits(
@@ -2966,7 +2969,7 @@ def toggle_plugin():
         content_type = request.content_type or ''
 
         if 'application/json' in content_type:
-            data = request.get_json()
+            data = request.get_json(silent=True)
             if not data or 'plugin_id' not in data or 'enabled' not in data:
                 return jsonify({'status': 'error', 'message': 'plugin_id and enabled required'}), 400
             plugin_id = data['plugin_id']
@@ -3837,7 +3840,7 @@ def install_plugin():
         if not api_v3.plugin_store_manager:
             return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data or 'plugin_id' not in data:
             return jsonify({'status': 'error', 'message': 'plugin_id required'}), 400
 
@@ -3971,10 +3974,15 @@ def install_plugin_from_url():
         if not api_v3.plugin_store_manager:
             return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data or 'repo_url' not in data:
             return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
 
+        # A non-string repo_url is a client mistake, not a server fault:
+        # .strip() would raise and the catch-all would report it as a 500.
+        if not isinstance(data['repo_url'], str) or not data['repo_url'].strip():
+            return jsonify({'status': 'error', 'message': 'repo_url must be a non-empty string'}), 400
+
         repo_url = data['repo_url'].strip()
         plugin_id = data.get('plugin_id')  # Optional, for monorepo installations
         plugin_path = data.get('plugin_path')  # Optional, for monorepo subdirectory
@@ -4026,10 +4034,15 @@ def get_registry_from_url():
         if not api_v3.plugin_store_manager:
             return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data or 'repo_url' not in data:
             return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
 
+        # A non-string repo_url is a client mistake, not a server fault:
+        # .strip() would raise and the catch-all would report it as a 500.
+        if not isinstance(data['repo_url'], str) or not data['repo_url'].strip():
+            return jsonify({'status': 'error', 'message': 'repo_url must be a non-empty string'}), 400
+
         repo_url = data['repo_url'].strip()
 
         # Get registry from the URL
@@ -4071,10 +4084,15 @@ def add_saved_repository():
         if not api_v3.saved_repositories_manager:
             return jsonify({'status': 'error', 'message': 'Saved repositories manager not initialized'}), 500
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data or 'repo_url' not in data:
             return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
 
+        # A non-string repo_url is a client mistake, not a server fault:
+        # .strip() would raise and the catch-all would report it as a 500.
+        if not isinstance(data['repo_url'], str) or not data['repo_url'].strip():
+            return jsonify({'status': 'error', 'message': 'repo_url must be a non-empty string'}), 400
+
         repo_url = data['repo_url'].strip()
         name = data.get('name')
 
@@ -4102,7 +4120,7 @@ def remove_saved_repository():
         if not api_v3.saved_repositories_manager:
             return jsonify({'status': 'error', 'message': 'Saved repositories manager not initialized'}), 500
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data or 'repo_url' not in data:
             return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
 
@@ -4236,7 +4254,7 @@ def refresh_plugin_store():
         if not api_v3.plugin_store_manager:
             return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
 
-        data = request.get_json() or {}
+        data = request.get_json(silent=True) or {}
         fetch_commit_info = data.get('fetch_commit_info', data.get('fetch_latest_versions', False))
 
         # Force refresh the registry
@@ -5822,7 +5840,7 @@ def reset_plugin_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        data = request.get_json() or {}
+        data = request.get_json(silent=True) or {}
         plugin_id = data.get('plugin_id')
         preserve_secrets = data.get('preserve_secrets', True)
 
@@ -6209,7 +6227,7 @@ sys.exit(proc.returncode)
 def authenticate_spotify():
     """Run Spotify authentication script"""
     try:
-        data = request.get_json() or {}
+        data = request.get_json(silent=True) or {}
         redirect_url = data.get('redirect_url', '').strip()
 
         # Get plugin directory
@@ -6272,7 +6290,6 @@ sys.exit(proc.returncode)
                     timeout=120,
                     env=env
                 )
-                os.unlink(wrapper_path)
 
                 if result.returncode == 0:
                     return jsonify({
@@ -6287,9 +6304,13 @@ sys.exit(proc.returncode)
                         'output': result.stdout + result.stderr
                     }), 400
             except subprocess.TimeoutExpired:
+                return jsonify({'status': 'error', 'message': 'Authentication timed out'}), 408
+            finally:
+                # The wrapper carries the user's redirect URL, so it must not
+                # survive the request on any path — including a failure to
+                # launch, which the previous per-branch unlinks missed.
                 if os.path.exists(wrapper_path):
                     os.unlink(wrapper_path)
-                return jsonify({'status': 'error', 'message': 'Authentication timed out'}), 408
         else:
             # Step 1: Get authorization URL
             # Import the script's functions directly to get the auth URL
@@ -6526,7 +6547,7 @@ def get_fonts_overrides():
 def save_fonts_overrides():
     """Save font overrides"""
     try:
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
@@ -7146,7 +7167,7 @@ def upload_of_the_day_json():
 def delete_of_the_day_json():
     """Delete a JSON file from of-the-day plugin"""
     try:
-        data = request.get_json() or {}
+        data = request.get_json(silent=True) or {}
         file_id = data.get('file_id')  # This is the category_name
 
         if not file_id:
@@ -7236,6 +7257,29 @@ def serve_plugin_static(plugin_id, file_path):
         return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
 
 
+_MAX_CREDENTIAL_BACKUPS = 5
+
+
+def _prune_credential_backups(plugin_dir: Path) -> None:
+    """Keep only the newest _MAX_CREDENTIAL_BACKUPS credential backups.
+
+    Every re-upload copies the previous credentials.json aside. Without
+    pruning those accumulate for the life of the install — each one a
+    complete set of OAuth client credentials sitting in the plugin
+    directory.
+    """
+    backups = sorted(
+        plugin_dir.glob('credentials.json.backup.*'),
+        key=lambda p: p.stat().st_mtime,
+        reverse=True,
+    )
+    for stale in backups[_MAX_CREDENTIAL_BACKUPS:]:
+        try:
+            stale.unlink()
+        except OSError:
+            logger.warning("Could not remove old credential backup %s", stale.name)
+
+
 @api_v3.route('/plugins/calendar/upload-credentials', methods=['POST'])
 def upload_calendar_credentials():
     """Upload credentials.json file for calendar plugin"""
@@ -7263,24 +7307,20 @@ def upload_calendar_credentials():
         try:
             file_content = file.read()
             file.seek(0)
-            json.loads(file_content)
+            creds_data = json.loads(file_content)
         except json.JSONDecodeError:
             return jsonify({'status': 'error', 'message': 'File is not valid JSON'}), 400
 
-        # Validate it looks like Google OAuth credentials
-        try:
-            file.seek(0)
-            creds_data = json.loads(file.read())
-            file.seek(0)
-
-            # Check for required Google OAuth fields
-            if 'installed' not in creds_data and 'web' not in creds_data:
+        # Validate it looks like Google OAuth credentials. A bare scalar, a
+        # list, true/null — all valid JSON, none of them credentials. Reject
+        # rather than save: a file written as credentials.json but unusable
+        # as credentials only fails later, somewhere less obvious.
+        if not isinstance(creds_data, dict) or not (
+                'installed' in creds_data or 'web' in creds_data):
             return jsonify({
                 'status': 'error',
                 'message': 'File does not appear to be a valid Google OAuth credentials file'
             }), 400
-        except Exception:
-            pass  # Continue even if validation fails
 
         # Get plugin directory
         plugin_id = 'calendar'
@@ -7300,6 +7340,7 @@ def upload_calendar_credentials():
             backup_path = Path(plugin_dir) / f'credentials.json.backup.{int(time.time())}'
             import shutil
             shutil.copy2(credentials_path, backup_path)
+            _prune_credential_backups(Path(plugin_dir))
 
         # Save new file
         file.save(str(credentials_path))
@@ -7603,7 +7644,7 @@ def connect_wifi():
     try:
         from src.wifi_manager import WiFiManager
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data:
             return jsonify({
                 'status': 'error',
@@ -7757,7 +7798,7 @@ def set_auto_enable_ap_mode():
     try:
         from src.wifi_manager import WiFiManager
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if data is None or 'auto_enable_ap_mode' not in data:
             return jsonify({
                 'status': 'error',
@@ -7886,7 +7927,7 @@ def delete_cache_file():
             from src.cache_manager import CacheManager
             api_v3.cache_manager = CacheManager()
 
-        data = request.get_json()
+        data = request.get_json(silent=True)
         if not data or 'key' not in data:
             return jsonify({'status': 'error', 'message': 'cache key is required'}), 400
 
@@ -8149,7 +8190,16 @@ def backup_restore():
         try:
             opts_dict = json.loads(options_raw)
         except json.JSONDecodeError:
-            opts_dict = {}
+            opts_dict = None
+        if not isinstance(opts_dict, dict):
+            # Every option defaults to True, so falling back to {} on a
+            # parse failure would silently perform a FULL restore —
+            # secrets and all — for a caller who asked for a narrow one
+            # and mis-serialized it. Refuse instead of guessing.
+            return jsonify({
+                'status': 'error',
+                'message': 'Invalid options: expected a JSON object',
+            }), 400
         options = RestoreOptions(
             restore_config=bool(opts_dict.get('restore_config', True)),
             restore_secrets=bool(opts_dict.get('restore_secrets', True)),