mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-11-10 12:38:32 +00:00
splitsec2
9d8e7eaad2
Now that the route moves have landed, this takes care of the code scanning alerts they brought along. Sixteen places in `queue_routes.py`, `release_routes.py`, `metadata_routes.py` and `system_routes.py` handed `str(e)` of an unexpected OSError, RuntimeError, TypeError or sqlite3 error back as the 500 body. It can carry file paths, URLs with tokens or driver text. Same idea as #1558 and #1562. Each now returns one fixed sentence ("Could not cancel the download", "Could not read the queue order", "Metadata search failed" and so on) and the full error and trace stay in the log. The routes covered are the cover proxy, cancel, retry, priority, reorder, queue order and active downloads; release download, search, source list and source record; the metadata providers, config, search and book routes; and the debug endpoint. I left alone the messages that are written for the person using the app: what a metadata provider raises as a ValueError or RuntimeError ("Hardcover API key is invalid"), `SourceUnavailableError`, and the reason `queue_release` gives. CodeQL still flags those, as I read it those are the ones that should be dismissed as intended, but that is your call. The same pattern is in the settings and onboarding routes, which aren't flagged, so I didn't touch them. There are 15 new tests, one per fixed route except the debug endpoint, which only exists in debug mode. All 15 fail without the change. Two queue tests that asserted the old text now assert the fixed message. The full suite, ruff and basedpyright are clean. @calibrain I pushed through the refactors keeping them the same functionality, this addresses "issues" flagged in them, I think some of these were dismissed before, but I gathered them all up here in a PR that you should have a say on as it is changing long standing code. <!-- This is an auto-generated comment: release notes by coderabbit.ai --> <!-- coderabbit-ai-generated:v1 --> ## Summary by CodeRabbit * **Bug Fixes** * Error responses now show generic messages instead of operational exception details across metadata, queue, release, and debug endpoints. HTTP status codes and error logging remain unchanged. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
496 lines
21 KiB
Python
496 lines
21 KiB
Python
"""Characterization tests for the queue routes, written before they moved out of main.py.
|
|
|
|
They pin what the routes do today: the status codes, the bodies and who may do what. The
|
|
guardrail tests cover the owner and non-owner cases for cancel, retry, priority and reorder,
|
|
so these fill in the rest: the local download route, the admin paths, the validation errors
|
|
and the failure responses.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import uuid
|
|
import zipfile
|
|
from io import BytesIO
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from shelfmark.core.models import DownloadTask
|
|
from shelfmark.core.viewer_redaction import viewer_download_id
|
|
|
|
|
|
@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 _login(client, *, user_id="alice", db_user_id: int | None = 7, is_admin=False) -> None:
|
|
with client.session_transaction() as sess:
|
|
sess["user_id"] = user_id
|
|
sess["is_admin"] = is_admin
|
|
if db_user_id is not None:
|
|
sess["db_user_id"] = db_user_id
|
|
|
|
|
|
def _create_user(main_module, *, prefix: str) -> dict:
|
|
return main_module.user_db.create_user(username=f"{prefix}-{uuid.uuid4().hex[:8]}", role="user")
|
|
|
|
|
|
def _task(task_id="task-1", *, user=None, request_id=None) -> DownloadTask:
|
|
return DownloadTask(
|
|
task_id=task_id,
|
|
source="direct_download",
|
|
title="A Book",
|
|
author="An Author",
|
|
user_id=user["id"] if user else None,
|
|
username=user["username"] if user else None,
|
|
request_id=request_id,
|
|
)
|
|
|
|
|
|
def _builtin(main_module):
|
|
return patch.object(main_module, "get_auth_mode", return_value="builtin")
|
|
|
|
|
|
class TestLocalDownload:
|
|
def test_missing_id_is_a_400(self, main_module, client):
|
|
_login(client, is_admin=True)
|
|
with _builtin(main_module):
|
|
resp = client.get("/api/localdownload")
|
|
|
|
assert resp.status_code == 400
|
|
assert resp.get_json() == {"error": "No book ID provided"}
|
|
|
|
def test_requires_a_login(self, main_module, client):
|
|
with _builtin(main_module):
|
|
resp = client.get("/api/localdownload?id=task-1")
|
|
|
|
assert resp.status_code == 401
|
|
|
|
def test_unknown_book_is_a_404_when_history_has_nothing(self, main_module, client):
|
|
_login(client, is_admin=True)
|
|
history = MagicMock()
|
|
history.get_by_task_id.return_value = None
|
|
with _builtin(main_module):
|
|
with patch.object(main_module.backend, "get_book_paths", return_value=([], None)):
|
|
with patch.object(main_module, "download_history_service", history):
|
|
resp = client.get("/api/localdownload?id=gone")
|
|
|
|
assert resp.status_code == 404
|
|
assert resp.get_json() == {"error": "File not found"}
|
|
|
|
def test_admin_gets_the_file_as_an_attachment(self, main_module, client, tmp_path):
|
|
book = tmp_path / "book.epub"
|
|
book.write_bytes(b"epub-bytes")
|
|
_login(client, is_admin=True)
|
|
task = _task()
|
|
with _builtin(main_module):
|
|
with patch.object(
|
|
main_module.backend, "get_book_paths", return_value=([str(book)], task)
|
|
):
|
|
resp = client.get("/api/localdownload?id=task-1")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.data == b"epub-bytes"
|
|
assert resp.headers["Content-Disposition"].startswith("attachment")
|
|
|
|
def test_owner_gets_the_file(self, main_module, client, tmp_path):
|
|
book = tmp_path / "book.epub"
|
|
book.write_bytes(b"mine")
|
|
user = _create_user(main_module, prefix="reader")
|
|
_login(client, user_id=user["username"], db_user_id=user["id"])
|
|
with _builtin(main_module):
|
|
with patch.object(
|
|
main_module.backend,
|
|
"get_book_paths",
|
|
return_value=([str(book)], _task(user=user)),
|
|
):
|
|
resp = client.get("/api/localdownload?id=task-1")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.data == b"mine"
|
|
|
|
def test_non_owner_gets_a_404_not_a_403(self, main_module, client, tmp_path):
|
|
book = tmp_path / "book.epub"
|
|
book.write_bytes(b"theirs")
|
|
owner = _create_user(main_module, prefix="owner")
|
|
other = _create_user(main_module, prefix="other")
|
|
_login(client, user_id=other["username"], db_user_id=other["id"])
|
|
with _builtin(main_module):
|
|
with patch.object(
|
|
main_module.backend,
|
|
"get_book_paths",
|
|
return_value=([str(book)], _task(user=owner)),
|
|
):
|
|
resp = client.get("/api/localdownload?id=task-1")
|
|
|
|
assert resp.status_code == 404
|
|
assert resp.get_json() == {"error": "File not found"}
|
|
|
|
def test_several_files_come_back_as_one_zip(self, main_module, client, tmp_path):
|
|
first = tmp_path / "01.mp3"
|
|
second = tmp_path / "02.mp3"
|
|
first.write_bytes(b"one")
|
|
second.write_bytes(b"two")
|
|
_login(client, is_admin=True)
|
|
with _builtin(main_module):
|
|
with patch.object(
|
|
main_module.backend,
|
|
"get_book_paths",
|
|
return_value=([str(first), str(second)], _task()),
|
|
):
|
|
resp = client.get("/api/localdownload?id=task-1")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.mimetype == "application/zip"
|
|
assert resp.headers["Content-Disposition"].startswith("attachment")
|
|
with zipfile.ZipFile(BytesIO(resp.data)) as archive:
|
|
assert sorted(archive.namelist()) == ["01.mp3", "02.mp3"]
|
|
|
|
def test_a_finished_download_is_served_from_history_to_its_owner(
|
|
self, main_module, client, tmp_path
|
|
):
|
|
book = tmp_path / "old.epub"
|
|
book.write_bytes(b"from-history")
|
|
user = _create_user(main_module, prefix="reader")
|
|
_login(client, user_id=user["username"], db_user_id=user["id"])
|
|
history = MagicMock()
|
|
history.get_by_task_id.return_value = {"user_id": user["id"], "download_path": str(book)}
|
|
with _builtin(main_module):
|
|
with patch.object(main_module.backend, "get_book_paths", return_value=([], None)):
|
|
with patch.object(main_module, "download_history_service", history):
|
|
with patch.object(
|
|
main_module.DownloadHistoryService,
|
|
"_resolve_existing_download_path",
|
|
return_value=str(book),
|
|
):
|
|
resp = client.get("/api/localdownload?id=task-1")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.data == b"from-history"
|
|
|
|
def test_a_history_entry_of_someone_else_is_a_404(self, main_module, client, tmp_path):
|
|
book = tmp_path / "old.epub"
|
|
book.write_bytes(b"not-yours")
|
|
user = _create_user(main_module, prefix="reader")
|
|
_login(client, user_id=user["username"], db_user_id=user["id"])
|
|
history = MagicMock()
|
|
history.get_by_task_id.return_value = {
|
|
"user_id": user["id"] + 1000,
|
|
"download_path": str(book),
|
|
}
|
|
with _builtin(main_module):
|
|
with patch.object(main_module.backend, "get_book_paths", return_value=([], None)):
|
|
with patch.object(main_module, "download_history_service", history):
|
|
resp = client.get("/api/localdownload?id=task-1")
|
|
|
|
assert resp.status_code == 404
|
|
|
|
def test_a_read_failure_returns_a_fixed_message(self, main_module, client):
|
|
_login(client, is_admin=True)
|
|
with _builtin(main_module):
|
|
with patch.object(
|
|
main_module.backend,
|
|
"get_book_paths",
|
|
side_effect=OSError("/srv/secret/path.epub is unreadable"),
|
|
):
|
|
resp = client.get("/api/localdownload?id=task-1")
|
|
|
|
assert resp.status_code == 500
|
|
assert resp.get_json() == {"error": "Could not read the file"}
|
|
|
|
|
|
class TestCancel:
|
|
def test_unknown_task_is_a_404(self, main_module, client):
|
|
_login(client, is_admin=True)
|
|
with _builtin(main_module):
|
|
with patch.object(main_module.backend.book_queue, "get_task", return_value=None):
|
|
resp = client.delete("/api/download/nope/cancel")
|
|
|
|
assert resp.status_code == 404
|
|
assert resp.get_json() == {"error": "Failed to cancel download or book not found"}
|
|
|
|
def test_admin_can_cancel_anyones_download(self, main_module, client):
|
|
owner = _create_user(main_module, prefix="owner")
|
|
_login(client, user_id="root", db_user_id=None, is_admin=True)
|
|
with _builtin(main_module):
|
|
with patch.object(
|
|
main_module.backend.book_queue, "get_task", return_value=_task(user=owner)
|
|
):
|
|
with patch.object(
|
|
main_module.backend, "cancel_download", return_value=True
|
|
) as cancel:
|
|
resp = client.delete("/api/download/task-1/cancel")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == {"status": "cancelled", "book_id": "task-1"}
|
|
cancel.assert_called_once_with("task-1")
|
|
|
|
def test_a_cancel_the_backend_refuses_is_a_404(self, main_module, client):
|
|
_login(client, is_admin=True)
|
|
with _builtin(main_module):
|
|
with patch.object(main_module.backend.book_queue, "get_task", return_value=_task()):
|
|
with patch.object(main_module.backend, "cancel_download", return_value=False):
|
|
resp = client.delete("/api/download/task-1/cancel")
|
|
|
|
assert resp.status_code == 404
|
|
|
|
def test_a_user_without_a_database_identity_is_a_403(self, main_module, client):
|
|
_login(client, user_id="ghost", db_user_id=None)
|
|
with _builtin(main_module):
|
|
with patch.object(main_module.backend.book_queue, "get_task", return_value=_task()):
|
|
resp = client.delete("/api/download/task-1/cancel")
|
|
|
|
assert resp.status_code == 403
|
|
assert resp.get_json() == {
|
|
"error": "User identity unavailable",
|
|
"code": "user_identity_unavailable",
|
|
}
|
|
|
|
def test_a_failure_in_the_backend_is_a_500_with_a_fixed_message(self, main_module, client):
|
|
_login(client, is_admin=True)
|
|
with _builtin(main_module):
|
|
with patch.object(
|
|
main_module.backend.book_queue, "get_task", side_effect=RuntimeError("queue broke")
|
|
):
|
|
resp = client.delete("/api/download/task-1/cancel")
|
|
|
|
assert resp.status_code == 500
|
|
assert resp.get_json() == {"error": "Could not cancel the download"}
|
|
|
|
def test_the_opaque_id_a_viewer_sees_cancels_the_real_task(self, main_module, client):
|
|
user = _create_user(main_module, prefix="reader")
|
|
_login(client, user_id=user["username"], db_user_id=user["id"])
|
|
real_id = "21:https://tracker.example/t/1"
|
|
viewer_id = viewer_download_id(real_id, 5)
|
|
assert viewer_id != real_id
|
|
task = _task(real_id, user=user, request_id=None)
|
|
live = MagicMock()
|
|
live.get_status.return_value = {
|
|
"queued": {real_id: _task(real_id, user=user, request_id=5)}
|
|
}
|
|
with _builtin(main_module):
|
|
with patch.object(main_module.backend, "book_queue", live):
|
|
live.get_task.return_value = task
|
|
with patch.object(main_module, "download_history_service", None):
|
|
with patch.object(
|
|
main_module.backend, "cancel_download", return_value=True
|
|
) as cancel:
|
|
resp = client.delete(f"/api/download/{viewer_id}/cancel")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == {"status": "cancelled", "book_id": viewer_id}
|
|
cancel.assert_called_once_with(real_id)
|
|
|
|
|
|
class TestPriority:
|
|
def test_a_missing_body_is_a_400(self, main_module, client):
|
|
_login(client, is_admin=True)
|
|
with _builtin(main_module):
|
|
resp = client.put("/api/queue/task-1/priority", json={})
|
|
|
|
assert resp.status_code == 400
|
|
assert resp.get_json() == {"error": "Priority not provided"}
|
|
|
|
def test_a_priority_that_is_not_a_number_is_a_400(self, main_module, client):
|
|
_login(client, is_admin=True)
|
|
with _builtin(main_module):
|
|
resp = client.put("/api/queue/task-1/priority", json={"priority": "high"})
|
|
|
|
assert resp.status_code == 400
|
|
assert resp.get_json() == {"error": "Invalid priority value"}
|
|
|
|
def test_a_user_without_a_database_identity_is_a_403(self, main_module, client):
|
|
_login(client, user_id="ghost", db_user_id=None)
|
|
with _builtin(main_module):
|
|
resp = client.put("/api/queue/task-1/priority", json={"priority": 1})
|
|
|
|
assert resp.status_code == 403
|
|
assert resp.get_json()["code"] == "user_identity_unavailable"
|
|
|
|
def test_unknown_task_is_a_404(self, main_module, client):
|
|
_login(client, is_admin=True)
|
|
with _builtin(main_module):
|
|
with patch.object(main_module.backend.book_queue, "get_task", return_value=None):
|
|
resp = client.put("/api/queue/nope/priority", json={"priority": 1})
|
|
|
|
assert resp.status_code == 404
|
|
assert resp.get_json() == {"error": "Failed to update priority or book not found"}
|
|
|
|
def test_admin_can_set_anyones_priority(self, main_module, client):
|
|
owner = _create_user(main_module, prefix="owner")
|
|
_login(client, user_id="root", db_user_id=None, is_admin=True)
|
|
with _builtin(main_module):
|
|
with patch.object(
|
|
main_module.backend.book_queue, "get_task", return_value=_task(user=owner)
|
|
):
|
|
with patch.object(
|
|
main_module.backend, "set_book_priority", return_value=True
|
|
) as set_priority:
|
|
resp = client.put("/api/queue/task-1/priority", json={"priority": "3"})
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == {"status": "updated", "book_id": "task-1", "priority": 3}
|
|
set_priority.assert_called_once_with("task-1", 3)
|
|
|
|
|
|
class TestReorder:
|
|
def test_a_missing_body_is_a_400(self, main_module, client):
|
|
_login(client, is_admin=True)
|
|
with _builtin(main_module):
|
|
resp = client.post("/api/queue/reorder", json={})
|
|
|
|
assert resp.status_code == 400
|
|
assert resp.get_json() == {"error": "book_priorities not provided"}
|
|
|
|
def test_priorities_that_are_not_a_dict_are_a_400(self, main_module, client):
|
|
_login(client, is_admin=True)
|
|
with _builtin(main_module):
|
|
resp = client.post("/api/queue/reorder", json={"book_priorities": [1, 2]})
|
|
|
|
assert resp.status_code == 400
|
|
assert resp.get_json() == {"error": "book_priorities must be a dictionary"}
|
|
|
|
def test_a_priority_that_is_not_an_integer_names_the_book(self, main_module, client):
|
|
_login(client, is_admin=True)
|
|
with _builtin(main_module):
|
|
resp = client.post("/api/queue/reorder", json={"book_priorities": {"a": "1"}})
|
|
|
|
assert resp.status_code == 400
|
|
assert resp.get_json() == {"error": "Invalid priority for book a"}
|
|
|
|
def test_admin_reorders_everything_it_names(self, main_module, client):
|
|
_login(client, user_id="root", db_user_id=None, is_admin=True)
|
|
with _builtin(main_module):
|
|
with patch.object(main_module.backend, "reorder_queue", return_value=True) as reorder:
|
|
resp = client.post("/api/queue/reorder", json={"book_priorities": {"a": 0, "b": 1}})
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == {"status": "reordered", "updated_count": 2}
|
|
reorder.assert_called_once_with({"a": 0, "b": 1})
|
|
|
|
def test_a_backend_that_refuses_is_a_500(self, main_module, client):
|
|
_login(client, is_admin=True)
|
|
with _builtin(main_module):
|
|
with patch.object(main_module.backend, "reorder_queue", return_value=False):
|
|
resp = client.post("/api/queue/reorder", json={"book_priorities": {"a": 0}})
|
|
|
|
assert resp.status_code == 500
|
|
assert resp.get_json() == {"error": "Failed to reorder queue"}
|
|
|
|
def test_a_non_admin_skips_ids_that_name_no_task(self, main_module, client):
|
|
user = _create_user(main_module, prefix="reader")
|
|
_login(client, user_id=user["username"], db_user_id=user["id"])
|
|
mine = _task("mine", user=user)
|
|
with _builtin(main_module):
|
|
with patch.object(
|
|
main_module.backend.book_queue,
|
|
"get_task",
|
|
side_effect=lambda task_id: {"mine": mine}.get(task_id),
|
|
):
|
|
with patch.object(
|
|
main_module.backend, "reorder_queue", return_value=True
|
|
) as reorder:
|
|
resp = client.post(
|
|
"/api/queue/reorder", json={"book_priorities": {"mine": 0, "gone": 1}}
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == {"status": "reordered", "updated_count": 1}
|
|
reorder.assert_called_once_with({"mine": 0})
|
|
|
|
|
|
class TestQueueViews:
|
|
def test_admin_sees_the_whole_queue_order(self, main_module, client):
|
|
_login(client, user_id="root", db_user_id=None, is_admin=True)
|
|
order = [{"id": "a", "priority": 0}, {"id": "b", "priority": 1}]
|
|
with _builtin(main_module):
|
|
with patch.object(main_module.backend, "get_queue_order", return_value=order):
|
|
resp = client.get("/api/queue/order")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == {"queue": order}
|
|
|
|
def test_admin_sees_every_active_download(self, main_module, client):
|
|
_login(client, user_id="root", db_user_id=None, is_admin=True)
|
|
with _builtin(main_module):
|
|
with patch.object(main_module.backend, "get_active_downloads", return_value=["a", "b"]):
|
|
resp = client.get("/api/downloads/active")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == {"active_downloads": ["a", "b"]}
|
|
|
|
def test_both_views_refuse_a_user_without_a_database_identity(self, main_module, client):
|
|
_login(client, user_id="ghost", db_user_id=None)
|
|
with _builtin(main_module):
|
|
order = client.get("/api/queue/order")
|
|
active = client.get("/api/downloads/active")
|
|
|
|
assert order.status_code == active.status_code == 403
|
|
assert order.get_json()["code"] == active.get_json()["code"] == "user_identity_unavailable"
|
|
|
|
def test_a_failure_is_a_500_with_a_fixed_message(self, main_module, client):
|
|
_login(client, is_admin=True)
|
|
with _builtin(main_module):
|
|
with patch.object(
|
|
main_module.backend, "get_queue_order", side_effect=RuntimeError("order broke")
|
|
):
|
|
resp = client.get("/api/queue/order")
|
|
|
|
assert resp.status_code == 500
|
|
assert resp.get_json() == {"error": "Could not read the queue order"}
|
|
|
|
|
|
class TestStatus:
|
|
def test_a_user_without_a_database_identity_gets_an_empty_object(self, main_module, client):
|
|
_login(client, user_id="ghost", db_user_id=None)
|
|
with _builtin(main_module):
|
|
resp = client.get("/api/status")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.get_json() == {}
|
|
|
|
def test_delivery_states_are_synced_and_announced_for_the_requests_that_changed(
|
|
self, main_module, client
|
|
):
|
|
_login(client, user_id="root", db_user_id=None, is_admin=True)
|
|
changed = [
|
|
{
|
|
"id": 4,
|
|
"status": "fulfilled",
|
|
"delivery_state": "complete",
|
|
"user_id": 9,
|
|
"book_data": {"title": "Dune"},
|
|
}
|
|
]
|
|
with _builtin(main_module):
|
|
with patch.object(main_module.backend, "queue_status", return_value={}):
|
|
with patch(
|
|
"shelfmark.core.queue_routes.sync_delivery_states_from_queue_status",
|
|
return_value=changed,
|
|
):
|
|
with patch.object(main_module, "emit_ws_event") as emit:
|
|
resp = client.get("/api/status")
|
|
|
|
assert resp.status_code == 200
|
|
rooms = {call.kwargs["room"] for call in emit.call_args_list}
|
|
assert rooms == {"user_9", "admins"}
|
|
payload = emit.call_args_list[0].kwargs["payload"]
|
|
assert payload == {
|
|
"request_id": 4,
|
|
"status": "fulfilled",
|
|
"delivery_state": "complete",
|
|
"title": "Dune",
|
|
}
|