mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-10-05 17:51:04 +01:00
CaliBrain
dce91e6972
Retrying an AudiobookBay download re-scraped the detail page for the magnet link before checking the torrent client, so a torrent that had already finished in the client could not be imported while AudiobookBay was down. The handler now keeps the magnet it resolved in the task's retry context, which is persisted with the download history, and a retry hands that magnet straight to the client's existing-download check. The context is handler-owned, so a client-supplied download URL cannot seed it. Fixes #1388
700 lines
26 KiB
Python
700 lines
26 KiB
Python
"""
|
|
Tests for AudiobookBay download handler.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from threading import Event
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from shelfmark.core.models import DownloadTask
|
|
from shelfmark.download.clients import (
|
|
DownloadState,
|
|
DownloadStatus,
|
|
)
|
|
from shelfmark.download.orchestrator import (
|
|
_restore_task_from_retry_payload,
|
|
serialize_task_for_retry,
|
|
)
|
|
from shelfmark.release_sources.audiobookbay.handler import AudiobookBayHandler
|
|
|
|
|
|
class ProgressRecorder:
|
|
"""Records progress and status updates during download."""
|
|
|
|
def __init__(self):
|
|
self.progress_values = []
|
|
self.status_updates = []
|
|
|
|
def progress_callback(self, progress: float):
|
|
self.progress_values.append(progress)
|
|
|
|
def status_callback(self, status: str, message=None):
|
|
self.status_updates.append((status, message))
|
|
|
|
@property
|
|
def last_status(self):
|
|
return self.status_updates[-1][0] if self.status_updates else None
|
|
|
|
@property
|
|
def last_message(self):
|
|
return self.status_updates[-1][1] if self.status_updates else None
|
|
|
|
@property
|
|
def statuses(self):
|
|
return [s[0] for s in self.status_updates]
|
|
|
|
|
|
class TestAudiobookBayHandlerDownload:
|
|
"""Tests for AudiobookBayHandler.download()."""
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
def test_download_success(self, mock_get_client, mock_extract_magnet):
|
|
"""Test successful download initiation."""
|
|
mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123"
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.name = "qbittorrent"
|
|
mock_client.find_existing.return_value = None
|
|
mock_client.add_download.return_value = "download_id_123"
|
|
mock_get_client.return_value = mock_client
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="https://audiobookbay.lu/abss/test-book/",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
)
|
|
cancel_flag = Event()
|
|
recorder = ProgressRecorder()
|
|
with patch.object(
|
|
AudiobookBayHandler, "_poll_and_complete", return_value=None
|
|
) as mock_poll:
|
|
result = handler.download(
|
|
task=task,
|
|
cancel_flag=cancel_flag,
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
assert result is None
|
|
mock_extract_magnet.assert_called_once_with(
|
|
"https://audiobookbay.lu/abss/test-book/", "audiobookbay.lu"
|
|
)
|
|
mock_client.add_download.assert_called_once()
|
|
mock_poll.assert_called_once()
|
|
assert "resolving" in recorder.statuses
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
def test_download_uses_source_url_for_hashed_task_id(
|
|
self, mock_get_client, mock_extract_magnet
|
|
):
|
|
"""Test release queue flow where task_id is source hash and source_url has detail URL."""
|
|
mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123"
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.name = "qbittorrent"
|
|
mock_client.find_existing.return_value = None
|
|
mock_client.add_download.return_value = "download_id_123"
|
|
mock_get_client.return_value = mock_client
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="35f56a3e5734bfa69c3169ee8e605a60",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
source_url="https://audiobookbay.lu/abss/test-book/",
|
|
)
|
|
cancel_flag = Event()
|
|
recorder = ProgressRecorder()
|
|
with patch.object(AudiobookBayHandler, "_poll_and_complete", return_value=None):
|
|
result = handler.download(
|
|
task=task,
|
|
cancel_flag=cancel_flag,
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
assert result is None
|
|
mock_extract_magnet.assert_called_once_with(
|
|
"https://audiobookbay.lu/abss/test-book/", "audiobookbay.lu"
|
|
)
|
|
assert "resolving" in recorder.statuses
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.config.get")
|
|
def test_download_rejects_source_url_host_mismatch(
|
|
self, mock_config_get, mock_get_client, mock_extract_magnet
|
|
):
|
|
"""Test hostile detail URLs are rejected before page fetch."""
|
|
mock_config_get.side_effect = lambda key, default="": (
|
|
"audiobookbay.lu" if key == "ABB_HOSTNAME" else default
|
|
)
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="35f56a3e5734bfa69c3169ee8e605a60",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
source_url="https://169.254.169.254/latest/meta-data/",
|
|
)
|
|
cancel_flag = Event()
|
|
recorder = ProgressRecorder()
|
|
|
|
result = handler.download(
|
|
task=task,
|
|
cancel_flag=cancel_flag,
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
assert result is None
|
|
assert recorder.last_status == "error"
|
|
assert "details url" in recorder.last_message.lower()
|
|
mock_extract_magnet.assert_not_called()
|
|
mock_get_client.assert_not_called()
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.config.get")
|
|
def test_download_allows_configured_source_url_host(
|
|
self, mock_config_get, mock_get_client, mock_extract_magnet
|
|
):
|
|
"""Test configured ABB host remains allowed for queued release URLs."""
|
|
mock_config_get.side_effect = lambda key, default="": (
|
|
"https://audiobookbay.lu/" if key == "ABB_HOSTNAME" else default
|
|
)
|
|
mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123"
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.name = "qbittorrent"
|
|
mock_client.find_existing.return_value = None
|
|
mock_client.add_download.return_value = "download_id_123"
|
|
mock_get_client.return_value = mock_client
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="35f56a3e5734bfa69c3169ee8e605a60",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
source_url="https://audiobookbay.lu/abss/test-book/",
|
|
)
|
|
cancel_flag = Event()
|
|
recorder = ProgressRecorder()
|
|
with patch.object(AudiobookBayHandler, "_poll_and_complete", return_value=None):
|
|
result = handler.download(
|
|
task=task,
|
|
cancel_flag=cancel_flag,
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
assert result is None
|
|
mock_extract_magnet.assert_called_once_with(
|
|
"https://audiobookbay.lu/abss/test-book/", "audiobookbay.lu"
|
|
)
|
|
mock_client.add_download.assert_called_once()
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
def test_download_existing_complete(self, mock_get_client, mock_extract_magnet):
|
|
"""Test handling existing complete download."""
|
|
mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123"
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.name = "qbittorrent"
|
|
mock_client.find_existing.return_value = (
|
|
"existing_id",
|
|
DownloadStatus(
|
|
progress=100,
|
|
state=DownloadState.COMPLETE,
|
|
message="Complete",
|
|
complete=True,
|
|
file_path="/path/to/book.m4b",
|
|
),
|
|
)
|
|
mock_client.get_download_path.return_value = "/path/to/book.m4b"
|
|
mock_get_client.return_value = mock_client
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="https://audiobookbay.lu/abss/test-book/",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
)
|
|
cancel_flag = Event()
|
|
recorder = ProgressRecorder()
|
|
|
|
with patch.object(
|
|
AudiobookBayHandler,
|
|
"_wait_for_completed_path",
|
|
return_value=(Path("/path/to/book.m4b"), None),
|
|
):
|
|
result = handler.download(
|
|
task=task,
|
|
cancel_flag=cancel_flag,
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
assert result == "/path/to/book.m4b"
|
|
mock_client.add_download.assert_not_called()
|
|
|
|
@pytest.mark.parametrize("restarted", [False, True], ids=["in_memory", "after_restart"])
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
def test_retry_imports_finished_torrent_while_site_is_down(
|
|
self, mock_get_client, mock_extract_magnet, restarted
|
|
):
|
|
"""A retry reuses the first attempt's magnet instead of re-scraping AudiobookBay (#1388)."""
|
|
magnet = "magnet:?xt=urn:btih:abc123"
|
|
mock_extract_magnet.return_value = magnet
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.name = "qbittorrent"
|
|
mock_client.find_existing.return_value = None
|
|
mock_client.add_download.return_value = "abc123"
|
|
mock_get_client.return_value = mock_client
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="35f56a3e5734bfa69c3169ee8e605a60",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
source_url="https://audiobookbay.lu/abss/test-book/",
|
|
)
|
|
recorder = ProgressRecorder()
|
|
|
|
# First attempt hands the magnet to the client, then gives up (e.g. stall timeout).
|
|
with patch.object(AudiobookBayHandler, "_poll_and_complete", return_value=None):
|
|
handler.download(
|
|
task=task,
|
|
cancel_flag=Event(),
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
mock_client.add_download.assert_called_once()
|
|
|
|
if restarted:
|
|
task = _restore_task_from_retry_payload(serialize_task_for_retry(task))
|
|
assert task is not None
|
|
|
|
# AudiobookBay goes down while the torrent finishes in the client.
|
|
mock_extract_magnet.reset_mock()
|
|
mock_extract_magnet.return_value = None
|
|
mock_client.find_existing.return_value = (
|
|
"abc123",
|
|
DownloadStatus(
|
|
progress=100,
|
|
state=DownloadState.COMPLETE,
|
|
message="Complete",
|
|
complete=True,
|
|
file_path="/path/to/book.m4b",
|
|
),
|
|
)
|
|
|
|
with patch.object(
|
|
AudiobookBayHandler,
|
|
"_wait_for_completed_path",
|
|
return_value=(Path("/path/to/book.m4b"), None),
|
|
):
|
|
result = handler.download(
|
|
task=task,
|
|
cancel_flag=Event(),
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
assert result == "/path/to/book.m4b"
|
|
mock_extract_magnet.assert_not_called()
|
|
assert mock_client.find_existing.call_args.args[0] == magnet
|
|
mock_client.add_download.assert_called_once()
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
def test_failed_scrape_caches_no_magnet(self, mock_get_client, mock_extract_magnet):
|
|
"""Only a magnet that was actually resolved is reused, so a retry scrapes again."""
|
|
mock_extract_magnet.return_value = None
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="35f56a3e5734bfa69c3169ee8e605a60",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
source_url="https://audiobookbay.lu/abss/test-book/",
|
|
)
|
|
recorder = ProgressRecorder()
|
|
|
|
for _ in range(2):
|
|
handler.download(
|
|
task=task,
|
|
cancel_flag=Event(),
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
assert mock_extract_magnet.call_count == 2
|
|
assert task.retry_source_context == {}
|
|
mock_get_client.assert_not_called()
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
def test_download_existing_in_progress(self, mock_get_client, mock_extract_magnet):
|
|
"""Test handling existing in-progress download."""
|
|
mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123"
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.name = "qbittorrent"
|
|
mock_client.find_existing.return_value = (
|
|
"existing_id",
|
|
DownloadStatus(
|
|
progress=50,
|
|
state=DownloadState.DOWNLOADING,
|
|
message="Downloading",
|
|
complete=False,
|
|
file_path=None,
|
|
),
|
|
)
|
|
mock_get_client.return_value = mock_client
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="https://audiobookbay.lu/abss/test-book/",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
)
|
|
cancel_flag = Event()
|
|
recorder = ProgressRecorder()
|
|
|
|
with patch.object(
|
|
AudiobookBayHandler, "_poll_and_complete", return_value=None
|
|
) as mock_poll:
|
|
result = handler.download(
|
|
task=task,
|
|
cancel_flag=cancel_flag,
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
assert result is None
|
|
assert "downloading" in recorder.statuses
|
|
mock_client.add_download.assert_not_called()
|
|
mock_poll.assert_called_once()
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
def test_download_cancellation(self, mock_get_client, mock_extract_magnet):
|
|
"""Test that cancellation is respected."""
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="https://audiobookbay.lu/abss/test-book/",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
)
|
|
cancel_flag = Event()
|
|
cancel_flag.set() # Set immediately
|
|
recorder = ProgressRecorder()
|
|
|
|
result = handler.download(
|
|
task=task,
|
|
cancel_flag=cancel_flag,
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
assert result is None
|
|
assert "cancelled" in recorder.statuses
|
|
mock_extract_magnet.assert_not_called()
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
def test_download_no_magnet_link(self, mock_get_client, mock_extract_magnet):
|
|
"""Test handling when magnet link extraction fails."""
|
|
mock_extract_magnet.return_value = None
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="https://audiobookbay.lu/abss/test-book/",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
)
|
|
cancel_flag = Event()
|
|
recorder = ProgressRecorder()
|
|
|
|
with patch.object(
|
|
AudiobookBayHandler,
|
|
"_wait_for_completed_path",
|
|
return_value=(None, "Could not locate existing download path"),
|
|
):
|
|
result = handler.download(
|
|
task=task,
|
|
cancel_flag=cancel_flag,
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
assert result is None
|
|
assert recorder.last_status == "error"
|
|
assert "magnet link" in recorder.last_message.lower()
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.list_configured_clients")
|
|
def test_download_no_client_configured(
|
|
self, mock_list_clients, mock_get_client, mock_extract_magnet
|
|
):
|
|
"""Test handling when no torrent client is configured."""
|
|
mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123"
|
|
mock_get_client.return_value = None
|
|
mock_list_clients.return_value = []
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="https://audiobookbay.lu/abss/test-book/",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
)
|
|
cancel_flag = Event()
|
|
recorder = ProgressRecorder()
|
|
|
|
result = handler.download(
|
|
task=task,
|
|
cancel_flag=cancel_flag,
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
assert result is None
|
|
assert recorder.last_status == "error"
|
|
assert "client" in recorder.last_message.lower()
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
def test_download_client_add_failure(self, mock_get_client, mock_extract_magnet):
|
|
"""Test handling when client.add_download fails."""
|
|
mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123"
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.name = "qbittorrent"
|
|
mock_client.find_existing.return_value = None
|
|
mock_client.add_download.side_effect = Exception("Client error")
|
|
mock_get_client.return_value = mock_client
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="https://audiobookbay.lu/abss/test-book/",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
)
|
|
cancel_flag = Event()
|
|
recorder = ProgressRecorder()
|
|
|
|
result = handler.download(
|
|
task=task,
|
|
cancel_flag=cancel_flag,
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
assert result is None
|
|
assert recorder.last_status == "error"
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
def test_download_existing_no_path(self, mock_get_client, mock_extract_magnet):
|
|
"""Test handling when existing download has no path."""
|
|
mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123"
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.name = "qbittorrent"
|
|
mock_client.find_existing.return_value = (
|
|
"existing_id",
|
|
DownloadStatus(
|
|
progress=100,
|
|
state=DownloadState.COMPLETE,
|
|
message="Complete",
|
|
complete=True,
|
|
file_path=None,
|
|
),
|
|
)
|
|
mock_client.get_download_path.return_value = None
|
|
mock_get_client.return_value = mock_client
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="https://audiobookbay.lu/abss/test-book/",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
)
|
|
cancel_flag = Event()
|
|
recorder = ProgressRecorder()
|
|
|
|
result = handler.download(
|
|
task=task,
|
|
cancel_flag=cancel_flag,
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
assert result is None
|
|
assert recorder.last_status == "error"
|
|
assert "path" in recorder.last_message.lower()
|
|
|
|
|
|
class TestAudiobookBayHandlerCategory:
|
|
"""Tests for category selection."""
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.config.get")
|
|
def test_category_selection_qbittorrent_audiobook(
|
|
self, mock_config_get, mock_get_client, mock_extract_magnet
|
|
):
|
|
"""Test audiobook category selection for qBittorrent."""
|
|
mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123"
|
|
|
|
def config_get(key, default=""):
|
|
if key == "QBITTORRENT_CATEGORY_AUDIOBOOK":
|
|
return "audiobooks"
|
|
return default
|
|
|
|
mock_config_get.side_effect = config_get
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.name = "qbittorrent"
|
|
mock_client.find_existing.return_value = None
|
|
mock_client.add_download.return_value = "download_id"
|
|
mock_get_client.return_value = mock_client
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="https://audiobookbay.lu/abss/test-book/",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
)
|
|
cancel_flag = Event()
|
|
recorder = ProgressRecorder()
|
|
|
|
with patch.object(AudiobookBayHandler, "_poll_and_complete", return_value=None):
|
|
handler.download(
|
|
task=task,
|
|
cancel_flag=cancel_flag,
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
# Verify category was passed
|
|
call_kwargs = mock_client.add_download.call_args.kwargs
|
|
assert call_kwargs["category"] == "audiobooks"
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.config.get")
|
|
def test_category_selection_transmission_general(
|
|
self, mock_config_get, mock_get_client, mock_extract_magnet
|
|
):
|
|
"""Test Transmission audiobook category does not fall back to general category."""
|
|
mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123"
|
|
|
|
def config_get(key, default=""):
|
|
if key == "TRANSMISSION_CATEGORY":
|
|
return "books"
|
|
return default
|
|
|
|
mock_config_get.side_effect = config_get
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.name = "transmission"
|
|
mock_client.find_existing.return_value = None
|
|
mock_client.add_download.return_value = "download_id"
|
|
mock_get_client.return_value = mock_client
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="https://audiobookbay.lu/abss/test-book/",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="audiobook",
|
|
)
|
|
cancel_flag = Event()
|
|
recorder = ProgressRecorder()
|
|
|
|
with patch.object(AudiobookBayHandler, "_poll_and_complete", return_value=None):
|
|
handler.download(
|
|
task=task,
|
|
cancel_flag=cancel_flag,
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
# Transmission audiobook downloads use only the audiobook category key.
|
|
call_kwargs = mock_client.add_download.call_args.kwargs
|
|
assert call_kwargs["category"] is None
|
|
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
|
|
@patch("shelfmark.release_sources.audiobookbay.handler.config.get")
|
|
def test_category_selection_non_audiobook(
|
|
self, mock_config_get, mock_get_client, mock_extract_magnet
|
|
):
|
|
"""Test that non-audiobook content types don't get category."""
|
|
mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123"
|
|
|
|
mock_config_get.return_value = ""
|
|
|
|
mock_client = MagicMock()
|
|
mock_client.name = "qbittorrent"
|
|
mock_client.find_existing.return_value = None
|
|
mock_client.add_download.return_value = "download_id"
|
|
mock_get_client.return_value = mock_client
|
|
|
|
handler = AudiobookBayHandler()
|
|
task = DownloadTask(
|
|
task_id="https://audiobookbay.lu/abss/test-book/",
|
|
source="audiobookbay",
|
|
title="Test Book",
|
|
content_type="ebook", # Not audiobook
|
|
)
|
|
cancel_flag = Event()
|
|
recorder = ProgressRecorder()
|
|
|
|
with patch.object(AudiobookBayHandler, "_poll_and_complete", return_value=None):
|
|
handler.download(
|
|
task=task,
|
|
cancel_flag=cancel_flag,
|
|
progress_callback=recorder.progress_callback,
|
|
status_callback=recorder.status_callback,
|
|
)
|
|
|
|
# Verify no category was passed
|
|
call_kwargs = mock_client.add_download.call_args.kwargs
|
|
assert call_kwargs["category"] is None
|
|
|
|
|
|
class TestAudiobookBayHandlerCancel:
|
|
"""Tests for AudiobookBayHandler.cancel()."""
|
|
|
|
def test_cancel_returns_false(self):
|
|
"""Test that cancel always returns False (torrents can't be cancelled)."""
|
|
handler = AudiobookBayHandler()
|
|
result = handler.cancel("test-task-id")
|
|
assert result is False
|