"""API tests for request routes and policy enforcement guards.""" from __future__ import annotations import importlib import uuid from unittest.mock import ANY, patch import pytest from shelfmark.core.notifications import NotificationEvent @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 _set_session(client, *, user_id: str, db_user_id: int | None, is_admin: bool) -> 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 elif "db_user_id" in sess: del sess["db_user_id"] def _create_user(main_module, *, prefix: str, role: str = "user") -> dict: username = f"{prefix}-{uuid.uuid4().hex[:8]}" return main_module.user_db.create_user(username=username, role=role) def _policy( *, requests_enabled: bool = True, default_ebook: str = "download", default_audiobook: str = "download", max_pending_requests_per_user: int = 20, requests_allow_notes: bool = True, rules: list[dict] | None = None, ) -> dict: return { "REQUESTS_ENABLED": requests_enabled, "REQUEST_POLICY_DEFAULT_EBOOK": default_ebook, "REQUEST_POLICY_DEFAULT_AUDIOBOOK": default_audiobook, "MAX_PENDING_REQUESTS_PER_USER": max_pending_requests_per_user, "REQUESTS_ALLOW_NOTES": requests_allow_notes, "REQUEST_POLICY_RULES": rules or [], } def _read_activity_log_row(main_module, snapshot_id: int): conn = main_module.user_db._connect() try: return conn.execute("SELECT * FROM activity_log WHERE id = ?", (snapshot_id,)).fetchone() finally: conn.close() class TestDownloadPolicyGuards: def test_download_endpoint_blocks_before_queue_when_policy_requires_request(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object( main_module, "_load_users_request_policy_settings", return_value=_policy(default_ebook="request_release"), ): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=_policy(default_ebook="request_release")): with patch.object(main_module.backend, "queue_book") as mock_queue_book: resp = client.get("/api/download?id=book-123") assert resp.status_code == 403 assert resp.json["code"] == "policy_requires_request" assert resp.json["required_mode"] == "request_release" mock_queue_book.assert_not_called() def test_release_download_endpoint_blocks_before_queue_when_policy_blocked(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object( main_module, "_load_users_request_policy_settings", return_value=_policy(default_ebook="blocked"), ): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=_policy(default_ebook="blocked")): with patch.object(main_module.backend, "queue_release") as mock_queue_release: resp = client.post( "/api/releases/download", json={"source": "direct_download", "source_id": "rel-1", "content_type": "ebook"}, ) assert resp.status_code == 403 assert resp.json["code"] == "policy_blocked" assert resp.json["required_mode"] == "blocked" mock_queue_release.assert_not_called() def test_admin_bypasses_policy_guards(self, main_module, client): admin = _create_user(main_module, prefix="admin", role="admin") _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object( main_module, "_load_users_request_policy_settings", return_value=_policy(default_ebook="blocked"), ): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=_policy(default_ebook="blocked")): with patch.object(main_module.backend, "queue_book", return_value=(True, None)) as mock_queue_book: resp = client.get("/api/download?id=book-123") assert resp.status_code == 200 assert resp.json["status"] == "queued" mock_queue_book.assert_called_once() def test_no_auth_mode_bypasses_policy_guards(self, main_module, client): with patch.object(main_module, "get_auth_mode", return_value="none"): with patch.object( main_module, "_load_users_request_policy_settings", return_value=_policy(default_ebook="blocked"), ): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=_policy(default_ebook="blocked")): with patch.object(main_module.backend, "queue_book", return_value=(True, None)) as mock_queue_book: resp = client.get("/api/download?id=book-123") assert resp.status_code == 200 assert resp.json["status"] == "queued" mock_queue_book.assert_called_once() class TestRequestRoutes: def test_request_endpoints_are_unavailable_in_no_auth_mode(self, main_module, client): with patch.object(main_module, "get_auth_mode", return_value="none"): resp = client.get("/api/requests") assert resp.status_code == 403 assert resp.json["code"] == "requests_unavailable" def test_request_policy_endpoint_returns_effective_policy(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_release") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.get("/api/request-policy") assert resp.status_code == 200 assert resp.json["requests_enabled"] is True assert resp.json["defaults"]["ebook"] == "request_release" assert "source_modes" in resp.json def test_create_list_and_cancel_request(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") payload = { "book_data": { "title": "The Pragmatic Programmer", "author": "Andrew Hunt", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-1", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "book", }, "note": "Please add this", } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): create_resp = client.post("/api/requests", json=payload) list_resp = client.get("/api/requests") assert create_resp.status_code == 201 request_id = create_resp.json["id"] assert create_resp.json["status"] == "pending" assert any(item["id"] == request_id for item in list_resp.json) cancel_resp = client.delete(f"/api/requests/{request_id}") assert cancel_resp.status_code == 200 assert cancel_resp.json["status"] == "cancelled" snapshot_id = main_module.activity_service.get_latest_activity_log_id( item_type="request", item_key=f"request:{request_id}", ) assert snapshot_id is not None log_row = _read_activity_log_row(main_module, snapshot_id) assert log_row is not None assert log_row["user_id"] == user["id"] assert log_row["final_status"] == "cancelled" assert log_row["origin"] == "request" def test_create_request_emits_websocket_events(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") payload = { "book_data": { "title": "Eventful Book", "author": "Event Author", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-event", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "book", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): with patch.object(main_module.ws_manager, "is_enabled", return_value=True): with patch.object(main_module.ws_manager.socketio, "emit") as mock_emit: resp = client.post("/api/requests", json=payload) assert resp.status_code == 201 request_id = resp.json["id"] assert mock_emit.call_count == 2 mock_emit.assert_any_call("new_request", ANY, to="admins") mock_emit.assert_any_call("request_update", ANY, to=f"user_{user['id']}") emitted_payloads = {call.args[0]: call.args[1] for call in mock_emit.call_args_list} assert emitted_payloads["new_request"]["request_id"] == request_id assert emitted_payloads["new_request"]["status"] == "pending" assert emitted_payloads["new_request"]["title"] == "Eventful Book" assert emitted_payloads["request_update"]["request_id"] == request_id def test_create_request_triggers_admin_notification(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") payload = { "book_data": { "title": "Notify Create Book", "author": "Notify Create Author", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-notify-create", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "book", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes.notify_admin") as mock_notify: with patch("shelfmark.core.request_routes.notify_user") as mock_notify_user: resp = client.post("/api/requests", json=payload) assert resp.status_code == 201 mock_notify.assert_called_once() event, context = mock_notify.call_args.args assert event == NotificationEvent.REQUEST_CREATED assert context.title == "Notify Create Book" assert context.author == "Notify Create Author" assert context.username == user["username"] mock_notify_user.assert_called_once() user_id, user_event, user_context = mock_notify_user.call_args.args assert user_id == user["id"] assert user_event == NotificationEvent.REQUEST_CREATED assert user_context.title == "Notify Create Book" def test_create_request_succeeds_when_notification_dispatch_raises(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") payload = { "book_data": { "title": "Resilient Notify Create Book", "author": "Resilient Notify Create Author", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-notify-resilience", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "book", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): with patch( "shelfmark.core.request_routes.notify_admin", side_effect=RuntimeError("admin notification unavailable"), ) as mock_notify_admin: with patch( "shelfmark.core.request_routes.notify_user", side_effect=RuntimeError("user notification unavailable"), ) as mock_notify_user: resp = client.post("/api/requests", json=payload) assert resp.status_code == 201 assert resp.json["status"] == "pending" mock_notify_admin.assert_called_once() mock_notify_user.assert_called_once() def test_cancel_request_emits_to_user_and_admin_rooms(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") payload = { "book_data": { "title": "Cancelable Book", "author": "Cancelable Author", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-cancel", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "book", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): with patch.object(main_module.ws_manager, "is_enabled", return_value=True): with patch.object(main_module.ws_manager.socketio, "emit") as mock_emit: create_resp = client.post("/api/requests", json=payload) request_id = create_resp.json["id"] mock_emit.reset_mock() cancel_resp = client.delete(f"/api/requests/{request_id}") assert create_resp.status_code == 201 assert cancel_resp.status_code == 200 assert cancel_resp.json["status"] == "cancelled" assert mock_emit.call_count == 2 mock_emit.assert_any_call("request_update", ANY, to=f"user_{user['id']}") mock_emit.assert_any_call("request_update", ANY, to="admins") def test_create_request_level_payload_mismatch_returns_400(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_release") payload = { "book_data": { "title": "Clean Code", "author": "Robert Martin", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-2", }, "context": { "source": "prowlarr", "content_type": "ebook", "request_level": "book", }, "release_data": { "source": "prowlarr", "source_id": "rel-2", "title": "Clean Code.epub", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", json=payload) assert resp.status_code == 400 assert "request_level=book requires null release_data" in resp.json["error"] def test_duplicate_pending_request_returns_409(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") payload = { "book_data": { "title": "Domain-Driven Design", "author": "Eric Evans", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-3", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "book", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): first_resp = client.post("/api/requests", json=payload) second_resp = client.post("/api/requests", json=payload) assert first_resp.status_code == 201 assert second_resp.status_code == 409 assert second_resp.json["code"] == "duplicate_pending_request" def test_create_request_enforces_max_pending_limit(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book", max_pending_requests_per_user=1) payload_1 = { "book_data": { "title": "Book A", "author": "Author A", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-a", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "book", }, } payload_2 = { "book_data": { "title": "Book B", "author": "Author B", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-b", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "book", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): first_resp = client.post("/api/requests", json=payload_1) second_resp = client.post("/api/requests", json=payload_2) assert first_resp.status_code == 201 assert second_resp.status_code == 409 assert second_resp.json["code"] == "max_pending_reached" def test_create_request_strips_note_when_notes_disabled(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book", requests_allow_notes=False) payload = { "book_data": { "title": "No Notes Book", "author": "No Notes Author", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-nonote", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "book", }, "note": "This should be dropped", } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", json=payload) assert resp.status_code == 201 assert resp.json["note"] is None def test_request_book_policy_requires_book_level_request(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") payload = { "book_data": { "title": "Refactoring", "author": "Martin Fowler", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-4", }, "context": { "source": "prowlarr", "content_type": "ebook", "request_level": "release", }, "release_data": { "source": "prowlarr", "source_id": "rel-4", "title": "Refactoring.epub", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", json=payload) assert resp.status_code == 403 assert resp.json["code"] == "policy_requires_request" assert resp.json["required_mode"] == "request_book" def test_request_book_policy_allows_direct_release_level_request(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") payload = { "book_data": { "title": "Direct Result", "author": "Direct Author", "content_type": "ebook", "provider": "direct_download", "provider_id": "dd-1", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "release", }, "release_data": { "source": "direct_download", "source_id": "dd-1", "title": "Direct Result.epub", "format": "epub", "size": "2 MB", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", json=payload) assert resp.status_code == 201 assert resp.json["request_level"] == "release" assert resp.json["policy_mode"] == "request_book" assert resp.json["release_data"]["source"] == "direct_download" assert resp.json["release_data"]["source_id"] == "dd-1" def test_non_admin_cannot_access_admin_request_routes(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) with patch.object(main_module, "get_auth_mode", return_value="builtin"): resp = client.get("/api/admin/requests") assert resp.status_code == 403 assert resp.json["error"] == "Admin access required" def test_admin_reject_and_terminal_conflict(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_book") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) create_payload = { "book_data": { "title": "Working Effectively with Legacy Code", "author": "Michael Feathers", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-5", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "book", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): create_resp = client.post("/api/requests", json=create_payload) request_id = create_resp.json["id"] _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) count_resp = client.get("/api/admin/requests/count") reject_resp = client.post( f"/api/admin/requests/{request_id}/reject", json={"admin_note": "Declined"}, ) reject_again_resp = client.post( f"/api/admin/requests/{request_id}/reject", json={"admin_note": "Declined again"}, ) assert count_resp.status_code == 200 assert count_resp.json["pending"] >= 1 assert reject_resp.status_code == 200 assert reject_resp.json["status"] == "rejected" assert reject_again_resp.status_code == 409 assert reject_again_resp.json["code"] == "stale_transition" snapshot_id = main_module.activity_service.get_latest_activity_log_id( item_type="request", item_key=f"request:{request_id}", ) assert snapshot_id is not None log_row = _read_activity_log_row(main_module, snapshot_id) assert log_row is not None assert log_row["user_id"] == user["id"] assert log_row["final_status"] == "rejected" assert log_row["origin"] == "request" def test_admin_reject_emits_update_to_user_and_admin_rooms(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_book") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) create_payload = { "book_data": { "title": "Reject Emit Book", "author": "Reject Emit Author", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-reject-emit", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "book", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): create_resp = client.post("/api/requests", json=create_payload) request_id = create_resp.json["id"] _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) with patch.object(main_module.ws_manager, "is_enabled", return_value=True): with patch.object(main_module.ws_manager.socketio, "emit") as mock_emit: reject_resp = client.post( f"/api/admin/requests/{request_id}/reject", json={"admin_note": "Rejected with event fanout"}, ) assert create_resp.status_code == 201 assert reject_resp.status_code == 200 assert reject_resp.json["status"] == "rejected" assert mock_emit.call_count == 2 mock_emit.assert_any_call("request_update", ANY, to=f"user_{user['id']}") mock_emit.assert_any_call("request_update", ANY, to="admins") def test_admin_reject_triggers_admin_notification(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_book") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) create_payload = { "book_data": { "title": "Reject Notify Book", "author": "Reject Notify Author", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-reject-notify", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "book", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): create_resp = client.post("/api/requests", json=create_payload) request_id = create_resp.json["id"] _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) with patch("shelfmark.core.request_routes.notify_admin") as mock_notify: with patch("shelfmark.core.request_routes.notify_user") as mock_notify_user: reject_resp = client.post( f"/api/admin/requests/{request_id}/reject", json={"admin_note": "Needs better metadata"}, ) assert create_resp.status_code == 201 assert reject_resp.status_code == 200 mock_notify.assert_called_once() event, context = mock_notify.call_args.args assert event == NotificationEvent.REQUEST_REJECTED assert context.title == "Reject Notify Book" assert context.admin_note == "Needs better metadata" assert context.username == user["username"] mock_notify_user.assert_called_once() user_id, user_event, user_context = mock_notify_user.call_args.args assert user_id == user["id"] assert user_event == NotificationEvent.REQUEST_REJECTED assert user_context.admin_note == "Needs better metadata" def test_admin_fulfil_queues_for_requesting_user(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_release") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) create_payload = { "book_data": { "title": "Patterns of Enterprise Application Architecture", "author": "Martin Fowler", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-6", }, "context": { "source": "prowlarr", "content_type": "ebook", "request_level": "release", }, "release_data": { "source": "prowlarr", "source_id": "rel-6", "title": "POEAA.epub", }, } captured: dict[str, object] = {} def fake_queue_release(release_data, priority, user_id=None, username=None): captured["release_data"] = release_data captured["priority"] = priority captured["user_id"] = user_id captured["username"] = username return True, None with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): create_resp = client.post("/api/requests", json=create_payload) request_id = create_resp.json["id"] _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) with patch.object(main_module.backend, "queue_release", side_effect=fake_queue_release): fulfil_resp = client.post( f"/api/admin/requests/{request_id}/fulfil", json={"admin_note": "Approved"}, ) assert fulfil_resp.status_code == 200 assert fulfil_resp.json["status"] == "fulfilled" assert captured["priority"] == 0 assert captured["user_id"] == user["id"] assert captured["username"] == user["username"] def test_admin_fulfil_emits_update_to_user_and_admin_rooms(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_release") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) create_payload = { "book_data": { "title": "Fulfil Emit Book", "author": "Fulfil Emit Author", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-fulfil-emit", }, "context": { "source": "prowlarr", "content_type": "ebook", "request_level": "release", }, "release_data": { "source": "prowlarr", "source_id": "rel-fulfil-emit", "title": "Fulfil Emit Book.epub", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): create_resp = client.post("/api/requests", json=create_payload) request_id = create_resp.json["id"] _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) with patch.object(main_module.backend, "queue_release", return_value=(True, None)): with patch.object(main_module.ws_manager, "is_enabled", return_value=True): with patch.object(main_module.ws_manager.socketio, "emit") as mock_emit: fulfil_resp = client.post( f"/api/admin/requests/{request_id}/fulfil", json={"admin_note": "Approved with event fanout"}, ) assert create_resp.status_code == 201 assert fulfil_resp.status_code == 200 assert fulfil_resp.json["status"] == "fulfilled" assert mock_emit.call_count == 2 mock_emit.assert_any_call("request_update", ANY, to=f"user_{user['id']}") mock_emit.assert_any_call("request_update", ANY, to="admins") def test_admin_fulfil_triggers_admin_notification(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_release") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) create_payload = { "book_data": { "title": "Fulfil Notify Book", "author": "Fulfil Notify Author", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-fulfil-notify", }, "context": { "source": "prowlarr", "content_type": "ebook", "request_level": "release", }, "release_data": { "source": "prowlarr", "source_id": "rel-fulfil-notify", "title": "Fulfil Notify Book.epub", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): create_resp = client.post("/api/requests", json=create_payload) request_id = create_resp.json["id"] _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) with patch.object(main_module.backend, "queue_release", return_value=(True, None)): with patch("shelfmark.core.request_routes.notify_admin") as mock_notify: with patch("shelfmark.core.request_routes.notify_user") as mock_notify_user: fulfil_resp = client.post( f"/api/admin/requests/{request_id}/fulfil", json={"admin_note": "Approved"}, ) assert create_resp.status_code == 201 assert fulfil_resp.status_code == 200 mock_notify.assert_called_once() event, context = mock_notify.call_args.args assert event == NotificationEvent.REQUEST_FULFILLED assert context.title == "Fulfil Notify Book" assert context.username == user["username"] mock_notify_user.assert_called_once() user_id, user_event, user_context = mock_notify_user.call_args.args assert user_id == user["id"] assert user_event == NotificationEvent.REQUEST_FULFILLED assert user_context.title == "Fulfil Notify Book" def test_admin_fulfil_book_level_request_requires_release_data(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_book") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) create_payload = { "book_data": { "title": "Designing Data-Intensive Applications", "author": "Martin Kleppmann", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-7", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "book", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): create_resp = client.post("/api/requests", json=create_payload) request_id = create_resp.json["id"] _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) fulfil_resp = client.post(f"/api/admin/requests/{request_id}/fulfil", json={}) assert fulfil_resp.status_code == 400 assert "release_data is required to fulfil book-level requests" in fulfil_resp.json["error"] def test_admin_fulfil_book_level_request_with_release_data(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_book") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) create_payload = { "book_data": { "title": "Book Level Fulfil", "author": "QA Author", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-book-fulfil", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "book", }, } captured: dict[str, object] = {} def fake_queue_release(release_data, priority, user_id=None, username=None): captured["release_data"] = release_data captured["priority"] = priority captured["user_id"] = user_id captured["username"] = username return True, None with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): create_resp = client.post("/api/requests", json=create_payload) request_id = create_resp.json["id"] _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) with patch.object(main_module.backend, "queue_release", side_effect=fake_queue_release): fulfil_resp = client.post( f"/api/admin/requests/{request_id}/fulfil", json={ "release_data": { "source": "direct_download", "source_id": "book-level-picked-release", "title": "Book Level Fulfil.epub", } }, ) assert create_resp.status_code == 201 assert fulfil_resp.status_code == 200 assert fulfil_resp.json["status"] == "fulfilled" assert fulfil_resp.json["request_level"] == "book" assert fulfil_resp.json["release_data"]["source_id"] == "book-level-picked-release" assert captured["release_data"]["source_id"] == "book-level-picked-release" assert captured["priority"] == 0 assert captured["user_id"] == user["id"] assert captured["username"] == user["username"] def test_admin_fulfil_uses_real_queue_and_preserves_requesting_identity(self, main_module, client): user = _create_user(main_module, prefix="reader") other_user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_release") source_id = f"real-queue-{uuid.uuid4().hex[:10]}" _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) create_payload = { "book_data": { "title": "Building Microservices", "author": "Sam Newman", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-8", }, "context": { "source": "direct_download", "content_type": "ebook", "request_level": "release", }, "release_data": { "source": "direct_download", "source_id": source_id, "title": "Building Microservices.epub", }, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): create_resp = client.post("/api/requests", json=create_payload) request_id = create_resp.json["id"] _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) fulfil_resp = client.post(f"/api/admin/requests/{request_id}/fulfil", json={}) assert fulfil_resp.status_code == 200 assert fulfil_resp.json["status"] == "fulfilled" user_status = main_module.backend.queue_status(user_id=user["id"]) assert source_id in user_status["queued"] assert user_status["queued"][source_id]["username"] == user["username"] other_status = main_module.backend.queue_status(user_id=other_user["id"]) assert source_id not in other_status["queued"] class TestRequestCreationEdgeCases: """Edge cases for POST /api/requests.""" def test_no_json_body_returns_400(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", content_type="text/plain", data="garbage") assert resp.status_code == 400 assert "No data provided" in resp.json["error"] def test_missing_book_data_returns_400(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", json={"context": {"source": "direct_download"}}) assert resp.status_code == 400 assert "book_data must be an object" in resp.json["error"] def test_non_dict_context_returns_400(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", json={ "context": "not-a-dict", "book_data": {"title": "X", "author": "Y", "provider": "z", "provider_id": "1"}, }) assert resp.status_code == 400 assert "context must be an object" in resp.json["error"] def test_book_data_missing_required_fields_returns_400(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", json={ "book_data": {"title": "Only a title"}, "context": {"source": "direct_download", "content_type": "ebook", "request_level": "book"}, }) assert resp.status_code == 400 assert "missing required field" in resp.json["error"] def test_book_data_payload_too_large_returns_400(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") payload = { "book_data": { "title": "Oversized Book", "author": "Big Payload", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-big", "description": "x" * 12000, }, "context": {"source": "direct_download", "content_type": "ebook", "request_level": "book"}, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", json=payload) assert resp.status_code == 400 assert "book_data must be <= 10240 bytes" in resp.json["error"] def test_disabled_requests_returns_403(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(requests_enabled=False, default_ebook="request_book") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", json={ "book_data": {"title": "T", "author": "A", "provider": "p", "provider_id": "1"}, "context": {"source": "direct_download", "content_type": "ebook", "request_level": "book"}, }) assert resp.status_code == 403 assert resp.json["code"] == "requests_unavailable" def test_blocked_policy_returns_403(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="blocked") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", json={ "book_data": {"title": "T", "author": "A", "provider": "p", "provider_id": "1", "content_type": "ebook"}, "context": {"source": "direct_download", "content_type": "ebook", "request_level": "book"}, }) assert resp.status_code == 403 assert resp.json["code"] == "policy_blocked" assert resp.json["required_mode"] == "blocked" def test_auto_infers_book_level_when_no_release_data(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") payload = { "book_data": { "title": "Auto Infer Book", "author": "Test Author", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-auto-1", }, "context": {"source": "direct_download", "content_type": "ebook"}, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", json=payload) assert resp.status_code == 201 assert resp.json["request_level"] == "book" def test_auto_infers_release_level_when_release_data_present(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_release") payload = { "book_data": { "title": "Auto Infer Release", "author": "Test Author", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-auto-2", }, "context": {"source": "prowlarr", "content_type": "ebook"}, "release_data": {"source": "prowlarr", "source_id": "auto-r", "title": "File.epub"}, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", json=payload) assert resp.status_code == 201 assert resp.json["request_level"] == "release" assert resp.json["release_data"]["source_id"] == "auto-r" def test_release_level_request_with_request_release_policy(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_release") payload = { "book_data": { "title": "Release Level Test", "author": "RLT Author", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-rl", }, "context": {"source": "prowlarr", "content_type": "ebook", "request_level": "release"}, "release_data": {"source": "prowlarr", "source_id": "rl-1", "title": "RL.epub"}, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", json=payload) assert resp.status_code == 201 assert resp.json["policy_mode"] == "request_release" assert resp.json["request_level"] == "release" def test_without_db_user_id_returns_403(self, main_module, client): with client.session_transaction() as sess: sess["user_id"] = "some-user" sess["is_admin"] = False if "db_user_id" in sess: del sess["db_user_id"] with patch.object(main_module, "get_auth_mode", return_value="builtin"): resp = client.post("/api/requests", json={ "book_data": {"title": "T", "author": "A", "provider": "p", "provider_id": "1"}, "context": {"source": "direct_download"}, }) assert resp.status_code == 403 assert resp.json["code"] == "user_identity_unavailable" def test_audiobook_content_type_request(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_audiobook="request_release") payload = { "book_data": { "title": "Audiobook Test", "author": "AB Author", "content_type": "audiobook", "provider": "hardcover", "provider_id": "hc-ab", }, "context": {"source": "prowlarr", "content_type": "audiobook", "request_level": "release"}, "release_data": {"source": "prowlarr", "source_id": "ab-1", "title": "AB.m4b"}, } with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.post("/api/requests", json=payload) assert resp.status_code == 201 assert resp.json["content_type"] == "audiobook" class TestRequestListAndFilterEdgeCases: """Edge cases for GET /api/requests and GET /api/admin/requests.""" def _seed_requests(self, main_module, client, user, policy, count=3): """Create multiple requests and return their IDs.""" ids = [] for i in range(count): payload = { "book_data": { "title": f"Seed Book {uuid.uuid4().hex[:6]}", "author": f"Author {i}", "content_type": "ebook", "provider": "openlibrary", "provider_id": f"ol-seed-{uuid.uuid4().hex[:6]}", }, "context": {"source": "direct_download", "content_type": "ebook", "request_level": "book"}, } resp = client.post("/api/requests", json=payload) assert resp.status_code == 201 ids.append(resp.json["id"]) return ids def test_list_requests_empty_result(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) with patch.object(main_module, "get_auth_mode", return_value="builtin"): resp = client.get("/api/requests") assert resp.status_code == 200 assert resp.json == [] def test_list_requests_with_status_filter(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): ids = self._seed_requests(main_module, client, user, policy, count=3) # Cancel the first request. client.delete(f"/api/requests/{ids[0]}") # Filter: only pending. pending_resp = client.get("/api/requests?status=pending") cancelled_resp = client.get("/api/requests?status=cancelled") assert pending_resp.status_code == 200 pending_ids = {r["id"] for r in pending_resp.json} assert ids[0] not in pending_ids assert ids[1] in pending_ids assert cancelled_resp.status_code == 200 cancelled_ids = {r["id"] for r in cancelled_resp.json} assert ids[0] in cancelled_ids def test_list_requests_with_pagination(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): self._seed_requests(main_module, client, user, policy, count=5) page1 = client.get("/api/requests?limit=2&offset=0") page2 = client.get("/api/requests?limit=2&offset=2") assert page1.status_code == 200 assert len(page1.json) == 2 assert page2.status_code == 200 assert len(page2.json) == 2 # Pages should not overlap. page1_ids = {r["id"] for r in page1.json} page2_ids = {r["id"] for r in page2.json} assert page1_ids.isdisjoint(page2_ids) def test_user_only_sees_own_requests(self, main_module, client): alice = _create_user(main_module, prefix="alice") bob = _create_user(main_module, prefix="bob") policy = _policy(default_ebook="request_book") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): # Alice creates a request. _set_session(client, user_id=alice["username"], db_user_id=alice["id"], is_admin=False) client.post("/api/requests", json={ "book_data": {"title": "Alice Book", "author": "A", "provider": "p", "provider_id": "a1", "content_type": "ebook"}, "context": {"source": "direct_download", "content_type": "ebook", "request_level": "book"}, }) # Bob creates a request. _set_session(client, user_id=bob["username"], db_user_id=bob["id"], is_admin=False) client.post("/api/requests", json={ "book_data": {"title": "Bob Book", "author": "B", "provider": "p", "provider_id": "b1", "content_type": "ebook"}, "context": {"source": "direct_download", "content_type": "ebook", "request_level": "book"}, }) # Bob lists — should only see his. bob_list = client.get("/api/requests") # Alice lists — should only see hers. _set_session(client, user_id=alice["username"], db_user_id=alice["id"], is_admin=False) alice_list = client.get("/api/requests") assert len(bob_list.json) == 1 assert bob_list.json[0]["book_data"]["title"] == "Bob Book" assert len(alice_list.json) == 1 assert alice_list.json[0]["book_data"]["title"] == "Alice Book" def test_admin_list_includes_username(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_book") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) client.post("/api/requests", json={ "book_data": {"title": "Admin View", "author": "AV", "provider": "p", "provider_id": "av1", "content_type": "ebook"}, "context": {"source": "direct_download", "content_type": "ebook", "request_level": "book"}, }) _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) resp = client.get("/api/admin/requests") assert resp.status_code == 200 matching = [r for r in resp.json if r["book_data"]["title"] == "Admin View"] assert len(matching) >= 1 assert matching[0]["username"] == user["username"] def test_admin_list_with_status_filter(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_book") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) create_resp = client.post("/api/requests", json={ "book_data": {"title": f"FilterTest-{uuid.uuid4().hex[:6]}", "author": "FT", "provider": "p", "provider_id": f"ft-{uuid.uuid4().hex[:6]}", "content_type": "ebook"}, "context": {"source": "direct_download", "content_type": "ebook", "request_level": "book"}, }) request_id = create_resp.json["id"] _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) client.post(f"/api/admin/requests/{request_id}/reject", json={}) pending_resp = client.get("/api/admin/requests?status=pending") rejected_resp = client.get("/api/admin/requests?status=rejected") pending_ids = {r["id"] for r in pending_resp.json} rejected_ids = {r["id"] for r in rejected_resp.json} assert request_id not in pending_ids assert request_id in rejected_ids class TestCancelEdgeCases: """Edge cases for DELETE /api/requests/.""" def test_cancel_nonexistent_request_returns_404(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) with patch.object(main_module, "get_auth_mode", return_value="builtin"): resp = client.delete("/api/requests/99999") assert resp.status_code == 404 def test_cancel_other_users_request_returns_403(self, main_module, client): alice = _create_user(main_module, prefix="alice") bob = _create_user(main_module, prefix="bob") policy = _policy(default_ebook="request_book") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): _set_session(client, user_id=alice["username"], db_user_id=alice["id"], is_admin=False) create_resp = client.post("/api/requests", json={ "book_data": {"title": "Alice Only", "author": "A", "provider": "p", "provider_id": "ao1", "content_type": "ebook"}, "context": {"source": "direct_download", "content_type": "ebook", "request_level": "book"}, }) request_id = create_resp.json["id"] _set_session(client, user_id=bob["username"], db_user_id=bob["id"], is_admin=False) cancel_resp = client.delete(f"/api/requests/{request_id}") assert cancel_resp.status_code == 403 def test_cancel_already_cancelled_returns_409(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): create_resp = client.post("/api/requests", json={ "book_data": {"title": "Cancel Twice", "author": "CT", "provider": "p", "provider_id": "ct1", "content_type": "ebook"}, "context": {"source": "direct_download", "content_type": "ebook", "request_level": "book"}, }) request_id = create_resp.json["id"] first = client.delete(f"/api/requests/{request_id}") second = client.delete(f"/api/requests/{request_id}") assert first.status_code == 200 assert second.status_code == 409 assert second.json["code"] == "stale_transition" class TestAdminFulfilEdgeCases: """Edge cases for POST /api/admin/requests//fulfil.""" def test_fulfil_nonexistent_request_returns_404(self, main_module, client): admin = _create_user(main_module, prefix="admin", role="admin") _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) with patch.object(main_module, "get_auth_mode", return_value="builtin"): resp = client.post("/api/admin/requests/99999/fulfil", json={ "release_data": {"source": "dd", "source_id": "r1", "title": "f.epub"}, }) assert resp.status_code == 404 def test_fulfil_with_queue_failure_returns_409(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_release") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): create_resp = client.post("/api/requests", json={ "book_data": {"title": "Queue Fail", "author": "QF", "provider": "p", "provider_id": "qf1", "content_type": "ebook"}, "context": {"source": "prowlarr", "content_type": "ebook", "request_level": "release"}, "release_data": {"source": "prowlarr", "source_id": "qf-r", "title": "QF.epub"}, }) request_id = create_resp.json["id"] _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) with patch.object(main_module.backend, "queue_release", return_value=(False, "Client offline")): fulfil_resp = client.post(f"/api/admin/requests/{request_id}/fulfil", json={}) assert fulfil_resp.status_code == 409 assert fulfil_resp.json["code"] == "queue_failed" def test_fulfil_with_admin_override_release_data(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_release") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) captured = {} def capture_queue(release_data, priority, **kwargs): captured["release_data"] = release_data return True, None with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): create_resp = client.post("/api/requests", json={ "book_data": {"title": "Override RD", "author": "OR", "provider": "p", "provider_id": "or1", "content_type": "ebook"}, "context": {"source": "prowlarr", "content_type": "ebook", "request_level": "release"}, "release_data": {"source": "prowlarr", "source_id": "original-r", "title": "Original.epub"}, }) request_id = create_resp.json["id"] _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) with patch.object(main_module.backend, "queue_release", side_effect=capture_queue): fulfil_resp = client.post(f"/api/admin/requests/{request_id}/fulfil", json={ "release_data": {"source": "direct_download", "source_id": "better-r", "title": "Better.epub"}, }) assert fulfil_resp.status_code == 200 assert captured["release_data"]["source_id"] == "better-r" assert fulfil_resp.json["release_data"]["source_id"] == "better-r" def test_admin_without_db_user_id_returns_403(self, main_module, client): with client.session_transaction() as sess: sess["user_id"] = "admin-user" sess["is_admin"] = True if "db_user_id" in sess: del sess["db_user_id"] with patch.object(main_module, "get_auth_mode", return_value="builtin"): resp = client.post("/api/admin/requests/1/fulfil", json={}) assert resp.status_code == 403 assert "Admin user identity unavailable" in resp.json["error"] class TestAdminRejectEdgeCases: """Edge cases for POST /api/admin/requests//reject.""" def test_reject_nonexistent_request_returns_404(self, main_module, client): admin = _create_user(main_module, prefix="admin", role="admin") _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) with patch.object(main_module, "get_auth_mode", return_value="builtin"): resp = client.post("/api/admin/requests/99999/reject", json={}) assert resp.status_code == 404 def test_reject_already_fulfilled_returns_409(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_release") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): create_resp = client.post("/api/requests", json={ "book_data": {"title": "Rej After Ful", "author": "RAF", "provider": "p", "provider_id": "raf1", "content_type": "ebook"}, "context": {"source": "prowlarr", "content_type": "ebook", "request_level": "release"}, "release_data": {"source": "prowlarr", "source_id": "raf-r", "title": "RAF.epub"}, }) request_id = create_resp.json["id"] _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) with patch.object(main_module.backend, "queue_release", return_value=(True, None)): client.post(f"/api/admin/requests/{request_id}/fulfil", json={}) reject_resp = client.post(f"/api/admin/requests/{request_id}/reject", json={}) assert reject_resp.status_code == 409 assert reject_resp.json["code"] == "stale_transition" class TestAdminCountEdgeCases: """Edge cases for GET /api/admin/requests/count.""" def test_count_reflects_all_statuses(self, main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") policy = _policy(default_ebook="request_book") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) # Create 3 requests. ids = [] for i in range(3): resp = client.post("/api/requests", json={ "book_data": { "title": f"Count Test {uuid.uuid4().hex[:6]}", "author": "CT", "provider": "p", "provider_id": f"ct-{uuid.uuid4().hex[:6]}", "content_type": "ebook", }, "context": {"source": "direct_download", "content_type": "ebook", "request_level": "book"}, }) ids.append(resp.json["id"]) # Cancel one. client.delete(f"/api/requests/{ids[0]}") # Admin rejects one. _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) client.post(f"/api/admin/requests/{ids[1]}/reject", json={}) count_resp = client.get("/api/admin/requests/count") assert count_resp.status_code == 200 by_status = count_resp.json["by_status"] # At least 1 of each status we created. assert by_status["cancelled"] >= 1 assert by_status["rejected"] >= 1 assert by_status["pending"] >= 1 assert count_resp.json["pending"] == by_status["pending"] assert count_resp.json["total"] == sum(by_status.values()) class TestPolicyEndpointEdgeCases: """Edge cases for GET /api/request-policy.""" def test_admin_view_shows_is_admin_true(self, main_module, client): admin = _create_user(main_module, prefix="admin", role="admin") _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) policy = _policy(default_ebook="download") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.get("/api/request-policy") assert resp.status_code == 200 assert resp.json["is_admin"] is True def test_policy_endpoint_reflects_per_user_overrides(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) # Global says download, but user override sets request_release for ebook. global_policy = _policy(default_ebook="download", default_audiobook="download") main_module.user_db.set_user_settings(user["id"], {"REQUEST_POLICY_DEFAULT_EBOOK": "request_release"}) with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=global_policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=global_policy): resp = client.get("/api/request-policy") assert resp.status_code == 200 assert resp.json["defaults"]["ebook"] == "request_release" assert resp.json["defaults"]["audiobook"] == "download" def test_policy_endpoint_without_session_returns_401(self, main_module, client): with patch.object(main_module, "get_auth_mode", return_value="builtin"): resp = client.get("/api/request-policy") assert resp.status_code == 401 def test_policy_endpoint_includes_allow_notes_from_effective_settings(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="download", requests_allow_notes=False) with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): resp = client.get("/api/request-policy") assert resp.status_code == 200 assert resp.json["allow_notes"] is False def test_policy_endpoint_allow_notes_reflects_per_user_override(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) global_policy = _policy(default_ebook="download", requests_allow_notes=False) main_module.user_db.set_user_settings(user["id"], {"REQUESTS_ALLOW_NOTES": True}) with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=global_policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=global_policy): resp = client.get("/api/request-policy") assert resp.status_code == 200 assert resp.json["allow_notes"] is True class TestDownloadPolicyGuardsExtended: """Extended policy enforcement tests for download endpoints.""" def test_download_allowed_when_requests_disabled(self, main_module, client): """When REQUESTS_ENABLED is false, policy is not enforced — downloads pass through.""" user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) # Even though default is blocked, requests are disabled so policy doesn't apply. policy = _policy(requests_enabled=False, default_ebook="blocked") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): with patch.object(main_module.backend, "queue_book", return_value=(True, None)): resp = client.get("/api/download?id=book-pass") assert resp.status_code == 200 assert resp.json["status"] == "queued" def test_download_allowed_when_policy_mode_is_download(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="download") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): with patch.object(main_module.backend, "queue_book", return_value=(True, None)): resp = client.get("/api/download?id=book-free") assert resp.status_code == 200 assert resp.json["status"] == "queued" def test_release_download_blocks_with_request_release_policy(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_release") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): with patch.object(main_module.backend, "queue_release") as mock_queue: resp = client.post("/api/releases/download", json={ "source": "prowlarr", "source_id": "rel-blocked", "content_type": "ebook", }) assert resp.status_code == 403 assert resp.json["code"] == "policy_requires_request" assert resp.json["required_mode"] == "request_release" mock_queue.assert_not_called() def test_release_download_infers_audiobook_type_from_format_when_content_type_missing(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="download", default_audiobook="blocked") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): with patch.object(main_module.backend, "queue_release") as mock_queue: resp = client.post("/api/releases/download", json={ "source": "prowlarr", "source_id": "audio-rel", "title": "Some Audio [m4b]", "format": "m4b", }) assert resp.status_code == 403 assert resp.json["code"] == "policy_blocked" assert resp.json["required_mode"] == "blocked" mock_queue.assert_not_called() def test_release_download_blocks_with_request_book_policy(self, main_module, client): user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy(default_ebook="request_book") with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): with patch.object(main_module.backend, "queue_release") as mock_queue: resp = client.post("/api/releases/download", json={ "source": "direct_download", "source_id": "rel-rbook", "content_type": "ebook", }) assert resp.status_code == 403 assert resp.json["code"] == "policy_requires_request" assert resp.json["required_mode"] == "request_book" mock_queue.assert_not_called() def test_release_download_with_per_source_matrix_rule(self, main_module, client): """Prowlarr blocked by matrix rule, but DD still allowed.""" user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) policy = _policy( default_ebook="download", rules=[{"source": "prowlarr", "content_type": "*", "mode": "blocked"}], ) with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=policy): with patch.object(main_module.backend, "queue_release") as mock_queue: # Prowlarr should be blocked. prowlarr_resp = client.post("/api/releases/download", json={ "source": "prowlarr", "source_id": "prowlarr-rel", "content_type": "ebook", }) with patch.object(main_module.backend, "queue_release", return_value=(True, None)) as mock_queue_dd: # DD should still be allowed. dd_resp = client.post("/api/releases/download", json={ "source": "direct_download", "source_id": "dd-rel", "content_type": "ebook", }) assert prowlarr_resp.status_code == 403 assert prowlarr_resp.json["code"] == "policy_blocked" mock_queue.assert_not_called() assert dd_resp.status_code == 200 mock_queue_dd.assert_called_once() def test_per_user_override_unlocks_blocked_source(self, main_module, client): """Global blocks prowlarr, per-user override unlocks it.""" user = _create_user(main_module, prefix="reader") _set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False) global_policy = _policy( default_ebook="download", rules=[{"source": "prowlarr", "content_type": "*", "mode": "blocked"}], ) # User override: unblock prowlarr. main_module.user_db.set_user_settings(user["id"], { "REQUEST_POLICY_RULES": [ {"source": "prowlarr", "content_type": "*", "mode": "download"}, ], }) with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module, "_load_users_request_policy_settings", return_value=global_policy): with patch("shelfmark.core.request_routes._load_users_request_policy_settings", return_value=global_policy): with patch.object(main_module.backend, "queue_release", return_value=(True, None)): resp = client.post("/api/releases/download", json={ "source": "prowlarr", "source_id": "prowlarr-unlocked", "content_type": "ebook", }) assert resp.status_code == 200 def test_clear_queue_does_not_mutate_fulfilled_request_delivery_state(main_module, client): user = _create_user(main_module, prefix="reader") admin = _create_user(main_module, prefix="admin", role="admin") _set_session(client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True) created = main_module.user_db.create_request( user_id=user["id"], content_type="ebook", request_level="release", policy_mode="request_release", book_data={ "title": "Clear Delivery State", "author": "QA", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-clear-delivery", }, release_data={ "source": "prowlarr", "source_id": "clear-delivery-source-id", "title": "Clear Delivery State.epub", }, status="fulfilled", delivery_state="complete", ) with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module.ws_manager, "is_enabled", return_value=False): with patch.object(main_module.ws_manager, "broadcast_status_update"): with patch.object(main_module.backend, "queue_status", return_value={}) as mock_queue_status: with patch.object(main_module.backend, "clear_completed", return_value=1) as mock_clear_completed: resp = client.delete("/api/queue/clear") assert resp.status_code == 200 assert resp.json["status"] == "cleared" assert resp.json["removed_count"] == 1 assert mock_queue_status.call_args_list[0].kwargs == {} mock_clear_completed.assert_called_once_with(user_id=None) refreshed = main_module.user_db.get_request(created["id"]) assert refreshed["delivery_state"] == "complete" def test_non_admin_clear_queue_is_scoped_without_mutating_request_delivery_state(main_module, client): alice = _create_user(main_module, prefix="alice") bob = _create_user(main_module, prefix="bob") _set_session(client, user_id=alice["username"], db_user_id=alice["id"], is_admin=False) alice_request = main_module.user_db.create_request( user_id=alice["id"], content_type="ebook", request_level="release", policy_mode="request_release", book_data={ "title": "Alice Clear Scope", "author": "QA", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-alice-scope", }, release_data={ "source": "prowlarr", "source_id": "shared-clear-scope-source-id", "title": "Alice Scope.epub", }, status="fulfilled", delivery_state="complete", ) bob_request = main_module.user_db.create_request( user_id=bob["id"], content_type="ebook", request_level="release", policy_mode="request_release", book_data={ "title": "Bob Clear Scope", "author": "QA", "content_type": "ebook", "provider": "openlibrary", "provider_id": "ol-bob-scope", }, release_data={ "source": "prowlarr", "source_id": "shared-clear-scope-source-id", "title": "Bob Scope.epub", }, status="fulfilled", delivery_state="complete", ) with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module.ws_manager, "is_enabled", return_value=False): with patch.object(main_module.ws_manager, "broadcast_status_update"): with patch.object(main_module.backend, "queue_status", return_value={}) as mock_queue_status: with patch.object(main_module.backend, "clear_completed", return_value=1) as mock_clear_completed: resp = client.delete("/api/queue/clear") assert resp.status_code == 200 assert resp.json["status"] == "cleared" assert resp.json["removed_count"] == 1 assert mock_queue_status.call_args_list[0].kwargs == {} mock_clear_completed.assert_called_once_with(user_id=alice["id"]) refreshed_alice = main_module.user_db.get_request(alice_request["id"]) refreshed_bob = main_module.user_db.get_request(bob_request["id"]) assert refreshed_alice["delivery_state"] == "complete" assert refreshed_bob["delivery_state"] == "complete" def test_non_admin_clear_queue_without_db_user_id_returns_403(main_module, client): _set_session(client, user_id="reader-no-db", db_user_id=None, is_admin=False) with patch.object(main_module, "get_auth_mode", return_value="builtin"): with patch.object(main_module.backend, "clear_completed") as mock_clear_completed: resp = client.delete("/api/queue/clear") assert resp.status_code == 403 assert resp.json["code"] == "user_identity_unavailable" mock_clear_completed.assert_not_called()