mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-13 22:58:06 +00:00
test(api): cover wifi and registry endpoints, and fix bodyless POSTs
The /wifi/* routes drive the host's real networking and the registry
routes reach GitHub, and neither had endpoint-level tests. Covering them
surfaced a bug affecting six endpoints.
Six handlers read their body as `request.get_json() or {}`. The `or {}`
says every field is optional and a missing body should fall back to
defaults — but get_json() without silent=True raises UnsupportedMediaType
when there is no JSON Content-Type, and it raises before `or {}` is ever
evaluated. Each handler's catch-all then reported that as a 500. So
POSTing with no body — what curl sends by default, and what a fetch()
without options sends — failed on /plugins/store/refresh,
/display/on-demand/start, /plugins/config/reset,
/plugins/of-the-day/json/delete, /plugins/{id}/limits and
/plugins/authenticate/spotify. The shipped UI always sends a JSON object,
which is why this stayed hidden.
All six now use silent=True. test_api_v3_optional_body.py covers the
affected endpoints and adds a source check, since the combination of
`or <default>` with a non-silent read is self-contradictory wherever it
appears and is easier to catch by inspection than by exercising each
endpoint by hand.
Also adds test/_api_v3_test_helpers.py: the blueprint holds its managers
on a module-level singleton rather than in Flask app state, so a test
that mocks them leaks into every later test unless the originals are
restored. The existing _make_client() does this for unittest classes;
this is the pytest-fixture equivalent, for the five suites still to come.
69 endpoint tests: connect/disconnect/AP/radio including the string-aware
boolean coercion these endpoints deliberately use, the radio's
lockout-refusal path, registry refresh and fetch-from-URL, and a guard
that WiFiManager is never constructed for real.
Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh
This commit is contained in:
@@ -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()
|
||||
@@ -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 <default>` is a contradiction without silent=True.
|
||||
|
||||
Writing `or {}` declares the body optional; omitting silent=True
|
||||
means the call raises before the default can apply. 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}")
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user