starred/shelfmark
1
0 Fork 0
mirror of https://github.com/calibrain/shelfmark.git synced 2026-11-10 12:28:27 +00:00
Code Issues Packages Projects Releases 10 Wiki Activity

refactor: move the system routes into core/system_routes.py (#1624)

Browse Source 
The system routes are the next group from the plan in Discussion #1536:
`/api/config`, `/api/health`, `/api/debug`, `/api/restart` and the
routes that serve the frontend (`/`, the SPA catch-all, `/assets/...`,
the logo, the favicons and `theme-init.js`) move from `main.py` to
`core/system_routes.py`, registered with `register_system_routes(app,
login_required, ...)`. `main.py` is about 250 lines shorter.

Same two-commit layout as before. The first commit adds 14 tests for the
frontend-serving routes and the `/api/config` error paths, which had
none, and they pass on current `main`. The second commit is the move,
and no existing test changed.

The helpers only these routes use moved with them (`admin_required`,
built on the `login_required` that is passed in, plus `_base_href` and
`_serve_index_html`). What `main.py` still owns, the base path, the
frontend folder, the auth mode and the session user id, is passed in as
callables evaluated per request, so the tests that patch them on `main`
still work. The debug routes are only registered when debug is on, as
before, with the stop-GUI and run-script hooks handed in.

The thing to look at is the SPA catch-all, which is now registered
earlier in `main.py` than it used to be (the old comment said it must be
last). Werkzeug picks the most specific rule whatever the registration
order, and I checked it: all 76 rule and method pairs in the URL map
still match their own endpoint, on `main` and on this branch, and an
unknown `/api/...` path still gets the JSON 404.

The Flask URL map is identical before and after, all 77 rules, and with
debug on (79). The full suite, ruff and basedpyright are clean. I also
ran the app from this branch: health, config, the logo, the favicon and
the theme script answer as before, a client-side path like
`/library/books` gets the index page, an unknown `/api/` path gets the
JSON 404, and the page loads in a browser. I did not exercise
`/api/restart` or a real debug script run, since nothing tests them.
This commit is contained in:
splitsec2 authored and GitHub committed 2026-11-01 14:12:11 -07:00
1 parent d41fa98749
commit d1e959758d
3 files changed
+466 -270
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

shelfmark/core/system_routes.py
+330
View File
@@ -0,0 +1,330 @@
"""System routes: the config and health endpoints, the debug tools, the frontend files and the SPA fallback."""
from __future__ import annotations
import os
import sqlite3
import subprocess
from functools import wraps
from pathlib import Path
from typing import TYPE_CHECKING, Any, NamedTuple, NoReturn
from flask import jsonify, send_file, send_from_directory, session
from werkzeug.wrappers import Response
from shelfmark.config.env import BUILD_VERSION, RELEASE_VERSION
from shelfmark.config.settings import _SUPPORTED_BOOK_LANGUAGE
from shelfmark.core import search_deadline
from shelfmark.core.config import config as app_config
from shelfmark.core.logger import setup_logger
from shelfmark.core.request_helpers import normalize_optional_text
from shelfmark.download import orchestrator as backend
if TYPE_CHECKING:
from collections.abc import Callable
from flask import Flask
logger = setup_logger(__name__)
_OPERATIONAL_ERRORS = (OSError, RuntimeError, TypeError, ValueError, sqlite3.Error)
_IMPORT_OPERATIONAL_ERRORS = (ImportError, *_OPERATIONAL_ERRORS)
_BASE_TAG = '<base href="/" data-shelfmark-base />'
def _raise_runtime_error(message: str) -> NoReturn:
raise RuntimeError(message)
class DebugHooks(NamedTuple):
"""What the DEBUG-only routes need from the app: stop the GUI, then run the debug script.
``run_script`` waits a second for the GUI to stop before it starts the script.
"""
stop_gui: Callable[[], None]
run_script: Callable[[], subprocess.CompletedProcess[str]]
def register_system_routes(
app: Flask,
login_required: Callable[..., Callable[..., Response | tuple[Response, int]]],
*,
resolve_auth_mode: Callable[[], str],
resolve_db_user_id: Callable[[], int | None],
get_base_path: Callable[[], str],
get_frontend_dist: Callable[[], Path],
debug_hooks: DebugHooks | None,
) -> None:
"""Register the system routes, the frontend files and the catch-all for client-side routes.
``login_required`` is the app's decorator. ``resolve_auth_mode``, ``resolve_db_user_id``,
``get_base_path`` and ``get_frontend_dist`` are called per request so a runtime patch of any of them still takes
effect. ``/api/debug`` and ``/api/restart`` are registered only when ``debug_hooks`` is given.
"""
def admin_required(
f: Callable[..., Response | tuple[Response, int]],
) -> Callable[..., Response | tuple[Response, int]]:
"""Require an authenticated admin session for a Flask route.
Builds on `login_required`, so auth mode "none" (no users, the single
operator owns the instance) still has access.
"""
@login_required
@wraps(f)
def decorated_function(*args: object, **kwargs: object) -> Response | tuple[Response, int]:
if resolve_auth_mode() != "none" and not session.get("is_admin", False):
return jsonify({"error": "Admin access required"}), 403
return f(*args, **kwargs)
return decorated_function
def _base_href() -> str:
base_path = get_base_path()
if not base_path:
return "/"
return f"{base_path}/"
def _serve_index_html() -> Response:
"""Serve index.html with an adjusted base tag for subpath deployments."""
frontend_dist = get_frontend_dist()
index_path = frontend_dist / "index.html"
try:
with index_path.open(encoding="utf-8") as handle:
html = handle.read()
except OSError:
return send_from_directory(frontend_dist, "index.html")
if get_base_path() and _BASE_TAG in html:
html = html.replace(_BASE_TAG, f'<base href="{_base_href()}" data-shelfmark-base />', 1)
return Response(html, mimetype="text/html")
# Serve frontend static files
@app.route("/assets/<path:filename>")
def serve_frontend_assets(filename: str) -> Response:
"""Serve static assets from the built frontend."""
return send_from_directory(get_frontend_dist() / "assets", filename)
@app.route("/")
def index() -> Response:
"""Serve the React frontend application.
Authentication is handled by the React app itself.
"""
return _serve_index_html()
@app.route("/theme-init.js")
def theme_init_js() -> Response:
"""Serve the blocking theme-init script."""
return send_from_directory(
get_frontend_dist(), "theme-init.js", mimetype="application/javascript"
)
@app.route("/logo.png")
def logo() -> Response:
"""Serve logo from built frontend assets."""
return send_from_directory(get_frontend_dist(), "logo.png", mimetype="image/png")
@app.route("/favicon.ico")
@app.route("/favico<path:_>")
def favicon(_: Any = None) -> Response:
"""Serve favicon from built frontend assets."""
return send_from_directory(
get_frontend_dist(), "favicon.ico", mimetype="image/vnd.microsoft.icon"
)
@app.route("/api/config", methods=["GET"])
@login_required
def api_config() -> Response | tuple[Response, int]:
"""Get application configuration for frontend.
Uses the dynamic config singleton to ensure settings changes
are reflected without requiring a container restart.
"""
try:
from shelfmark.config.env import _is_config_dir_writable
from shelfmark.core.onboarding import is_onboarding_complete as _get_onboarding_complete
from shelfmark.metadata_providers import (
get_provider_default_sort,
get_provider_search_fields,
get_provider_sort_options,
)
db_user_id = resolve_db_user_id()
search_mode = app_config.get("SEARCH_MODE", "universal", user_id=db_user_id)
default_release_source = app_config.get(
"DEFAULT_RELEASE_SOURCE",
"",
user_id=db_user_id,
)
default_release_source_audiobook = app_config.get(
"DEFAULT_RELEASE_SOURCE_AUDIOBOOK",
"",
user_id=db_user_id,
)
configured_metadata_provider = normalize_optional_text(
app_config.get(
"METADATA_PROVIDER",
"",
user_id=db_user_id,
)
)
_configured_metadata_provider_audiobook = normalize_optional_text(
app_config.get(
"METADATA_PROVIDER_AUDIOBOOK",
"",
user_id=db_user_id,
)
)
metadata_ui_provider = (
configured_metadata_provider or _configured_metadata_provider_audiobook
)
config = {
"calibre_web_url": app_config.get("CALIBRE_WEB_URL", ""),
"audiobook_library_url": app_config.get("AUDIOBOOK_LIBRARY_URL", ""),
"search_page_title": app_config.get("SEARCH_PAGE_TITLE", "Shelfmark"),
"debug": app_config.get("DEBUG", False),
"build_version": BUILD_VERSION,
"release_version": RELEASE_VERSION,
"book_languages": _SUPPORTED_BOOK_LANGUAGE,
"default_language": app_config.get("BOOK_LANGUAGE", ["en"], user_id=db_user_id),
"default_content_type": app_config.get(
"DEFAULT_CONTENT_TYPE", "ebook", user_id=db_user_id
),
"supported_formats": app_config.SUPPORTED_FORMATS,
"supported_audiobook_formats": app_config.SUPPORTED_AUDIOBOOK_FORMATS,
"search_mode": search_mode,
"metadata_sort_options": get_provider_sort_options(metadata_ui_provider),
"metadata_search_fields": get_provider_search_fields(metadata_ui_provider),
"default_release_source": default_release_source,
"default_release_source_audiobook": default_release_source_audiobook,
"show_release_source_links": app_config.get("SHOW_RELEASE_SOURCE_LINKS", True),
"show_combined_selector": app_config.get(
"SHOW_COMBINED_SELECTOR", True, user_id=db_user_id
),
"force_combined_search": app_config.get(
"FORCE_COMBINED_SEARCH", False, user_id=db_user_id
),
"books_output_mode": app_config.get("BOOKS_OUTPUT_MODE", "folder"),
"auto_open_downloads_sidebar": app_config.get("AUTO_OPEN_DOWNLOADS_SIDEBAR", True),
"hardcover_auto_remove_on_download": app_config.get(
"HARDCOVER_AUTO_REMOVE_ON_DOWNLOAD", True
),
"download_to_browser_content_types": app_config.get(
"DOWNLOAD_TO_BROWSER_CONTENT_TYPES",
[],
user_id=db_user_id,
),
# The client must not give up before this budget does. `/api/releases`
# answers a spent budget with a message naming the real cause (a protection
# challenge nobody could solve); a browser that aborted first replaces it
# with a generic network/proxy error and RELEASE_SEARCH_TIMEOUT becomes a
# setting the user can raise with no visible effect. See issue #1285.
"release_search_timeout": search_deadline.budget_seconds(),
"settings_enabled": _is_config_dir_writable(),
"onboarding_complete": _get_onboarding_complete(),
# Default sort orders
"default_sort": app_config.get(
"AA_DEFAULT_SORT", ""
), # For direct mode (Anna's Archive) — empty means use local downloads sort
"metadata_default_sort": get_provider_default_sort(
metadata_ui_provider
), # For universal mode
}
return jsonify(config)
except _IMPORT_OPERATIONAL_ERRORS as e:
logger.error_trace(f"Config error: {e}")
return jsonify({"error": "Could not load the configuration"}), 500
@app.route("/api/health", methods=["GET"])
def api_health() -> Response | tuple[Response, int]:
"""Health check endpoint for container orchestration.
No authentication required.
Returns:
flask.Response: JSON with status "ok" and optional degraded features. Answers 503 with
status "error" when the download coordinator is not running, since nothing is
downloaded without it.
"""
response: dict[str, object] = {"status": "ok"}
# Report degraded features
if not backend.WEBSOCKET_AVAILABLE:
response["degraded"] = {
"websocket": "WebSocket unavailable - real-time updates disabled"
}
coordinator_up = backend.coordinator_running()
response["coordinator"] = "running" if coordinator_up else "stopped"
if not coordinator_up:
response["status"] = "error"
return jsonify(response), 503
return jsonify(response)
if debug_hooks is not None:
@app.route("/api/debug", methods=["GET"])
@admin_required
def debug() -> Response | tuple[Response, int]:
"""Run `/app/genDebug.sh`, generate a debug zip, and return it.
The file is written to `/tmp/shelfmark-debug.zip` before being returned.
"""
try:
logger.info("Debug endpoint called, stopping GUI and generating debug info...")
debug_hooks.stop_gui()
result = debug_hooks.run_script()
if result.returncode != 0:
_raise_runtime_error(f"Debug script failed: {result.stderr}")
logger.info("Debug script executed: %s", result.stdout)
debug_file_path = result.stdout.strip().split("\n")[-1]
if not Path(debug_file_path).exists():
logger.error("Debug zip file not found at: %s", debug_file_path)
return jsonify({"error": "Failed to generate debug information"}), 500
logger.info("Sending debug file: %s", debug_file_path)
return send_file(
debug_file_path,
mimetype="application/zip",
download_name=Path(debug_file_path).name,
as_attachment=True,
)
except subprocess.CalledProcessError as e:
logger.error_trace(
f"Debug script error: {e}, stdout: {e.stdout}, stderr: {e.stderr}"
)
return jsonify({"error": f"Debug script failed: {e.stderr}"}), 500
except _OPERATIONAL_ERRORS as e:
logger.error_trace(f"Debug endpoint error: {e}")
return jsonify({"error": str(e)}), 500
@app.route("/api/restart", methods=["GET"])
@admin_required
def restart() -> Response | tuple[Response, int]:
"""Restart the application."""
os._exit(0)
# Catch-all route for React Router. Werkzeug tries static rules before dynamic ones, so it
# does not have to be registered last.
# This handles client-side routing by serving index.html for any unmatched routes
@app.route("/<path:path>")
def catch_all(path: str) -> Response | tuple[Response, int]:
"""Serve the React app for any route not matched by API endpoints.
This allows React Router to handle client-side routing.
Authentication is handled by the React app itself.
"""
# If the request is for an API endpoint or static file, let it 404
if path.startswith(("api/", "assets/")):
return jsonify({"error": "Resource not found"}), 404
# Otherwise serve the React app
return _serve_index_html()
shelfmark/main.py
+16 -270
View File
@@ -14,7 +14,7 @@ from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable, NoReturn, cast
from urllib.parse import quote
from flask import Flask, g, jsonify, request, send_file, send_from_directory, session
from flask import Flask, g, jsonify, request, send_file, session
from flask_cors import CORS
from flask_socketio import SocketIO, emit
from werkzeug.middleware.proxy_fix import ProxyFix
@@ -23,7 +23,6 @@ from werkzeug.wrappers import Response
from shelfmark.api.websocket import ws_manager
from shelfmark.config.env import (
BUILD_VERSION,
CONFIG_DIR,
CWA_DB_PATH,
DISABLE_LOCAL_AUTH,
@@ -31,7 +30,6 @@ from shelfmark.config.env import (
FLASK_PORT,
HIDE_LOCAL_AUTH,
OIDC_AUTO_REDIRECT,
RELEASE_VERSION,
SESSION_COOKIE_NAME,
SESSION_COOKIE_SECURE_ENV,
SOCKETIO_ALLOWED_ORIGINS,
@@ -40,7 +38,6 @@ from shelfmark.config.env import (
)
from shelfmark.config.security import _migrate_security_settings
from shelfmark.config.settings import (
_SUPPORTED_BOOK_LANGUAGE,
migrate_audiobook_format_settings,
)
from shelfmark.core import api_key as api_key_module # module access lets tests monkeypatch the key
@@ -97,6 +94,7 @@ from shelfmark.core.requests_service import (
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.system_routes import DebugHooks, register_system_routes
from shelfmark.core.user_db import UserDB
from shelfmark.core.utils import AUDIOBOOK_FORMATS, normalize_base_path
from shelfmark.core.viewer_redaction import (
@@ -1102,84 +1100,7 @@ def login_required(
return decorated_function
def admin_required(
f: Callable[..., Response | tuple[Response, int]],
) -> Callable[..., Response | tuple[Response, int]]:
"""Require an authenticated admin session for a Flask route.
Builds on `login_required`, so auth mode "none" (no users, the single
operator owns the instance) still has access.
"""
@login_required
@wraps(f)
def decorated_function(*args: object, **kwargs: object) -> Response | tuple[Response, int]:
if get_auth_mode() != "none" and not session.get("is_admin", False):
return jsonify({"error": "Admin access required"}), 403
return f(*args, **kwargs)
return decorated_function
_BASE_TAG = '<base href="/" data-shelfmark-base />'
def _base_href() -> str:
if not BASE_PATH:
return "/"
return f"{BASE_PATH}/"
def _serve_index_html() -> Response:
"""Serve index.html with an adjusted base tag for subpath deployments."""
index_path = FRONTEND_DIST / "index.html"
try:
with index_path.open(encoding="utf-8") as handle:
html = handle.read()
except OSError:
return send_from_directory(FRONTEND_DIST, "index.html")
if BASE_PATH and _BASE_TAG in html:
html = html.replace(_BASE_TAG, f'<base href="{_base_href()}" data-shelfmark-base />', 1)
return Response(html, mimetype="text/html")
# Serve frontend static files
@app.route("/assets/<path:filename>")
def serve_frontend_assets(filename: str) -> Response:
"""Serve static assets from the built frontend."""
return send_from_directory(FRONTEND_DIST / "assets", filename)
@app.route("/")
def index() -> Response:
"""Serve the React frontend application.
Authentication is handled by the React app itself.
"""
return _serve_index_html()
@app.route("/theme-init.js")
def theme_init_js() -> Response:
"""Serve the blocking theme-init script."""
return send_from_directory(FRONTEND_DIST, "theme-init.js", mimetype="application/javascript")
@app.route("/logo.png")
def logo() -> Response:
"""Serve logo from built frontend assets."""
return send_from_directory(FRONTEND_DIST, "logo.png", mimetype="image/png")
@app.route("/favicon.ico")
@app.route("/favico<path:_>")
def favicon(_: Any = None) -> Response:
"""Serve favicon from built frontend assets."""
return send_from_directory(FRONTEND_DIST, "favicon.ico", mimetype="image/vnd.microsoft.icon")
_debug_hooks: DebugHooks | None = None
if _is_debug_enabled():
import subprocess
@@ -1194,47 +1115,21 @@ if _is_debug_enabled():
def _stop_gui() -> None:
_cleanup_orphan_processes()
@app.route("/api/debug", methods=["GET"])
@admin_required
def debug() -> Response | tuple[Response, int]:
"""Run `/app/genDebug.sh`, generate a debug zip, and return it.
def _run_debug_script() -> subprocess.CompletedProcess[str]:
time.sleep(1)
return subprocess.run(["/app/genDebug.sh"], capture_output=True, text=True, check=True)
The file is written to `/tmp/shelfmark-debug.zip` before being returned.
"""
try:
logger.info("Debug endpoint called, stopping GUI and generating debug info...")
_stop_gui()
time.sleep(1)
result = subprocess.run(
["/app/genDebug.sh"], capture_output=True, text=True, check=True
)
if result.returncode != 0:
_raise_runtime_error(f"Debug script failed: {result.stderr}")
logger.info("Debug script executed: %s", result.stdout)
debug_file_path = result.stdout.strip().split("\n")[-1]
if not Path(debug_file_path).exists():
logger.error("Debug zip file not found at: %s", debug_file_path)
return jsonify({"error": "Failed to generate debug information"}), 500
_debug_hooks = DebugHooks(stop_gui=_stop_gui, run_script=_run_debug_script)
logger.info("Sending debug file: %s", debug_file_path)
return send_file(
debug_file_path,
mimetype="application/zip",
download_name=Path(debug_file_path).name,
as_attachment=True,
)
except subprocess.CalledProcessError as e:
logger.error_trace(f"Debug script error: {e}, stdout: {e.stdout}, stderr: {e.stderr}")
return jsonify({"error": f"Debug script failed: {e.stderr}"}), 500
except _OPERATIONAL_ERRORS as e:
logger.error_trace(f"Debug endpoint error: {e}")
return jsonify({"error": str(e)}), 500
@app.route("/api/restart", methods=["GET"])
@admin_required
def restart() -> Response | tuple[Response, int]:
"""Restart the application."""
os._exit(0)
register_system_routes(
app,
login_required,
resolve_auth_mode=_resolve_auth_mode_for_routes,
resolve_db_user_id=lambda: get_session_db_user_id(session),
get_base_path=lambda: BASE_PATH,
get_frontend_dist=lambda: FRONTEND_DIST,
debug_hooks=_debug_hooks,
)
def _parse_search_filters_from_request() -> SearchFilters:
@@ -1388,112 +1283,6 @@ def api_download_release() -> Response | tuple[Response, int]:
return jsonify({"error": str(e)}), 500
@app.route("/api/config", methods=["GET"])
@login_required
def api_config() -> Response | tuple[Response, int]:
"""Get application configuration for frontend.
Uses the dynamic config singleton to ensure settings changes
are reflected without requiring a container restart.
"""
try:
from shelfmark.config.env import _is_config_dir_writable
from shelfmark.core.onboarding import is_onboarding_complete as _get_onboarding_complete
from shelfmark.metadata_providers import (
get_provider_default_sort,
get_provider_search_fields,
get_provider_sort_options,
)
db_user_id = get_session_db_user_id(session)
search_mode = app_config.get("SEARCH_MODE", "universal", user_id=db_user_id)
default_release_source = app_config.get(
"DEFAULT_RELEASE_SOURCE",
"",
user_id=db_user_id,
)
default_release_source_audiobook = app_config.get(
"DEFAULT_RELEASE_SOURCE_AUDIOBOOK",
"",
user_id=db_user_id,
)
configured_metadata_provider = normalize_optional_text(
app_config.get(
"METADATA_PROVIDER",
"",
user_id=db_user_id,
)
)
_configured_metadata_provider_audiobook = normalize_optional_text(
app_config.get(
"METADATA_PROVIDER_AUDIOBOOK",
"",
user_id=db_user_id,
)
)
metadata_ui_provider = (
configured_metadata_provider or _configured_metadata_provider_audiobook
)
config = {
"calibre_web_url": app_config.get("CALIBRE_WEB_URL", ""),
"audiobook_library_url": app_config.get("AUDIOBOOK_LIBRARY_URL", ""),
"search_page_title": app_config.get("SEARCH_PAGE_TITLE", "Shelfmark"),
"debug": app_config.get("DEBUG", False),
"build_version": BUILD_VERSION,
"release_version": RELEASE_VERSION,
"book_languages": _SUPPORTED_BOOK_LANGUAGE,
"default_language": app_config.get("BOOK_LANGUAGE", ["en"], user_id=db_user_id),
"default_content_type": app_config.get(
"DEFAULT_CONTENT_TYPE", "ebook", user_id=db_user_id
),
"supported_formats": app_config.SUPPORTED_FORMATS,
"supported_audiobook_formats": app_config.SUPPORTED_AUDIOBOOK_FORMATS,
"search_mode": search_mode,
"metadata_sort_options": get_provider_sort_options(metadata_ui_provider),
"metadata_search_fields": get_provider_search_fields(metadata_ui_provider),
"default_release_source": default_release_source,
"default_release_source_audiobook": default_release_source_audiobook,
"show_release_source_links": app_config.get("SHOW_RELEASE_SOURCE_LINKS", True),
"show_combined_selector": app_config.get(
"SHOW_COMBINED_SELECTOR", True, user_id=db_user_id
),
"force_combined_search": app_config.get(
"FORCE_COMBINED_SEARCH", False, user_id=db_user_id
),
"books_output_mode": app_config.get("BOOKS_OUTPUT_MODE", "folder"),
"auto_open_downloads_sidebar": app_config.get("AUTO_OPEN_DOWNLOADS_SIDEBAR", True),
"hardcover_auto_remove_on_download": app_config.get(
"HARDCOVER_AUTO_REMOVE_ON_DOWNLOAD", True
),
"download_to_browser_content_types": app_config.get(
"DOWNLOAD_TO_BROWSER_CONTENT_TYPES",
[],
user_id=db_user_id,
),
# The client must not give up before this budget does. `/api/releases`
# answers a spent budget with a message naming the real cause (a protection
# challenge nobody could solve); a browser that aborted first replaces it
# with a generic network/proxy error and RELEASE_SEARCH_TIMEOUT becomes a
# setting the user can raise with no visible effect. See issue #1285.
"release_search_timeout": search_deadline.budget_seconds(),
"settings_enabled": _is_config_dir_writable(),
"onboarding_complete": _get_onboarding_complete(),
# Default sort orders
"default_sort": app_config.get(
"AA_DEFAULT_SORT", ""
), # For direct mode (Anna's Archive) — empty means use local downloads sort
"metadata_default_sort": get_provider_default_sort(
metadata_ui_provider
), # For universal mode
}
return jsonify(config)
except _IMPORT_OPERATIONAL_ERRORS as e:
logger.error_trace(f"Config error: {e}")
return jsonify({"error": "Could not load the configuration"}), 500
register_stats_routes(
app,
get_user_db=lambda: user_db,
@@ -1501,33 +1290,6 @@ register_stats_routes(
)
@app.route("/api/health", methods=["GET"])
def api_health() -> Response | tuple[Response, int]:
"""Health check endpoint for container orchestration.
No authentication required.
Returns:
flask.Response: JSON with status "ok" and optional degraded features. Answers 503 with
status "error" when the download coordinator is not running, since nothing is
downloaded without it.
"""
response: dict[str, object] = {"status": "ok"}
# Report degraded features
if not backend.WEBSOCKET_AVAILABLE:
response["degraded"] = {"websocket": "WebSocket unavailable - real-time updates disabled"}
coordinator_up = backend.coordinator_running()
response["coordinator"] = "running" if coordinator_up else "stopped"
if not coordinator_up:
response["status"] = "error"
return jsonify(response), 503
return jsonify(response)
def _resolve_status_scope(*, require_authenticated: bool = True) -> tuple[bool, int | None, bool]:
"""Resolve queue-status visibility from session state.
@@ -3080,22 +2842,6 @@ register_settings_routes(app, login_required)
register_onboarding_routes(app, login_required)
# Catch-all route for React Router (must be last)
# This handles client-side routing by serving index.html for any unmatched routes
@app.route("/<path:path>")
def catch_all(path: str) -> Response | tuple[Response, int]:
"""Serve the React app for any route not matched by API endpoints.
This allows React Router to handle client-side routing.
Authentication is handled by the React app itself.
"""
# If the request is for an API endpoint or static file, let it 404
if path.startswith(("api/", "assets/")):
return jsonify({"error": "Resource not found"}), 404
# Otherwise serve the React app
return _serve_index_html()
def _get_request_sid() -> str | None:
"""Return the Socket.IO session id for the active request when available."""
sid = getattr(request, "sid", None)
tests/core/test_system_routes.py
+120
View File
@@ -0,0 +1,120 @@
"""Tests for the frontend-serving routes and the gaps in /api/config coverage."""
from __future__ import annotations
import importlib
from pathlib import Path
from unittest.mock import patch
import pytest
_INDEX = '<html><head><base href="/" data-shelfmark-base /></head><body>app</body></html>'
@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()
@pytest.fixture
def dist(main_module, tmp_path: Path):
"""A throwaway frontend build in place of the real `frontend-dist`."""
(tmp_path / "assets").mkdir()
(tmp_path / "assets" / "app.js").write_text("console.log(1)")
(tmp_path / "index.html").write_text(_INDEX)
(tmp_path / "logo.png").write_bytes(b"\x89PNG")
(tmp_path / "favicon.ico").write_bytes(b"ico")
(tmp_path / "theme-init.js").write_text("// theme")
with (
patch.object(main_module, "FRONTEND_DIST", tmp_path),
patch.object(main_module, "BASE_PATH", ""),
):
yield tmp_path
def test_index_serves_the_app_with_the_default_base_tag(client, dist):
resp = client.get("/")
assert resp.status_code == 200
assert resp.mimetype == "text/html"
assert resp.get_data(as_text=True) == _INDEX
def test_index_rewrites_the_base_tag_for_a_subpath(main_module, client, dist):
with patch.object(main_module, "BASE_PATH", "/books"):
resp = client.get("/")
assert '<base href="/books/" data-shelfmark-base />' in resp.get_data(as_text=True)
def test_index_without_a_frontend_build_is_a_404(client, dist):
(dist / "index.html").unlink()
assert client.get("/").status_code == 404
def test_catch_all_serves_the_app_for_client_side_routes(client, dist):
resp = client.get("/settings/general")
assert resp.status_code == 200
assert resp.get_data(as_text=True) == _INDEX
@pytest.mark.parametrize("path", ["/api/no-such-route", "/api/deeper/no-such-route"])
def test_catch_all_keeps_unknown_api_paths_a_404(client, dist, path):
resp = client.get(path)
assert resp.status_code == 404
assert resp.get_json() == {"error": "Resource not found"}
def test_assets_are_served_from_the_build(client, dist):
resp = client.get("/assets/app.js")
assert resp.status_code == 200
assert resp.get_data(as_text=True) == "console.log(1)"
def test_a_missing_asset_is_a_404_not_the_app(client, dist):
assert client.get("/assets/missing.js").status_code == 404
@pytest.mark.parametrize(
("path", "mimetype"),
[
("/logo.png", "image/png"),
("/favicon.ico", "image/vnd.microsoft.icon"),
("/favicon-32.png", "image/vnd.microsoft.icon"),
("/theme-init.js", "application/javascript"),
],
)
def test_root_files_are_served_with_their_mimetype(client, dist, path, mimetype):
resp = client.get(path)
assert resp.status_code == 200
assert resp.mimetype == mimetype
def test_config_requires_a_session_when_auth_is_on(main_module, client):
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
resp = client.get("/api/config")
assert resp.status_code == 401
def test_config_reports_an_operational_error(client):
with patch("shelfmark.core.onboarding.is_onboarding_complete", side_effect=RuntimeError("x")):
resp = client.get("/api/config")
assert resp.status_code == 500
assert resp.get_json() == {"error": "Could not load the configuration"}