diff --git a/shelfmark/core/download_recording.py b/shelfmark/core/download_recording.py new file mode 100644 index 00000000..5d400787 --- /dev/null +++ b/shelfmark/core/download_recording.py @@ -0,0 +1,271 @@ +"""Download recording: write each queued download to the history and, when it reaches a +terminal status, finalize the row, notify, push the activity update and reopen a failed request. +""" + +from __future__ import annotations + +import sqlite3 +from typing import TYPE_CHECKING, Any + +from shelfmark.core.logger import setup_logger +from shelfmark.core.models import TERMINAL_QUEUE_STATUSES, QueueStatus +from shelfmark.core.notifications import ( + NotificationContext, + NotificationEvent, + notify_admin, + notify_user, +) +from shelfmark.core.request_helpers import ( + emit_ws_event, + normalize_optional_text, + normalize_positive_int, +) +from shelfmark.core.request_policy import normalize_content_type, normalize_source +from shelfmark.core.requests_service import reopen_failed_request +from shelfmark.core.viewer_redaction import viewer_download_id +from shelfmark.download import orchestrator as backend +from shelfmark.release_sources import get_source_display_name + +if TYPE_CHECKING: + from collections.abc import Callable + + from shelfmark.core.activity_view_state_service import ActivityViewStateService + from shelfmark.core.download_history_service import DownloadHistoryService + from shelfmark.core.user_db import UserDB + +logger = setup_logger(__name__) + +_OPERATIONAL_ERRORS = (OSError, RuntimeError, TypeError, ValueError, sqlite3.Error) + + +def _queue_status_to_final_activity_status(status: QueueStatus) -> str | None: + return status.value if status in TERMINAL_QUEUE_STATUSES else None + + +def _queue_status_to_notification_event(status: QueueStatus) -> NotificationEvent | None: + if status == QueueStatus.COMPLETE: + return NotificationEvent.DOWNLOAD_COMPLETE + if status == QueueStatus.ERROR: + return NotificationEvent.DOWNLOAD_FAILED + return None + + +def _notify_admin_for_terminal_download_status( + *, task_id: str, status: QueueStatus, task: Any +) -> None: + event = _queue_status_to_notification_event(status) + if event is None: + return + + raw_owner_user_id = getattr(task, "user_id", None) + try: + owner_user_id = int(raw_owner_user_id) if raw_owner_user_id is not None else None + except TypeError, ValueError: + owner_user_id = None + + content_type = normalize_optional_text(getattr(task, "content_type", None)) + context = NotificationContext( + event=event, + title=str(getattr(task, "title", "Unknown title") or "Unknown title"), + author=str(getattr(task, "author", "Unknown author") or "Unknown author"), + username=normalize_optional_text(getattr(task, "username", None)), + content_type=normalize_content_type(content_type) if content_type is not None else None, + format=normalize_optional_text(getattr(task, "format", None)), + source=normalize_source(getattr(task, "source", None)), + error_message=( + normalize_optional_text(getattr(task, "status_message", None)) + if event == NotificationEvent.DOWNLOAD_FAILED + else None + ), + ) + try: + notify_admin(event, context) + except (RuntimeError, TypeError, ValueError) as exc: + logger.warning( + "Failed to trigger admin notification for download %s (%s): %s", + task_id, + status.value, + exc, + ) + if owner_user_id is None: + return + try: + notify_user(owner_user_id, event, context) + except (RuntimeError, TypeError, ValueError) as exc: + logger.warning( + "Failed to trigger user notification for download %s (%s, user_id=%s): %s", + task_id, + status.value, + owner_user_id, + exc, + ) + + +def register_download_recording( + *, + get_user_db: Callable[[], UserDB | None], + get_download_history_service: Callable[[], DownloadHistoryService | None], + get_activity_view_state_service: Callable[[], ActivityViewStateService | None], + emit_request_update_events: Callable[[list[dict[str, Any]]], None], + ws_manager: object, +) -> None: + """Hook the download queue so every queued and finished download is recorded.""" + + def _emit_activity_update_for_task(*, payload: dict[str, Any], task: Any) -> None: + owner_user_id = normalize_positive_int(getattr(task, "user_id", None)) + emit_ws_event( + ws_manager, + event_name="activity_update", + room="admins", + payload=payload, + ) + if owner_user_id is None: + return + # The owner knows a request-linked download by its opaque id, as /api/status + # shows it, since the task id can carry a private tracker's passkey (#1418). + task_id = payload.get("task_id") + if isinstance(task_id, str): + payload = { + **payload, + "task_id": viewer_download_id(task_id, getattr(task, "request_id", None)), + } + emit_ws_event( + ws_manager, + event_name="activity_update", + room=f"user_{owner_user_id}", + payload=payload, + ) + + def _record_download_queued(task_id: str, task: Any) -> None: + """Persist initial download record when a task enters the queue.""" + history_service = get_download_history_service() + if history_service is None: + return + + owner_user_id = normalize_positive_int(getattr(task, "user_id", None)) + request_id = normalize_positive_int(getattr(task, "request_id", None)) + origin = "requested" if request_id else "direct" + + source_name = normalize_source(getattr(task, "source", None)) + source_display = get_source_display_name(source_name) + + try: + history_service.record_download( + task_id=task_id, + user_id=owner_user_id, + username=normalize_optional_text(getattr(task, "username", None)), + request_id=request_id, + source=source_name, + source_display_name=source_display, + title=str(getattr(task, "title", "Unknown title") or "Unknown title"), + author=normalize_optional_text(getattr(task, "author", None)), + file_format=normalize_optional_text(getattr(task, "format", None)), + size=normalize_optional_text(getattr(task, "size", None)), + preview=normalize_optional_text(getattr(task, "preview", None)), + content_type=normalize_optional_text(getattr(task, "content_type", None)), + downloads=getattr(task, "downloads", None), + origin=origin, + retry_payload=backend.serialize_task_for_retry(task), + ) + except _OPERATIONAL_ERRORS as exc: + logger.warning("Failed to record download at queue time for task %s: %s", task_id, exc) + return + + view_state_service = get_activity_view_state_service() + if view_state_service is None: + return + + try: + cleared_view_state = 0 + cleared_view_state += view_state_service.clear_item_for_all_viewers( + item_type="download", + item_key=f"download:{task_id}", + ) + if request_id is not None: + cleared_view_state += view_state_service.clear_item_for_all_viewers( + item_type="request", + item_key=f"request:{request_id}", + ) + if cleared_view_state > 0: + _emit_activity_update_for_task( + task=task, + payload={ + "kind": "activity_reset", + "task_id": task_id, + }, + ) + except _OPERATIONAL_ERRORS as exc: + logger.warning("Failed to reset activity viewer state for task %s: %s", task_id, exc) + + def _record_download_terminal_snapshot(task_id: str, status: QueueStatus, task: Any) -> None: + _notify_admin_for_terminal_download_status(task_id=task_id, status=status, task=task) + + final_status = _queue_status_to_final_activity_status(status) + if final_status is None: + return + + history_service = get_download_history_service() + view_state_service = get_activity_view_state_service() + user_db = get_user_db() + finalized_download = False + if history_service is not None: + try: + history_service.finalize_download( + task_id=task_id, + final_status=final_status, + status_message=normalize_optional_text(getattr(task, "status_message", None)), + download_path=normalize_optional_text(getattr(task, "download_path", None)), + retry_payload=backend.serialize_task_for_retry(task), + ) + finalized_download = True + except _OPERATIONAL_ERRORS as exc: + logger.warning("Failed to finalize download history for task %s: %s", task_id, exc) + + if finalized_download: + _emit_activity_update_for_task( + task=task, + payload={ + "kind": "download_terminal", + "task_id": task_id, + "status": final_status, + }, + ) + + if user_db is None or status != QueueStatus.ERROR: + return + + request_id = normalize_positive_int(getattr(task, "request_id", None)) + if request_id is None: + return + if backend.can_retry_download_task(task, status): + return + + raw_error_message = getattr(task, "status_message", None) + fallback_reason = ( + raw_error_message.strip() + if isinstance(raw_error_message, str) and raw_error_message.strip() + else "Download failed" + ) + try: + reopened_request = reopen_failed_request( + user_db, + request_id=request_id, + failure_reason=fallback_reason, + ) + if reopened_request is not None: + if view_state_service is not None: + view_state_service.clear_item_for_all_viewers( + item_type="request", + item_key=f"request:{request_id}", + ) + emit_request_update_events([reopened_request]) + except _OPERATIONAL_ERRORS as exc: + logger.warning( + "Failed to reopen request %s after terminal download error %s: %s", + request_id, + task_id, + exc, + ) + + backend.book_queue.set_queue_hook(_record_download_queued) + backend.book_queue.set_terminal_status_hook(_record_download_terminal_snapshot) diff --git a/shelfmark/main.py b/shelfmark/main.py index 406b75c5..f65d6bb2 100644 --- a/shelfmark/main.py +++ b/shelfmark/main.py @@ -60,18 +60,9 @@ from shelfmark.core.auth_routes import ( # noqa: F401 ) from shelfmark.core.config import config as app_config from shelfmark.core.download_history_service import DownloadHistoryService +from shelfmark.core.download_recording import register_download_recording from shelfmark.core.logger import setup_logger from shelfmark.core.metadata_routes import register_metadata_routes -from shelfmark.core.models import ( - TERMINAL_QUEUE_STATUSES, - QueueStatus, -) -from shelfmark.core.notifications import ( - NotificationContext, - NotificationEvent, - notify_admin, - notify_user, -) from shelfmark.core.onboarding_routes import register_onboarding_routes from shelfmark.core.prefix_middleware import PrefixMiddleware from shelfmark.core.queue_routes import register_queue_routes @@ -81,14 +72,8 @@ from shelfmark.core.request_helpers import ( get_session_db_user_id, load_users_request_policy_settings, normalize_optional_text, - normalize_positive_int, -) -from shelfmark.core.request_policy import ( - normalize_content_type, - normalize_source, ) from shelfmark.core.requests_service import ( - reopen_failed_request, sync_delivery_states_from_queue_status, ) from shelfmark.core.settings_routes import register_settings_routes @@ -101,13 +86,9 @@ from shelfmark.core.utils import normalize_base_path from shelfmark.core.viewer_redaction import ( configure_public_id_key, redact_status, - viewer_download_id, ) from shelfmark.download import orchestrator as backend from shelfmark.download import warmup -from shelfmark.release_sources import ( - get_source_display_name, -) if TYPE_CHECKING: from werkzeug.wrappers import Response @@ -521,225 +502,14 @@ def _resolve_status_scope(*, require_authenticated: bool = True) -> tuple[bool, return False, db_user_id, True -def _queue_status_to_final_activity_status(status: QueueStatus) -> str | None: - return status.value if status in TERMINAL_QUEUE_STATUSES else None - - -def _queue_status_to_notification_event(status: QueueStatus) -> NotificationEvent | None: - if status == QueueStatus.COMPLETE: - return NotificationEvent.DOWNLOAD_COMPLETE - if status == QueueStatus.ERROR: - return NotificationEvent.DOWNLOAD_FAILED - return None - - -def _notify_admin_for_terminal_download_status( - *, task_id: str, status: QueueStatus, task: Any -) -> None: - event = _queue_status_to_notification_event(status) - if event is None: - return - - raw_owner_user_id = getattr(task, "user_id", None) - try: - owner_user_id = int(raw_owner_user_id) if raw_owner_user_id is not None else None - except TypeError, ValueError: - owner_user_id = None - - content_type = normalize_optional_text(getattr(task, "content_type", None)) - context = NotificationContext( - event=event, - title=str(getattr(task, "title", "Unknown title") or "Unknown title"), - author=str(getattr(task, "author", "Unknown author") or "Unknown author"), - username=normalize_optional_text(getattr(task, "username", None)), - content_type=normalize_content_type(content_type) if content_type is not None else None, - format=normalize_optional_text(getattr(task, "format", None)), - source=normalize_source(getattr(task, "source", None)), - error_message=( - normalize_optional_text(getattr(task, "status_message", None)) - if event == NotificationEvent.DOWNLOAD_FAILED - else None - ), - ) - try: - notify_admin(event, context) - except (RuntimeError, TypeError, ValueError) as exc: - logger.warning( - "Failed to trigger admin notification for download %s (%s): %s", - task_id, - status.value, - exc, - ) - if owner_user_id is None: - return - try: - notify_user(owner_user_id, event, context) - except (RuntimeError, TypeError, ValueError) as exc: - logger.warning( - "Failed to trigger user notification for download %s (%s, user_id=%s): %s", - task_id, - status.value, - owner_user_id, - exc, - ) - - -def _emit_activity_update_for_task(*, payload: dict[str, Any], task: Any) -> None: - owner_user_id = normalize_positive_int(getattr(task, "user_id", None)) - emit_ws_event( - ws_manager, - event_name="activity_update", - room="admins", - payload=payload, - ) - if owner_user_id is None: - return - # The owner knows a request-linked download by its opaque id, as /api/status - # shows it, since the task id can carry a private tracker's passkey (#1418). - task_id = payload.get("task_id") - if isinstance(task_id, str): - payload = { - **payload, - "task_id": viewer_download_id(task_id, getattr(task, "request_id", None)), - } - emit_ws_event( - ws_manager, - event_name="activity_update", - room=f"user_{owner_user_id}", - payload=payload, - ) - - -def _record_download_queued(task_id: str, task: Any) -> None: - """Persist initial download record when a task enters the queue.""" - if download_history_service is None: - return - - owner_user_id = normalize_positive_int(getattr(task, "user_id", None)) - request_id = normalize_positive_int(getattr(task, "request_id", None)) - origin = "requested" if request_id else "direct" - - source_name = normalize_source(getattr(task, "source", None)) - source_display = get_source_display_name(source_name) - - try: - download_history_service.record_download( - task_id=task_id, - user_id=owner_user_id, - username=normalize_optional_text(getattr(task, "username", None)), - request_id=request_id, - source=source_name, - source_display_name=source_display, - title=str(getattr(task, "title", "Unknown title") or "Unknown title"), - author=normalize_optional_text(getattr(task, "author", None)), - file_format=normalize_optional_text(getattr(task, "format", None)), - size=normalize_optional_text(getattr(task, "size", None)), - preview=normalize_optional_text(getattr(task, "preview", None)), - content_type=normalize_optional_text(getattr(task, "content_type", None)), - downloads=getattr(task, "downloads", None), - origin=origin, - retry_payload=backend.serialize_task_for_retry(task), - ) - except _OPERATIONAL_ERRORS as exc: - logger.warning("Failed to record download at queue time for task %s: %s", task_id, exc) - return - - if activity_view_state_service is None: - return - - try: - cleared_view_state = 0 - cleared_view_state += activity_view_state_service.clear_item_for_all_viewers( - item_type="download", - item_key=f"download:{task_id}", - ) - if request_id is not None: - cleared_view_state += activity_view_state_service.clear_item_for_all_viewers( - item_type="request", - item_key=f"request:{request_id}", - ) - if cleared_view_state > 0: - _emit_activity_update_for_task( - task=task, - payload={ - "kind": "activity_reset", - "task_id": task_id, - }, - ) - except _OPERATIONAL_ERRORS as exc: - logger.warning("Failed to reset activity viewer state for task %s: %s", task_id, exc) - - -def _record_download_terminal_snapshot(task_id: str, status: QueueStatus, task: Any) -> None: - _notify_admin_for_terminal_download_status(task_id=task_id, status=status, task=task) - - final_status = _queue_status_to_final_activity_status(status) - if final_status is None: - return - - finalized_download = False - if download_history_service is not None: - try: - download_history_service.finalize_download( - task_id=task_id, - final_status=final_status, - status_message=normalize_optional_text(getattr(task, "status_message", None)), - download_path=normalize_optional_text(getattr(task, "download_path", None)), - retry_payload=backend.serialize_task_for_retry(task), - ) - finalized_download = True - except _OPERATIONAL_ERRORS as exc: - logger.warning("Failed to finalize download history for task %s: %s", task_id, exc) - - if finalized_download: - _emit_activity_update_for_task( - task=task, - payload={ - "kind": "download_terminal", - "task_id": task_id, - "status": final_status, - }, - ) - - if user_db is None or status != QueueStatus.ERROR: - return - - request_id = normalize_positive_int(getattr(task, "request_id", None)) - if request_id is None: - return - if backend.can_retry_download_task(task, status): - return - - raw_error_message = getattr(task, "status_message", None) - fallback_reason = ( - raw_error_message.strip() - if isinstance(raw_error_message, str) and raw_error_message.strip() - else "Download failed" - ) - try: - reopened_request = reopen_failed_request( - user_db, - request_id=request_id, - failure_reason=fallback_reason, - ) - if reopened_request is not None: - if activity_view_state_service is not None: - activity_view_state_service.clear_item_for_all_viewers( - item_type="request", - item_key=f"request:{request_id}", - ) - _emit_request_update_events([reopened_request]) - except _OPERATIONAL_ERRORS as exc: - logger.warning( - "Failed to reopen request %s after terminal download error %s: %s", - request_id, - task_id, - exc, - ) - - -backend.book_queue.set_queue_hook(_record_download_queued) -backend.book_queue.set_terminal_status_hook(_record_download_terminal_snapshot) +register_download_recording( + get_user_db=lambda: user_db, + get_download_history_service=lambda: download_history_service, + get_activity_view_state_service=lambda: activity_view_state_service, + # Looked up when called: _emit_request_update_events is defined just below. + emit_request_update_events=lambda rows: _emit_request_update_events(rows), # noqa: PLW0108 + ws_manager=ws_manager, +) def _emit_request_update_events(updated_requests: list[dict[str, Any]]) -> None: diff --git a/tests/core/test_activity_routes_api.py b/tests/core/test_activity_routes_api.py index 8447c4eb..cf1e33e5 100644 --- a/tests/core/test_activity_routes_api.py +++ b/tests/core/test_activity_routes_api.py @@ -1661,7 +1661,7 @@ class TestActivityRoutes: viewer_scope=f"user:{user['id']}", ) - main_module._record_download_queued( + main_module.backend.book_queue._queue_hook( "task-reused-1", SimpleNamespace( user_id=user["id"], diff --git a/tests/core/test_activity_terminal_snapshots.py b/tests/core/test_activity_terminal_snapshots.py index f60bce98..05f69446 100644 --- a/tests/core/test_activity_terminal_snapshots.py +++ b/tests/core/test_activity_terminal_snapshots.py @@ -9,6 +9,7 @@ from unittest.mock import ANY, patch import pytest +from shelfmark.core import download_recording from shelfmark.core.models import DownloadTask, QueueStatus from shelfmark.core.notifications import NotificationEvent from shelfmark.core.viewer_redaction import public_download_id @@ -172,8 +173,8 @@ class TestTerminalSnapshotCapture: assert main_module.backend.book_queue.add(task) is True try: - with patch.object(main_module, "notify_admin") as mock_notify: - with patch.object(main_module, "notify_user") as mock_notify_user: + with patch.object(download_recording, "notify_admin") as mock_notify: + with patch.object(download_recording, "notify_user") as mock_notify_user: main_module.backend.book_queue.update_status(task_id, QueueStatus.COMPLETE) mock_notify.assert_called_once() @@ -264,8 +265,8 @@ class TestTerminalSnapshotCapture: try: main_module.backend.book_queue.update_status_message(task_id, "Resolver timed out") - with patch.object(main_module, "notify_admin") as mock_notify: - with patch.object(main_module, "notify_user") as mock_notify_user: + with patch.object(download_recording, "notify_admin") as mock_notify: + with patch.object(download_recording, "notify_user") as mock_notify_user: main_module.backend.book_queue.update_status(task_id, QueueStatus.ERROR) mock_notify.assert_called_once() @@ -324,7 +325,7 @@ class TestTerminalSnapshotCapture: main_module.backend.book_queue.update_status_message( task_id, "Destination not writable" ) - with patch.object(main_module, "reopen_failed_request") as mock_reopen: + with patch.object(download_recording, "reopen_failed_request") as mock_reopen: main_module.backend.book_queue.update_status(task_id, QueueStatus.ERROR) mock_reopen.assert_not_called() @@ -463,7 +464,7 @@ class TestTerminalSnapshotCapture: with patch.object(main_module.ws_manager, "is_enabled", return_value=True): with patch.object(main_module.ws_manager.socketio, "emit") as mock_emit: - main_module._record_download_queued(task_id, task) + main_module.backend.book_queue._queue_hook(task_id, task) mock_emit.assert_any_call( "activity_update", @@ -501,7 +502,7 @@ class TestTerminalSnapshotCapture: with patch.object(main_module.ws_manager, "is_enabled", return_value=True): with patch.object(main_module.ws_manager.socketio, "emit") as mock_emit: - main_module._record_download_queued(task_id, task) + main_module.backend.book_queue._queue_hook(task_id, task) payloads = _activity_update_payloads(mock_emit) assert payloads["admins"] == {"kind": "activity_reset", "task_id": task_id} @@ -524,8 +525,8 @@ class TestTerminalSnapshotCapture: assert main_module.backend.book_queue.add(task) is True try: - with patch.object(main_module, "notify_admin") as mock_notify: - with patch.object(main_module, "notify_user") as mock_notify_user: + with patch.object(download_recording, "notify_admin") as mock_notify: + with patch.object(download_recording, "notify_user") as mock_notify_user: main_module.backend.book_queue.update_status(task_id, QueueStatus.CANCELLED) mock_notify.assert_not_called() diff --git a/tests/core/test_download_api_guardrails.py b/tests/core/test_download_api_guardrails.py index 51521602..3d416cf6 100644 --- a/tests/core/test_download_api_guardrails.py +++ b/tests/core/test_download_api_guardrails.py @@ -8,7 +8,7 @@ from unittest.mock import patch import pytest -from shelfmark.core.models import DownloadTask +from shelfmark.core.models import DownloadTask, QueueStatus @pytest.fixture(scope="module") @@ -687,7 +687,7 @@ class TestRetryDownloadEndpointGuardrails: with patch.object( main_module.backend.book_queue, "get_task_status", - return_value=main_module.QueueStatus.ERROR, + return_value=QueueStatus.ERROR, ): with patch.object( main_module.backend, "retry_download", return_value=(True, None) diff --git a/tests/core/test_download_recording.py b/tests/core/test_download_recording.py new file mode 100644 index 00000000..e3ccd7d6 --- /dev/null +++ b/tests/core/test_download_recording.py @@ -0,0 +1,455 @@ +"""Pin the download recording hooks that main.py registers on the download queue.""" + +from __future__ import annotations + +import importlib +import sqlite3 +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from shelfmark.core import download_recording +from shelfmark.core.models import QueueStatus +from shelfmark.core.notifications import NotificationEvent +from shelfmark.download import orchestrator as backend + + +@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 + + +def _recording(): + """The module that holds the recording code (notify_*, reopen_failed_request live there).""" + return download_recording + + +def _queued_hook(): + hook = backend.book_queue._queue_hook + assert hook is not None + return hook + + +def _terminal_hook(): + hook = backend.book_queue._terminal_status_hook + assert hook is not None + return hook + + +def _task(**overrides): + values = { + "user_id": 7, + "username": "alice", + "request_id": None, + "source": "direct_download", + "title": "A Title", + "author": "An Author", + "format": "epub", + "size": "1 MB", + "preview": "http://x/p.jpg", + "content_type": "ebook", + "downloads": 3, + "status_message": None, + "download_path": None, + } + values.update(overrides) + return SimpleNamespace(**values) + + +@pytest.fixture +def services(main_module): + history = MagicMock() + view_state = MagicMock() + view_state.clear_item_for_all_viewers.return_value = 0 + user_db = MagicMock() + with ( + patch.object(main_module, "download_history_service", history), + patch.object(main_module, "activity_view_state_service", view_state), + patch.object(main_module, "user_db", user_db), + patch.object(backend, "serialize_task_for_retry", return_value={"k": "v"}), + ): + yield SimpleNamespace(history=history, view_state=view_state, user_db=user_db) + + +@pytest.fixture +def ws_emit(main_module): + with ( + patch.object(main_module.ws_manager, "is_enabled", return_value=True), + patch.object(main_module.ws_manager.socketio, "emit") as emit, + ): + yield emit + + +class TestHookRegistration: + def test_hooks_are_registered_on_the_book_queue(self, main_module): + assert callable(backend.book_queue._queue_hook) + assert callable(backend.book_queue._terminal_status_hook) + + +class TestQueuedRecord: + def test_direct_task_is_recorded_with_all_fields(self, main_module, services): + _queued_hook()("t1", _task()) + + services.history.record_download.assert_called_once_with( + task_id="t1", + user_id=7, + username="alice", + request_id=None, + source="direct_download", + source_display_name=services.history.record_download.call_args.kwargs[ + "source_display_name" + ], + title="A Title", + author="An Author", + file_format="epub", + size="1 MB", + preview="http://x/p.jpg", + content_type="ebook", + downloads=3, + origin="direct", + retry_payload={"k": "v"}, + ) + + def test_request_linked_task_has_requested_origin(self, main_module, services): + _queued_hook()("t2", _task(request_id=42)) + + kwargs = services.history.record_download.call_args.kwargs + assert kwargs["origin"] == "requested" + assert kwargs["request_id"] == 42 + + def test_missing_title_and_bad_user_id_are_normalised(self, main_module, services): + _queued_hook()("t3", _task(title="", user_id="bogus", author=" ")) + + kwargs = services.history.record_download.call_args.kwargs + assert kwargs["title"] == "Unknown title" + assert kwargs["user_id"] is None + assert kwargs["author"] is None + + def test_no_history_service_is_a_noop(self, main_module, services): + with patch.object(main_module, "download_history_service", None): + _queued_hook()("t4", _task()) + + services.view_state.clear_item_for_all_viewers.assert_not_called() + + def test_record_failure_is_swallowed_and_skips_view_state_reset(self, main_module, services): + services.history.record_download.side_effect = sqlite3.OperationalError("locked") + + _queued_hook()("t5", _task()) + + services.view_state.clear_item_for_all_viewers.assert_not_called() + + def test_unexpected_error_type_propagates(self, main_module, services): + services.history.record_download.side_effect = KeyError("boom") + + with pytest.raises(KeyError): + _queued_hook()("t6", _task()) + + def test_no_view_state_service_skips_reset(self, main_module, services, ws_emit): + with patch.object(main_module, "activity_view_state_service", None): + _queued_hook()("t7", _task()) + + services.history.record_download.assert_called_once() + ws_emit.assert_not_called() + + def test_view_state_cleared_for_download_and_request(self, main_module, services, ws_emit): + services.view_state.clear_item_for_all_viewers.return_value = 1 + + _queued_hook()("t8", _task(request_id=42)) + + calls = services.view_state.clear_item_for_all_viewers.call_args_list + assert [c.kwargs for c in calls] == [ + {"item_type": "download", "item_key": "download:t8"}, + {"item_type": "request", "item_key": "request:42"}, + ] + rooms = [c.kwargs["to"] for c in ws_emit.call_args_list] + assert rooms == ["admins", "user_7"] + assert ws_emit.call_args_list[0].args[0] == "activity_update" + assert ws_emit.call_args_list[0].args[1] == {"kind": "activity_reset", "task_id": "t8"} + + def test_nothing_cleared_means_no_ws_event(self, main_module, services, ws_emit): + _queued_hook()("t9", _task()) + + ws_emit.assert_not_called() + + def test_ownerless_task_only_notifies_admins(self, main_module, services, ws_emit): + services.view_state.clear_item_for_all_viewers.return_value = 1 + + _queued_hook()("t10", _task(user_id=None)) + + assert [c.kwargs["to"] for c in ws_emit.call_args_list] == ["admins"] + + def test_view_state_failure_is_swallowed(self, main_module, services, ws_emit): + services.view_state.clear_item_for_all_viewers.side_effect = RuntimeError("x") + + _queued_hook()("t11", _task()) + + ws_emit.assert_not_called() + + +class TestTerminalSnapshot: + @pytest.mark.parametrize( + ("status", "expected"), + [ + (QueueStatus.COMPLETE, "complete"), + (QueueStatus.ERROR, "error"), + (QueueStatus.CANCELLED, "cancelled"), + ], + ) + def test_each_terminal_status_finalizes_history( + self, main_module, services, ws_emit, status, expected + ): + task = _task(status_message=" msg ", download_path="/tmp/x.epub") + with ( + patch.object(_recording(), "notify_admin"), + patch.object(_recording(), "notify_user"), + patch.object(backend, "can_retry_download_task", return_value=True), + ): + _terminal_hook()("t1", status, task) + + services.history.finalize_download.assert_called_once_with( + task_id="t1", + final_status=expected, + status_message="msg", + download_path="/tmp/x.epub", + retry_payload={"k": "v"}, + ) + assert [c.kwargs["to"] for c in ws_emit.call_args_list] == ["admins", "user_7"] + assert ws_emit.call_args_list[0].args[1] == { + "kind": "download_terminal", + "task_id": "t1", + "status": expected, + } + + @pytest.mark.parametrize( + "status", + [QueueStatus.QUEUED, QueueStatus.DOWNLOADING, QueueStatus.RESOLVING], + ) + def test_non_terminal_status_is_ignored(self, main_module, services, ws_emit, status): + with ( + patch.object(_recording(), "notify_admin") as notify_admin, + patch.object(_recording(), "notify_user") as notify_user, + ): + _terminal_hook()("t1", status, _task()) + + services.history.finalize_download.assert_not_called() + notify_admin.assert_not_called() + notify_user.assert_not_called() + ws_emit.assert_not_called() + + def test_finalize_failure_is_swallowed_and_emits_nothing(self, main_module, services, ws_emit): + services.history.finalize_download.side_effect = sqlite3.OperationalError("locked") + with patch.object(_recording(), "notify_admin"), patch.object(_recording(), "notify_user"): + _terminal_hook()("t1", QueueStatus.COMPLETE, _task()) + + ws_emit.assert_not_called() + + def test_no_history_service_still_notifies_but_emits_nothing( + self, main_module, services, ws_emit + ): + with ( + patch.object(main_module, "download_history_service", None), + patch.object(_recording(), "notify_admin") as notify_admin, + patch.object(_recording(), "notify_user"), + ): + _terminal_hook()("t1", QueueStatus.COMPLETE, _task()) + + notify_admin.assert_called_once() + ws_emit.assert_not_called() + + def test_unexpected_finalize_error_propagates(self, main_module, services): + services.history.finalize_download.side_effect = KeyError("boom") + with patch.object(_recording(), "notify_admin"), patch.object(_recording(), "notify_user"): + with pytest.raises(KeyError): + _terminal_hook()("t1", QueueStatus.COMPLETE, _task()) + + +class TestNotifications: + def test_complete_notifies_admin_and_owner(self, main_module, services): + task = _task(title="T", author="A", format="epub", content_type="ebook") + with ( + patch.object(_recording(), "notify_admin") as notify_admin, + patch.object(_recording(), "notify_user") as notify_user, + ): + _terminal_hook()("t1", QueueStatus.COMPLETE, task) + + event, context = notify_admin.call_args.args + assert event == NotificationEvent.DOWNLOAD_COMPLETE + assert (context.title, context.author, context.username) == ("T", "A", "alice") + assert context.format == "epub" + assert context.error_message is None + user_id, user_event, user_context = notify_user.call_args.args + assert (user_id, user_event) == (7, NotificationEvent.DOWNLOAD_COMPLETE) + assert user_context is context + + def test_error_carries_the_error_message(self, main_module, services): + task = _task(status_message="disk full") + with ( + patch.object(_recording(), "notify_admin") as notify_admin, + patch.object(_recording(), "notify_user"), + patch.object(backend, "can_retry_download_task", return_value=True), + ): + _terminal_hook()("t1", QueueStatus.ERROR, task) + + event, context = notify_admin.call_args.args + assert event == NotificationEvent.DOWNLOAD_FAILED + assert context.error_message == "disk full" + + def test_cancelled_sends_no_notification(self, main_module, services): + with ( + patch.object(_recording(), "notify_admin") as notify_admin, + patch.object(_recording(), "notify_user") as notify_user, + ): + _terminal_hook()("t1", QueueStatus.CANCELLED, _task()) + + notify_admin.assert_not_called() + notify_user.assert_not_called() + + def test_ownerless_task_notifies_admin_only(self, main_module, services): + with ( + patch.object(_recording(), "notify_admin") as notify_admin, + patch.object(_recording(), "notify_user") as notify_user, + ): + _terminal_hook()("t1", QueueStatus.COMPLETE, _task(user_id=None)) + + notify_admin.assert_called_once() + notify_user.assert_not_called() + + def test_unparseable_owner_id_notifies_admin_only(self, main_module, services): + with ( + patch.object(_recording(), "notify_admin") as notify_admin, + patch.object(_recording(), "notify_user") as notify_user, + ): + _terminal_hook()("t1", QueueStatus.COMPLETE, _task(user_id="nope")) + + notify_admin.assert_called_once() + notify_user.assert_not_called() + + def test_missing_fields_fall_back_to_unknown(self, main_module, services): + with ( + patch.object(_recording(), "notify_admin") as notify_admin, + patch.object(_recording(), "notify_user"), + ): + _terminal_hook()("t1", QueueStatus.COMPLETE, SimpleNamespace()) + + context = notify_admin.call_args.args[1] + assert context.title == "Unknown title" + assert context.author == "Unknown author" + assert context.content_type is None + + def test_admin_notify_failure_does_not_stop_user_notify_or_finalize( + self, main_module, services + ): + with ( + patch.object(_recording(), "notify_admin", side_effect=RuntimeError("down")), + patch.object(_recording(), "notify_user") as notify_user, + ): + _terminal_hook()("t1", QueueStatus.COMPLETE, _task()) + + notify_user.assert_called_once() + services.history.finalize_download.assert_called_once() + + def test_user_notify_failure_is_swallowed(self, main_module, services): + with ( + patch.object(_recording(), "notify_admin"), + patch.object(_recording(), "notify_user", side_effect=ValueError("bad")), + ): + _terminal_hook()("t1", QueueStatus.COMPLETE, _task()) + + services.history.finalize_download.assert_called_once() + + def test_unexpected_notify_error_propagates(self, main_module, services): + with ( + patch.object(_recording(), "notify_admin", side_effect=KeyError("boom")), + patch.object(_recording(), "notify_user"), + pytest.raises(KeyError), + ): + _terminal_hook()("t1", QueueStatus.COMPLETE, _task()) + + +class TestReopenAfterTerminalError: + def _run(self, main_module, services, *, task, can_retry=False, status=QueueStatus.ERROR): + emitted = MagicMock() + with ( + patch.object(_recording(), "notify_admin"), + patch.object(_recording(), "notify_user"), + patch.object(backend, "can_retry_download_task", return_value=can_retry), + patch.object( + _recording(), "reopen_failed_request", return_value={"id": task.request_id} + ) as reopen, + patch.object(main_module, "_emit_request_update_events", emitted), + ): + _terminal_hook()("t1", status, task) + return reopen, emitted + + def test_error_on_request_linked_task_reopens_and_broadcasts(self, main_module, services): + task = _task(request_id=42, status_message=" kaput ") + reopen, emitted = self._run(main_module, services, task=task) + + reopen.assert_called_once_with(services.user_db, request_id=42, failure_reason="kaput") + services.view_state.clear_item_for_all_viewers.assert_called_once_with( + item_type="request", item_key="request:42" + ) + emitted.assert_called_once_with([{"id": 42}]) + + def test_blank_message_uses_default_reason(self, main_module, services): + reopen, _ = self._run(main_module, services, task=_task(request_id=42, status_message=" ")) + + assert reopen.call_args.kwargs["failure_reason"] == "Download failed" + + def test_retryable_task_is_not_reopened(self, main_module, services): + reopen, emitted = self._run( + main_module, services, task=_task(request_id=42), can_retry=True + ) + + reopen.assert_not_called() + emitted.assert_not_called() + + def test_task_without_request_is_not_reopened(self, main_module, services): + reopen, _ = self._run(main_module, services, task=_task(request_id=None)) + + reopen.assert_not_called() + + def test_complete_is_not_reopened(self, main_module, services): + reopen, _ = self._run( + main_module, services, task=_task(request_id=42), status=QueueStatus.COMPLETE + ) + + reopen.assert_not_called() + + def test_no_user_db_is_not_reopened(self, main_module, services): + with patch.object(main_module, "user_db", None): + reopen, _ = self._run(main_module, services, task=_task(request_id=42)) + + reopen.assert_not_called() + + def test_reopen_returning_none_broadcasts_nothing(self, main_module, services): + task = _task(request_id=42) + emitted = MagicMock() + with ( + patch.object(_recording(), "notify_admin"), + patch.object(_recording(), "notify_user"), + patch.object(backend, "can_retry_download_task", return_value=False), + patch.object(_recording(), "reopen_failed_request", return_value=None), + patch.object(main_module, "_emit_request_update_events", emitted), + ): + _terminal_hook()("t1", QueueStatus.ERROR, task) + + emitted.assert_not_called() + services.view_state.clear_item_for_all_viewers.assert_not_called() + + def test_reopen_failure_is_swallowed(self, main_module, services): + with ( + patch.object(_recording(), "notify_admin"), + patch.object(_recording(), "notify_user"), + patch.object(backend, "can_retry_download_task", return_value=False), + patch.object( + _recording(), "reopen_failed_request", side_effect=sqlite3.OperationalError("x") + ), + ): + _terminal_hook()("t1", QueueStatus.ERROR, _task(request_id=42)) diff --git a/tests/core/test_queue_routes.py b/tests/core/test_queue_routes.py index 7cdda99c..0b2cf8b7 100644 --- a/tests/core/test_queue_routes.py +++ b/tests/core/test_queue_routes.py @@ -17,6 +17,7 @@ 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") @@ -272,7 +273,7 @@ class TestCancel: 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 = main_module.viewer_download_id(real_id, 5) + viewer_id = viewer_download_id(real_id, 5) assert viewer_id != real_id task = _task(real_id, user=user, request_id=None) live = MagicMock()