mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-14 07:08:05 +00:00
test(api): cover the install endpoints, and make 14 dead guards reachable
/plugins/install and /plugins/install-from-url were tested only at the
PluginStoreManager layer, so the route logic — the queue-versus-direct
branch, schema invalidation, discovery, state and history recording — was
unexercised.
Covering them surfaced the wider form of the body-parsing bug fixed for
the `or {}` handlers in the previous commit. Fourteen handlers read
`data = request.get_json()` and immediately guard with `if not data:
return 400, 'No data provided'`. That guard cannot run: get_json()
without silent=True raises UnsupportedMediaType for a request with no
JSON body, so the catch-all answered 500 "an error occurred; see logs
for details" where the handler plainly meant to answer 400 and say
which field was missing. Every one of these endpoints told a caller who
simply forgot the body to go read the server logs.
All fourteen now use silent=True, so the guard each author already wrote
is the one that runs. This covers /config/raw/main and /config/raw/secrets
among them, whose own bodyless case had the same shape.
The two remaining bare reads are left alone: neither declares what a
missing body should do, so there is no stated intent to honour.
31 install tests plus 17 body tests. The install pair is checked against
each other rather than only individually — the same install logic is
written twice, once in the queue callback and once in the fallback, so
the tests assert both produce identical schema, discovery, state and
history effects. They agree today; the one difference is the success
message wording, which is characterized rather than changed.
Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh
This commit is contained in:
@@ -18,6 +18,7 @@ This file covers the endpoints whose bodyless behaviour is not already
|
||||
tested in their own suite.
|
||||
"""
|
||||
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
@@ -70,21 +71,66 @@ class TestPluginLimits:
|
||||
assert api_v3_client.post(self.URL).status_code != 500
|
||||
|
||||
|
||||
class TestNoToleratedBodyReadIsUnguarded:
|
||||
def test_every_or_default_body_read_uses_silent(self):
|
||||
class TestMissingBodyGivesTheDeclaredError:
|
||||
"""Handlers that answer "No data provided" must actually be able to.
|
||||
|
||||
A second group of handlers reads `data = request.get_json()` and then
|
||||
guards with `if not data: return 400`. That guard is unreachable for a
|
||||
request with no JSON body, because get_json() raises first — so the
|
||||
caller got a 500 "an error occurred; see logs for details" instead of
|
||||
the 400 the handler plainly intends to send.
|
||||
"""
|
||||
|
||||
@pytest.mark.parametrize("url", [
|
||||
"/api/v3/plugins/install",
|
||||
"/api/v3/plugins/install-from-url",
|
||||
"/api/v3/plugins/registry-from-url",
|
||||
"/api/v3/config/raw/main",
|
||||
"/api/v3/config/raw/secrets",
|
||||
"/api/v3/cache/delete",
|
||||
])
|
||||
def test_bodyless_post_gets_a_400_not_a_500(self, api_v3_client, api_v3_module, url):
|
||||
response = api_v3_client.post(url)
|
||||
assert response.status_code == 400, (
|
||||
f"{url} answered {response.status_code}: "
|
||||
f"{response.get_data(as_text=True)[:200]}")
|
||||
|
||||
@pytest.mark.parametrize("url", [
|
||||
"/api/v3/plugins/install",
|
||||
"/api/v3/config/raw/main",
|
||||
])
|
||||
def test_malformed_json_gets_a_400_not_a_500(self, api_v3_client, api_v3_module, url):
|
||||
response = api_v3_client.post(
|
||||
url, data="{not json", content_type="application/json")
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
class TestNoBodyReadContradictsItsOwnGuard:
|
||||
SOURCE = Path(__file__).parent.parent / "web_interface/blueprints/api_v3.py"
|
||||
|
||||
def test_no_or_default_read_is_unguarded(self):
|
||||
"""`get_json() or <default>` is a contradiction without silent=True.
|
||||
|
||||
Writing `or {}` declares the body optional; omitting silent=True
|
||||
means the call raises before the default can apply. Catch the
|
||||
combination here rather than waiting for each endpoint to be
|
||||
exercised by hand.
|
||||
means the call raises before the default can apply.
|
||||
"""
|
||||
source = Path(__file__).parent.parent.joinpath(
|
||||
"web_interface/blueprints/api_v3.py").read_text()
|
||||
offenders = [
|
||||
line.strip() for line in source.splitlines()
|
||||
line.strip() for line in self.SOURCE.read_text().splitlines()
|
||||
if "request.get_json()" in line and " or " in line
|
||||
]
|
||||
assert offenders == [], (
|
||||
"these reads declare a default but raise before reaching it; "
|
||||
f"use get_json(silent=True): {offenders}")
|
||||
|
||||
def test_no_not_data_guard_is_unreachable(self):
|
||||
"""A `if not data:` guard needs a read that can actually return None."""
|
||||
lines = self.SOURCE.read_text().splitlines()
|
||||
offenders = []
|
||||
for i, line in enumerate(lines):
|
||||
if re.search(r"=\s*request\.get_json\(\)\s*$", line):
|
||||
window = "\n".join(lines[i + 1:i + 3])
|
||||
if re.search(r"if\s+(not\s+data\b|data\s+is\s+None)", window):
|
||||
offenders.append(f"line {i + 1}: {line.strip()}")
|
||||
assert offenders == [], (
|
||||
"these handlers guard on a missing body but raise before the "
|
||||
f"guard runs; use get_json(silent=True): {offenders}")
|
||||
|
||||
@@ -0,0 +1,302 @@
|
||||
"""
|
||||
Endpoint tests for POST /plugins/install and POST /plugins/install-from-url.
|
||||
|
||||
Both were only ever tested at the PluginStoreManager layer, so the route
|
||||
logic — the queue-vs-direct branch, schema invalidation, plugin discovery,
|
||||
state and history recording — was unexercised.
|
||||
|
||||
/plugins/install carries the same install logic twice: once inside the
|
||||
operation-queue callback and once in the direct fallback. The paired
|
||||
tests below assert both branches produce the same side effects, so the
|
||||
duplication cannot quietly drift.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from test._api_v3_test_helpers import api_v3_client, api_v3_module # noqa: F401,E402
|
||||
|
||||
INSTALL = "/api/v3/plugins/install"
|
||||
FROM_URL = "/api/v3/plugins/install-from-url"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def queued(api_v3_module):
|
||||
"""Enable the operation queue and run its callback synchronously."""
|
||||
queue = MagicMock()
|
||||
|
||||
def enqueue(operation_type, plugin_id, operation_callback=None):
|
||||
queue.callback_result = operation_callback(MagicMock())
|
||||
return "op-123"
|
||||
|
||||
queue.enqueue_operation.side_effect = enqueue
|
||||
api_v3_module.api_v3.operation_queue = queue
|
||||
return queue
|
||||
|
||||
|
||||
def side_effects(module):
|
||||
"""The manager calls a successful install is expected to make."""
|
||||
api = module.api_v3
|
||||
return {
|
||||
"schema_invalidated": api.schema_manager.invalidate_cache.call_args_list,
|
||||
"discovered": api.plugin_manager.discover_plugins.call_count,
|
||||
"loaded": api.plugin_manager.load_plugin.call_args_list,
|
||||
"state_set": api.plugin_state_manager.set_plugin_installed.call_args_list,
|
||||
"history": api.operation_history.record_operation.call_args_list,
|
||||
}
|
||||
|
||||
|
||||
class TestInstallValidation:
|
||||
def test_uninitialized_store_manager_is_a_500(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_store_manager = None
|
||||
response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
|
||||
assert response.status_code == 500
|
||||
assert "not initialized" in response.get_json()["message"]
|
||||
|
||||
def test_missing_plugin_id_is_a_400(self, api_v3_client, api_v3_module):
|
||||
response = api_v3_client.post(INSTALL, json={})
|
||||
assert response.status_code == 400
|
||||
assert "plugin_id required" in response.get_json()["message"]
|
||||
api_v3_module.api_v3.plugin_store_manager.install_plugin.assert_not_called()
|
||||
|
||||
def test_empty_body_is_a_400(self, api_v3_client, api_v3_module):
|
||||
assert api_v3_client.post(INSTALL, json=None).status_code == 400
|
||||
|
||||
|
||||
class TestInstallDirectPath:
|
||||
"""operation_queue is None — the fallback branch."""
|
||||
|
||||
def test_success(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
|
||||
response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
|
||||
assert response.status_code == 200
|
||||
assert response.get_json()["status"] == "success"
|
||||
|
||||
def test_success_side_effects(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
|
||||
api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
|
||||
effects = side_effects(api_v3_module)
|
||||
assert effects["schema_invalidated"] == [(("clock",), {})]
|
||||
assert effects["discovered"] == 1
|
||||
assert effects["loaded"] == [(("clock",), {})]
|
||||
assert effects["state_set"] == [(("clock",), {})]
|
||||
assert effects["history"][0].kwargs["status"] == "success"
|
||||
|
||||
def test_branch_forwarded_to_the_manager(self, api_v3_client, api_v3_module):
|
||||
manager = api_v3_module.api_v3.plugin_store_manager
|
||||
manager.install_plugin.return_value = True
|
||||
api_v3_client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
|
||||
manager.install_plugin.assert_called_once_with("clock", branch="dev")
|
||||
|
||||
def test_branch_named_in_the_message(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
|
||||
response = api_v3_client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
|
||||
assert "(branch: dev)" in response.get_json()["message"]
|
||||
|
||||
def test_failure_is_a_500(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
|
||||
response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
|
||||
assert response.status_code == 500
|
||||
assert "Failed to install" in response.get_json()["message"]
|
||||
|
||||
def test_failure_mentions_missing_registry_entry(self, api_v3_client, api_v3_module):
|
||||
manager = api_v3_module.api_v3.plugin_store_manager
|
||||
manager.install_plugin.return_value = False
|
||||
manager.get_plugin_info.return_value = None
|
||||
response = api_v3_client.post(INSTALL, json={"plugin_id": "ghost"})
|
||||
assert "not found in registry" in response.get_json()["message"]
|
||||
|
||||
def test_failure_omits_registry_note_when_plugin_is_known(
|
||||
self, api_v3_client, api_v3_module):
|
||||
manager = api_v3_module.api_v3.plugin_store_manager
|
||||
manager.install_plugin.return_value = False
|
||||
manager.get_plugin_info.return_value = {"id": "clock"}
|
||||
response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
|
||||
assert "not found in registry" not in response.get_json()["message"]
|
||||
|
||||
def test_failure_recorded_in_history(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
|
||||
api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
|
||||
record = api_v3_module.api_v3.operation_history.record_operation.call_args
|
||||
assert record.kwargs["status"] == "failed"
|
||||
|
||||
def test_no_side_effects_on_failure(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
|
||||
api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
|
||||
effects = side_effects(api_v3_module)
|
||||
assert effects["schema_invalidated"] == []
|
||||
assert effects["loaded"] == []
|
||||
assert effects["state_set"] == []
|
||||
|
||||
|
||||
class TestInstallQueuedPath:
|
||||
"""operation_queue present — the callback branch."""
|
||||
|
||||
def test_returns_an_operation_id(self, api_v3_client, api_v3_module, queued):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
|
||||
response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
|
||||
assert response.status_code == 200
|
||||
assert response.get_json()["data"]["operation_id"] == "op-123"
|
||||
|
||||
def test_message_says_queued(self, api_v3_client, api_v3_module, queued):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
|
||||
response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
|
||||
assert "queued" in response.get_json()["message"]
|
||||
|
||||
def test_callback_success_side_effects(self, api_v3_client, api_v3_module, queued):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
|
||||
api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
|
||||
effects = side_effects(api_v3_module)
|
||||
assert effects["schema_invalidated"] == [(("clock",), {})]
|
||||
assert effects["discovered"] == 1
|
||||
assert effects["loaded"] == [(("clock",), {})]
|
||||
assert effects["state_set"] == [(("clock",), {})]
|
||||
assert effects["history"][0].kwargs["status"] == "success"
|
||||
|
||||
def test_callback_reports_success(self, api_v3_client, api_v3_module, queued):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
|
||||
api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
|
||||
assert queued.callback_result["success"] is True
|
||||
|
||||
def test_callback_failure_raises_for_the_queue(self, api_v3_client, api_v3_module, queued):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
|
||||
# The callback signals failure by raising, so the queue can mark the
|
||||
# operation failed; the route's catch-all turns it into a 500.
|
||||
response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
|
||||
assert response.status_code == 500
|
||||
|
||||
def test_callback_failure_recorded_in_history(self, api_v3_client, api_v3_module, queued):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
|
||||
api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
|
||||
record = api_v3_module.api_v3.operation_history.record_operation.call_args
|
||||
assert record.kwargs["status"] == "failed"
|
||||
|
||||
def test_branch_forwarded_from_the_callback(self, api_v3_client, api_v3_module, queued):
|
||||
manager = api_v3_module.api_v3.plugin_store_manager
|
||||
manager.install_plugin.return_value = True
|
||||
api_v3_client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
|
||||
manager.install_plugin.assert_called_once_with("clock", branch="dev")
|
||||
|
||||
|
||||
class TestInstallPathsAgree:
|
||||
"""The queue callback and the direct fallback duplicate the same logic."""
|
||||
|
||||
def _run(self, client, module, install_ok, queue):
|
||||
module.api_v3.plugin_store_manager.install_plugin.return_value = install_ok
|
||||
client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
|
||||
return side_effects(module)
|
||||
|
||||
def test_success_side_effects_match(self, api_v3_client, api_v3_module):
|
||||
direct = self._run(api_v3_client, api_v3_module, True, None)
|
||||
|
||||
# Reset and re-run through the queue.
|
||||
for mock in (api_v3_module.api_v3.schema_manager,
|
||||
api_v3_module.api_v3.plugin_manager,
|
||||
api_v3_module.api_v3.plugin_state_manager,
|
||||
api_v3_module.api_v3.operation_history):
|
||||
mock.reset_mock()
|
||||
queue = MagicMock()
|
||||
queue.enqueue_operation.side_effect = (
|
||||
lambda t, p, operation_callback=None: operation_callback(MagicMock()) and "op")
|
||||
api_v3_module.api_v3.operation_queue = queue
|
||||
queued = self._run(api_v3_client, api_v3_module, True, queue)
|
||||
|
||||
assert direct["schema_invalidated"] == queued["schema_invalidated"]
|
||||
assert direct["discovered"] == queued["discovered"]
|
||||
assert direct["loaded"] == queued["loaded"]
|
||||
assert direct["state_set"] == queued["state_set"]
|
||||
assert (direct["history"][0].kwargs["status"]
|
||||
== queued["history"][0].kwargs["status"])
|
||||
assert (direct["history"][0].kwargs["details"]
|
||||
== queued["history"][0].kwargs["details"])
|
||||
|
||||
def test_only_the_message_wording_differs(self, api_v3_client, api_v3_module):
|
||||
# Characterized: the direct path says "Plugin installed
|
||||
# successfully" while the queue callback says "Plugin clock
|
||||
# installed successfully". Cosmetic, and the queue's text is
|
||||
# internal to the operation record rather than the HTTP response.
|
||||
api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
|
||||
direct = api_v3_client.post(INSTALL, json={"plugin_id": "clock"}).get_json()
|
||||
assert direct["message"] == "Plugin installed successfully"
|
||||
|
||||
|
||||
class TestInstallFromUrl:
|
||||
def test_uninitialized_store_manager_is_a_500(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_store_manager = None
|
||||
assert api_v3_client.post(FROM_URL, json={"repo_url": "http://x"}).status_code == 500
|
||||
|
||||
def test_missing_repo_url_is_a_400(self, api_v3_client, api_v3_module):
|
||||
response = api_v3_client.post(FROM_URL, json={})
|
||||
assert response.status_code == 400
|
||||
assert "repo_url required" in response.get_json()["message"]
|
||||
|
||||
def test_success(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
|
||||
"success": True, "plugin_id": "clock", "name": "Clock"}
|
||||
response = api_v3_client.post(FROM_URL, json={"repo_url": "https://github.com/o/r"})
|
||||
assert response.status_code == 200
|
||||
body = response.get_json()
|
||||
assert body["plugin_id"] == "clock"
|
||||
assert body["name"] == "Clock"
|
||||
|
||||
def test_all_optional_arguments_forwarded(self, api_v3_client, api_v3_module):
|
||||
manager = api_v3_module.api_v3.plugin_store_manager
|
||||
manager.install_from_url.return_value = {"success": True, "plugin_id": "clock"}
|
||||
api_v3_client.post(FROM_URL, json={
|
||||
"repo_url": " https://github.com/o/r ",
|
||||
"plugin_id": "clock",
|
||||
"plugin_path": "plugins/clock",
|
||||
"branch": "dev",
|
||||
})
|
||||
manager.install_from_url.assert_called_once_with(
|
||||
repo_url="https://github.com/o/r",
|
||||
plugin_id="clock",
|
||||
plugin_path="plugins/clock",
|
||||
branch="dev",
|
||||
)
|
||||
|
||||
def test_success_invalidates_schema_and_loads_plugin(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
|
||||
"success": True, "plugin_id": "clock"}
|
||||
api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
|
||||
api_v3_module.api_v3.schema_manager.invalidate_cache.assert_called_once_with("clock")
|
||||
api_v3_module.api_v3.plugin_manager.load_plugin.assert_called_once_with("clock")
|
||||
|
||||
def test_success_without_plugin_id_skips_discovery(self, api_v3_client, api_v3_module):
|
||||
# install_from_url can succeed without naming the plugin; there is
|
||||
# then nothing to invalidate or load.
|
||||
api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
|
||||
"success": True, "plugin_id": None}
|
||||
api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
|
||||
api_v3_module.api_v3.schema_manager.invalidate_cache.assert_not_called()
|
||||
api_v3_module.api_v3.plugin_manager.load_plugin.assert_not_called()
|
||||
|
||||
def test_branch_from_result_included(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
|
||||
"success": True, "plugin_id": "clock", "branch": "dev"}
|
||||
body = api_v3_client.post(FROM_URL, json={"repo_url": "http://x"}).get_json()
|
||||
assert body["branch"] == "dev"
|
||||
assert "(branch: dev)" in body["message"]
|
||||
|
||||
def test_failure_reports_the_managers_error(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
|
||||
"success": False, "error": "repo not found"}
|
||||
response = api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
|
||||
assert response.status_code == 500
|
||||
assert response.get_json()["message"] == "repo not found"
|
||||
|
||||
def test_failure_without_error_uses_fallback_text(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
|
||||
"success": False}
|
||||
response = api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
|
||||
assert "Failed to install plugin from URL" in response.get_json()["message"]
|
||||
|
||||
def test_manager_exception_is_a_500(self, api_v3_client, api_v3_module):
|
||||
api_v3_module.api_v3.plugin_store_manager.install_from_url.side_effect = (
|
||||
RuntimeError("boom"))
|
||||
assert api_v3_client.post(FROM_URL, json={"repo_url": "http://x"}).status_code == 500
|
||||
@@ -328,7 +328,7 @@ def save_schedule_config():
|
||||
if not api_v3.config_manager:
|
||||
return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
|
||||
|
||||
data = request.get_json()
|
||||
data = request.get_json(silent=True)
|
||||
if not data:
|
||||
return jsonify({'status': 'error', 'message': 'No data provided'}), 400
|
||||
|
||||
@@ -536,7 +536,7 @@ def save_dim_schedule_config():
|
||||
if not api_v3.config_manager:
|
||||
return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
|
||||
|
||||
data = request.get_json()
|
||||
data = request.get_json(silent=True)
|
||||
if not data:
|
||||
return jsonify({'status': 'error', 'message': 'No data provided'}), 400
|
||||
|
||||
@@ -1345,7 +1345,7 @@ def save_raw_main_config():
|
||||
if not api_v3.config_manager:
|
||||
return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
|
||||
|
||||
data = request.get_json()
|
||||
data = request.get_json(silent=True)
|
||||
if not data:
|
||||
return jsonify({'status': 'error', 'message': 'No data provided'}), 400
|
||||
|
||||
@@ -1391,7 +1391,7 @@ def save_raw_secrets_config():
|
||||
if not api_v3.config_manager:
|
||||
return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
|
||||
|
||||
data = request.get_json()
|
||||
data = request.get_json(silent=True)
|
||||
if not data:
|
||||
return jsonify({'status': 'error', 'message': 'No data provided'}), 400
|
||||
|
||||
@@ -2966,7 +2966,7 @@ def toggle_plugin():
|
||||
content_type = request.content_type or ''
|
||||
|
||||
if 'application/json' in content_type:
|
||||
data = request.get_json()
|
||||
data = request.get_json(silent=True)
|
||||
if not data or 'plugin_id' not in data or 'enabled' not in data:
|
||||
return jsonify({'status': 'error', 'message': 'plugin_id and enabled required'}), 400
|
||||
plugin_id = data['plugin_id']
|
||||
@@ -3837,7 +3837,7 @@ def install_plugin():
|
||||
if not api_v3.plugin_store_manager:
|
||||
return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
|
||||
|
||||
data = request.get_json()
|
||||
data = request.get_json(silent=True)
|
||||
if not data or 'plugin_id' not in data:
|
||||
return jsonify({'status': 'error', 'message': 'plugin_id required'}), 400
|
||||
|
||||
@@ -3971,7 +3971,7 @@ def install_plugin_from_url():
|
||||
if not api_v3.plugin_store_manager:
|
||||
return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
|
||||
|
||||
data = request.get_json()
|
||||
data = request.get_json(silent=True)
|
||||
if not data or 'repo_url' not in data:
|
||||
return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
|
||||
|
||||
@@ -4026,7 +4026,7 @@ def get_registry_from_url():
|
||||
if not api_v3.plugin_store_manager:
|
||||
return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
|
||||
|
||||
data = request.get_json()
|
||||
data = request.get_json(silent=True)
|
||||
if not data or 'repo_url' not in data:
|
||||
return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
|
||||
|
||||
@@ -4071,7 +4071,7 @@ def add_saved_repository():
|
||||
if not api_v3.saved_repositories_manager:
|
||||
return jsonify({'status': 'error', 'message': 'Saved repositories manager not initialized'}), 500
|
||||
|
||||
data = request.get_json()
|
||||
data = request.get_json(silent=True)
|
||||
if not data or 'repo_url' not in data:
|
||||
return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
|
||||
|
||||
@@ -4102,7 +4102,7 @@ def remove_saved_repository():
|
||||
if not api_v3.saved_repositories_manager:
|
||||
return jsonify({'status': 'error', 'message': 'Saved repositories manager not initialized'}), 500
|
||||
|
||||
data = request.get_json()
|
||||
data = request.get_json(silent=True)
|
||||
if not data or 'repo_url' not in data:
|
||||
return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
|
||||
|
||||
@@ -6529,7 +6529,7 @@ def get_fonts_overrides():
|
||||
def save_fonts_overrides():
|
||||
"""Save font overrides"""
|
||||
try:
|
||||
data = request.get_json()
|
||||
data = request.get_json(silent=True)
|
||||
if not data:
|
||||
return jsonify({'status': 'error', 'message': 'No data provided'}), 400
|
||||
|
||||
@@ -7635,7 +7635,7 @@ def connect_wifi():
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
data = request.get_json()
|
||||
data = request.get_json(silent=True)
|
||||
if not data:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
@@ -7789,7 +7789,7 @@ def set_auto_enable_ap_mode():
|
||||
try:
|
||||
from src.wifi_manager import WiFiManager
|
||||
|
||||
data = request.get_json()
|
||||
data = request.get_json(silent=True)
|
||||
if data is None or 'auto_enable_ap_mode' not in data:
|
||||
return jsonify({
|
||||
'status': 'error',
|
||||
@@ -7918,7 +7918,7 @@ def delete_cache_file():
|
||||
from src.cache_manager import CacheManager
|
||||
api_v3.cache_manager = CacheManager()
|
||||
|
||||
data = request.get_json()
|
||||
data = request.get_json(silent=True)
|
||||
if not data or 'key' not in data:
|
||||
return jsonify({'status': 'error', 'message': 'cache key is required'}), 400
|
||||
|
||||
|
||||
Reference in New Issue
Block a user