From 622f6ac6776ca2e4e7cd3a7e0c927b6ebd761a7a Mon Sep 17 00:00:00 2001 From: splitsec2 <35583321+splitsec2@users.noreply.github.com> Date: Sun, 1 Nov 2026 20:09:01 -0600 Subject: [PATCH] refactor: move the download recording into core/download_recording.py (#1642) Last of the planned moves is the download recording (row 9 in Discussion #1536). The queue hooks that write a download to the history when it is queued, finalize it on a terminal status, notify, push the activity update and reopen a failed request move from `main.py` to `core/download_recording.py`, registered with `register_download_recording(...)`. None of it was tested directly, so the first commit adds 38 tests: the queued record for direct and request linked downloads, the terminal snapshot for complete, error and cancelled, ignoring non terminal statuses, who gets notified, reopening a failed request, and what happens when a service is missing or a call fails. They pass on current `main`. The second commit is the move. `_resolve_status_scope` and the request update emitter stay in `main.py` because other code uses them, so the emitter is passed in, and the user database and history service are looked up per call as before. The hooks are registered at the same spot. Some existing test lines changed with the move and no assertion did: the `notify_admin`, `notify_user` and `reopen_failed_request` patches point at the new module, the direct calls to `_record_download_queued` go through the queue's hook, and two tests import `QueueStatus` and `viewer_download_id` themselves now that `main.py` no longer imports them. The URL map is identical, all 77 rules, and the full suite, ruff and basedpyright are clean. ## Summary by CodeRabbit * **Improvements** * Download history and activity updates reflect queued downloads and their completion statuses; activity updates are emitted after successful finalization. * Admins and eligible download owners receive notifications when downloads reach a terminal status. * Requests linked to downloads that fail and cannot be retried are reopened with a failure reason. * Activity updates use an opaque download identifier rather than exposing the underlying task identifier. --- shelfmark/core/download_recording.py | 271 +++++++++++ shelfmark/main.py | 248 +--------- tests/core/test_activity_routes_api.py | 2 +- .../core/test_activity_terminal_snapshots.py | 19 +- tests/core/test_download_api_guardrails.py | 4 +- tests/core/test_download_recording.py | 455 ++++++++++++++++++ tests/core/test_queue_routes.py | 3 +- 7 files changed, 750 insertions(+), 252 deletions(-) create mode 100644 shelfmark/core/download_recording.py create mode 100644 tests/core/test_download_recording.py 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()