diff --git a/shelfmark/core/settings_routes.py b/shelfmark/core/settings_routes.py new file mode 100644 index 00000000..7d6ef468 --- /dev/null +++ b/shelfmark/core/settings_routes.py @@ -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/", 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/", 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//action/", 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 diff --git a/shelfmark/main.py b/shelfmark/main.py index 099951be..cfbbf209 100644 --- a/shelfmark/main.py +++ b/shelfmark/main.py @@ -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/", 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/", 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//action/", 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) # ============================================================================= diff --git a/tests/core/test_settings_routes.py b/tests/core/test_settings_routes.py new file mode 100644 index 00000000..5f45f85d --- /dev/null +++ b/tests/core/test_settings_routes.py @@ -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/ +# --------------------------------------------------------------------------- # + + +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/ +# --------------------------------------------------------------------------- # + + +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//action/ +# --------------------------------------------------------------------------- # + + +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