diff --git a/test/_api_v3_test_helpers.py b/test/_api_v3_test_helpers.py new file mode 100644 index 00000000..9bcb7984 --- /dev/null +++ b/test/_api_v3_test_helpers.py @@ -0,0 +1,75 @@ +""" +Shared scaffolding for api_v3 blueprint tests. + +Not a test module (the leading underscore keeps pytest from collecting +it). It is the pytest-fixture equivalent of ``_make_client()`` in +test_uninstall_and_reconcile_endpoint.py, which is unittest-style and +requires ``self.addCleanup``. + +The api_v3 blueprint keeps its managers as attributes on a module-level +singleton, not in Flask app state, so replacing them with mocks leaks +into every later test that imports api_v3 unless the originals are put +back. ``api_v3_client`` snapshots and restores them around each test. +""" + +from unittest.mock import MagicMock + +import pytest +from flask import Flask + + +# Every manager attribute the blueprint reads. Anything missing here keeps +# whatever a previously-run test left on the singleton. +API_V3_MANAGER_ATTRS = ( + 'config_manager', 'plugin_manager', 'plugin_store_manager', + 'plugin_state_manager', 'saved_repositories_manager', 'schema_manager', + 'operation_queue', 'operation_history', 'cache_manager', +) + +_SENTINEL = object() + + +def build_app(blueprint): + app = Flask(__name__) + app.config['TESTING'] = True + app.config['SECRET_KEY'] = 'test' + app.register_blueprint(blueprint, url_prefix='/api/v3') + return app + + +@pytest.fixture +def api_v3_module(): + """The api_v3 module with every manager replaced by a MagicMock. + + Restores the original attributes afterwards. Tests point individual + managers at real objects (a ConfigManager over tmp_path, say) or set + them to None to exercise the not-initialized branches. + """ + from web_interface.blueprints import api_v3 as module + + originals = { + name: getattr(module.api_v3, name, _SENTINEL) + for name in API_V3_MANAGER_ATTRS + } + for name in API_V3_MANAGER_ATTRS: + setattr(module.api_v3, name, MagicMock()) + # Default to the direct path; queue tests opt in explicitly. + module.api_v3.operation_queue = None + + yield module + + for name, original in originals.items(): + if original is _SENTINEL: + if hasattr(module.api_v3, name): + try: + delattr(module.api_v3, name) + except AttributeError: + pass + else: + setattr(module.api_v3, name, original) + + +@pytest.fixture +def api_v3_client(api_v3_module): + """Flask test client wired to the mocked blueprint.""" + return build_app(api_v3_module.api_v3).test_client() diff --git a/test/test_api_v3_optional_body.py b/test/test_api_v3_optional_body.py new file mode 100644 index 00000000..c03db5a8 --- /dev/null +++ b/test/test_api_v3_optional_body.py @@ -0,0 +1,90 @@ +""" +Regression tests: POST endpoints whose body is optional must accept a +request that has no body at all. + +Six handlers in api_v3 read their body as ``request.get_json() or {}``. +The ``or {}`` states the intent plainly — every field is optional, so a +bodyless POST should fall back to defaults. But ``get_json()`` without +``silent=True`` raises ``UnsupportedMediaType`` when the request carries +no JSON Content-Type, and it raises *before* ``or {}`` is evaluated. Each +handler's catch-all then turned that into a 500. + +So the natural way to call these endpoints — a POST with no body, which +is what curl, a fetch() without options, and most HTTP clients send by +default — failed on every one of them. The shipped UI always sends a JSON +object, which is why this went unnoticed. + +This file covers the endpoints whose bodyless behaviour is not already +tested in their own suite. +""" + +import sys +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from test._api_v3_test_helpers import api_v3_client, api_v3_module # noqa: F401,E402 + + +class TestOnDemandStart: + URL = "/api/v3/display/on-demand/start" + + def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module): + response = api_v3_client.post(self.URL) + # The endpoint may still reject the request on its own terms (no + # plugin_id, nothing to display); what it must not do is fail with + # a 500 raised out of body parsing. + assert response.status_code != 500 + + def test_json_body_still_works(self, api_v3_client, api_v3_module): + assert api_v3_client.post(self.URL, json={}).status_code != 500 + + +class TestResetPluginConfig: + URL = "/api/v3/plugins/config/reset" + + def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module): + assert api_v3_client.post(self.URL).status_code != 500 + + def test_json_body_still_works(self, api_v3_client, api_v3_module): + assert api_v3_client.post(self.URL, json={}).status_code != 500 + + +class TestDeleteOfTheDayJson: + URL = "/api/v3/plugins/of-the-day/json/delete" + + def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module): + assert api_v3_client.post(self.URL).status_code != 500 + + def test_json_body_still_works(self, api_v3_client, api_v3_module): + assert api_v3_client.post(self.URL, json={}).status_code != 500 + + +class TestPluginLimits: + URL = "/api/v3/plugins/clock/limits" + + def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module): + assert api_v3_client.post(self.URL).status_code != 500 + + +class TestNoToleratedBodyReadIsUnguarded: + def test_every_or_default_body_read_uses_silent(self): + """`get_json() or ` is a contradiction without silent=True. + + Writing `or {}` declares the body optional; omitting silent=True + means the call raises before the default can apply. Catch the + combination here rather than waiting for each endpoint to be + exercised by hand. + """ + source = Path(__file__).parent.parent.joinpath( + "web_interface/blueprints/api_v3.py").read_text() + offenders = [ + line.strip() for line in source.splitlines() + if "request.get_json()" in line and " or " in line + ] + assert offenders == [], ( + "these reads declare a default but raise before reaching it; " + f"use get_json(silent=True): {offenders}") diff --git a/test/test_api_v3_registry_endpoints.py b/test/test_api_v3_registry_endpoints.py new file mode 100644 index 00000000..e1968c39 --- /dev/null +++ b/test/test_api_v3_registry_endpoints.py @@ -0,0 +1,174 @@ +""" +Endpoint tests for the plugin-registry routes in api_v3: +POST /plugins/store/refresh and POST /plugins/registry-from-url. + +Both reach out to the network through PluginStoreManager (mocked here) and +had no endpoint-level coverage; registry-from-url in particular takes a +user-supplied URL and hands it straight to the manager. +""" + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from test._api_v3_test_helpers import api_v3_client, api_v3_module # noqa: F401,E402 + + +class TestRefreshPluginStore: + URL = "/api/v3/plugins/store/refresh" + + def test_uninitialized_manager_is_a_500(self, api_v3_client, api_v3_module): + api_v3_module.api_v3.plugin_store_manager = None + response = api_v3_client.post(self.URL, json={}) + assert response.status_code == 500 + assert "not initialized" in response.get_json()["message"] + + def test_success_reports_plugin_count(self, api_v3_client, api_v3_module): + api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = { + "plugins": [{"id": "a"}, {"id": "b"}, {"id": "c"}]} + response = api_v3_client.post(self.URL, json={}) + assert response.status_code == 200 + assert response.get_json()["plugin_count"] == 3 + + def test_forces_a_refresh_rather_than_using_cache(self, api_v3_client, api_v3_module): + manager = api_v3_module.api_v3.plugin_store_manager + manager.fetch_registry.return_value = {"plugins": []} + api_v3_client.post(self.URL, json={}) + manager.fetch_registry.assert_called_once_with(force_refresh=True) + + def test_empty_registry_reports_zero(self, api_v3_client, api_v3_module): + api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {} + response = api_v3_client.post(self.URL, json={}) + assert response.get_json()["plugin_count"] == 0 + + def test_no_body_is_accepted(self, api_v3_client, api_v3_module): + # Regression: `request.get_json() or {}` says a missing body is + # fine, but get_json() raises UnsupportedMediaType before `or {}` + # is reached, so a bodyless POST — the natural way to call a + # refresh endpoint — came back 500. + api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []} + assert api_v3_client.post(self.URL).status_code == 200 + + def test_body_without_json_content_type_is_accepted( + self, api_v3_client, api_v3_module): + api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []} + response = api_v3_client.post(self.URL, data="", content_type="text/plain") + assert response.status_code == 200 + + def test_malformed_json_body_falls_back_to_defaults( + self, api_v3_client, api_v3_module): + api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []} + response = api_v3_client.post( + self.URL, data="{not json", content_type="application/json") + assert response.status_code == 200 + + @pytest.mark.parametrize("key", ["fetch_commit_info", "fetch_latest_versions"]) + def test_either_commit_info_key_extends_the_message( + self, api_v3_client, api_v3_module, key): + # fetch_latest_versions is the older spelling; both must work. + api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []} + response = api_v3_client.post(self.URL, json={key: True}) + assert "commit metadata" in response.get_json()["message"] + + def test_message_stays_plain_without_the_flag(self, api_v3_client, api_v3_module): + api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []} + response = api_v3_client.post(self.URL, json={}) + assert response.get_json()["message"] == "Plugin store refreshed" + + def test_network_failure_is_a_500(self, api_v3_client, api_v3_module): + api_v3_module.api_v3.plugin_store_manager.fetch_registry.side_effect = ( + ConnectionError("github unreachable")) + response = api_v3_client.post(self.URL, json={}) + assert response.status_code == 500 + assert response.get_json()["message"] == "An error occurred; see logs for details" + + def test_failure_body_carries_no_traceback_or_paths( + self, api_v3_client, api_v3_module): + api_v3_module.api_v3.plugin_store_manager.fetch_registry.side_effect = ( + RuntimeError("failed at /home/user/LEDMatrix/src/secret.py line 42")) + body = api_v3_client.post(self.URL, json={}).get_json() + assert "Traceback" not in str(body) + # `details` is describe_exception output: one line, type-named, + # credential-redacted. It may quote the message, but never a stack. + assert body["details"].startswith("RuntimeError:") + assert "\n" not in body["details"] + + +class TestRegistryFromUrl: + URL = "/api/v3/plugins/registry-from-url" + + def test_uninitialized_manager_is_a_500(self, api_v3_client, api_v3_module): + api_v3_module.api_v3.plugin_store_manager = None + response = api_v3_client.post(self.URL, json={"repo_url": "http://x"}) + assert response.status_code == 500 + + def test_missing_repo_url_is_a_400(self, api_v3_client, api_v3_module): + response = api_v3_client.post(self.URL, json={}) + assert response.status_code == 400 + assert "repo_url required" in response.get_json()["message"] + api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.assert_not_called() + + def test_success_returns_the_plugin_list(self, api_v3_client, api_v3_module): + api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = { + "plugins": [{"id": "clock"}]} + response = api_v3_client.post( + self.URL, json={"repo_url": "https://github.com/o/r"}) + assert response.status_code == 200 + body = response.get_json() + assert body["plugins"] == [{"id": "clock"}] + assert body["registry_url"] == "https://github.com/o/r" + + def test_url_is_trimmed_before_use(self, api_v3_client, api_v3_module): + manager = api_v3_module.api_v3.plugin_store_manager + manager.fetch_registry_from_url.return_value = {"plugins": []} + api_v3_client.post(self.URL, json={"repo_url": " https://github.com/o/r "}) + manager.fetch_registry_from_url.assert_called_once_with("https://github.com/o/r") + + def test_registry_without_plugins_key_returns_empty_list( + self, api_v3_client, api_v3_module): + api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = { + "other": 1} + response = api_v3_client.post(self.URL, json={"repo_url": "http://x"}) + assert response.get_json()["plugins"] == [] + + def test_no_registry_found_is_a_400(self, api_v3_client, api_v3_module): + api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = None + response = api_v3_client.post(self.URL, json={"repo_url": "http://x/not-a-registry"}) + assert response.status_code == 400 + assert "Failed to fetch registry" in response.get_json()["message"] + + @pytest.mark.parametrize("url", [ + "not a url", + "javascript:alert(1)", + "file:///etc/passwd", + "http://localhost:8080/admin", + ]) + def test_unusable_urls_fail_cleanly(self, api_v3_client, api_v3_module, url): + # Characterization: the handler performs no URL validation of its + # own — whatever the manager makes of the URL decides the outcome. + # What is pinned here is that a rejected URL produces a clean 400 + # rather than a traceback or a 500. + api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = None + response = api_v3_client.post(self.URL, json={"repo_url": url}) + assert response.status_code == 400 + assert "Traceback" not in str(response.get_json()) + + def test_fetch_exception_is_a_500_without_internals( + self, api_v3_client, api_v3_module): + api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.side_effect = ( + ValueError("parse failed in /srv/app/internal.py")) + response = api_v3_client.post(self.URL, json={"repo_url": "http://x"}) + assert response.status_code == 500 + body = response.get_json() + assert body["message"] == "An error occurred; see logs for details" + assert "Traceback" not in str(body) + + def test_non_string_repo_url_is_a_500_not_a_crash( + self, api_v3_client, api_v3_module): + # .strip() on a non-string raises; the handler's catch-all turns + # that into a 500 rather than propagating. + response = api_v3_client.post(self.URL, json={"repo_url": 12345}) + assert response.status_code == 500 diff --git a/test/test_api_v3_wifi_endpoints.py b/test/test_api_v3_wifi_endpoints.py new file mode 100644 index 00000000..0c713ae2 --- /dev/null +++ b/test/test_api_v3_wifi_endpoints.py @@ -0,0 +1,240 @@ +""" +Endpoint tests for the /wifi/* routes in api_v3. + +These routes drive the host's actual networking — connecting, dropping a +connection, switching the radio off — and had no endpoint-level tests at +all. WiFiManager is mocked throughout; nothing here may touch real +networking. + +Each handler does `from src.wifi_manager import WiFiManager` inside the +function body, so the patch target is the class at its definition site. +""" + +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from test._api_v3_test_helpers import api_v3_client, api_v3_module # noqa: F401,E402 + + +@pytest.fixture +def wifi_manager(): + """Patch WiFiManager where it is defined; yield the instance mock.""" + with patch("src.wifi_manager.WiFiManager") as cls: + instance = MagicMock() + cls.return_value = instance + yield instance + + +class TestConnect: + URL = "/api/v3/wifi/connect" + + def test_success(self, api_v3_client, wifi_manager): + wifi_manager.connect_to_network.return_value = (True, "Connected to HomeNet") + response = api_v3_client.post(self.URL, json={"ssid": "HomeNet", "password": "pw"}) + assert response.status_code == 200 + assert response.get_json()["message"] == "Connected to HomeNet" + wifi_manager.connect_to_network.assert_called_once_with("HomeNet", "pw") + + def test_missing_body_rejected(self, api_v3_client, wifi_manager): + response = api_v3_client.post(self.URL, json={}) + assert response.status_code == 400 + wifi_manager.connect_to_network.assert_not_called() + + def test_missing_ssid_rejected(self, api_v3_client, wifi_manager): + response = api_v3_client.post(self.URL, json={"password": "pw"}) + assert response.status_code == 400 + assert "SSID is required" in response.get_json()["message"] + wifi_manager.connect_to_network.assert_not_called() + + @pytest.mark.parametrize("ssid", ["", " ", "\t"]) + def test_blank_ssid_rejected(self, api_v3_client, wifi_manager, ssid): + response = api_v3_client.post(self.URL, json={"ssid": ssid}) + assert response.status_code == 400 + wifi_manager.connect_to_network.assert_not_called() + + def test_ssid_is_trimmed(self, api_v3_client, wifi_manager): + wifi_manager.connect_to_network.return_value = (True, "ok") + api_v3_client.post(self.URL, json={"ssid": " HomeNet "}) + wifi_manager.connect_to_network.assert_called_once_with("HomeNet", "") + + def test_missing_password_becomes_empty_string(self, api_v3_client, wifi_manager): + wifi_manager.connect_to_network.return_value = (True, "ok") + api_v3_client.post(self.URL, json={"ssid": "OpenNet"}) + wifi_manager.connect_to_network.assert_called_once_with("OpenNet", "") + + def test_null_password_becomes_empty_string(self, api_v3_client, wifi_manager): + wifi_manager.connect_to_network.return_value = (True, "ok") + api_v3_client.post(self.URL, json={"ssid": "OpenNet", "password": None}) + wifi_manager.connect_to_network.assert_called_once_with("OpenNet", "") + + def test_failure_reports_the_managers_reason(self, api_v3_client, wifi_manager): + wifi_manager.connect_to_network.return_value = (False, "Bad password") + response = api_v3_client.post(self.URL, json={"ssid": "HomeNet"}) + assert response.status_code == 400 + assert response.get_json()["message"] == "Bad password" + + def test_failure_without_reason_uses_fallback_text(self, api_v3_client, wifi_manager): + wifi_manager.connect_to_network.return_value = (False, None) + response = api_v3_client.post(self.URL, json={"ssid": "HomeNet"}) + assert response.status_code == 400 + assert response.get_json()["message"] == "Failed to connect to network" + + def test_manager_exception_is_a_500_without_leaking_internals( + self, api_v3_client, wifi_manager): + wifi_manager.connect_to_network.side_effect = RuntimeError( + "/usr/lib/secret/path blew up") + response = api_v3_client.post(self.URL, json={"ssid": "HomeNet"}) + assert response.status_code == 500 + body = response.get_json() + assert body["message"] == "An error occurred; see logs for details" + # `details` comes from describe_exception, which is deliberately + # safe to return (redacted, capped) — it names the type. + assert "RuntimeError" in body["details"] + + +class TestDisconnect: + URL = "/api/v3/wifi/disconnect" + + def test_success(self, api_v3_client, wifi_manager): + wifi_manager.disconnect_from_network.return_value = (True, "Disconnected") + response = api_v3_client.post(self.URL) + assert response.status_code == 200 + assert response.get_json()["message"] == "Disconnected" + + def test_failure(self, api_v3_client, wifi_manager): + wifi_manager.disconnect_from_network.return_value = (False, "Not connected") + response = api_v3_client.post(self.URL) + assert response.status_code == 400 + assert response.get_json()["message"] == "Not connected" + + def test_failure_without_reason_uses_fallback(self, api_v3_client, wifi_manager): + wifi_manager.disconnect_from_network.return_value = (False, "") + response = api_v3_client.post(self.URL) + assert response.get_json()["message"] == "Failed to disconnect from network" + + def test_exception_is_a_500(self, api_v3_client, wifi_manager): + wifi_manager.disconnect_from_network.side_effect = OSError("nmcli missing") + assert api_v3_client.post(self.URL).status_code == 500 + + +class TestApMode: + ENABLE = "/api/v3/wifi/ap/enable" + DISABLE = "/api/v3/wifi/ap/disable" + + def test_enable_success(self, api_v3_client, wifi_manager): + wifi_manager.enable_ap_mode.return_value = (True, "AP enabled") + response = api_v3_client.post(self.ENABLE, json={}) + assert response.status_code == 200 + wifi_manager.enable_ap_mode.assert_called_once_with(force=False) + + @pytest.mark.parametrize("raw,expected", [ + (True, True), (False, False), + ("true", True), ("TRUE", True), ("1", True), + ("false", False), ("no", False), ("yes", False), + (1, False), # only real True or the listed strings count + ]) + def test_force_coercion(self, api_v3_client, wifi_manager, raw, expected): + wifi_manager.enable_ap_mode.return_value = (True, "ok") + api_v3_client.post(self.ENABLE, json={"force": raw}) + wifi_manager.enable_ap_mode.assert_called_once_with(force=expected) + + def test_enable_without_body(self, api_v3_client, wifi_manager): + wifi_manager.enable_ap_mode.return_value = (True, "ok") + assert api_v3_client.post(self.ENABLE).status_code == 200 + + def test_enable_failure(self, api_v3_client, wifi_manager): + wifi_manager.enable_ap_mode.return_value = (False, "hostapd missing") + response = api_v3_client.post(self.ENABLE, json={}) + assert response.status_code == 400 + assert response.get_json()["message"] == "hostapd missing" + + def test_disable_success(self, api_v3_client, wifi_manager): + wifi_manager.disable_ap_mode.return_value = (True, "AP disabled") + assert api_v3_client.post(self.DISABLE).status_code == 200 + + def test_disable_failure(self, api_v3_client, wifi_manager): + wifi_manager.disable_ap_mode.return_value = (False, "not running") + assert api_v3_client.post(self.DISABLE).status_code == 400 + + def test_enable_exception_is_a_500(self, api_v3_client, wifi_manager): + wifi_manager.enable_ap_mode.side_effect = RuntimeError("boom") + assert api_v3_client.post(self.ENABLE, json={}).status_code == 500 + + +class TestRadio: + URL = "/api/v3/wifi/radio" + + def test_get_state(self, api_v3_client, wifi_manager): + wifi_manager.get_wifi_radio_state.return_value = { + "enabled": True, "ethernet_connected": False} + response = api_v3_client.get(self.URL) + assert response.status_code == 200 + assert response.get_json()["data"]["enabled"] is True + + def test_get_state_exception_is_a_500(self, api_v3_client, wifi_manager): + wifi_manager.get_wifi_radio_state.side_effect = OSError("rfkill missing") + assert api_v3_client.get(self.URL).status_code == 500 + + def test_enabled_is_required(self, api_v3_client, wifi_manager): + response = api_v3_client.post(self.URL, json={}) + assert response.status_code == 400 + assert "enabled is required" in response.get_json()["message"] + wifi_manager.set_wifi_radio.assert_not_called() + + def test_enable_success(self, api_v3_client, wifi_manager): + wifi_manager.set_wifi_radio.return_value = (True, "Radio on", None) + wifi_manager.get_wifi_radio_state.return_value = {"enabled": True} + response = api_v3_client.post(self.URL, json={"enabled": True}) + assert response.status_code == 200 + wifi_manager.set_wifi_radio.assert_called_once_with(True, force=False) + + @pytest.mark.parametrize("raw,expected", [ + (True, True), ("true", True), ("1", True), ("yes", True), + (False, False), ("false", False), ("off", False), (0, False), + ]) + def test_enabled_coercion_is_string_aware( + self, api_v3_client, wifi_manager, raw, expected): + # bool("false") is True, so the endpoint parses strings explicitly + # rather than trusting truthiness — it is a public contract, not + # only the shipped UI which always sends real JSON booleans. + wifi_manager.set_wifi_radio.return_value = (True, "ok", None) + wifi_manager.get_wifi_radio_state.return_value = {} + api_v3_client.post(self.URL, json={"enabled": raw}) + wifi_manager.set_wifi_radio.assert_called_once_with(expected, force=False) + + def test_force_passed_through(self, api_v3_client, wifi_manager): + wifi_manager.set_wifi_radio.return_value = (True, "ok", None) + wifi_manager.get_wifi_radio_state.return_value = {} + api_v3_client.post(self.URL, json={"enabled": False, "force": "true"}) + wifi_manager.set_wifi_radio.assert_called_once_with(False, force=True) + + def test_refusal_reports_reason(self, api_v3_client, wifi_manager): + # Disabling the radio without Ethernet would lock the user out of + # this very interface, so the manager can refuse with a reason. + wifi_manager.set_wifi_radio.return_value = ( + False, "Refusing: no wired fallback", "no_ethernet") + response = api_v3_client.post(self.URL, json={"enabled": False}) + assert response.status_code == 400 + body = response.get_json() + assert body["reason"] == "no_ethernet" + assert "Refusing" in body["message"] + + def test_exception_is_a_500(self, api_v3_client, wifi_manager): + wifi_manager.set_wifi_radio.side_effect = RuntimeError("boom") + assert api_v3_client.post(self.URL, json={"enabled": True}).status_code == 500 + + +class TestNoRealNetworking: + def test_wifi_manager_is_never_constructed_for_real(self, api_v3_client): + # Guard against a future refactor moving the import to module level, + # where the fixture's patch of the definition site would stop + # applying and the tests would start driving real networking. + with patch("src.wifi_manager.WiFiManager") as cls: + cls.return_value.disconnect_from_network.return_value = (True, "ok") + api_v3_client.post("/api/v3/wifi/disconnect") + assert cls.called diff --git a/web_interface/blueprints/api_v3.py b/web_interface/blueprints/api_v3.py index 06fef117..7f6e108c 100644 --- a/web_interface/blueprints/api_v3.py +++ b/web_interface/blueprints/api_v3.py @@ -2411,7 +2411,7 @@ def get_on_demand_status(): def start_on_demand_display(): """Request the display controller to run a specific plugin on-demand.""" try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} plugin_id = data.get('plugin_id') mode = data.get('mode') duration = data.get('duration') @@ -2935,7 +2935,7 @@ def manage_plugin_limits(plugin_id): }) else: # POST - Set limits - data = request.get_json() or {} + data = request.get_json(silent=True) or {} from src.plugin_system.resource_monitor import ResourceLimits limits = ResourceLimits( @@ -4236,7 +4236,7 @@ def refresh_plugin_store(): if not api_v3.plugin_store_manager: return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500 - data = request.get_json() or {} + data = request.get_json(silent=True) or {} fetch_commit_info = data.get('fetch_commit_info', data.get('fetch_latest_versions', False)) # Force refresh the registry @@ -5822,7 +5822,7 @@ def reset_plugin_config(): if not api_v3.config_manager: return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500 - data = request.get_json() or {} + data = request.get_json(silent=True) or {} plugin_id = data.get('plugin_id') preserve_secrets = data.get('preserve_secrets', True) @@ -6209,7 +6209,7 @@ sys.exit(proc.returncode) def authenticate_spotify(): """Run Spotify authentication script""" try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} redirect_url = data.get('redirect_url', '').strip() # Get plugin directory @@ -7146,7 +7146,7 @@ def upload_of_the_day_json(): def delete_of_the_day_json(): """Delete a JSON file from of-the-day plugin""" try: - data = request.get_json() or {} + data = request.get_json(silent=True) or {} file_id = data.get('file_id') # This is the category_name if not file_id: