mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-11-10 12:28:27 +00:00
refactor: move the settings routes into core/settings_routes.py (#1595)
Second route move from the plan in Discussion #1536: the four settings routes (`GET /api/settings`, `GET` and `PUT /api/settings/<tab>`, and `POST /api/settings/<tab>/action/<key>`) go from `main.py` to `core/settings_routes.py`, registered with `register_settings_routes(app, login_required)`. `main.py` is 143 lines shorter. These handlers call `settings_registry` directly, so nothing else has to be passed in. The handler bodies are the same code, and the only difference I found after de-indenting them is that `ruff format` wrapped one signature that no longer fits on a line. Same two-commit layout as the onboarding one (#1594). The first commit adds 27 tests for these routes and they pass on current `main` before anything moves. They pin the 404 for an unknown tab, the 400 when a `PUT` body is not a JSON object (a list, a string, a number, or no body at all), the "No changes to save" answer for an empty object, the unsaved form values reaching an action, the result passed through with a 400 on failure, the 500 with the error text on an operational error, a 401 without a session, a 403 for a signed-in non-admin on all four routes, and open access with no auth. The second commit is the move and no test changed in it. The admin check for these paths lives in `login_required`, so the same decorator is passed in. What I checked: - The Flask URL map (every rule, its methods and its endpoint name) is identical before and after, all 77 rules. - Full suite, ruff and basedpyright are clean. - I ran the app from this branch and opened Settings in a browser: the modal lists every tab and the General tab shows its values, which come from the moved `GET`. Over HTTP, a `PUT` of a new search page title saved and read back, a list body got the 400 and an unknown tab the 404. As with #1594, log lines for these routes now come from `shelfmark.core.settings_routes`, and the error tuple is a local copy (`_SETTINGS_ERRORS`) so the module does not import `main`. The two PRs touch neighbouring lines of `main.py` and I checked that they merge together without a conflict, in either order.
This commit is contained in:
splitsec2 authored and 
GitHub committed
1 parent
b2e3c1faf1
commit
69cbf72698
3 files changed
+454
-145
No files matched your search
@@ -0,0 +1,177 @@
|
||||
"""Settings: ``/api/settings`` and its per-tab read, update and action routes."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
from importlib import import_module
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from flask import jsonify, request
|
||||
|
||||
from shelfmark.core.logger import setup_logger
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
|
||||
from flask import Flask
|
||||
from werkzeug.wrappers import Response
|
||||
|
||||
logger = setup_logger(__name__)
|
||||
|
||||
_SETTINGS_ERRORS = (ImportError, OSError, RuntimeError, TypeError, ValueError, sqlite3.Error)
|
||||
|
||||
|
||||
def register_settings_routes(
|
||||
app: Flask,
|
||||
login_required: Callable[..., Callable[..., Response | tuple[Response, int]]],
|
||||
) -> None:
|
||||
"""Register the settings routes.
|
||||
|
||||
``login_required`` is the app's decorator. It is also what limits these paths to an
|
||||
admin session when authentication is on, so it has to be the same one the other routes use.
|
||||
"""
|
||||
|
||||
@app.route("/api/settings", methods=["GET"])
|
||||
@login_required
|
||||
def api_settings_get_all() -> Response | tuple[Response, int]:
|
||||
"""Get all settings tabs with their fields and current values.
|
||||
|
||||
Returns:
|
||||
flask.Response: JSON with all settings tabs.
|
||||
|
||||
"""
|
||||
try:
|
||||
import_module("shelfmark.config.notifications_settings")
|
||||
import_module("shelfmark.config.security")
|
||||
|
||||
# Ensure settings are registered by importing settings modules
|
||||
# This triggers the @register_settings decorators
|
||||
import_module("shelfmark.config.settings")
|
||||
import_module("shelfmark.config.users_settings")
|
||||
from shelfmark.core.settings_registry import serialize_all_settings
|
||||
|
||||
data = serialize_all_settings(include_values=True)
|
||||
return jsonify(data)
|
||||
except _SETTINGS_ERRORS as e:
|
||||
logger.error_trace(f"Settings get error: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
@app.route("/api/settings/<tab_name>", methods=["GET"])
|
||||
@login_required
|
||||
def api_settings_get_tab(tab_name: str) -> Response | tuple[Response, int]:
|
||||
"""Get settings for a specific tab.
|
||||
|
||||
Path Parameters:
|
||||
tab_name (str): Settings tab name (e.g., "general", "hardcover")
|
||||
|
||||
Returns:
|
||||
flask.Response: JSON with tab settings and values.
|
||||
|
||||
"""
|
||||
try:
|
||||
import_module("shelfmark.config.notifications_settings")
|
||||
import_module("shelfmark.config.security")
|
||||
|
||||
# Ensure settings are registered
|
||||
import_module("shelfmark.config.settings")
|
||||
import_module("shelfmark.config.users_settings")
|
||||
from shelfmark.core.settings_registry import (
|
||||
get_settings_tab,
|
||||
serialize_tab,
|
||||
)
|
||||
|
||||
tab = get_settings_tab(tab_name)
|
||||
if not tab:
|
||||
return jsonify({"error": f"Unknown settings tab: {tab_name}"}), 404
|
||||
|
||||
return jsonify(serialize_tab(tab, include_values=True))
|
||||
except _SETTINGS_ERRORS as e:
|
||||
logger.error_trace(f"Settings get tab error: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
@app.route("/api/settings/<tab_name>", methods=["PUT"])
|
||||
@login_required
|
||||
def api_settings_update_tab(tab_name: str) -> Response | tuple[Response, int]:
|
||||
"""Update settings for a specific tab.
|
||||
|
||||
Path Parameters:
|
||||
tab_name (str): Settings tab name
|
||||
|
||||
Request Body:
|
||||
JSON object with setting keys and values to update.
|
||||
|
||||
Returns:
|
||||
flask.Response: JSON with update result.
|
||||
|
||||
"""
|
||||
try:
|
||||
import_module("shelfmark.config.notifications_settings")
|
||||
import_module("shelfmark.config.security")
|
||||
|
||||
# Ensure settings are registered
|
||||
import_module("shelfmark.config.settings")
|
||||
import_module("shelfmark.config.users_settings")
|
||||
from shelfmark.core.settings_registry import (
|
||||
get_settings_tab,
|
||||
update_settings,
|
||||
)
|
||||
|
||||
tab = get_settings_tab(tab_name)
|
||||
if not tab:
|
||||
return jsonify({"error": f"Unknown settings tab: {tab_name}"}), 404
|
||||
|
||||
values = request.get_json(silent=True)
|
||||
if values is None or not isinstance(values, dict):
|
||||
return jsonify({"error": "Request body must be a JSON object"}), 400
|
||||
|
||||
# If no values to update, return success with empty updated list
|
||||
if not values:
|
||||
return jsonify({"success": True, "message": "No changes to save", "updated": []})
|
||||
|
||||
result = update_settings(tab_name, values)
|
||||
|
||||
if result["success"]:
|
||||
return jsonify(result)
|
||||
return jsonify(result), 400
|
||||
except _SETTINGS_ERRORS as e:
|
||||
logger.error_trace(f"Settings update error: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
@app.route("/api/settings/<tab_name>/action/<action_key>", methods=["POST"])
|
||||
@login_required
|
||||
def api_settings_execute_action(
|
||||
tab_name: str, action_key: str
|
||||
) -> Response | tuple[Response, int]:
|
||||
"""Execute a settings action (e.g., test connection).
|
||||
|
||||
Path Parameters:
|
||||
tab_name (str): Settings tab name
|
||||
action_key (str): Action key to execute
|
||||
|
||||
Request Body (optional):
|
||||
JSON object with current form values (unsaved)
|
||||
|
||||
Returns:
|
||||
flask.Response: JSON with action result.
|
||||
|
||||
"""
|
||||
try:
|
||||
import_module("shelfmark.config.notifications_settings")
|
||||
import_module("shelfmark.config.security")
|
||||
|
||||
# Ensure settings are registered
|
||||
import_module("shelfmark.config.settings")
|
||||
import_module("shelfmark.config.users_settings")
|
||||
from shelfmark.core.settings_registry import execute_action
|
||||
|
||||
# Get current form values if provided (for testing with unsaved values)
|
||||
current_values = request.get_json(silent=True) or {}
|
||||
|
||||
result = execute_action(tab_name, action_key, current_values)
|
||||
|
||||
if result["success"]:
|
||||
return jsonify(result)
|
||||
return jsonify(result), 400
|
||||
except _SETTINGS_ERRORS as e:
|
||||
logger.error_trace(f"Settings action error: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
+2
-145
@@ -93,6 +93,7 @@ from shelfmark.core.requests_service import (
|
||||
reopen_failed_request,
|
||||
sync_delivery_states_from_queue_status,
|
||||
)
|
||||
from shelfmark.core.settings_routes import register_settings_routes
|
||||
from shelfmark.core.socketio_origins import build_cors_allowed_origins
|
||||
from shelfmark.core.stats_routes import STATS_PATH, register_stats_routes
|
||||
from shelfmark.core.user_db import UserDB
|
||||
@@ -3564,151 +3565,7 @@ def api_release_source_record(source_name: str, record_id: str) -> Response | tu
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
|
||||
@app.route("/api/settings", methods=["GET"])
|
||||
@login_required
|
||||
def api_settings_get_all() -> Response | tuple[Response, int]:
|
||||
"""Get all settings tabs with their fields and current values.
|
||||
|
||||
Returns:
|
||||
flask.Response: JSON with all settings tabs.
|
||||
|
||||
"""
|
||||
try:
|
||||
import_module("shelfmark.config.notifications_settings")
|
||||
import_module("shelfmark.config.security")
|
||||
|
||||
# Ensure settings are registered by importing settings modules
|
||||
# This triggers the @register_settings decorators
|
||||
import_module("shelfmark.config.settings")
|
||||
import_module("shelfmark.config.users_settings")
|
||||
from shelfmark.core.settings_registry import serialize_all_settings
|
||||
|
||||
data = serialize_all_settings(include_values=True)
|
||||
return jsonify(data)
|
||||
except _IMPORT_OPERATIONAL_ERRORS as e:
|
||||
logger.error_trace(f"Settings get error: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
|
||||
@app.route("/api/settings/<tab_name>", methods=["GET"])
|
||||
@login_required
|
||||
def api_settings_get_tab(tab_name: str) -> Response | tuple[Response, int]:
|
||||
"""Get settings for a specific tab.
|
||||
|
||||
Path Parameters:
|
||||
tab_name (str): Settings tab name (e.g., "general", "hardcover")
|
||||
|
||||
Returns:
|
||||
flask.Response: JSON with tab settings and values.
|
||||
|
||||
"""
|
||||
try:
|
||||
import_module("shelfmark.config.notifications_settings")
|
||||
import_module("shelfmark.config.security")
|
||||
|
||||
# Ensure settings are registered
|
||||
import_module("shelfmark.config.settings")
|
||||
import_module("shelfmark.config.users_settings")
|
||||
from shelfmark.core.settings_registry import (
|
||||
get_settings_tab,
|
||||
serialize_tab,
|
||||
)
|
||||
|
||||
tab = get_settings_tab(tab_name)
|
||||
if not tab:
|
||||
return jsonify({"error": f"Unknown settings tab: {tab_name}"}), 404
|
||||
|
||||
return jsonify(serialize_tab(tab, include_values=True))
|
||||
except _IMPORT_OPERATIONAL_ERRORS as e:
|
||||
logger.error_trace(f"Settings get tab error: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
|
||||
@app.route("/api/settings/<tab_name>", methods=["PUT"])
|
||||
@login_required
|
||||
def api_settings_update_tab(tab_name: str) -> Response | tuple[Response, int]:
|
||||
"""Update settings for a specific tab.
|
||||
|
||||
Path Parameters:
|
||||
tab_name (str): Settings tab name
|
||||
|
||||
Request Body:
|
||||
JSON object with setting keys and values to update.
|
||||
|
||||
Returns:
|
||||
flask.Response: JSON with update result.
|
||||
|
||||
"""
|
||||
try:
|
||||
import_module("shelfmark.config.notifications_settings")
|
||||
import_module("shelfmark.config.security")
|
||||
|
||||
# Ensure settings are registered
|
||||
import_module("shelfmark.config.settings")
|
||||
import_module("shelfmark.config.users_settings")
|
||||
from shelfmark.core.settings_registry import (
|
||||
get_settings_tab,
|
||||
update_settings,
|
||||
)
|
||||
|
||||
tab = get_settings_tab(tab_name)
|
||||
if not tab:
|
||||
return jsonify({"error": f"Unknown settings tab: {tab_name}"}), 404
|
||||
|
||||
values = request.get_json(silent=True)
|
||||
if values is None or not isinstance(values, dict):
|
||||
return jsonify({"error": "Request body must be a JSON object"}), 400
|
||||
|
||||
# If no values to update, return success with empty updated list
|
||||
if not values:
|
||||
return jsonify({"success": True, "message": "No changes to save", "updated": []})
|
||||
|
||||
result = update_settings(tab_name, values)
|
||||
|
||||
if result["success"]:
|
||||
return jsonify(result)
|
||||
return jsonify(result), 400
|
||||
except _IMPORT_OPERATIONAL_ERRORS as e:
|
||||
logger.error_trace(f"Settings update error: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
|
||||
@app.route("/api/settings/<tab_name>/action/<action_key>", methods=["POST"])
|
||||
@login_required
|
||||
def api_settings_execute_action(tab_name: str, action_key: str) -> Response | tuple[Response, int]:
|
||||
"""Execute a settings action (e.g., test connection).
|
||||
|
||||
Path Parameters:
|
||||
tab_name (str): Settings tab name
|
||||
action_key (str): Action key to execute
|
||||
|
||||
Request Body (optional):
|
||||
JSON object with current form values (unsaved)
|
||||
|
||||
Returns:
|
||||
flask.Response: JSON with action result.
|
||||
|
||||
"""
|
||||
try:
|
||||
import_module("shelfmark.config.notifications_settings")
|
||||
import_module("shelfmark.config.security")
|
||||
|
||||
# Ensure settings are registered
|
||||
import_module("shelfmark.config.settings")
|
||||
import_module("shelfmark.config.users_settings")
|
||||
from shelfmark.core.settings_registry import execute_action
|
||||
|
||||
# Get current form values if provided (for testing with unsaved values)
|
||||
current_values = request.get_json(silent=True) or {}
|
||||
|
||||
result = execute_action(tab_name, action_key, current_values)
|
||||
|
||||
if result["success"]:
|
||||
return jsonify(result)
|
||||
return jsonify(result), 400
|
||||
except _IMPORT_OPERATIONAL_ERRORS as e:
|
||||
logger.error_trace(f"Settings action error: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
register_settings_routes(app, login_required)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
|
||||
@@ -0,0 +1,275 @@
|
||||
"""API tests for the settings endpoints: auth gating, validation and error handling."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def main_module():
|
||||
"""Import `shelfmark.main` with background startup disabled."""
|
||||
with patch("shelfmark.download.orchestrator.start"):
|
||||
import shelfmark.main as main
|
||||
|
||||
importlib.reload(main)
|
||||
return main
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(main_module):
|
||||
return main_module.app.test_client()
|
||||
|
||||
|
||||
def _sign_in(client, *, is_admin: bool) -> None:
|
||||
with client.session_transaction() as sess:
|
||||
sess["user_id"] = "someone"
|
||||
sess["db_user_id"] = 7
|
||||
sess["is_admin"] = is_admin
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def builtin_auth(main_module):
|
||||
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def admin(client, builtin_auth):
|
||||
_sign_in(client, is_admin=True)
|
||||
return client
|
||||
|
||||
|
||||
_REGISTRY = "shelfmark.core.settings_registry"
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# GET /api/settings
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_get_all_returns_every_tab_with_values(admin):
|
||||
with patch(f"{_REGISTRY}.serialize_all_settings", return_value={"tabs": []}) as serialize:
|
||||
resp = admin.get("/api/settings")
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json == {"tabs": []}
|
||||
serialize.assert_called_once_with(include_values=True)
|
||||
|
||||
|
||||
def test_get_all_reports_an_operational_error(admin):
|
||||
with patch(f"{_REGISTRY}.serialize_all_settings", side_effect=OSError("disk")):
|
||||
resp = admin.get("/api/settings")
|
||||
|
||||
assert resp.status_code == 500
|
||||
assert resp.json == {"error": "disk"}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# GET /api/settings/<tab>
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_get_tab_returns_the_serialized_tab(admin):
|
||||
tab = object()
|
||||
with (
|
||||
patch(f"{_REGISTRY}.get_settings_tab", return_value=tab) as get_tab,
|
||||
patch(f"{_REGISTRY}.serialize_tab", return_value={"name": "general"}) as serialize,
|
||||
):
|
||||
resp = admin.get("/api/settings/general")
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json == {"name": "general"}
|
||||
get_tab.assert_called_once_with("general")
|
||||
serialize.assert_called_once_with(tab, include_values=True)
|
||||
|
||||
|
||||
def test_get_tab_is_a_404_for_an_unknown_tab(admin):
|
||||
with patch(f"{_REGISTRY}.get_settings_tab", return_value=None):
|
||||
resp = admin.get("/api/settings/nope")
|
||||
|
||||
assert resp.status_code == 404
|
||||
assert resp.json == {"error": "Unknown settings tab: nope"}
|
||||
|
||||
|
||||
def test_get_tab_reports_an_operational_error(admin):
|
||||
with patch(f"{_REGISTRY}.get_settings_tab", side_effect=RuntimeError("boom")):
|
||||
resp = admin.get("/api/settings/general")
|
||||
|
||||
assert resp.status_code == 500
|
||||
assert resp.json == {"error": "boom"}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# PUT /api/settings/<tab>
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_update_tab_saves_the_values_and_returns_the_result(admin):
|
||||
result = {"success": True, "message": "Saved", "updated": ["A"]}
|
||||
with (
|
||||
patch(f"{_REGISTRY}.get_settings_tab", return_value=object()),
|
||||
patch(f"{_REGISTRY}.update_settings", return_value=result) as update,
|
||||
):
|
||||
resp = admin.put("/api/settings/general", json={"A": 1})
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json == result
|
||||
update.assert_called_once_with("general", {"A": 1})
|
||||
|
||||
|
||||
def test_update_tab_returns_400_with_the_result_when_it_fails(admin):
|
||||
result = {"success": False, "message": "Invalid", "errors": ["A"]}
|
||||
with (
|
||||
patch(f"{_REGISTRY}.get_settings_tab", return_value=object()),
|
||||
patch(f"{_REGISTRY}.update_settings", return_value=result),
|
||||
):
|
||||
resp = admin.put("/api/settings/general", json={"A": 1})
|
||||
|
||||
assert resp.status_code == 400
|
||||
assert resp.json == result
|
||||
|
||||
|
||||
def test_update_tab_with_no_values_succeeds_without_saving(admin):
|
||||
with (
|
||||
patch(f"{_REGISTRY}.get_settings_tab", return_value=object()),
|
||||
patch(f"{_REGISTRY}.update_settings") as update,
|
||||
):
|
||||
resp = admin.put("/api/settings/general", json={})
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json == {"success": True, "message": "No changes to save", "updated": []}
|
||||
update.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("body", [[1, 2], "text", 5])
|
||||
def test_update_tab_needs_a_json_object(admin, body):
|
||||
with (
|
||||
patch(f"{_REGISTRY}.get_settings_tab", return_value=object()),
|
||||
patch(f"{_REGISTRY}.update_settings") as update,
|
||||
):
|
||||
resp = admin.put("/api/settings/general", json=body)
|
||||
|
||||
assert resp.status_code == 400
|
||||
assert resp.json == {"error": "Request body must be a JSON object"}
|
||||
update.assert_not_called()
|
||||
|
||||
|
||||
def test_update_tab_without_a_body_is_a_400(admin):
|
||||
with (
|
||||
patch(f"{_REGISTRY}.get_settings_tab", return_value=object()),
|
||||
patch(f"{_REGISTRY}.update_settings") as update,
|
||||
):
|
||||
resp = admin.put("/api/settings/general")
|
||||
|
||||
assert resp.status_code == 400
|
||||
update.assert_not_called()
|
||||
|
||||
|
||||
def test_update_tab_is_a_404_for_an_unknown_tab_before_it_reads_the_body(admin):
|
||||
with (
|
||||
patch(f"{_REGISTRY}.get_settings_tab", return_value=None),
|
||||
patch(f"{_REGISTRY}.update_settings") as update,
|
||||
):
|
||||
resp = admin.put("/api/settings/nope", json={"A": 1})
|
||||
|
||||
assert resp.status_code == 404
|
||||
assert resp.json == {"error": "Unknown settings tab: nope"}
|
||||
update.assert_not_called()
|
||||
|
||||
|
||||
def test_update_tab_reports_an_operational_error(admin):
|
||||
with (
|
||||
patch(f"{_REGISTRY}.get_settings_tab", return_value=object()),
|
||||
patch(f"{_REGISTRY}.update_settings", side_effect=OSError("disk")),
|
||||
):
|
||||
resp = admin.put("/api/settings/general", json={"A": 1})
|
||||
|
||||
assert resp.status_code == 500
|
||||
assert resp.json == {"error": "disk"}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# POST /api/settings/<tab>/action/<key>
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_action_runs_with_the_unsaved_form_values(admin):
|
||||
with patch(f"{_REGISTRY}.execute_action", return_value={"success": True}) as run:
|
||||
resp = admin.post("/api/settings/general/action/test_it", json={"A": "typed"})
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json == {"success": True}
|
||||
run.assert_called_once_with("general", "test_it", {"A": "typed"})
|
||||
|
||||
|
||||
def test_action_without_a_body_runs_with_no_values(admin):
|
||||
with patch(f"{_REGISTRY}.execute_action", return_value={"success": True}) as run:
|
||||
resp = admin.post("/api/settings/general/action/test_it")
|
||||
|
||||
assert resp.status_code == 200
|
||||
run.assert_called_once_with("general", "test_it", {})
|
||||
|
||||
|
||||
def test_action_returns_400_with_the_result_when_it_fails(admin):
|
||||
result = {"success": False, "message": "Could not connect"}
|
||||
with patch(f"{_REGISTRY}.execute_action", return_value=result):
|
||||
resp = admin.post("/api/settings/general/action/test_it", json={})
|
||||
|
||||
assert resp.status_code == 400
|
||||
assert resp.json == result
|
||||
|
||||
|
||||
def test_action_reports_an_operational_error(admin):
|
||||
with patch(f"{_REGISTRY}.execute_action", side_effect=RuntimeError("boom")):
|
||||
resp = admin.post("/api/settings/general/action/test_it", json={})
|
||||
|
||||
assert resp.status_code == 500
|
||||
assert resp.json == {"error": "boom"}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Auth
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
_ROUTES = [
|
||||
("get", "/api/settings"),
|
||||
("get", "/api/settings/general"),
|
||||
("put", "/api/settings/general"),
|
||||
("post", "/api/settings/general/action/test_it"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("method", "path"), _ROUTES)
|
||||
def test_settings_need_a_session_when_auth_is_on(client, builtin_auth, method, path):
|
||||
resp = getattr(client, method)(path, json={"A": 1})
|
||||
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("method", "path"), _ROUTES)
|
||||
def test_settings_are_admin_only_for_a_signed_in_user(client, builtin_auth, method, path):
|
||||
_sign_in(client, is_admin=False)
|
||||
with (
|
||||
patch(f"{_REGISTRY}.serialize_all_settings") as serialize_all,
|
||||
patch(f"{_REGISTRY}.get_settings_tab") as get_tab,
|
||||
patch(f"{_REGISTRY}.update_settings") as update,
|
||||
patch(f"{_REGISTRY}.execute_action") as run,
|
||||
):
|
||||
resp = getattr(client, method)(path, json={"A": 1})
|
||||
|
||||
assert resp.status_code == 403
|
||||
assert resp.json == {"error": "Admin access required"}
|
||||
for mock in (serialize_all, get_tab, update, run):
|
||||
mock.assert_not_called()
|
||||
|
||||
|
||||
def test_settings_are_open_when_no_auth_is_configured(main_module, client):
|
||||
with (
|
||||
patch.object(main_module, "get_auth_mode", return_value="none"),
|
||||
patch(f"{_REGISTRY}.serialize_all_settings", return_value={"tabs": []}),
|
||||
):
|
||||
resp = client.get("/api/settings")
|
||||
|
||||
assert resp.status_code == 200
|
||||
Reference in new issue
Block a user