From 6cfcf2e3845cafdaa7e6a3f9a0d5b1abb3acadb5 Mon Sep 17 00:00:00 2001 From: Chuck <33324927+ChuckBuilds@users.noreply.github.com> Date: Mon, 28 Sep 2026 10:42:07 -0400 Subject: [PATCH] fix(web): plugin dir resolver in routes, nmcli AP detection, daemon config reload, upload safety, BDF preview (#655) * fix(web): plugin dir resolver in routes, nmcli AP detection, daemon config reload, upload safety - Route plugin lookups (installed list, update, recorded version, config form, web UI pages) through the plugin manager's resolver so plugins in ledmatrix- directories work. - Captive-portal detection also sees the nmcli fallback AP (cached). - WiFi monitor daemon re-reads wifi_config.json when its mtime changes. - Drop the AP check in disconnect_from_network that could never fire. - LED status file per WiFiManager; config path falls back to this checkout. - BDF font preview via src.common.bdf_font. - Asset uploads validate every file before saving; metadata and calendar credentials written atomically; no absolute path in the response; asset delete answers 400 for a missing body. - Coerce string booleans in plugin toggle, on-demand start and AP force. - SSE broadcaster clears its thread handle before exiting. - start.py log filter handles every exc_info form. - Cleanups: unused plugins/fonts partial work, duplicate backup catch-alls, raw-config error helper, update-route tidy, redundant imports. Co-Authored-By: Claude Opus 5.5 * fix(web): request BDF font previews now that the server renders them The Fonts tab skipped the preview request for .bdf files because the server used to refuse them; /fonts/preview now draws BDF with the shared loader. Co-Authored-By: Claude Opus 5.5 * fix(web): take the update route's plugin directory from a directory listing CodeQL flagged the path built from the request's plugin_id (the id was already validated with safe_path_component, which CodeQL doesn't model; the same flow on main is alerts 738/739). The directory is now the entry of plugins_dir matched by name, so nothing built from user input reaches the filesystem; an id with nothing installed goes to the store manager, which reports it not found as before. Co-Authored-By: Claude Opus 5.5 * fix(web): read the blueprint's plugin_manager defensively in _plugin_directory _get_plugin_version now goes through _plugin_directory, which read api_v3.plugin_manager directly; the attribute exists only once the app sets it, so test_path_traversal_guards::test_a_real_manifest_is_read failed when run on its own (order-dependent in the full suite). Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- CHANGELOG.md | 12 + scripts/utils/wifi_monitor_daemon.py | 31 +- src/wifi_manager.py | 73 +++-- test/test_api_v3_bool_coercion.py | 76 +++++ test/test_api_v3_upload_writes.py | 138 +++++++++ test/test_api_v3_wifi_endpoints.py | 9 +- test/test_web_plugin_dir_resolution.py | 200 ++++++++++++ test/test_wifi_ap_state_handling.py | 198 ++++++++++++ test/web_interface/test_api_v3_config_raw.py | 19 ++ .../test_api_v3_unhandled_errors.py | 19 +- test/web_interface/test_update_all_plugins.py | 1 - .../test_web_backend_small_fixes.py | 144 +++++++++ web_interface/app.py | 76 +++-- web_interface/blueprints/api_v3/__init__.py | 19 +- web_interface/blueprints/api_v3/backup.py | 292 +++++++++--------- web_interface/blueprints/api_v3/config.py | 60 ++-- web_interface/blueprints/api_v3/display.py | 8 +- web_interface/blueprints/api_v3/fonts.py | 49 ++- web_interface/blueprints/api_v3/plugins.py | 155 ++++++---- web_interface/blueprints/api_v3/wifi.py | 5 +- web_interface/blueprints/pages_v3.py | 117 ++----- web_interface/start.py | 17 +- .../templates/v3/partials/fonts.html | 8 - 23 files changed, 1300 insertions(+), 426 deletions(-) create mode 100644 test/test_api_v3_bool_coercion.py create mode 100644 test/test_api_v3_upload_writes.py create mode 100644 test/test_web_plugin_dir_resolution.py create mode 100644 test/test_wifi_ap_state_handling.py create mode 100644 test/web_interface/test_web_backend_small_fixes.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 66125ae3..10fa767e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,18 @@ accepts both, but the store flags the old spelling as deprecated ## Unreleased +- Web backend and WiFi fixes: + - Plugins installed as `ledmatrix-` (or in a directory not named after their id) work in the installed list, the update button, recorded versions, the plugin config form and plugin web UI pages. Those routes built `plugins_dir/` themselves instead of asking the plugin manager. + - The captive-portal checks (`/generate_204` and friends) also detect an access point brought up through NetworkManager, the fallback `enable_ap_mode` uses without hostapd; only hostapd was checked, so phones on that AP were told the internet worked. + - The WiFi monitor daemon re-reads `wifi_config.json` when it changes, so the "auto-enable AP mode" toggle takes effect without restarting the daemon. + - Disconnecting from WiFi in the web UI no longer runs an AP-mode check that could never enable the AP; it only added seconds of waiting. The daemon still enables the AP after its grace period. + - The WiFi status message file follows each WiFi manager's own config directory, and the config path falls back to this checkout rather than `/home/ledpi/LEDMatrix`. + - Fonts tab: the preview endpoint renders BDF fonts with the panel's own rasterizer instead of refusing them. (The Fonts page still skips the request for `.bdf`; enabling it there is a separate template change.) + - Uploading several plugin images checks every file before saving any, so a rejected file no longer leaves the others saved; the images' `.metadata.json` and the calendar plugin's `credentials.json` are written atomically, and the credentials upload no longer returns the server's absolute path. + - `"false"` sent as a string no longer counts as true when toggling a plugin (including Starlark apps) or starting on-demand mode (`pinned`, `start_service`); `force` on the AP-enable route is parsed like every other WiFi boolean (`"yes"` and `1` now force). + - The live-preview stream starts a new broadcast thread for a client that connects while the previous one is shutting down; that client got no updates. + - The web server's log filter no longer raises when werkzeug logs with `exc_info=True`. + - The raw secrets editor's save errors carry `error_code` like the main config's; the asset delete route answers 400 for a missing body instead of 415/500. Dead code removed: an unused manifest scan on each Plugins-tab load, backup routes' duplicate catch-alls, redundant imports. - Plugin system fixes: - Unloading a plugin waits (up to 5s) for an in-flight `update()` before running `cleanup()`/`on_disable()`, and an update that finishes after the unload no longer puts the plugin back to ENABLED. - A plugin whose load fails after its module was imported (constructor, `validate_config()` or `on_enable()` raising) no longer leaves that module cached: fixing the plugin and reloading it runs the new code without a restart. Its font registrations are dropped too. diff --git a/scripts/utils/wifi_monitor_daemon.py b/scripts/utils/wifi_monitor_daemon.py index 397bcf63..6c96db76 100755 --- a/scripts/utils/wifi_monitor_daemon.py +++ b/scripts/utils/wifi_monitor_daemon.py @@ -42,6 +42,8 @@ class WiFiMonitorDaemon: """ self.check_interval = check_interval self.wifi_manager = WiFiManager() + # mtime of wifi_config.json as last loaded; see _reload_config_if_changed. + self._config_mtime = self._config_file_mtime() self.running = True self.last_state = None # Counts consecutive checks where nmcli says "connected" but internet is unreachable. @@ -57,7 +59,32 @@ class WiFiMonitorDaemon: """Handle shutdown signals""" logger.info(f"Received signal {signum}, shutting down...") self.running = False - + + def _config_file_mtime(self): + try: + return self.wifi_manager.config_path.stat().st_mtime_ns + except OSError: + return None + + def _reload_config_if_changed(self): + """Re-read wifi_config.json when it has changed on disk. + + The web UI's auto-enable toggle (POST /api/v3/wifi/ap/auto-enable) + only writes the file; this process read it once at startup, so the + toggle did nothing until the daemon restarted. One stat per check. + """ + mtime = self._config_file_mtime() + if mtime is None or mtime == self._config_mtime: + return + before = self.wifi_manager.config.get("auto_enable_ap_mode", True) + self.wifi_manager._load_config() + # _load_config can itself save (it fills in missing keys), so take + # the mtime after it, or that save would trigger another reload. + self._config_mtime = self._config_file_mtime() + after = self.wifi_manager.config.get("auto_enable_ap_mode", True) + if after != before: + logger.info(f"wifi_config.json changed: auto_enable_ap_mode={after}") + def run(self): """Main daemon loop""" logger.info("WiFi Monitor Daemon started") @@ -78,6 +105,8 @@ class WiFiMonitorDaemon: while self.running: try: + self._reload_config_if_changed() + # One combined check that also returns the state it observed — # the previous flow fetched status before AND after the check # on top of the check's own internal fetch, each one several diff --git a/src/wifi_manager.py b/src/wifi_manager.py index 88f3c420..12d98797 100644 --- a/src/wifi_manager.py +++ b/src/wifi_manager.py @@ -44,14 +44,12 @@ def get_wifi_config_path(): # Try to determine project root project_root = os.environ.get('LEDMATRIX_ROOT') if not project_root: - # Try to find project root by looking for config directory - current = Path(__file__).resolve().parent.parent - if (current / 'config').exists(): - project_root = str(current) - else: - # Fallback to common location - project_root = "/home/ledpi/LEDMatrix" - + # This file is /src/wifi_manager.py. The root is used even when + # config/ does not exist yet (WiFiManager creates it): the old + # fallback was a hardcoded /home/ledpi/LEDMatrix, which is some other + # user's checkout, or nothing, on any install not made as ledpi. + project_root = str(Path(__file__).resolve().parents[1]) + return Path(project_root) / "config" / "wifi_config.json" @@ -90,8 +88,11 @@ AP_PROFILE_NAME = "LEDMatrix-Setup-AP" #: Deleted by name only, never by SSID, so a saved home network is never hit. AP_PROFILE_NAMES = (AP_PROFILE_NAME, "Hotspot", "TickerSetup-AP") -# LED status message file (for display_controller integration) -LED_STATUS_FILE = None # Will be set dynamically +# LED status message file (for display_controller integration). None means +# each WiFiManager writes next to its own config file, which for the default +# config is get_wifi_status_path(), the file the display reads. Set only to +# redirect it (tests). +LED_STATUS_FILE = None @dataclass @@ -131,12 +132,12 @@ class WiFiManager: self.config_path.parent.mkdir(parents=True, exist_ok=True) self._load_config() - # Set LED status file path (for display_controller integration) - global LED_STATUS_FILE - if LED_STATUS_FILE is None: - project_root = self.config_path.parent.parent - LED_STATUS_FILE = project_root / "config" / "wifi_status.json" - + # LED status file (for display_controller integration): next to this + # manager's config. It used to be a module global set by whichever + # manager was built first, so a later one with a different config + # path wrote its messages to the first one's directory. + self._led_status_file = self.config_path.parent / "wifi_status.json" + # Check which tools are available self.has_nmcli = self._check_command("nmcli") self.has_iwlist = self._check_command("iwlist") @@ -182,21 +183,22 @@ class WiFiManager: duration: How long to show message (seconds) """ try: - if LED_STATUS_FILE is None: + status_file = LED_STATUS_FILE or getattr(self, '_led_status_file', None) + if status_file is None: return - + status = { 'message': message, 'timestamp': time.time(), 'duration': duration } - LED_STATUS_FILE.parent.mkdir(parents=True, exist_ok=True) + status_file.parent.mkdir(parents=True, exist_ok=True) # Write-then-rename: the display reads this at ~1 Hz and deletes # a file it can't parse, so a half-written one would lose the message. - tmp_path = LED_STATUS_FILE.with_name(LED_STATUS_FILE.name + '.tmp') + tmp_path = status_file.with_name(status_file.name + '.tmp') with open(tmp_path, 'w') as f: json.dump(status, f) - os.replace(tmp_path, LED_STATUS_FILE) + os.replace(tmp_path, status_file) logger.info(f"LED message: {message}") except Exception as e: logger.debug(f"Could not write LED status message: {e}") @@ -204,8 +206,9 @@ class WiFiManager: def _clear_led_message(self): """Clear any WiFi status message from LED display.""" try: - if LED_STATUS_FILE and LED_STATUS_FILE.exists(): - LED_STATUS_FILE.unlink() + status_file = LED_STATUS_FILE or getattr(self, '_led_status_file', None) + if status_file and status_file.exists(): + status_file.unlink() except Exception as e: logger.debug(f"Could not clear LED status message: {e}") @@ -1668,8 +1671,9 @@ class WiFiManager: Disconnect from the current WiFi network Args: - skip_ap_check: If True, skip auto-enabling AP mode after disconnect - (useful when switching networks) + skip_ap_check: Accepted for callers that still pass it; ignored. + Auto-enabling AP mode after a disconnect is the + wifi monitor daemon's job. Returns: Tuple of (success, message) @@ -1705,18 +1709,13 @@ class WiFiManager: logger.info("Successfully disconnected from WiFi network") # Wait longer for the disconnect to fully complete time.sleep(2) - - # Check if AP mode should be auto-enabled - # Skip if we're switching networks (skip_ap_check=True) - if not skip_ap_check: - auto_enable = self.config.get("auto_enable_ap_mode", True) - if auto_enable: - # Give it a moment, then check if we should enable AP mode - time.sleep(1) - self.check_and_manage_ap_mode() - else: - logger.debug("Skipping AP mode check (network switch in progress)") - + + # No AP-mode check here. It used to run one, but the web + # routes build a fresh WiFiManager per request, so its + # grace-period counter started at 0 and one check could + # never reach the 3 needed to enable the AP: it only added + # seconds of sleeps and nmcli calls. The monitor daemon's + # long-lived manager is what counts consecutive checks. return True, "Disconnected from WiFi network" else: error_msg = result.stderr.strip() or result.stdout.strip() diff --git a/test/test_api_v3_bool_coercion.py b/test/test_api_v3_bool_coercion.py new file mode 100644 index 00000000..c8a60502 --- /dev/null +++ b/test/test_api_v3_bool_coercion.py @@ -0,0 +1,76 @@ +"""Boolean request fields are coerced, not used raw. + +``bool("false")`` is True. /plugins/toggle stored a JSON ``"enabled": "false"`` +as-is in config.json (a truthy string the display then treats as enabled) and +passed it to the Starlark toggle the same way, and /display/on-demand/start +pinned the mode and restarted the service for ``"pinned": "false"`` / +``"start_service": "false"``. +""" + +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 + + +class TestPluginToggle: + @pytest.fixture + def saved(self, api_v3_module, monkeypatch): + monkeypatch.setattr(api_v3_module, '_discovered_plugin_manifests', + lambda *a, **k: {'clock': {}}) + api_v3_module.api_v3.config_manager.load_config = MagicMock(return_value={}) + captured = {} + + def save(config_manager, config, create_backup=True): + captured.update(config) + return True, None + monkeypatch.setattr(api_v3_module, '_save_config_atomic', save) + return captured + + @pytest.mark.parametrize("raw,expected", [ + ("false", False), ("true", True), (False, False), (True, True), (0, False), + ]) + def test_enabled_is_stored_as_a_real_bool(self, api_v3_client, saved, raw, expected): + response = api_v3_client.post('/api/v3/plugins/toggle', + json={'plugin_id': 'clock', 'enabled': raw}) + assert response.status_code == 200, response.get_json() + assert saved['clock']['enabled'] is expected + + def test_a_starlark_app_gets_the_coerced_value(self, api_v3_client, api_v3_module): + toggle = MagicMock(return_value=({'status': 'success'}, 200)) + with patch('web_interface.blueprints.api_v3.plugins._toggle_starlark_app', toggle): + api_v3_client.post('/api/v3/plugins/toggle', + json={'plugin_id': 'starlark:clock', 'enabled': 'false'}) + toggle.assert_called_once_with('clock', False) + + +class TestOnDemandStart: + @pytest.fixture + def service(self, api_v3_module): + api_v3_module.api_v3.plugin_manager = None + api_v3_module.api_v3.config_manager = None + with patch("web_interface.blueprints.api_v3.display._get_display_service_status", + return_value={"active": True}), \ + patch("web_interface.blueprints.api_v3.display._stop_display_service") as stop, \ + patch("web_interface.blueprints.api_v3.display._ensure_display_service_running", + return_value={"active": True}) as ensure: + yield stop, ensure + + def test_string_false_neither_pins_nor_restarts(self, api_v3_client, service): + stop, ensure = service + response = api_v3_client.post('/api/v3/display/on-demand/start', json={ + 'plugin_id': 'weather', 'pinned': 'false', 'start_service': 'false'}) + assert response.status_code == 200, response.get_json() + assert response.get_json()['data']['pinned'] is False + stop.assert_not_called() + ensure.assert_not_called() + + def test_string_true_pins(self, api_v3_client, service): + response = api_v3_client.post('/api/v3/display/on-demand/start', json={ + 'plugin_id': 'weather', 'pinned': 'true', 'start_service': False}) + assert response.get_json()['data']['pinned'] is True diff --git a/test/test_api_v3_upload_writes.py b/test/test_api_v3_upload_writes.py new file mode 100644 index 00000000..92f446ea --- /dev/null +++ b/test/test_api_v3_upload_writes.py @@ -0,0 +1,138 @@ +"""Plugin asset uploads and the calendar credentials upload write safely. + +* An upload of several files checked and saved them one at a time, so a bad + third file answered 400 after the first two were already on disk and in + .metadata.json -- the user was told it failed and the images appeared anyway. +* .metadata.json and credentials.json were written in place (open 'w' / + FileStorage.save), so a failure mid-write left a truncated file. +* The asset delete route called get_json() without silent=True. +* The credentials route returned the server's absolute path; nothing reads it. +""" + +import io +import json +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 + +PNG = b"\x89PNG\r\n\x1a\n" + b"0" * 20 + + +@pytest.fixture +def project(tmp_path, api_v3_module, monkeypatch): + import web_interface.blueprints.api_v3.plugins as plugins_module + monkeypatch.setattr(plugins_module, "PROJECT_ROOT", tmp_path) + return tmp_path / "assets" / "plugins" / "static-image" / "uploads" + + +def _upload(client, files): + return client.post( + "/api/v3/plugins/assets/upload", + data={"plugin_id": "static-image", + "files": [(io.BytesIO(body), name) for name, body in files]}, + content_type="multipart/form-data", + ) + + +class TestAssetUpload: + def test_a_bad_file_saves_none_of_the_batch(self, api_v3_client, project): + response = _upload(api_v3_client, [ + ("a.png", PNG), ("b.png", PNG), ("c.png", b"not an image at all"), + ]) + assert response.status_code == 400 + saved = [p.name for p in project.iterdir()] if project.exists() else [] + assert saved == [], "files before the bad one were saved anyway" + + def test_a_batch_over_the_total_limit_saves_none(self, api_v3_client, project, monkeypatch): + # The total-size check also runs before anything is written. + project.mkdir(parents=True) + (project / ".metadata.json").write_text(json.dumps( + {"old": {"id": "old", "size": 50 * 1024 * 1024 - 30}}), encoding="utf-8") + response = _upload(api_v3_client, [("a.png", PNG), ("b.png", PNG)]) + assert response.status_code == 400 + assert sorted(p.name for p in project.iterdir()) == [".metadata.json"] + + def test_a_good_batch_is_all_saved_and_recorded(self, api_v3_client, project): + response = _upload(api_v3_client, [("a.png", PNG), ("b.png", PNG)]) + assert response.status_code == 200, response.get_json() + body = response.get_json() + assert len(body["uploaded_files"]) == 2 + metadata = json.loads((project / ".metadata.json").read_text(encoding="utf-8")) + assert sorted(metadata) == sorted(f["id"] for f in body["uploaded_files"]) + for entry in body["uploaded_files"]: + assert (project / entry["filename"]).exists() + + def test_metadata_is_replaced_by_rename(self, api_v3_client, project, monkeypatch): + import src.config_manager_atomic as atomic + replaced = [] + real = atomic.os.replace + monkeypatch.setattr(atomic.os, "replace", + lambda s, d: (replaced.append(Path(d).name), real(s, d))) + _upload(api_v3_client, [("a.png", PNG)]) + assert ".metadata.json" in replaced + + +class TestAssetDelete: + @pytest.mark.parametrize("kwargs", [ + {}, # no body at all + {"data": "plugin_id=x", "content_type": "application/x-www-form-urlencoded"}, + {"json": ["plugin_id", "image_id"]}, # JSON, not an object + ]) + def test_a_missing_or_non_object_body_is_a_400(self, api_v3_client, project, kwargs): + response = api_v3_client.post("/api/v3/plugins/assets/delete", **kwargs) + assert response.status_code == 400 + assert response.get_json()["status"] == "error" + + def test_metadata_is_replaced_by_rename(self, api_v3_client, project, monkeypatch): + uploaded = _upload(api_v3_client, [("a.png", PNG)]).get_json()["uploaded_files"][0] + import src.config_manager_atomic as atomic + replaced = [] + real = atomic.os.replace + monkeypatch.setattr(atomic.os, "replace", + lambda s, d: (replaced.append(Path(d).name), real(s, d))) + response = api_v3_client.post("/api/v3/plugins/assets/delete", + json={"plugin_id": "static-image", + "image_id": uploaded["id"]}) + assert response.status_code == 200 + assert ".metadata.json" in replaced + assert json.loads((project / ".metadata.json").read_text(encoding="utf-8")) == {} + + +class TestCalendarCredentials: + CREDS = {"installed": {"client_id": "x", "client_secret": "y"}} + + @pytest.fixture + def plugin_dir(self, 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 _post(self, client): + return client.post( + "/api/v3/plugins/calendar/upload-credentials", + data={"file": (io.BytesIO(json.dumps(self.CREDS).encode()), "credentials.json")}, + content_type="multipart/form-data", + ) + + def test_the_absolute_server_path_is_not_returned(self, api_v3_client, plugin_dir): + body = self._post(api_v3_client).get_json() + assert body["status"] == "success" + assert body["path"] == "credentials.json" + assert str(plugin_dir) not in json.dumps(body) + + def test_the_file_is_replaced_by_rename(self, api_v3_client, plugin_dir, monkeypatch): + (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"old": 1}})) + import src.config_manager_atomic as atomic + replaced = [] + real = atomic.os.replace + monkeypatch.setattr(atomic.os, "replace", + lambda s, d: (replaced.append(Path(d).name), real(s, d))) + assert self._post(api_v3_client).status_code == 200 + assert replaced == ["credentials.json"] + assert json.loads((plugin_dir / "credentials.json").read_text()) == self.CREDS diff --git a/test/test_api_v3_wifi_endpoints.py b/test/test_api_v3_wifi_endpoints.py index 38814fb4..0e40bb25 100644 --- a/test/test_api_v3_wifi_endpoints.py +++ b/test/test_api_v3_wifi_endpoints.py @@ -261,8 +261,13 @@ class TestApMode: @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 + ("false", False), ("no", False), + # Parsed by _parse_bool_ish like every other boolean on these routes + # (the radio route's force included); this one used to have its own + # rules, under which "yes" and 1 meant False. + ("yes", True), (1, True), (0, False), + # Anything unrecognised is not a request to force. + ("typo", False), (None, False), (2, False), ]) def test_force_coercion(self, api_v3_client, wifi_manager, raw, expected): wifi_manager.enable_ap_mode.return_value = (True, "ok") diff --git a/test/test_web_plugin_dir_resolution.py b/test/test_web_plugin_dir_resolution.py new file mode 100644 index 00000000..135a935c --- /dev/null +++ b/test/test_web_plugin_dir_resolution.py @@ -0,0 +1,200 @@ +"""Web routes find a plugin through the plugin manager's resolver. + +Several routes built ``plugins_dir/`` themselves. A plugin whose +directory is ``ledmatrix-`` (the store's repo naming), or whose directory +name is not its manifest id, is not there, so those routes silently worked +on nothing: the installed list never refreshed its manifest or read its git +info, the recorded version was '', the update route compared manifests and +commits of a directory that does not exist, and the plugin pages 404'd or +rendered no schema. ``_plugin_directory()`` / ``get_plugin_directory()`` +(src/plugin_system/plugin_dirs.py) is the one answer to "where is plugin X". +""" + +import json +import sys +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +from flask import Flask + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from src.plugin_system.plugin_dirs import resolve_plugin_dir # noqa: E402 +from test._api_v3_test_helpers import api_v3_client, api_v3_module # noqa: F401,E402 + + +def _resolver(plugins_dir): + """What PluginManager.get_plugin_directory does for an id discovery has + not mapped: , then ledmatrix-, in plugins_dir.""" + def get_plugin_directory(plugin_id): + found = resolve_plugin_dir(plugin_id, [plugins_dir], prefix=True, + by_manifest=False) + return str(found) if found else None + return get_plugin_directory + + +@pytest.fixture +def prefixed_plugin(tmp_path, api_v3_module): + """A 'demo' plugin installed as plugins_dir/ledmatrix-demo.""" + plugins_dir = tmp_path / "plugin-repos" + plugin_dir = plugins_dir / "ledmatrix-demo" + plugin_dir.mkdir(parents=True) + (plugin_dir / "manifest.json").write_text(json.dumps({ + "id": "demo", "name": "Demo", "version": "2.0.0", + "description": "fresh from disk", "last_updated": "2026-09-01", + }), encoding="utf-8") + + api = api_v3_module.api_v3 + api.plugin_manager.plugins_dir = str(plugins_dir) + api.plugin_manager.get_plugin_directory = MagicMock(side_effect=_resolver(plugins_dir)) + api.plugin_store_manager.plugins_dir = str(plugins_dir) + return plugin_dir + + +class TestInstalledList: + def test_the_manifest_is_refreshed_from_the_prefixed_directory( + self, api_v3_client, api_v3_module, prefixed_plugin): + api = api_v3_module.api_v3 + info = {"id": "demo", "name": "Demo", "version": "1.0.0", + "description": "stale cached copy", "loaded": False} + api.plugin_manager.get_all_plugin_info = MagicMock(return_value=[info]) + api.plugin_manager.get_plugin = MagicMock(return_value=None) + api.plugin_store_manager.get_registry_info = MagicMock(return_value=None) + api.plugin_store_manager._get_local_git_info = MagicMock(return_value=None) + api.config_manager.load_config = MagicMock(return_value={}) + + response = api_v3_client.get("/api/v3/plugins/installed") + + assert response.status_code == 200 + [entry] = [p for p in response.get_json()["data"]["plugins"] if p["id"] == "demo"] + assert entry["description"] == "fresh from disk" + # And git info is read from the same directory. + api.plugin_store_manager._get_local_git_info.assert_called_once_with(prefixed_plugin) + + +class TestRecordedVersion: + def test_the_version_is_read_from_the_prefixed_directory( + self, api_v3_module, prefixed_plugin): + assert api_v3_module._get_plugin_version("demo") == "2.0.0" + + def test_an_unsafe_id_is_still_refused(self, api_v3_module, prefixed_plugin): + assert api_v3_module._get_plugin_version("../ledmatrix-demo") == "" + + +class TestUpdateRoute: + def _update(self, client, api): + api.plugin_store_manager.get_plugin_info = MagicMock(return_value=None) + api.plugin_store_manager.update_plugin = MagicMock(return_value=True) + api.plugin_store_manager._get_local_git_info = MagicMock(return_value=None) + api.plugin_manager.plugins = {} + api.schema_manager = None + api.plugin_state_manager = None + api.operation_history = None + return client.post("/api/v3/plugins/update", json={"plugin_id": "demo"}) + + def test_it_works_on_the_prefixed_directory( + self, api_v3_client, api_v3_module, prefixed_plugin): + api = api_v3_module.api_v3 + response = self._update(api_v3_client, api) + + assert response.status_code == 200, response.get_json() + # The manifest it reports from is the plugin's, not a missing one. + assert response.get_json()["data"]["last_updated"] == "2026-09-01" + called_with = {c.args[0] for c in api.plugin_store_manager._get_local_git_info.call_args_list} + assert called_with == {prefixed_plugin} + + def test_git_info_is_read_once_before_and_once_after( + self, api_v3_client, api_v3_module, prefixed_plugin): + # A third read existed only to feed a debug log line. (A plain + # plugins_dir/demo, so the old code's existence check let it run.) + (prefixed_plugin.parent / "demo").mkdir() + api = api_v3_module.api_v3 + self._update(api_v3_client, api) + assert api.plugin_store_manager._get_local_git_info.call_count == 2 + + def test_a_plugin_that_is_not_installed_touches_no_path( + self, api_v3_client, api_v3_module, prefixed_plugin): + # The directory is taken from a listing of plugins_dir, so an id + # with nothing by that name installed never becomes a path at all. + api = api_v3_module.api_v3 + api.plugin_store_manager.get_plugin_info = MagicMock(return_value=None) + api.plugin_store_manager.update_plugin = MagicMock(return_value=False) + api.plugin_store_manager._get_local_git_info = MagicMock(return_value=None) + api.plugin_manager.plugins = {} + api.operation_history = None + + response = api_v3_client.post("/api/v3/plugins/update", json={"plugin_id": "ghost"}) + + assert response.status_code >= 400 + assert "not found" in response.get_json()["message"] + api.plugin_store_manager._get_local_git_info.assert_not_called() + api.plugin_store_manager.update_plugin.assert_called_once_with("ghost") + + +@pytest.fixture +def pages(tmp_path, monkeypatch): + from web_interface.blueprints import pages_v3 as module + + plugins_dir = tmp_path / "plugin-repos" + plugins_dir.mkdir() + plugin_manager = MagicMock() + plugin_manager.plugins_dir = plugins_dir + plugin_manager.get_plugin.return_value = None + monkeypatch.setattr(module.pages_v3, "plugin_manager", plugin_manager, raising=False) + monkeypatch.setattr(module.pages_v3, "config_manager", + MagicMock(load_config=lambda: {}), raising=False) + monkeypatch.setattr(module.pages_v3, "schema_manager", None, raising=False) + monkeypatch.setattr(module.pages_v3, "plugin_store_manager", MagicMock(), raising=False) + app = Flask(__name__, template_folder=str( + Path(module.__file__).resolve().parents[1] / "templates")) + app.config["TESTING"] = True + app.register_blueprint(module.pages_v3) + return module, plugins_dir, plugin_manager, app.test_client() + + +class TestPluginPages: + def test_web_ui_is_served_from_the_directory_discovery_found(self, pages): + # A directory name that is neither nor ledmatrix-: only the + # plugin manager's discovery map knows it holds 'radar'. + _, plugins_dir, plugin_manager, client = pages + web_ui = plugins_dir / "Radar-Checkout" / "web_ui" + web_ui.mkdir(parents=True) + (web_ui / "panel.html").write_text("

radar panel

", encoding="utf-8") + plugin_manager.get_plugin_directory.side_effect = ( + lambda pid: str(plugins_dir / "Radar-Checkout") if pid == "radar" else None) + + response = client.get("/plugin-ui/radar/web-ui/panel.html") + + assert response.status_code == 200 + assert "radar panel" in response.get_data(as_text=True) + + def test_the_config_form_reads_the_prefixed_directorys_schema(self, pages): + _, plugins_dir, plugin_manager, client = pages + plugin_dir = plugins_dir / "ledmatrix-weather" + plugin_dir.mkdir() + (plugin_dir / "config_schema.json").write_text(json.dumps({ + "type": "object", + "properties": {"enabled": {"type": "boolean"}, + "units": {"type": "string", "title": "Turn It On"}}, + }), encoding="utf-8") + (plugin_dir / "manifest.json").write_text( + json.dumps({"id": "weather", "name": "Weather"}), encoding="utf-8") + plugin_manager.get_plugin_info.return_value = {"id": "weather", "name": "Weather"} + plugin_manager.get_plugin_directory.side_effect = _resolver(plugins_dir) + + response = client.get("/partials/plugin-config/weather") + + # plugins_dir/weather has no schema, which rendered as a 500 + # "schema unavailable". + assert response.status_code == 200, response.get_data(as_text=True)[:300] + assert "Turn It On" in response.get_data(as_text=True) + + +class TestPartialsDoNoUnusedWork: + def test_the_plugins_tab_reads_no_plugin_data(self, pages): + # plugins.html takes no plugin list; plugins_manager.js fetches it. + _, _, plugin_manager, client = pages + response = client.get("/partials/plugins") + assert response.status_code == 200 + plugin_manager.get_all_plugin_info.assert_not_called() diff --git a/test/test_wifi_ap_state_handling.py b/test/test_wifi_ap_state_handling.py new file mode 100644 index 00000000..b608d029 --- /dev/null +++ b/test/test_wifi_ap_state_handling.py @@ -0,0 +1,198 @@ +"""WiFi / AP-mode fixes around the web UI and the monitor daemon. + +* disconnect_from_network ran an AP-mode check that could never enable the + AP: the web routes build a fresh WiFiManager per request, whose grace + counter starts at 0 and needs 3 consecutive checks. It only added sleeps. +* The monitor daemon read wifi_config.json once at startup, so the web UI's + auto-enable toggle (which only writes the file) had no effect until the + daemon restarted. +* The captive-portal endpoints only checked hostapd, but enable_ap_mode falls + back to an nmcli AP; with that AP up, phones got "internet works". +* The LED status file path was a module global set by whichever WiFiManager + was built first, and the config path fell back to /home/ledpi/LEDMatrix. +""" + +import importlib.util +import json +import logging +import os +import pathlib +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import src.wifi_manager as wm # noqa: E402 +from src.wifi_manager import WiFiManager, WiFiStatus # noqa: E402 + +REPO = Path(__file__).resolve().parents[1] + + +def _ok(stdout=""): + return SimpleNamespace(returncode=0, stdout=stdout, stderr="") + + +def _manager(config_path): + with patch("src.wifi_manager.subprocess.run", return_value=_ok("wlan0\n")): + manager = WiFiManager(config_path=config_path) + manager._wifi_interface = "wlan0" + manager.has_nmcli = True + return manager + + +class TestDisconnect: + def test_no_ap_check_and_no_extra_sleep(self, tmp_path): + manager = _manager(tmp_path / "config" / "wifi_config.json") + manager.get_wifi_status = MagicMock( + return_value=WiFiStatus(connected=True, ssid="home")) + manager._find_profile_for_ssid = MagicMock(return_value=None) + manager.check_and_manage_ap_mode = MagicMock() + sleeps = [] + with patch("src.wifi_manager.subprocess.run", return_value=_ok()), \ + patch("src.wifi_manager.time.sleep", side_effect=sleeps.append): + ok, _message = manager.disconnect_from_network() + assert ok + manager.check_and_manage_ap_mode.assert_not_called() + assert sum(sleeps) == 2 + + +class TestLedStatusFile: + def test_each_manager_writes_next_to_its_own_config(self, tmp_path, monkeypatch): + monkeypatch.setattr(wm, "LED_STATUS_FILE", None) + first = _manager(tmp_path / "a" / "wifi_config.json") + second = _manager(tmp_path / "b" / "wifi_config.json") + second._show_led_message("hello", duration=3) + assert json.loads((tmp_path / "b" / "wifi_status.json").read_text())["message"] == "hello" + assert not (tmp_path / "a" / "wifi_status.json").exists() + first._show_led_message("from a") + assert (tmp_path / "a" / "wifi_status.json").exists() + + def test_the_default_manager_writes_where_the_display_reads(self, tmp_path): + manager = _manager(tmp_path / "config" / "wifi_config.json") + assert manager._led_status_file == tmp_path / "config" / "wifi_status.json" + # And for the default config, that is get_wifi_status_path(). + assert wm.get_wifi_config_path().parent / "wifi_status.json" == wm.get_wifi_status_path() + + +class TestConfigPathFallback: + def test_without_a_config_dir_the_project_root_is_used(self, monkeypatch): + monkeypatch.delenv("LEDMATRIX_ROOT", raising=False) + # A fresh checkout has no config/ yet; the fallback was a hardcoded + # /home/ledpi/LEDMatrix, which is nobody's checkout on most installs. + monkeypatch.setattr(pathlib.Path, "exists", lambda self: False) + assert wm.get_wifi_config_path() == REPO / "config" / "wifi_config.json" + + +def _load_daemon(): + """Import scripts/utils/wifi_monitor_daemon.py without its /var/log + FileHandler, which cannot be opened off the Pi.""" + path = REPO / "scripts" / "utils" / "wifi_monitor_daemon.py" + spec = importlib.util.spec_from_file_location("wifi_monitor_daemon_under_test", path) + module = importlib.util.module_from_spec(spec) + with patch.object(logging, "FileHandler", lambda *a, **k: logging.NullHandler()): + spec.loader.exec_module(module) + return module + + +class TestDaemonReloadsConfig: + @pytest.fixture + def daemon(self, tmp_path): + module = _load_daemon() + config_path = tmp_path / "wifi_config.json" + config_path.write_text(json.dumps({"auto_enable_ap_mode": True})) + + manager = SimpleNamespace(config_path=config_path, + config={"auto_enable_ap_mode": True}) + manager._load_config = MagicMock(side_effect=lambda: manager.config.update( + json.loads(config_path.read_text()))) + daemon = module.WiFiMonitorDaemon.__new__(module.WiFiMonitorDaemon) + daemon.wifi_manager = manager + daemon._config_mtime = daemon._config_file_mtime() + return daemon, manager, config_path + + def _rewrite(self, config_path, data): + before = config_path.stat().st_mtime_ns + config_path.write_text(json.dumps(data)) + os.utime(config_path, ns=(before + 10**9, before + 10**9)) + + def test_a_changed_file_is_reread(self, daemon): + daemon, manager, config_path = daemon + self._rewrite(config_path, {"auto_enable_ap_mode": False}) + daemon._reload_config_if_changed() + assert manager.config["auto_enable_ap_mode"] is False + + def test_an_unchanged_file_is_not(self, daemon): + daemon, manager, _ = daemon + daemon._reload_config_if_changed() + daemon._reload_config_if_changed() + manager._load_config.assert_not_called() + + def test_the_loop_rereads_before_each_check(self, tmp_path): + module = _load_daemon() + daemon = module.WiFiMonitorDaemon.__new__(module.WiFiMonitorDaemon) + daemon.check_interval = 0 + daemon.running = True + daemon.last_state = None + daemon._consecutive_internet_failures = 0 + daemon._nm_restart_threshold = 5 + order = [] + manager = MagicMock() + manager.config = {} + manager.get_wifi_status.return_value = WiFiStatus(connected=False) + manager._is_ethernet_connected.return_value = False + manager.check_and_manage_ap_mode_with_state.side_effect = lambda: ( + order.append("check"), (False, WiFiStatus(connected=False), False, False))[1] + daemon.wifi_manager = manager + daemon._reload_config_if_changed = lambda: order.append("reload") + + def stop(_seconds): + daemon.running = False + with patch.object(module.time, "sleep", side_effect=stop): + daemon.run() + assert order == ["reload", "check"] + + +class TestCaptivePortalSeesTheNmcliAp: + @pytest.fixture + def web_app(self, monkeypatch): + import web_interface.app as web_app + from web_interface.cache import TTLCache + monkeypatch.setattr(web_app, "_service_status_cache", TTLCache()) + monkeypatch.setattr(web_app, "_SYSTEMCTL", "/bin/systemctl") + monkeypatch.setattr(web_app, "_NMCLI", "/usr/bin/nmcli") + return web_app + + def _run(self, active_connections): + calls = [] + + def run(argv, **kwargs): + calls.append(argv) + if "is-active" in argv: + return _ok("inactive\n") + return _ok(active_connections) + return calls, run + + def test_an_nmcli_ap_counts_as_ap_mode(self, web_app, monkeypatch): + calls, run = self._run("LEDMatrix-Setup-AP:802-11-wireless\n") + monkeypatch.setattr(web_app.subprocess, "run", run) + assert web_app.is_ap_mode_active() is True + # Cached: these endpoints are hit per request. + web_app.is_ap_mode_active() + assert len(calls) == 2 + + def test_an_ordinary_wifi_connection_does_not(self, web_app, monkeypatch): + _, run = self._run("home:802-11-wireless\nWired connection 1:802-3-ethernet\n") + monkeypatch.setattr(web_app.subprocess, "run", run) + assert web_app.is_ap_mode_active() is False + + def test_the_detection_endpoints_redirect(self, web_app, monkeypatch): + _, run = self._run("LEDMatrix-Setup-AP:802-11-wireless\n") + monkeypatch.setattr(web_app.subprocess, "run", run) + web_app.app.config["TESTING"] = True + with web_app.app.test_client() as client: + response = client.get("/generate_204") + assert response.status_code == 302 diff --git a/test/web_interface/test_api_v3_config_raw.py b/test/web_interface/test_api_v3_config_raw.py index bceebd13..ff9e0bbb 100644 --- a/test/web_interface/test_api_v3_config_raw.py +++ b/test/web_interface/test_api_v3_config_raw.py @@ -178,6 +178,25 @@ class TestSaveRawSecrets: monkeypatch.setattr(env.config_manager, "save_raw_file_content", boom) assert env.client.post(SECRETS, json={"a": 1}).status_code == 500 + @pytest.mark.parametrize("error,code", [ + (ConfigError("cannot write", config_path="/etc/s.json"), "CONFIG_SAVE_FAILED"), + (RuntimeError("nope"), "UNKNOWN_ERROR"), + ]) + def test_errors_answer_in_the_main_routes_shape(self, env, monkeypatch, error, code): + # Both raw routes build their 500 with one helper now; this one used + # to hand-roll a body without error_code or context. raw_json.html + # reads only `message`, which both shapes carry. + def boom(kind, data): + raise error + monkeypatch.setattr(env.config_manager, "save_raw_file_content", boom) + secrets = env.client.post(SECRETS, json={"a": 1}) + main = env.client.post(MAIN, json={"a": 1}) + assert secrets.status_code == main.status_code == 500 + body = secrets.get_json() + assert body["error_code"] == code + assert body["message"] == main.get_json()["message"] + assert set(body) == set(main.get_json()) + class TestRawEndpointsBypassSecretSeparation: """Pinned behaviour, deliberately not "fixed". diff --git a/test/web_interface/test_api_v3_unhandled_errors.py b/test/web_interface/test_api_v3_unhandled_errors.py index 5bd14150..6b90c944 100644 --- a/test/web_interface/test_api_v3_unhandled_errors.py +++ b/test/web_interface/test_api_v3_unhandled_errors.py @@ -122,8 +122,22 @@ def test_credentials_are_redacted_from_the_detail(client): assert body["details"].startswith("RuntimeError: forced failure") -def test_a_client_error_keeps_its_own_status(client): +def _raise_415(): + raise UnsupportedMediaType( + "Did not attempt to load JSON data because the request Content-Type " + "was not 'application/json'.") + + +# An api_v3 route raising a 415 from inside. No route does that on its own +# any more (the asset delete route, which used to, now reads its body with +# get_json(silent=True)), so one route's view is swapped for one that does; +# the endpoint stays api_v3's, so its error handler is the one that answers. +_SWAPPED_ENDPOINT = "api_v3.delete_plugin_asset" + + +def test_a_client_error_keeps_its_own_status(client, monkeypatch): """HTTPExceptions subclass Exception; a 415 must not become a 500.""" + monkeypatch.setitem(client.application.view_functions, _SWAPPED_ENDPOINT, _raise_415) resp = client.post("/api/v3/plugins/assets/delete", data="not json", content_type="text/plain") assert resp.status_code == 415 @@ -151,8 +165,9 @@ class TestInTheRealApp: assert resp.get_json() == EXPECTED def test_client_errors_read_the_same_as_the_global_handler( - self, web_app, exploding_managers): + self, web_app, exploding_managers, monkeypatch): """The blueprint's 4xx shape must not drift from app.py's.""" + monkeypatch.setitem(web_app.app.view_functions, _SWAPPED_ENDPOINT, _raise_415) resp = web_app.app.test_client().post( "/api/v3/plugins/assets/delete", data="not json", content_type="text/plain") diff --git a/test/web_interface/test_update_all_plugins.py b/test/web_interface/test_update_all_plugins.py index 899092ad..39386948 100644 --- a/test/web_interface/test_update_all_plugins.py +++ b/test/web_interface/test_update_all_plugins.py @@ -126,7 +126,6 @@ class TestUpdateRouteReportsNoOps: self._install(tmp_path, 'clock', '1.0.0') store._get_local_git_info.side_effect = [ {'sha': 'aaaaaaa000', 'branch': 'main'}, # before - {'sha': 'aaaaaaa000', 'branch': 'main'}, # is-git check {'sha': 'bbbbbbb111', 'branch': 'main'}, # after ] body = client.post('/api/v3/plugins/update', json={'plugin_id': 'clock'}).get_json() diff --git a/test/web_interface/test_web_backend_small_fixes.py b/test/web_interface/test_web_backend_small_fixes.py new file mode 100644 index 00000000..178b424e --- /dev/null +++ b/test/web_interface/test_web_backend_small_fixes.py @@ -0,0 +1,144 @@ +"""Small web-backend fixes: BDF font preview, the SSE broadcaster restart +window, start.py's werkzeug log filter, and the backup routes' errors.""" + +import base64 +import io +import shutil +import sys +import threading +from pathlib import Path +from unittest.mock import patch + +import pytest +from PIL import Image + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from test._api_v3_test_helpers import api_v3_client, api_v3_module # noqa: F401,E402 + +REPO = Path(__file__).resolve().parents[2] + + +class TestBdfFontPreview: + """The route refused every BDF font ("needs complex rendering") although + src/common/bdf_font.py draws them for the panel.""" + + @pytest.fixture + def fonts_root(self, tmp_path, api_v3_module, monkeypatch): + import web_interface.blueprints.api_v3.fonts as fonts_module + fonts_dir = tmp_path / "assets" / "fonts" + fonts_dir.mkdir(parents=True) + shutil.copy(REPO / "assets" / "fonts" / "6x10.bdf", fonts_dir / "6x10.bdf") + monkeypatch.setattr(fonts_module, "PROJECT_ROOT", tmp_path) + return fonts_dir + + def _preview(self, client, **params): + query = {"font": "6x10.bdf", "text": "Hi", "size": 10, + "bg": "000000", "fg": "ffffff"} + query.update(params) + return client.get("/api/v3/fonts/preview", query_string=query) + + def test_a_bdf_font_renders(self, api_v3_client, fonts_root): + response = self._preview(api_v3_client) + assert response.status_code == 200, response.get_json() + data = response.get_json()["data"] + assert data["image"].startswith("data:image/png;base64,") + image = Image.open(io.BytesIO(base64.b64decode(data["image"].split(",", 1)[1]))) + assert image.size == (data["width"], data["height"]) + colors = {c for _, c in image.convert("RGB").getcolors(10**6)} + assert (255, 255, 255) in colors + + def test_the_glyphs_are_the_panels(self, api_v3_client, fonts_root): + # The same rasterizer as the panel: exactly the pixels draw_bdf_text + # lights for this string, nothing anti-aliased. + from PIL import ImageDraw + from src.common.bdf_font import draw_bdf_text, load_bdf_face + data = self._preview(api_v3_client, text="A").get_json()["data"] + preview = Image.open(io.BytesIO(base64.b64decode(data["image"].split(",", 1)[1]))).convert("RGB") + + face, _ = load_bdf_face(str(fonts_root / "6x10.bdf"), 10) + glyph = Image.new("RGB", (20, 20)) + draw_bdf_text(ImageDraw.Draw(glyph), "A", 0, 0, face, color=(255, 255, 255)) + lit = sum(1 for p in glyph.getdata() if p == (255, 255, 255)) + assert lit > 0 + assert sum(1 for p in preview.getdata() if p == (255, 255, 255)) == lit + assert set(preview.getdata()) <= {(0, 0, 0), (255, 255, 255)} + + def test_multi_line_text_is_taller(self, api_v3_client, fonts_root): + one = self._preview(api_v3_client, text="Hi", size=10).get_json()["data"] + # Tall enough to exceed the 30px minimum. + three = self._preview(api_v3_client, text="a\nb\nc", size=10).get_json()["data"] + assert three["height"] > one["height"] + + +class TestStreamBroadcasterRestart: + """Between the broadcast thread's break and its exit, is_alive() was + still True, so a client subscribing in that window got no thread.""" + + def test_a_subscriber_during_shutdown_gets_a_new_thread(self): + import web_interface.app as web_app + + closing = threading.Event() + release = threading.Event() + produced = threading.Event() + started = [] + + def factory(): + started.append(threading.current_thread()) + + def gen(): + try: + while True: + produced.set() + yield {"n": 1} + finally: + # Closing the generator after the break: hold the thread + # alive here, which is the window in question. + closing.set() + release.wait(5) + return gen() + + broadcaster = web_app._StreamBroadcaster(factory) + first = broadcaster.subscribe() + assert produced.wait(5) + broadcaster.unsubscribe(first) + assert closing.wait(5), "the broadcast thread never noticed it had no clients" + try: + second = broadcaster.subscribe() + assert second.get(timeout=5) == {"n": 1} + assert len(started) == 2 + finally: + release.set() + broadcaster.unsubscribe(second) + + +class TestStartLogFilterExcInfo: + """logging takes exc_info as a tuple, an exception, or True; the filter + unpacked it as a 3-tuple, so the other two raised TypeError.""" + + def test_every_form_yields_the_exception(self): + from web_interface.start import _exc_info_value + err = BrokenPipeError(32, "Broken pipe") + assert _exc_info_value(err) is err + assert _exc_info_value((type(err), err, None)) is err + try: + raise err + except BrokenPipeError: + assert _exc_info_value(True) is err + + def test_true_outside_an_except_is_none(self): + from web_interface.start import _exc_info_value + assert _exc_info_value(True) is None + + +class TestBackupErrors: + """The backup routes' own catch-alls are gone; the blueprint handler + answers with the fields backup_restore.html reads.""" + + def test_a_failed_export_is_a_500_with_a_message(self, api_v3_client): + with patch("src.backup_manager.create_backup", side_effect=OSError(28, "No space left")): + response = api_v3_client.post("/api/v3/backup/export") + assert response.status_code == 500 + body = response.get_json() + assert body["status"] == "error" + assert body["message"] diff --git a/web_interface/app.py b/web_interface/app.py index 46266e31..14037180 100644 --- a/web_interface/app.py +++ b/web_interface/app.py @@ -44,6 +44,7 @@ from src.plugin_system.operation_history import OperationHistory _JOURNALCTL = shutil.which('journalctl') _SYSTEMCTL = shutil.which('systemctl') +_NMCLI = shutil.which('nmcli') _VCGENCMD = shutil.which('vcgencmd') from web_interface import display_preview @@ -288,6 +289,7 @@ except ImportError: # (AP mode) or per SSE tick (display service). A failed check keeps the last # known answer for the same TTL rather than retrying on every request. from web_interface.cache import TTLCache +from src.wifi_manager import AP_PROFILE_NAME _service_status_cache = TTLCache() _AP_MODE_CACHE_TTL = 30 # seconds — AP mode is user-initiated; 30s is fine _LEDMATRIX_SERVICE_CACHE_TTL = 15 # seconds @@ -319,12 +321,49 @@ def _unit_is_active(unit, ttl): _service_status_cache.set(unit, active, ttl=ttl) return active +def _nmcli_ap_is_active(ttl): + """Whether NetworkManager has our access-point profile up, cached for + ``ttl`` seconds. + + WiFiManager.enable_ap_mode falls back to an nmcli AP when hostapd is not + available, and there is no hostapd unit running then. The match is + WiFiManager._get_ap_status_nmcli's: our profile name, or a connection of + a hotspot type. False where there is no nmcli; on a failed check, the + last known answer. + """ + active = _service_status_cache.get('nmcli-ap') + if active is not None: + return active + active = _service_status_cache.peek('nmcli-ap', False) + if _NMCLI: + try: + result = subprocess.run( # nosec B603 - fixed argv # nosemgrep + [_NMCLI, '-t', '-f', 'NAME,TYPE', 'connection', 'show', '--active'], + capture_output=True, text=True, timeout=2) + active = False + for line in result.stdout.splitlines(): + parts = line.split(':') + if len(parts) < 2: + continue + if parts[0].strip() == AP_PROFILE_NAME or 'hotspot' in parts[1].strip().lower(): + active = True + break + except (subprocess.SubprocessError, OSError) as e: + logging.getLogger('web_interface').warning( + "nmcli active-connection check failed: %s", e) + _service_status_cache.set('nmcli-ap', active, ttl=ttl) + return active + def is_ap_mode_active(): """ Check if access point mode is currently active (cached, 30s TTL). - Uses a direct systemctl check instead of instantiating WiFiManager. + Uses direct systemctl/nmcli checks instead of instantiating WiFiManager, + and the same two WiFiManager._is_ap_mode_active uses: hostapd, else the + nmcli AP it falls back to. Checking hostapd alone meant the captive + portal never triggered on a Pi whose AP came up through nmcli. """ - return _unit_is_active('hostapd', _AP_MODE_CACHE_TTL) + return (_unit_is_active('hostapd', _AP_MODE_CACHE_TTL) + or _nmcli_ap_is_active(_AP_MODE_CACHE_TTL)) # Captive portal detection endpoints # When AP mode is active, return responses that TRIGGER the captive portal popup. @@ -377,7 +416,6 @@ def not_found_error(error): @app.errorhandler(500) def internal_error(error): """Handle 500 errors.""" - import logging logger = logging.getLogger('web_interface') logger.error("Internal server error", exc_info=True) payload = { @@ -416,7 +454,6 @@ def handle_exception(error): 'message': error.description, }), error.code or 500 - import logging logger = logging.getLogger('web_interface') logger.error("Unhandled exception", exc_info=True) return jsonify({ @@ -620,6 +657,11 @@ class _StreamBroadcaster: if not self._clients: # No subscribers — exit so the thread doesn't spin indefinitely. # subscribe() will restart it when a new client arrives. + # Drop the handle here, under the lock: between this break + # and the thread actually ending (closing the generator + # can take a while) is_alive() is still True, and a + # client subscribing then got no thread at all. + self._thread = None break for q in self._clients: try: @@ -870,8 +912,7 @@ def favicon(): return '', 204 _reconciliation_started = False -import threading as _threading -_reconciliation_lock = _threading.Lock() +_reconciliation_lock = threading.Lock() def _run_startup_reconciliation() -> None: """Run state reconciliation in background to auto-repair missing plugins. @@ -911,7 +952,7 @@ def _run_startup_reconciliation() -> None: # Write status file so the web UI can surface unresolved issues as a # banner without the user having to read journalctl. Mirrors the # hw_status pattern (/tmp/led_matrix_hw_status.json). - import json as _json, tempfile as _tempfile, os as _os + import tempfile _recon_status = { "done": True, "successful": result.reconciliation_successful, @@ -925,21 +966,21 @@ def _run_startup_reconciliation() -> None: for inc in result.inconsistencies_manual ], } - _recon_path = _os.path.join(_tempfile.gettempdir(), "ledmatrix_reconciliation.json") + _recon_path = os.path.join(tempfile.gettempdir(), "ledmatrix_reconciliation.json") _tmp = None try: - if not _os.path.islink(_recon_path): - _fd, _tmp = _tempfile.mkstemp(dir=_tempfile.gettempdir(), prefix=".led_recon_") - with _os.fdopen(_fd, "w") as _f: - _json.dump(_recon_status, _f) - _os.replace(_tmp, _recon_path) + if not os.path.islink(_recon_path): + _fd, _tmp = tempfile.mkstemp(dir=tempfile.gettempdir(), prefix=".led_recon_") + with os.fdopen(_fd, "w") as _f: + json.dump(_recon_status, _f) + os.replace(_tmp, _recon_path) _tmp = None # Rename succeeded; nothing to clean up except (OSError, ValueError, TypeError) as _e: _logger.warning("[Reconciliation] Could not write status file: %s", _e) finally: - if _tmp is not None and _os.path.exists(_tmp): + if _tmp is not None and os.path.exists(_tmp): try: - _os.unlink(_tmp) + os.unlink(_tmp) except OSError: pass except Exception as e: @@ -953,7 +994,7 @@ def start_startup_reconciliation(): with _reconciliation_lock: if not _reconciliation_started: _reconciliation_started = True - _threading.Thread(target=_run_startup_reconciliation, daemon=True).start() + threading.Thread(target=_run_startup_reconciliation, daemon=True).start() _auto_updater = None @@ -983,10 +1024,9 @@ def start_auto_update_scheduler(): if __name__ == '__main__': - import os as _os start_auto_update_scheduler() # threaded=True is Flask's default since 1.0 but stated explicitly so that # long-lived /api/v3/stream/* SSE connections don't starve other requests. # Debug mode is off by default; opt in with FLASK_DEBUG=1 in the environment. - _debug = _os.environ.get('FLASK_DEBUG', '0') == '1' + _debug = os.environ.get('FLASK_DEBUG', '0') == '1' app.run(host='0.0.0.0', port=5000, debug=_debug, threaded=True) # nosec B104 - intentional; local network device diff --git a/web_interface/blueprints/api_v3/__init__.py b/web_interface/blueprints/api_v3/__init__.py index abf0397e..80b109af 100644 --- a/web_interface/blueprints/api_v3/__init__.py +++ b/web_interface/blueprints/api_v3/__init__.py @@ -170,9 +170,16 @@ def _get_plugin_version(plugin_id: str) -> str: that arrived in a request body, so the name is validated here rather than relying on each of them to have done it. """ - manifest_path = resolve_under( - api_v3.plugin_store_manager.plugins_dir, plugin_id, "manifest.json" - ) + # The resolver first: a plugin installed as ledmatrix-, or whose + # directory is named differently from its manifest id, is not at + # plugins_dir/, and recording '' as its version hid that it worked. + plugin_dir = _plugin_directory(plugin_id) + if plugin_dir is not None: + manifest_path = plugin_dir / "manifest.json" + else: + manifest_path = resolve_under( + api_v3.plugin_store_manager.plugins_dir, plugin_id, "manifest.json" + ) if manifest_path is None: logger.warning("[PluginVersion] Rejected unsafe plugin id %r", plugin_id) return '' @@ -1420,9 +1427,11 @@ def _plugin_directory(plugin_id: str) -> Optional[Path]: is no fallback to the legacy plugins/ directory: the loader never scans it, so a plugin found only there is one that never runs. """ - if not api_v3.plugin_manager: + # getattr: the blueprint only has plugin_manager once the app has set it. + manager = getattr(api_v3, 'plugin_manager', None) + if not manager: return None - plugin_dir = api_v3.plugin_manager.get_plugin_directory(plugin_id) + plugin_dir = manager.get_plugin_directory(plugin_id) if not plugin_dir or not Path(plugin_dir).exists(): return None return Path(plugin_dir) diff --git a/web_interface/blueprints/api_v3/backup.py b/web_interface/blueprints/api_v3/backup.py index edc53865..2184861b 100644 --- a/web_interface/blueprints/api_v3/backup.py +++ b/web_interface/blueprints/api_v3/backup.py @@ -2,6 +2,10 @@ Routes decorate the shared `api_v3` Blueprint from the package `__init__`, so their endpoint names are unchanged by living here. + +No route here catches Exception: the blueprint's handler in `__init__` +logs it and answers 500 with the same `status`/`message` fields the +backup page reads (backup_restore.html), plus `details`. """ from web_interface.blueprints.api_v3 import ( PROJECT_ROOT, Path, _coerce_to_bool, _safe_backup_path, api_v3, @@ -17,78 +21,62 @@ import web_interface.blueprints.api_v3 as _pkg @api_v3.route('/backup/preview', methods=['GET']) def backup_preview(): """Return a summary of what a new backup would include.""" - try: - from src.backup_manager import preview_backup_contents - data = preview_backup_contents(PROJECT_ROOT) - return jsonify({'status': 'success', 'data': data}) - except Exception as e: - logger.error("backup_preview failed: %s", e, exc_info=True) - return jsonify({'status': 'error', 'message': 'An internal error occurred; see logs for details'}), 500 + from src.backup_manager import preview_backup_contents + data = preview_backup_contents(PROJECT_ROOT) + return jsonify({'status': 'success', 'data': data}) @api_v3.route('/backup/list', methods=['GET']) def backup_list(): """List backup ZIPs stored in the export directory.""" - try: - _pkg._BACKUP_EXPORT_DIR.mkdir(parents=True, exist_ok=True) - entries = [] - for p in sorted(_pkg._BACKUP_EXPORT_DIR.iterdir(), key=lambda x: x.stat().st_mtime, reverse=True): - if not p.is_file() or p.suffix != '.zip': - continue - st = p.stat() - entries.append({ - 'filename': p.name, - 'size': st.st_size, - 'created_at': datetime.fromtimestamp(st.st_mtime).strftime('%Y-%m-%d %H:%M:%S'), - }) - return jsonify({'status': 'success', 'data': entries}) - except Exception as e: - logger.error("backup_list failed: %s", e, exc_info=True) - return jsonify({'status': 'error', 'message': 'An internal error occurred; see logs for details'}), 500 + _pkg._BACKUP_EXPORT_DIR.mkdir(parents=True, exist_ok=True) + entries = [] + for p in sorted(_pkg._BACKUP_EXPORT_DIR.iterdir(), key=lambda x: x.stat().st_mtime, reverse=True): + if not p.is_file() or p.suffix != '.zip': + continue + st = p.stat() + entries.append({ + 'filename': p.name, + 'size': st.st_size, + 'created_at': datetime.fromtimestamp(st.st_mtime).strftime('%Y-%m-%d %H:%M:%S'), + }) + return jsonify({'status': 'success', 'data': entries}) @api_v3.route('/backup/export', methods=['POST']) def backup_export(): """Create a new backup ZIP and return its filename.""" - try: - from src.backup_manager import create_backup - zip_path = create_backup(PROJECT_ROOT, output_dir=_pkg._BACKUP_EXPORT_DIR) - return jsonify({'status': 'success', 'filename': zip_path.name}) - except Exception as e: - logger.error("backup_export failed: %s", e, exc_info=True) - return jsonify({'status': 'error', 'message': 'An internal error occurred; see logs for details'}), 500 + from src.backup_manager import create_backup + zip_path = create_backup(PROJECT_ROOT, output_dir=_pkg._BACKUP_EXPORT_DIR) + return jsonify({'status': 'success', 'filename': zip_path.name}) @api_v3.route('/backup/validate', methods=['POST']) def backup_validate(): """Validate an uploaded backup ZIP and return its manifest.""" + from src.backup_manager import validate_backup + if 'backup_file' not in request.files: + return jsonify({'status': 'error', 'message': 'No backup_file in request'}), 400 + f = request.files['backup_file'] + with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp: + tmp_path = tmp.name + f.save(tmp_path) try: - from src.backup_manager import validate_backup - if 'backup_file' not in request.files: - return jsonify({'status': 'error', 'message': 'No backup_file in request'}), 400 - f = request.files['backup_file'] - with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp: - tmp_path = tmp.name - f.save(tmp_path) + ok, err_msg, manifest = validate_backup(Path(tmp_path)) + finally: try: - ok, err_msg, manifest = validate_backup(Path(tmp_path)) - finally: - try: - os.unlink(tmp_path) - except OSError: - pass - if not ok: - logger.warning("Backup validation failed: %s", err_msg) - return jsonify({'status': 'error', 'message': 'Invalid or corrupted backup file'}), 400 - safe_manifest = { - 'schema_version': manifest.get('schema_version'), - 'created_at': manifest.get('created_at'), - 'ledmatrix_version': manifest.get('ledmatrix_version'), - 'hostname': manifest.get('hostname'), - 'contents': manifest.get('contents', []), - 'detected_contents': manifest.get('detected_contents', []), - 'plugins': manifest.get('plugins', []), - 'total_uncompressed': manifest.get('total_uncompressed'), - 'file_count': manifest.get('file_count'), - } - return jsonify({'status': 'success', 'data': safe_manifest}) - except Exception as e: - logger.error("backup_validate failed: %s", e, exc_info=True) - return jsonify({'status': 'error', 'message': 'An internal error occurred; see logs for details'}), 500 + os.unlink(tmp_path) + except OSError: + pass + if not ok: + logger.warning("Backup validation failed: %s", err_msg) + return jsonify({'status': 'error', 'message': 'Invalid or corrupted backup file'}), 400 + safe_manifest = { + 'schema_version': manifest.get('schema_version'), + 'created_at': manifest.get('created_at'), + 'ledmatrix_version': manifest.get('ledmatrix_version'), + 'hostname': manifest.get('hostname'), + 'contents': manifest.get('contents', []), + 'detected_contents': manifest.get('detected_contents', []), + 'plugins': manifest.get('plugins', []), + 'total_uncompressed': manifest.get('total_uncompressed'), + 'file_count': manifest.get('file_count'), + } + return jsonify({'status': 'success', 'data': safe_manifest}) #: The only keys RestoreOptions recognizes. A typo'd or renamed key (e.g. #: "restoreSecrets") would otherwise be silently ignored by opts_dict.get(), #: leaving that flag at its True default -- restoring secrets a caller's @@ -100,105 +88,101 @@ _RESTORE_OPTION_KEYS = frozenset(( @api_v3.route('/backup/restore', methods=['POST']) def backup_restore(): """Restore a backup ZIP with optional RestoreOptions.""" + from src.backup_manager import restore_backup, RestoreOptions + if 'backup_file' not in request.files: + return jsonify({'status': 'error', 'message': 'No backup_file in request'}), 400 + f = request.files['backup_file'] + options_raw = request.form.get('options', '{}') try: - from src.backup_manager import restore_backup, RestoreOptions - if 'backup_file' not in request.files: - return jsonify({'status': 'error', 'message': 'No backup_file in request'}), 400 - f = request.files['backup_file'] - options_raw = request.form.get('options', '{}') + opts_dict = json.loads(options_raw) + except json.JSONDecodeError: + 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 + unknown_keys = set(opts_dict) - _RESTORE_OPTION_KEYS + if unknown_keys: + return jsonify({ + 'status': 'error', + 'message': f'Unknown restore option(s): {", ".join(sorted(unknown_keys))}', + }), 400 + # _coerce_to_bool (not bare bool()) because a request can send these + # as JSON strings: bool("false") is True in Python, so a caller who + # explicitly asked to skip secrets would have had them restored + # anyway. + options = RestoreOptions( + restore_config=_coerce_to_bool(opts_dict.get('restore_config', True)), + restore_secrets=_coerce_to_bool(opts_dict.get('restore_secrets', True)), + restore_wifi=_coerce_to_bool(opts_dict.get('restore_wifi', True)), + restore_fonts=_coerce_to_bool(opts_dict.get('restore_fonts', True)), + restore_plugin_uploads=_coerce_to_bool(opts_dict.get('restore_plugin_uploads', True)), + reinstall_plugins=_coerce_to_bool(opts_dict.get('reinstall_plugins', True)), + ) + with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp: + tmp_path = tmp.name + f.save(tmp_path) + try: + result = restore_backup(Path(tmp_path), PROJECT_ROOT, options) + finally: try: - opts_dict = json.loads(options_raw) - except json.JSONDecodeError: - 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 - unknown_keys = set(opts_dict) - _RESTORE_OPTION_KEYS - if unknown_keys: - return jsonify({ - 'status': 'error', - 'message': f'Unknown restore option(s): {", ".join(sorted(unknown_keys))}', - }), 400 - # _coerce_to_bool (not bare bool()) because a request can send these - # as JSON strings: bool("false") is True in Python, so a caller who - # explicitly asked to skip secrets would have had them restored - # anyway. - options = RestoreOptions( - restore_config=_coerce_to_bool(opts_dict.get('restore_config', True)), - restore_secrets=_coerce_to_bool(opts_dict.get('restore_secrets', True)), - restore_wifi=_coerce_to_bool(opts_dict.get('restore_wifi', True)), - restore_fonts=_coerce_to_bool(opts_dict.get('restore_fonts', True)), - restore_plugin_uploads=_coerce_to_bool(opts_dict.get('restore_plugin_uploads', True)), - reinstall_plugins=_coerce_to_bool(opts_dict.get('reinstall_plugins', True)), - ) - with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp: - tmp_path = tmp.name - f.save(tmp_path) - try: - result = restore_backup(Path(tmp_path), PROJECT_ROOT, options) - finally: + os.unlink(tmp_path) + except OSError: + pass + + # Reinstall plugins if requested and store manager available + if options.reinstall_plugins and result.plugins_to_install: + psm = getattr(api_v3, 'plugin_store_manager', None) + for plug in result.plugins_to_install: + pid = plug.get('plugin_id') + if not pid: + continue try: - os.unlink(tmp_path) - except OSError: - pass - - # Reinstall plugins if requested and store manager available - if options.reinstall_plugins and result.plugins_to_install: - psm = getattr(api_v3, 'plugin_store_manager', None) - for plug in result.plugins_to_install: - pid = plug.get('plugin_id') - if not pid: - continue - try: - if psm and hasattr(psm, 'install_plugin'): - ok = psm.install_plugin(pid) - if ok: - result.plugins_installed.append(pid) - else: - result.plugins_failed.append({'plugin_id': pid, 'error': 'install_plugin returned False'}) + if psm and hasattr(psm, 'install_plugin'): + ok = psm.install_plugin(pid) + if ok: + result.plugins_installed.append(pid) else: - result.plugins_failed.append({'plugin_id': pid, 'error': 'Store manager unavailable'}) - except Exception as pe: - logger.error( - "[Backup] Failed to reinstall plugin %r: %s", pid, pe, exc_info=True - ) - result.plugins_failed.append({'plugin_id': pid, 'error': 'Installation failed; see server logs'}) + result.plugins_failed.append({'plugin_id': pid, 'error': 'install_plugin returned False'}) + else: + result.plugins_failed.append({'plugin_id': pid, 'error': 'Store manager unavailable'}) + except Exception as pe: + logger.error( + "[Backup] Failed to reinstall plugin %r: %s", pid, pe, exc_info=True + ) + result.plugins_failed.append({'plugin_id': pid, 'error': 'Installation failed; see server logs'}) - # A restore that dropped files can still report success if the only - # failures were plugin reinstalls, since those don't touch result.errors. - if result.plugins_failed: - result.success = False + # A restore that dropped files can still report success if the only + # failures were plugin reinstalls, since those don't touch result.errors. + if result.plugins_failed: + result.success = False - data = result.to_dict() - if not result.success: - # Name what failed, and what nonetheless landed. A restore is - # partial far more often than it is total -- a fresh install can - # leave config_secrets.json unwritable by the web service, so - # config restores and secrets do not. "Restore had errors" alone - # left the user unable to tell a wholly failed restore from one - # that quietly dropped their API keys. - failed_plugins = [ - str(p.get('plugin_id')) for p in (result.plugins_failed or []) if p.get('plugin_id') - ] - parts = [] - if result.restored: - parts.append(f"restored: {', '.join(result.restored)}") - if result.errors: - parts.append(f"failed: {'; '.join(result.errors)}") - if failed_plugins: - parts.append(f"plugins not reinstalled: {', '.join(failed_plugins)}") - message = 'Restore incomplete — ' + ('. '.join(parts) if parts else 'see logs') - return jsonify({'status': 'error', 'message': message, 'data': data}), 500 - return jsonify({'status': 'success', 'data': data}) - except Exception as e: - logger.error("backup_restore failed: %s", e, exc_info=True) - return jsonify({'status': 'error', 'message': 'An internal error occurred; see logs for details'}), 500 + data = result.to_dict() + if not result.success: + # Name what failed, and what nonetheless landed. A restore is + # partial far more often than it is total -- a fresh install can + # leave config_secrets.json unwritable by the web service, so + # config restores and secrets do not. "Restore had errors" alone + # left the user unable to tell a wholly failed restore from one + # that quietly dropped their API keys. + failed_plugins = [ + str(p.get('plugin_id')) for p in (result.plugins_failed or []) if p.get('plugin_id') + ] + parts = [] + if result.restored: + parts.append(f"restored: {', '.join(result.restored)}") + if result.errors: + parts.append(f"failed: {'; '.join(result.errors)}") + if failed_plugins: + parts.append(f"plugins not reinstalled: {', '.join(failed_plugins)}") + message = 'Restore incomplete — ' + ('. '.join(parts) if parts else 'see logs') + return jsonify({'status': 'error', 'message': message, 'data': data}), 500 + return jsonify({'status': 'success', 'data': data}) @api_v3.route('/backup/download/', methods=['GET']) def backup_download(filename): """Stream a backup ZIP to the browser.""" diff --git a/web_interface/blueprints/api_v3/config.py b/web_interface/blueprints/api_v3/config.py index bdd97207..7904dbb4 100644 --- a/web_interface/blueprints/api_v3/config.py +++ b/web_interface/blueprints/api_v3/config.py @@ -1150,6 +1150,27 @@ def get_secrets_config(): # alone so a client can still tell "set" from "not set". return jsonify({'status': 'success', 'data': mask_all_secret_values(config)}) +def _raw_config_save_error(e): + """The 500 both raw-config save routes answer a failed save with. + + A ConfigError that names its file says which one; the rest is the same + for every failure. raw_json.html reads only ``message``, which this + keeps, alongside ``status`` and ``details`` as before. + """ + from src.exceptions import ConfigError + error_message = 'An error occurred; see logs for details' + config_path = getattr(e, 'config_path', None) if isinstance(e, ConfigError) else None + if config_path: + error_message = f"{error_message} (config_path: {config_path})" + return error_response( + ErrorCode.CONFIG_SAVE_FAILED if isinstance(e, ConfigError) else ErrorCode.UNKNOWN_ERROR, + error_message, + details=describe_exception(e), + context={'config_path': config_path} if config_path else None, + status_code=500 + ) + + @api_v3.route('/config/raw/main', methods=['POST']) def save_raw_main_config(): """Save raw main configuration JSON""" @@ -1193,31 +1214,8 @@ def save_raw_main_config(): logger.warning("Automatic update setup could not be started", exc_info=True) return jsonify({'status': 'success', 'message': message}) except Exception as e: - from src.exceptions import ConfigError logger.error("Error saving raw main config", exc_info=True) - - # Extract more specific error message if it's a ConfigError - if isinstance(e, ConfigError): - error_message = 'An error occurred; see logs for details' - if hasattr(e, 'config_path') and e.config_path: - error_message = f"{error_message} (config_path: {e.config_path})" - return error_response( - ErrorCode.CONFIG_SAVE_FAILED, - error_message, - details=describe_exception(e), - - context={'config_path': e.config_path} if hasattr(e, 'config_path') and e.config_path else None, - status_code=500 - ) - else: - error_message = 'An error occurred; see logs for details' - return error_response( - ErrorCode.UNKNOWN_ERROR, - error_message, - details=describe_exception(e), - - status_code=500 - ) + return _raw_config_save_error(e) @api_v3.route('/config/raw/secrets', methods=['POST']) def save_raw_secrets_config(): """Save raw secrets configuration JSON""" @@ -1257,17 +1255,5 @@ def save_raw_secrets_config(): return jsonify({'status': 'success', 'message': 'Secrets configuration saved successfully'}) except Exception as e: - from src.exceptions import ConfigError logger.error("Error saving raw secrets config", exc_info=True) - - # Extract more specific error message if it's a ConfigError - if isinstance(e, ConfigError): - # ConfigError has a message attribute and may have context - error_message = 'An error occurred; see logs for details' - if hasattr(e, 'config_path') and e.config_path: - error_message = f"{error_message} (config_path: {e.config_path})" - else: - error_message = 'An error occurred; see logs for details' - - return jsonify({'status': 'error', 'message': error_message, - 'details': describe_exception(e)}), 500 + return _raw_config_save_error(e) diff --git a/web_interface/blueprints/api_v3/display.py b/web_interface/blueprints/api_v3/display.py index 0d99c0df..3de25a2f 100644 --- a/web_interface/blueprints/api_v3/display.py +++ b/web_interface/blueprints/api_v3/display.py @@ -4,7 +4,7 @@ Routes decorate the shared `api_v3` Blueprint from the package `__init__`, so their endpoint names are unchanged by living here. """ from web_interface.blueprints.api_v3 import ( - _ensure_display_service_running, + _coerce_to_bool, _ensure_display_service_running, _get_display_service_status, _stop_display_service, api_v3, jsonify, logger, request, uuid, ) @@ -151,8 +151,10 @@ def start_on_demand_display(): plugin_id = data.get('plugin_id') mode = data.get('mode') duration = data.get('duration') - pinned = bool(data.get('pinned', False)) - start_service = data.get('start_service', True) + # _coerce_to_bool: bool("false") is True, so a string "false" pinned the + # mode or (re)started the service it asked to leave alone. + pinned = _coerce_to_bool(data.get('pinned', False)) + start_service = _coerce_to_bool(data.get('start_service', True)) if not plugin_id and not mode: return jsonify({'status': 'error', 'message': 'plugin_id or mode is required'}), 400 diff --git a/web_interface/blueprints/api_v3/fonts.py b/web_interface/blueprints/api_v3/fonts.py index 2eb28a8a..7585bb49 100644 --- a/web_interface/blueprints/api_v3/fonts.py +++ b/web_interface/blueprints/api_v3/fonts.py @@ -309,13 +309,30 @@ def get_font_preview() -> tuple[Response, int] | Response: # Load font font = None - if str(font_path).endswith('.bdf'): - # BDF fonts require complex per-glyph rendering via freetype - # Return explicit error rather than showing misleading preview with default font - return jsonify({ - 'status': 'error', - 'message': 'BDF font preview not supported. BDF fonts will render correctly on the LED matrix.' - }), 400 + bdf_face = None + lines = text.split('\n') + line_height = 0 + if font_path.suffix.lower() == '.bdf': + # PIL cannot draw a BDF strike, so it goes through the same loader + # and glyph rasterizer the panel uses (src/common/bdf_font.py). The + # face comes back at the file's native size when it has no strike at + # the one asked for -- which is also what the panel would draw. + from src.common.bdf_font import load_bdf_face, draw_bdf_text + try: + bdf_face, realised_px = load_bdf_face(str(font_path), size) + except Exception as e: + logger.warning("[FontPreview] Failed to load BDF font %s: %s", font_path, e) + return jsonify({'status': 'error', 'message': 'Could not load BDF font'}), 400 + line_height = (bdf_face.size.height >> 6) or realised_px + line_widths = [] + for line in lines: + width = 0 + for char in line: + bdf_face.load_char(char) + width += bdf_face.glyph.advance.x >> 6 + line_widths.append(width) + text_width = max(line_widths) + text_height = line_height * len(lines) else: # TTF/OTF fonts try: @@ -325,12 +342,12 @@ def get_font_preview() -> tuple[Response, int] | Response: logger.warning("[FontPreview] Failed to load font %s: %s", font_path, e) font = ImageFont.load_default() - # Calculate text size - temp_img = Image.new('RGB', (1, 1)) - temp_draw = ImageDraw.Draw(temp_img) - bbox = temp_draw.textbbox((0, 0), text, font=font) - text_width = bbox[2] - bbox[0] - text_height = bbox[3] - bbox[1] + # Calculate text size + temp_img = Image.new('RGB', (1, 1)) + temp_draw = ImageDraw.Draw(temp_img) + bbox = temp_draw.textbbox((0, 0), text, font=font) + text_width = bbox[2] - bbox[0] + text_height = bbox[3] - bbox[1] # Create image with padding padding = 10 @@ -350,7 +367,11 @@ def get_font_preview() -> tuple[Response, int] | Response: x = (img_width - text_width) // 2 y = (img_height - text_height) // 2 - draw.text((x, y), text, font=font, fill=fg_rgb) + if bdf_face is not None: + for index, line in enumerate(lines): + draw_bdf_text(draw, line, x, y + index * line_height, bdf_face, color=fg_rgb) + else: + draw.text((x, y), text, font=font, fill=fg_rgb) # Convert to base64 buffer = io.BytesIO() diff --git a/web_interface/blueprints/api_v3/plugins.py b/web_interface/blueprints/api_v3/plugins.py index 310a783f..c8ee1d49 100644 --- a/web_interface/blueprints/api_v3/plugins.py +++ b/web_interface/blueprints/api_v3/plugins.py @@ -26,7 +26,9 @@ from src.common.path_safety import ( ) from src.web_interface.config_arrays import coerce_array_shapes from src.web_interface.validators import dedup_unique_arrays +from src.config_manager_atomic import atomic_write_text import web_interface.blueprints.api_v3 as _pkg +from typing import Optional # Read through the module rather than bound by value: tests patch these # as module attributes, and a value binding would not see the patch. # Several are also called from helpers that live in __init__, so the @@ -70,9 +72,12 @@ def get_installed_plugins(): plugin_state = state_info.get('state') plugin_error_info = state_info.get('error_info') - # Re-read manifest from disk to ensure we have the latest metadata - manifest_path = Path(api_v3.plugin_manager.plugins_dir) / plugin_id / "manifest.json" - if manifest_path.exists(): + # Re-read manifest from disk to ensure we have the latest metadata. + # Through the resolver, not plugins_dir/: a plugin installed as + # ledmatrix- otherwise never had its manifest refreshed here. + plugin_path = _plugin_directory(plugin_id) + manifest_path = plugin_path / "manifest.json" if plugin_path else None + if manifest_path is not None and manifest_path.exists(): try: with open(manifest_path, 'r', encoding='utf-8') as f: fresh_manifest = json.load(f) @@ -103,8 +108,7 @@ def get_installed_plugins(): update_available = _is_plugin_update_available(installed_version, latest_version) # Local git info (single subprocess on cache miss, zero on hit) - plugin_path = Path(api_v3.plugin_manager.plugins_dir) / plugin_id - local_git_info = api_v3.plugin_store_manager._get_local_git_info(plugin_path) if plugin_path.exists() else None + local_git_info = api_v3.plugin_store_manager._get_local_git_info(plugin_path) if plugin_path else None if local_git_info: sha = local_git_info.get('sha', '') @@ -411,7 +415,10 @@ def toggle_plugin(): 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'] - enabled = data['enabled'] + # Coerced, not stored raw: "false" is a truthy string, and this + # value is written to config.json and handed to the Starlark + # toggle, so {"enabled": "false"} used to switch a plugin ON. + enabled = _coerce_to_bool(data['enabled']) else: # Form data or query string (HTMX submission) plugin_id = request.args.get('plugin_id') or request.form.get('plugin_id') @@ -863,6 +870,22 @@ def get_plugin_config(): return success_response(data=plugin_config) except Exception as e: return exception_error_response(e, ErrorCode.CONFIG_LOAD_FAILED) +def _listed_plugin_dir(base: Path, name: str) -> Optional[Path]: + """The entry of ``base`` called ``name``, or None. + + The path returned comes from listing ``base``, not from joining ``name`` + onto it, so a caller that validated ``name`` doesn't have to rely on that + validation alone: nothing reaches the filesystem unless it's already there. + """ + try: + for entry in base.iterdir(): + if entry.name == name: + return entry + except OSError: + pass + return None + + @api_v3.route('/plugins/update', methods=['POST']) def update_plugin(): """Update plugin""" @@ -922,22 +945,31 @@ def update_plugin(): # Always do direct updates (they're fast git pull operations) # Operation queue is reserved for longer operations like install/uninstall - plugins_base = Path(api_v3.plugin_store_manager.plugins_dir) - plugin_dir = plugins_base / plugin_id - manifest_path = resolve_under(plugin_dir, "manifest.json") - if manifest_path is None: - return error_response( - ErrorCode.INVALID_INPUT, - 'Invalid plugin_id', - status_code=400 - ) + # The resolver finds a plugin installed as ledmatrix-. Either way + # the directory used is taken from a listing of plugins_dir, matched by + # name, never built from the request value -- so no path here depends + # on user input. None means nothing by that name is installed; the + # store manager still gets the id and reports that itself. + resolved = _plugin_directory(plugin_id) + plugin_dir = _listed_plugin_dir( + Path(api_v3.plugin_store_manager.plugins_dir), + resolved.name if resolved else plugin_id) + manifest_path = None + if plugin_dir is not None: + manifest_path = resolve_under(plugin_dir, "manifest.json") + if manifest_path is None: + return error_response( + ErrorCode.INVALID_INPUT, + 'Invalid plugin_id', + status_code=400 + ) current_last_updated = None current_version = None current_commit = None current_branch = None - if manifest_path.exists(): + if manifest_path is not None and manifest_path.exists(): try: with open(manifest_path, 'r', encoding='utf-8') as f: manifest = json.load(f) @@ -958,22 +990,12 @@ def update_plugin(): except Exception as e: logger.debug("Could not read local manifest for plugin: %s", e) - if api_v3.plugin_store_manager: - git_info_before = api_v3.plugin_store_manager._get_local_git_info(plugin_dir) - if git_info_before: - current_commit = git_info_before.get('sha') - current_branch = git_info_before.get('branch') - - # Check if plugin is a git repo first (for better error messages) - # plugin_id is validated above; reuse the same directory rather - # than rebuilding it from a value that might not match. - plugin_path_dir = plugin_dir - is_git_repo = False - if plugin_path_dir.exists(): - git_info = api_v3.plugin_store_manager._get_local_git_info(plugin_path_dir) - is_git_repo = git_info is not None - if is_git_repo: - logger.debug("Plugin is a git repository, will update via git pull") + git_info_before = (api_v3.plugin_store_manager._get_local_git_info(plugin_dir) + if plugin_dir is not None else None) + if git_info_before: + current_commit = git_info_before.get('sha') + current_branch = git_info_before.get('branch') + logger.debug("Plugin is a git repository, will update via git pull") remote_info = api_v3.plugin_store_manager.get_plugin_info(plugin_id, fetch_latest_from_github=True) remote_commit = remote_info.get('last_commit_sha') if remote_info else None @@ -986,7 +1008,7 @@ def update_plugin(): updated_last_updated = current_last_updated updated_version = current_version try: - if manifest_path.exists(): + if manifest_path is not None and manifest_path.exists(): with open(manifest_path, 'r', encoding='utf-8') as f: manifest = json.load(f) updated_last_updated = manifest.get('last_updated', current_last_updated) @@ -996,11 +1018,11 @@ def update_plugin(): updated_commit = None updated_branch = remote_branch or current_branch - if api_v3.plugin_store_manager: - git_info_after = api_v3.plugin_store_manager._get_local_git_info(plugin_dir) - if git_info_after: - updated_commit = git_info_after.get('sha') - updated_branch = git_info_after.get('branch') or updated_branch + git_info_after = (api_v3.plugin_store_manager._get_local_git_info(plugin_dir) + if plugin_dir is not None else None) + if git_info_after: + updated_commit = git_info_after.get('sha') + updated_branch = git_info_after.get('branch') or updated_branch # update_plugin() answers True for "nothing to do" as well as for # a real update (a ZIP-installed monorepo plugin already at the @@ -1069,11 +1091,10 @@ def update_plugin(): message=message ) else: - plugin_path_dir = plugin_dir - if not plugin_path_dir.exists(): + if plugin_dir is None or not plugin_dir.exists(): client_msg = 'Plugin update failed: plugin not found' else: - git_info = api_v3.plugin_store_manager._get_local_git_info(plugin_path_dir) + git_info = api_v3.plugin_store_manager._get_local_git_info(plugin_dir) if not git_info: plugin_info = api_v3.plugin_store_manager.get_plugin_info(plugin_id) if not plugin_info: @@ -2759,6 +2780,16 @@ def _plugin_uploads_dir(plugin_id): return resolve_under(PROJECT_ROOT / 'assets' / 'plugins', plugin_id, 'uploads') +def _write_upload_metadata(metadata_file, metadata): + """Replace an uploads directory's .metadata.json atomically. + + The plugin config page reads this file to list a plugin's images; one cut + off mid-write (a power loss on a Pi) failed to parse, and every upload it + recorded dropped out of the list while the files stayed on disk. + """ + atomic_write_text(metadata_file, json.dumps(metadata, indent=2)) + + @api_v3.route('/plugins/assets/upload', methods=['POST']) def upload_plugin_asset(): """Upload asset files for a plugin""" @@ -2804,6 +2835,11 @@ def upload_plugin_asset(): if 'size' in entry: total_size += entry.get('size', 0) + # Every file is checked before any is saved. Checking and saving in one + # loop meant a bad third file answered 400 after the first two were + # already written -- on disk and in the metadata the UI lists, though + # the user was told the upload failed. + accepted = [] for file in files: if not file.filename: continue @@ -2853,6 +2889,10 @@ def upload_plugin_asset(): 'message': f'File {file.filename} is not a valid image file' }), 400 + total_size += file_size + accepted.append((file, file_ext, file_size, file_content)) + + for file, file_ext, file_size, file_content in accepted: # Generate unique filename timestamp = int(_pkg.time.time()) file_hash = hashlib.md5(file_content + file.filename.encode()).hexdigest()[:8] @@ -2894,11 +2934,7 @@ def upload_plugin_asset(): 'uploaded_at': metadata[image_id]['uploaded_at'] }) - total_size += file_size - - # Save metadata - with open(metadata_file, 'w') as f: - json.dump(metadata, f, indent=2) + _write_upload_metadata(metadata_file, metadata) return jsonify({ 'status': 'success', @@ -3018,16 +3054,20 @@ def upload_calendar_credentials(): shutil.copy2(credentials_path, backup_path) _prune_credential_backups(Path(plugin_dir)) - # Save new file - file.save(str(credentials_path)) - - # Set proper permissions - os.chmod(credentials_path, 0o600) # Read/write for owner only + # Save new file: atomically, and created 0o600 (read/write for owner + # only) rather than chmod-ed after. file.save() truncated the old file + # first, so a failure mid-write left the plugin with a broken + # credentials.json, and the secret sat world-readable until the chmod. + atomic_write_text(credentials_path, + file_content.decode(json.detect_encoding(file_content)), + mode=0o600) return jsonify({ 'status': 'success', 'message': 'Credentials file uploaded successfully', - 'path': str(credentials_path) + # Relative to the plugin: nothing reads this field, and the server's + # absolute layout is not the client's business. + 'path': credentials_path.name }) @api_v3.route('/plugins/calendar/authenticate', methods=['POST']) @@ -3170,7 +3210,12 @@ def list_calendar_calendars(): @api_v3.route('/plugins/assets/delete', methods=['POST']) def delete_plugin_asset(): """Delete an asset file for a plugin""" - data = request.get_json() + # silent=True: without it a missing or non-JSON body raised inside + # get_json() and came back as a 415 in the generic error shape, or, for + # a JSON array, an AttributeError 500. + data = request.get_json(silent=True) + if not isinstance(data, dict): + return jsonify({'status': 'error', 'message': 'plugin_id and image_id are required'}), 400 plugin_id = data.get('plugin_id') image_id = data.get('image_id') @@ -3209,9 +3254,7 @@ def delete_plugin_asset(): # Remove from metadata del metadata[image_id] - # Save metadata - with open(metadata_file, 'w') as f: - json.dump(metadata, f, indent=2) + _write_upload_metadata(metadata_file, metadata) return jsonify({'status': 'success', 'message': 'Image deleted successfully'}) diff --git a/web_interface/blueprints/api_v3/wifi.py b/web_interface/blueprints/api_v3/wifi.py index f504a8a8..c38f0462 100644 --- a/web_interface/blueprints/api_v3/wifi.py +++ b/web_interface/blueprints/api_v3/wifi.py @@ -305,8 +305,9 @@ def enable_ap_mode(): from src.wifi_manager import WiFiManager wifi_manager = WiFiManager() - _force_raw = (request.get_json(silent=True) or {}).get('force', False) - force = _force_raw is True or (isinstance(_force_raw, str) and _force_raw.lower() in ('true', '1')) + # The same parsing as every other boolean on these routes. Anything it + # does not recognise (None) is not a request to force. + force = _parse_bool_ish((request.get_json(silent=True) or {}).get('force', False)) is True success, message = wifi_manager.enable_ap_mode(force=force) if success: diff --git a/web_interface/blueprints/pages_v3.py b/web_interface/blueprints/pages_v3.py index 3f55bc4c..e12c40ed 100644 --- a/web_interface/blueprints/pages_v3.py +++ b/web_interface/blueprints/pages_v3.py @@ -320,13 +320,31 @@ def serve_plugin_web_ui(plugin_id, filename): return 'Error serving file', 500, {'Content-Type': 'text/plain'} +def _resolved_plugin_dir(plugin_id): + """The plugin manager's answer for where ``plugin_id`` lives, or None. + + Its discovery map is authoritative (a plugin whose directory name is not + its id is found there), and it refuses anything that is not one plain + path segment. See src/plugin_system/plugin_dirs.py. + """ + found = pages_v3.plugin_manager.get_plugin_directory(plugin_id) + if isinstance(found, (str, Path)) and Path(found).exists(): + return Path(found) + return None + + def _plugin_dir_for(safe_id): """A sanitised plugin id's directory, which may not exist. - Contained under the configured plugins directory, with PluginManager's - ``ledmatrix-`` prefix fallback. Raises ValueError for an id that would - leave it; the routes answer that with a 403. + The plugin manager's resolver first; otherwise contained under the + configured plugins directory, with PluginManager's ``ledmatrix-`` prefix + fallback. Raises ValueError for an id that would leave it; the routes + answer that with a 403. """ + resolved = _resolved_plugin_dir(safe_id) + if resolved is not None: + return resolved + plugins_base = Path(pages_v3.plugin_manager.plugins_dir).resolve() plugin_dir = resolve_under(plugins_base, safe_id) if plugin_dir is None: @@ -556,91 +574,15 @@ def _load_schedule_partial(): def _load_plugins_partial(): """Load plugins management partial""" - # Load plugin data from the plugin system - plugins_data = [] - - # Get installed plugins if managers are available - if pages_v3.plugin_manager and pages_v3.plugin_store_manager: - try: - # Get all installed plugin info - all_plugin_info = pages_v3.plugin_manager.get_all_plugin_info() - - # Load config once before the loop (not per-plugin) - full_config = pages_v3.config_manager.load_config() if pages_v3.config_manager else {} - - # Format for the web interface - for plugin_info in all_plugin_info: - plugin_id = plugin_info.get('id') - - # Re-read manifest from disk to ensure we have the latest metadata - manifest_path = Path(pages_v3.plugin_manager.plugins_dir) / plugin_id / "manifest.json" - if manifest_path.exists(): - try: - with open(manifest_path, 'r', encoding='utf-8') as f: - fresh_manifest = json.load(f) - # Update plugin_info with fresh manifest data - plugin_info.update(fresh_manifest) - except Exception: - # If we can't read the fresh manifest, use the cached one - logger.warning("Could not read fresh manifest for plugin: %s", plugin_id) - - # Get enabled status from config (source of truth) - # Read from config file first, fall back to plugin instance if config doesn't have the key - enabled = None - if pages_v3.config_manager: - plugin_config = full_config.get(plugin_id, {}) - # Check if 'enabled' key exists in config (even if False) - if 'enabled' in plugin_config: - enabled = bool(plugin_config['enabled']) - - # Fallback to plugin instance if config doesn't have enabled key - if enabled is None: - plugin_instance = pages_v3.plugin_manager.get_plugin(plugin_id) - if plugin_instance: - enabled = plugin_instance.enabled - else: - # Default to True if no config key and plugin not loaded (matches BasePlugin default) - enabled = True - - # Get verified status from store registry (no GitHub API calls needed) - store_info = pages_v3.plugin_store_manager.get_registry_info(plugin_id) - verified = store_info.get('verified', False) if store_info else False - - last_updated = plugin_info.get('last_updated') - last_commit = plugin_info.get('last_commit') or plugin_info.get('last_commit_sha') - branch = plugin_info.get('branch') - - if store_info: - last_updated = last_updated or store_info.get('last_updated') or store_info.get('last_updated_iso') - last_commit = last_commit or store_info.get('last_commit') or store_info.get('last_commit_sha') - branch = branch or store_info.get('branch') or store_info.get('default_branch') - - plugins_data.append({ - 'id': plugin_id, - 'name': plugin_info.get('name', plugin_id), - 'author': plugin_info.get('author', 'Unknown'), - 'category': plugin_info.get('category', 'General'), - 'description': plugin_info.get('description', 'No description available'), - 'tags': plugin_info.get('tags', []), - 'enabled': enabled, - 'verified': verified, - 'loaded': plugin_info.get('loaded', False), - 'last_updated': last_updated, - 'last_commit': last_commit, - 'branch': branch - }) - except Exception: - logger.error("Error loading plugin data", exc_info=True) - - return render_template('v3/partials/plugins.html', - plugins=plugins_data) + # plugins.html takes no plugin data: plugins_manager.js fetches the list + # from /api/v3/plugins/installed. Building it here read every manifest on + # disk on each Plugins-tab load for nothing. + return render_template('v3/partials/plugins.html') def _load_fonts_partial(): """Load fonts management partial""" - # This would load font data from the font system - fonts_data = {} # Placeholder for font data - return render_template('v3/partials/fonts.html', - fonts=fonts_data) + # The page fetches its font data from /api/v3/fonts/* itself. + return render_template('v3/partials/fonts.html') def _load_logs_partial(): """Load logs viewer partial""" @@ -749,6 +691,11 @@ def _load_plugin_config_partial(plugin_id): if not plugin_info: return '
Plugin not found
', 404 + # The containment check above only validates the id. The files are + # read from wherever the plugin manager says the plugin lives, which + # for one installed as ledmatrix- is not plugins_dir/. + _plugin_dir = _resolved_plugin_dir(plugin_id) or _plugin_dir + # Get plugin instance (may be None if not loaded) plugin_instance = pages_v3.plugin_manager.get_plugin(plugin_id) diff --git a/web_interface/start.py b/web_interface/start.py index 7b37de78..4d2e41a5 100644 --- a/web_interface/start.py +++ b/web_interface/start.py @@ -17,6 +17,21 @@ logger = logging.getLogger('web_interface.start') _CLIENT_DISCONNECT_ERRNOS = (113, 32, 104) +def _exc_info_value(exc_info): + """The exception a logging call's ``exc_info`` argument refers to. + + logging accepts a (type, value, traceback) tuple, an exception instance, + or True for the exception being handled. The werkzeug filter below used + to unpack it as a tuple, so True or an instance raised TypeError from + inside the logging call itself. + """ + if isinstance(exc_info, BaseException): + return exc_info + if isinstance(exc_info, tuple): + return exc_info[1] if len(exc_info) > 1 else None + return sys.exc_info()[1] + + def get_local_ips(): """Get list of local IP addresses the service will be accessible on.""" ips = [] @@ -94,7 +109,7 @@ def main(): return # For exceptions, check if it's a socket error if 'exc_info' in kwargs and kwargs['exc_info']: - exc_type, exc_value, exc_tb = kwargs['exc_info'] + exc_value = _exc_info_value(kwargs['exc_info']) if isinstance(exc_value, OSError): if exc_value.errno in _CLIENT_DISCONNECT_ERRNOS: werkzeug_logger.debug(message, *args, **kwargs) diff --git a/web_interface/templates/v3/partials/fonts.html b/web_interface/templates/v3/partials/fonts.html index 3a6df195..c559162a 100644 --- a/web_interface/templates/v3/partials/fonts.html +++ b/web_interface/templates/v3/partials/fonts.html @@ -586,14 +586,6 @@ async function updateFontPreview() { return; } - // BDF bitmap fonts cannot be rendered server-side — skip the API call - if (family.toLowerCase().endsWith('.bdf')) { - previewImage.style.display = 'none'; - loadingText.style.display = 'block'; - loadingText.textContent = 'Preview not available for BDF bitmap fonts'; - return; - } - // Show loading state loadingText.textContent = 'Loading preview...'; loadingText.style.display = 'block';