starred/shelfmark
1
0
Fork 0
mirror of https://github.com/calibrain/shelfmark.git synced 2026-09-24 20:50:28 +01:00
Code Issues Packages Projects Releases 10 Wiki Activity
Files
e7007865a4b8b1d8b2a2403f9b64796d7168b449
shelfmark/tests/prowlarr/test_source.py
T
CaliBrain e7007865a4 fix(prowlarr): stop turning indexer failures into empty results and 404s (#1251) 
Two independent bugs, both from an indexer that Prowlarr proxies rather
than answers for itself: the search never reported that it had failed,
and the grab never resolved what it was handed.

Search. A Torznab search is Prowlarr proxying a live request out to the
tracker, so for a Cloudflare-fronted indexer it waits on FlareSolverr.
The client gave it the 30s budget sized for Prowlarr's own JSON
endpoints, then swallowed every failure -- the timeout, the 429 Prowlarr
returns once it has disabled an indexer, a parse error -- into the same
empty list that means "this indexer has nothing". A cold challenge
routinely runs past a minute, so the UI said "No releases found for this
book" while FlareSolverr was still solving. That empty list also drove
the auto-expand retry, which fires on "no results with the category
filter". A timeout satisfies it, so Shelfmark sent a second search to an
indexer still busy with the first -- two Chromes at once, enough to take
FlareSolverr's down on a small host.

torznab_search now raises ProwlarrSearchError, and an empty list
strictly
means the indexer answered with no matches. The source records which
indexer searches failed: one dead indexer no longer sinks the others,
auto-expand runs only when every indexer genuinely answered, and zero
results with at least one failure raises SourceUnavailableError, which
the releases endpoint already turns into a 503 carrying a real message.
Prowlarr being unreachable was the same lie by another route -- the
indexer list came back empty, leaving nothing to query -- and now says
so.

Indexer searches also get their own timeout, PROWLARR_INDEXER_TIMEOUT,
defaulting to 90s and clamped to 5-300. Prowlarr's status and indexer
list keep 30s so Test Connection stays responsive, and the connect
timeout is split out at 10s so an unreachable Prowlarr fails fast rather
than hanging for the whole read budget. The overall per-request search
budget now scales to twice the indexer timeout, capped at 240s, so
raising the setting is not undone by the cap one level up while staying
under the 300s gunicorn worker timeout.

Grab. Prowlarr hands out a proxy download URL, with no magnetUrl and no
infoHash, for any indexer that only publishes torrent files. The native
Real-Debrid client built its magnet as "if not
url.startswith('magnet:') and expected_hash", so with no hash to work
from it left the URL alone and POSTed it to /torrents/addMagnet as the
magnet field. Real-Debrid answered 404 and the grab died on a raw HTTP
error. AllDebrid carried the same line and the same bug.

Both now resolve the URL first, through the extract_torrent_info path
the
torrent clients have used since #1108: pass a magnet through untouched,
follow a redirect or a response body that turns out to be a magnet,
otherwise upload the fetched .torrent, and fall back to a magnet built
from the infoHash only when the fetch failed. The file is preferred over
a synthesized urn:btih: magnet because it carries the tracker list; a
bare hash leaves the service to find the swarm on DHT alone. Fetches are
shared with the rest of the add path through the torrent fetch cache, so
resolving costs at most one request. Real-Debrid takes the file on PUT
/torrents/addTorrent with the raw bytes as the request body, AllDebrid
on
POST /magnet/upload/file as multipart files[]. A URL that resolves to
neither form now raises before any request reaches the service, so the
user reads why instead of a 404. Neither debrid client had any test
coverage; both have some now.

Fixes #1249
Fixes #1250
2026-08-21 09:07:03 -04:00

1401 lines
51 KiB
Python
Raw Blame History

"""
Tests for the Prowlarr source module.
Tests the utility functions for parsing release metadata.
"""
# Import the functions to test
import pytest
import requests
from shelfmark.metadata_providers import BookMetadata
from shelfmark.release_sources import SourceUnavailableError
from shelfmark.release_sources.prowlarr.api import ProwlarrClient, ProwlarrSearchError
from shelfmark.release_sources.prowlarr.source import (
PROWLARR_SEARCH_TIMEOUT_SECONDS,
ProwlarrSource,
_build_indexer_priority,
_collapse_duplicate_indexer_results,
_detect_content_type_from_categories,
_extract_format,
_extract_mam_language,
_fetch_indexer_seed_settings,
_last_known_seed_settings,
_parse_size,
_release_identity,
_result_dedup_key,
_search_budget_seconds,
)
from shelfmark.release_sources.prowlarr.utils import (
build_source_id,
get_protocol_display,
sanitize_download_url,
)
class _AvailableSource:
display_name = "Prowlarr"
def is_available(self):
return True
@pytest.fixture(autouse=True)
def source_available_by_default(monkeypatch):
import shelfmark.download.orchestrator as orchestrator
monkeypatch.setattr(orchestrator, "get_source", lambda _source: _AvailableSource())
class TestParseSize:
"""Tests for the _parse_size function."""
def test_parse_size_bytes(self):
"""Test parsing small byte sizes."""
assert _parse_size(100) == "100 B"
assert _parse_size(512) == "512 B"
def test_parse_size_kilobytes(self):
"""Test parsing kilobyte sizes."""
assert _parse_size(1024) == "1.0 KB"
assert _parse_size(2048) == "2.0 KB"
assert _parse_size(1536) == "1.5 KB"
def test_parse_size_megabytes(self):
"""Test parsing megabyte sizes."""
assert _parse_size(1048576) == "1.0 MB"
assert _parse_size(5242880) == "5.0 MB"
assert _parse_size(1572864) == "1.5 MB"
def test_parse_size_gigabytes(self):
"""Test parsing gigabyte sizes."""
assert _parse_size(1073741824) == "1.0 GB"
assert _parse_size(2147483648) == "2.0 GB"
def test_parse_size_terabytes(self):
"""Test parsing terabyte sizes."""
assert _parse_size(1099511627776) == "1.0 TB"
def test_parse_size_none(self):
"""Test that None returns None."""
assert _parse_size(None) is None
def test_parse_size_zero(self):
"""Test that zero returns None."""
assert _parse_size(0) is None
def test_parse_size_negative(self):
"""Test that negative values return None."""
assert _parse_size(-100) is None
class TestExtractFormat:
"""Tests for the _extract_format function."""
def test_extract_format_from_extension(self):
"""Test extracting format from file extension."""
assert _extract_format("The Book.epub") == "epub"
assert _extract_format("The Book.mobi") == "mobi"
assert _extract_format("The Book.pdf") == "pdf"
assert _extract_format("The Book.azw3") == "azw3"
def test_extract_format_from_brackets(self):
"""Test extracting format from brackets."""
assert _extract_format("The Book [EPUB]") == "epub"
assert _extract_format("The Book (PDF)") == "pdf"
assert _extract_format("The Book {MOBI}") == "mobi"
def test_extract_format_from_word(self):
"""Test extracting format as standalone word."""
assert _extract_format("The Book epub version") == "epub"
assert _extract_format("mobi edition of the book") == "mobi"
def test_extract_format_priority_extension_over_bracket(self):
"""Test that file extension takes priority over brackets."""
# Extension is more reliable
assert _extract_format("The Book [PDF].epub") == "epub"
def test_extract_format_case_insensitive(self):
"""Test that format extraction is case insensitive."""
assert _extract_format("The Book.EPUB") == "epub"
assert _extract_format("The Book [PDF]") == "pdf"
assert _extract_format("The Book.Mobi") == "mobi"
def test_extract_format_none_when_no_format(self):
"""Test that None is returned when no format found."""
assert _extract_format("The Book by Author") is None
assert _extract_format("") is None
def test_extract_format_cbz_cbr(self):
"""Test comic book formats."""
assert _extract_format("Comic Issue 1.cbz") == "cbz"
assert _extract_format("Comic Issue 2.cbr") == "cbr"
def test_extract_format_fb2(self):
"""Test FB2 format (common in Russian ebooks)."""
assert _extract_format("Russian Book.fb2") == "fb2"
assert _extract_format("Book [FB2]") == "fb2"
def test_extract_format_djvu(self):
"""Test DjVu format."""
assert _extract_format("Scanned Book.djvu") == "djvu"
def test_extract_format_avoids_false_positives(self):
"""Test that format extraction doesn't match partial words."""
# "republic" should not match "pdf" or other formats
assert _extract_format("The Republic by Plato") is None
# "literal" should not match "lit"
assert _extract_format("Literal Translation") is None
class TestGetProtocolDisplay:
"""Tests for the get_protocol_display function."""
def test_get_protocol_from_protocol_field_torrent(self):
"""Test extracting torrent protocol from protocol field."""
result = {"protocol": "torrent", "downloadUrl": "https://example.com"}
assert get_protocol_display(result) == "torrent"
def test_get_protocol_from_protocol_field_usenet(self):
"""Test extracting usenet protocol from protocol field."""
result = {"protocol": "usenet", "downloadUrl": "https://example.com"}
assert get_protocol_display(result) == "nzb"
def test_get_protocol_from_magnet_url(self):
"""Test inferring torrent from magnet URL."""
result = {"downloadUrl": "magnet:?xt=urn:btih:abc123"}
assert get_protocol_display(result) == "torrent"
def test_get_protocol_from_torrent_url(self):
"""Test inferring torrent from .torrent URL."""
result = {"downloadUrl": "https://example.com/file.torrent"}
assert get_protocol_display(result) == "torrent"
def test_get_protocol_from_nzb_url(self):
"""Test inferring NZB from .nzb URL."""
result = {"downloadUrl": "https://example.com/file.nzb"}
assert get_protocol_display(result) == "nzb"
def test_get_protocol_fallback_to_magnet_url(self):
"""Test fallback to magnetUrl field."""
result = {"magnetUrl": "magnet:?xt=urn:btih:abc123"}
assert get_protocol_display(result) == "torrent"
def test_get_protocol_unknown(self):
"""Test unknown protocol for unclear URLs."""
result = {"downloadUrl": "https://example.com/download"}
assert get_protocol_display(result) == "unknown"
def test_get_protocol_case_insensitive(self):
"""Test protocol detection is case insensitive."""
result = {"protocol": "TORRENT"}
assert get_protocol_display(result) == "torrent"
result = {"protocol": "Usenet"}
assert get_protocol_display(result) == "nzb"
class TestSanitizeDownloadUrl:
"""Tests for the sanitize_download_url helper."""
def test_sanitizes_apikey_whitespace(self):
"""Strip whitespace around apikey separators."""
url = "http://prowlarr:9696/5/download?apikey = 12345"
assert sanitize_download_url(url) == "http://prowlarr:9696/5/download?apikey=12345"
def test_sanitizes_multiple_query_params(self):
"""Sanitize all query pairs while keeping params."""
url = "http://prowlarr:9696/5/download?apikey = 12345&indexer = 7"
assert (
sanitize_download_url(url) == "http://prowlarr:9696/5/download?apikey=12345&indexer=7"
)
def test_leaves_non_http_urls_untouched(self):
"""Do not modify magnet or other non-http URLs."""
url = "magnet:?xt=urn:btih:abc123"
assert sanitize_download_url(url) == url
def test_leaves_clean_urls_untouched(self):
"""Return clean URLs as-is."""
url = "https://prowlarr:9696/5/download?apikey=12345"
assert sanitize_download_url(url) == url
class TestDetectContentType:
"""Tests for the _detect_content_type_from_categories function."""
def test_fallback_without_categories(self):
assert _detect_content_type_from_categories([], "ebook") == "book"
assert _detect_content_type_from_categories([], "audiobook") == "audiobook"
def test_audiobook_categories(self):
assert _detect_content_type_from_categories([{"id": 3030}], "ebook") == "audiobook"
assert _detect_content_type_from_categories([3000], "ebook") == "audiobook"
def test_book_category_range(self):
assert _detect_content_type_from_categories([{"id": 7000}], "ebook") == "book"
assert _detect_content_type_from_categories([7020], "audiobook") == "book"
assert _detect_content_type_from_categories([7030], "ebook") == "book"
def test_non_book_categories_return_other(self):
assert _detect_content_type_from_categories([{"id": 2000}], "ebook") == "other"
class FakeTorznabClient:
def __init__(self, search_results=None, seed_settings=None):
self.calls: list[tuple[str, object]] = []
self.queries: list[str] = []
self.seed_settings_calls: list[object] = []
self.search_results = search_results or []
self.seed_settings = seed_settings or {}
self.indexer_timeout = 90
def get_enabled_indexers_detailed(self, *, raise_on_error=False):
return [
{
"id": 1,
"enable": True,
"capabilities": {
"categories": [
{"id": 7000, "subCategories": []},
{"id": 3030, "subCategories": []},
]
},
}
]
def torznab_search(
self,
*,
indexer_id: int,
query: str,
categories=None,
search_type="book",
limit=100,
offset=0,
):
del indexer_id, search_type, limit, offset
self.calls.append((query, categories))
self.queries.append(query)
return self.search_results
def get_enriched_indexer_ids(self, restrict_to=None, indexers=None):
del restrict_to, indexers
return []
def get_indexer_seed_settings(self, restrict_to=None):
self.seed_settings_calls.append(restrict_to)
return self.seed_settings
class TestProwlarrIndexerSeedSettings:
def test_get_indexer_seed_settings_reads_prowlarr_minutes_field(self, monkeypatch):
client = ProwlarrClient("http://prowlarr:9696", "apikey")
monkeypatch.setattr(
client,
"get_enabled_indexers_detailed",
lambda *, raise_on_error=False: [
{
"id": 13,
"protocol": "torrent",
"fields": [
{"name": "torrentBaseSettings.seedRatio", "value": "2.5"},
{"name": "torrentBaseSettings.seedTime", "value": "7200"},
],
},
{
"id": 14,
"protocol": "usenet",
"fields": [
{"name": "torrentBaseSettings.seedRatio", "value": "3"},
{"name": "torrentBaseSettings.seedTime", "value": "9999"},
],
},
],
)
assert client.get_indexer_seed_settings() == {
13: {"ratio_limit": 2.5, "seeding_time_limit_minutes": 7200}
}
class TestProwlarrLocalizedQueries:
def test_manual_query_still_applies_content_type_categories(self, monkeypatch):
import shelfmark.release_sources.prowlarr.source as prowlarr_source
def fake_get(key: str, default=None):
values = {
"PROWLARR_INDEXERS": "",
"PROWLARR_AUTO_EXPAND": False,
}
return values.get(key, default)
monkeypatch.setattr(prowlarr_source.config, "get", fake_get)
fake_client = FakeTorznabClient()
source = ProwlarrSource()
monkeypatch.setattr(source, "_get_client", lambda: fake_client)
book = BookMetadata(
provider="hardcover",
provider_id="123",
title="Anything",
authors=["Someone"],
)
from shelfmark.core.search_plan import build_release_search_plan
plan = build_release_search_plan(book, languages=["en"], manual_query="my custom")
source.search(book, plan, content_type="audiobook")
assert fake_client.calls == [("my custom", [3030])]
def test_manual_query_expand_removes_categories(self, monkeypatch):
import shelfmark.release_sources.prowlarr.source as prowlarr_source
def fake_get(key: str, default=None):
values = {
"PROWLARR_INDEXERS": "",
"PROWLARR_AUTO_EXPAND": False,
}
return values.get(key, default)
monkeypatch.setattr(prowlarr_source.config, "get", fake_get)
fake_client = FakeTorznabClient()
source = ProwlarrSource()
monkeypatch.setattr(source, "_get_client", lambda: fake_client)
book = BookMetadata(
provider="hardcover",
provider_id="123",
title="Anything",
authors=["Someone"],
)
from shelfmark.core.search_plan import build_release_search_plan
plan = build_release_search_plan(book, languages=["en"], manual_query="my custom")
source.search(book, plan, expand_search=True, content_type="audiobook")
assert fake_client.calls == [("my custom", None)]
def test_search_attaches_configured_seed_time_minutes_to_release(self, monkeypatch):
import shelfmark.release_sources.prowlarr.source as prowlarr_source
def fake_get(key: str, default=None):
values = {
"PROWLARR_INDEXERS": "",
"PROWLARR_AUTO_EXPAND": False,
"PROWLARR_USE_SEED_PREFERENCES": True,
}
return values.get(key, default)
monkeypatch.setattr(prowlarr_source.config, "get", fake_get)
fake_client = FakeTorznabClient(
search_results=[
{
"guid": "mam-result-1",
"protocol": "torrent",
"title": "Test Release",
"magnetUrl": "magnet:?xt=urn:btih:abc123",
"indexerId": 1,
"indexer": "MyAnonamouse",
"minimumSeedTime": 259200,
"minimumRatio": 1,
}
],
seed_settings={1: {"ratio_limit": 2.0, "seeding_time_limit_minutes": 7200}},
)
source = ProwlarrSource()
monkeypatch.setattr(source, "_get_client", lambda: fake_client)
book = BookMetadata(
provider="hardcover",
provider_id="123",
title="Anything",
authors=["Someone"],
)
from shelfmark.core.search_plan import build_release_search_plan
plan = build_release_search_plan(book, languages=["en"])
releases = source.search(book, plan, content_type="ebook")
assert len(releases) == 1
assert fake_client.seed_settings_calls == [None]
assert releases[0].extra["configured_ratio_limit"] == 2.0
assert releases[0].extra["configured_seed_time_minutes"] == 7200
assert "minimum_seed_time" not in releases[0].extra
assert "minimum_ratio" not in releases[0].extra
def test_search_ignores_configured_seed_time_when_disabled(self, monkeypatch):
import shelfmark.release_sources.prowlarr.source as prowlarr_source
def fake_get(key: str, default=None):
values = {
"PROWLARR_INDEXERS": "",
"PROWLARR_AUTO_EXPAND": False,
"PROWLARR_USE_SEED_PREFERENCES": False,
}
return values.get(key, default)
monkeypatch.setattr(prowlarr_source.config, "get", fake_get)
fake_client = FakeTorznabClient(
search_results=[
{
"guid": "mam-result-1",
"protocol": "torrent",
"title": "Test Release",
"magnetUrl": "magnet:?xt=urn:btih:abc123",
"indexerId": 1,
"indexer": "MyAnonamouse",
}
],
seed_settings={1: {"ratio_limit": 2.0, "seeding_time_limit_minutes": 7200}},
)
source = ProwlarrSource()
monkeypatch.setattr(source, "_get_client", lambda: fake_client)
book = BookMetadata(
provider="hardcover",
provider_id="123",
title="Anything",
authors=["Someone"],
)
from shelfmark.core.search_plan import build_release_search_plan
plan = build_release_search_plan(book, languages=["en"])
releases = source.search(book, plan, content_type="ebook")
assert len(releases) == 1
assert fake_client.seed_settings_calls == []
assert releases[0].extra["configured_ratio_limit"] is None
assert releases[0].extra["configured_seed_time_minutes"] is None
def test_redacted_search_result_still_builds_private_retry_payload(self, monkeypatch):
import shelfmark.download.orchestrator as orchestrator
import shelfmark.release_sources.prowlarr.source as prowlarr_source
secret_download_url = "https://prowlarr.example.com/1/download?apikey=secret"
def fake_get(key: str, default=None, user_id=None):
values = {
"PROWLARR_INDEXERS": "",
"PROWLARR_AUTO_EXPAND": False,
"PROWLARR_USE_SEED_PREFERENCES": False,
}
return values.get(key, default)
monkeypatch.setattr(prowlarr_source.config, "get", fake_get)
monkeypatch.setattr(orchestrator.config, "get", fake_get)
fake_client = FakeTorznabClient(
search_results=[
{
"guid": "secret-prowlarr-release",
"protocol": "usenet",
"title": "Secret Bearing Release",
"downloadUrl": secret_download_url,
}
]
)
source = ProwlarrSource()
monkeypatch.setattr(source, "_get_client", lambda: fake_client)
book = BookMetadata(
provider="hardcover",
provider_id="123",
title="Anything",
authors=["Someone"],
)
from shelfmark.core.search_plan import build_release_search_plan
plan = build_release_search_plan(book, languages=["en"])
releases = source.search(book, plan, content_type="ebook")
assert len(releases) == 1
assert releases[0].download_url is None
captured_tasks = []
def fake_add(task):
captured_tasks.append(task)
return True
monkeypatch.setattr(orchestrator.book_queue, "add", fake_add)
success, error = orchestrator.queue_release(releases[0].__dict__)
assert success is True
assert error is None
assert captured_tasks[0].retry_download_url is None
assert captured_tasks[0].retry_download_protocol is None
assert captured_tasks[0].retry_source_context == {
"source_id": "secret-prowlarr-release",
"info_url": "secret-prowlarr-release",
}
assert "retry_download_url" not in orchestrator._task_to_dict(captured_tasks[0])
def test_search_uses_localized_titles_when_available(self, monkeypatch):
import shelfmark.release_sources.prowlarr.source as prowlarr_source
def fake_get(key: str, default=None):
values = {
"PROWLARR_INDEXERS": "",
"PROWLARR_AUTO_EXPAND": False,
}
return values.get(key, default)
monkeypatch.setattr(prowlarr_source.config, "get", fake_get)
fake_client = FakeTorznabClient()
source = ProwlarrSource()
monkeypatch.setattr(source, "_get_client", lambda: fake_client)
book = BookMetadata(
provider="hardcover",
provider_id="219252",
title="The Lightning Thief",
authors=["Rick Riordan"],
titles_by_language={"hu": "A villámtolvaj"},
)
from shelfmark.core.search_plan import build_release_search_plan
plan = build_release_search_plan(book, languages=["en", "hu"])
source.search(book, plan, content_type="ebook")
assert "The Lightning Thief" in fake_client.queries
assert "A villámtolvaj" in fake_client.queries
assert len(fake_client.queries) == 2
def test_search_does_not_override_search_title_for_english(self, monkeypatch):
import shelfmark.release_sources.prowlarr.source as prowlarr_source
def fake_get(key: str, default=None):
values = {
"PROWLARR_INDEXERS": "",
"PROWLARR_AUTO_EXPAND": False,
}
return values.get(key, default)
monkeypatch.setattr(prowlarr_source.config, "get", fake_get)
fake_client = FakeTorznabClient()
source = ProwlarrSource()
monkeypatch.setattr(source, "_get_client", lambda: fake_client)
book = BookMetadata(
provider="hardcover",
provider_id="123",
title="Mistborn: The Final Empire",
search_title="The Final Empire",
search_author="Brandon Sanderson",
authors=["Brandon Sanderson"],
titles_by_language={
"en": "Mistborn: The Final Empire",
"hu": "A végső birodalom",
},
)
from shelfmark.core.search_plan import build_release_search_plan
plan = build_release_search_plan(book, languages=["en", "hu"])
source.search(book, plan, content_type="ebook")
assert "The Final Empire" in fake_client.queries
assert "A végső birodalom" in fake_client.queries
assert "Mistborn: The Final Empire" not in fake_client.queries
def test_auto_expand_logs_query_argument(self, monkeypatch):
import shelfmark.release_sources.prowlarr.source as prowlarr_source
def fake_get(key: str, default=None):
values = {
"PROWLARR_INDEXERS": "",
"PROWLARR_AUTO_EXPAND": True,
}
return values.get(key, default)
info_calls: list[tuple[str, tuple[object, ...]]] = []
monkeypatch.setattr(prowlarr_source.config, "get", fake_get)
monkeypatch.setattr(
prowlarr_source.logger,
"info",
lambda message, *args: info_calls.append((str(message), args)),
)
fake_client = FakeTorznabClient()
source = ProwlarrSource()
monkeypatch.setattr(source, "_get_client", lambda: fake_client)
book = BookMetadata(
provider="hardcover",
provider_id="123",
title="Anything",
authors=["Someone"],
)
from shelfmark.core.search_plan import build_release_search_plan
plan = build_release_search_plan(book, languages=["en"])
source.search(book, plan, content_type="ebook")
query = fake_client.calls[0][0]
assert fake_client.calls == [(query, [7000]), (query, None)]
assert info_calls == [
(
"Prowlarr: no results for query '%s' with category filter, auto-expanding search",
(query,),
)
]
def test_get_column_config_ignores_indexer_lookup_failure(self, monkeypatch):
source = ProwlarrSource()
class FailingClient:
def get_enabled_indexers_detailed(self, *, raise_on_error=False):
raise RuntimeError("indexers unavailable")
monkeypatch.setattr(source, "_get_client", lambda: FailingClient())
monkeypatch.setattr(source, "_get_selected_indexer_ids", lambda: None)
config = source.get_column_config()
assert config.available_indexers is None
assert config.default_indexers is None
def test_resolve_indexer_ids_from_names_returns_none_on_lookup_failure(self):
source = ProwlarrSource()
class FailingClient:
def get_enabled_indexers_detailed(self, *, raise_on_error=False):
raise RuntimeError("indexers unavailable")
assert source._resolve_indexer_ids_from_names(FailingClient(), ["Alpha"]) is None
def test_get_search_indexer_ids_returns_empty_on_lookup_failure(self):
source = ProwlarrSource()
class FailingClient:
def get_enabled_indexers_detailed(self, *, raise_on_error=False):
raise RuntimeError("indexers unavailable")
assert source._get_search_indexer_ids(FailingClient(), None, [7000]) == []
class TestFetchIndexerSeedSettingsFallback:
"""Regression tests for #795: transient Prowlarr API failures must not
silently strip per-indexer share limits from search results."""
@pytest.fixture(autouse=True)
def _clear_last_known_seed_settings(self):
_last_known_seed_settings.clear()
yield
_last_known_seed_settings.clear()
def test_success_updates_last_known_good(self):
class Client:
def get_indexer_seed_settings(self, restrict_to=None):
del restrict_to
return {13: {"seeding_time_limit_minutes": 4320, "ratio_limit": 1.0}}
fetched = _fetch_indexer_seed_settings(Client(), None)
assert fetched == {13: {"seeding_time_limit_minutes": 4320, "ratio_limit": 1.0}}
assert _last_known_seed_settings == fetched
def test_failure_falls_back_to_last_known_good(self):
_last_known_seed_settings[13] = {"seeding_time_limit_minutes": 4320}
class FailingClient:
def get_indexer_seed_settings(self, restrict_to=None):
del restrict_to
raise RuntimeError("indexers unavailable")
fetched = _fetch_indexer_seed_settings(FailingClient(), None)
assert fetched == {13: {"seeding_time_limit_minutes": 4320}}
# The fallback must be a copy so callers cannot mutate the cache.
fetched[99] = {"ratio_limit": 2.0}
assert 99 not in _last_known_seed_settings
def test_failure_with_no_history_returns_empty(self):
class FailingClient:
def get_indexer_seed_settings(self, restrict_to=None):
del restrict_to
raise RuntimeError("indexers unavailable")
assert _fetch_indexer_seed_settings(FailingClient(), None) == {}
class TestMamLanguageCoverage:
"""MyAnonamouse offers 62 languages; an unmapped code is dropped entirely,
which would leave {Language} empty and different-language editions colliding."""
def test_unmapped_language_is_dropped_not_passed_through(self):
# Documents why coverage matters: there is no raw fallback.
assert _extract_mam_language("Some Book [XYZ / EPUB]") is None
def test_maps_the_common_three_letter_codes(self):
cases = {
"ENG": "en",
"SWE": "sv",
"GER": "de",
"DEU": "de",
"FRE": "fr",
"FRA": "fr",
"CZE": "cs",
"CES": "cs",
}
for tag, expected in cases.items():
assert _extract_mam_language(f"Book [{tag} / EPUB]") == expected, tag
def test_maps_languages_added_for_mam_parity(self):
cases = {
"LAT": "la",
"PER": "fa",
"FAS": "fa",
"TAM": "ta",
"URD": "ur",
"EST": "et",
"ICE": "is",
"ISL": "is",
"GLE": "ga",
"TGL": "fil",
"BEN": "bn",
"BOS": "bs",
"SAN": "sa",
"GLA": "gd",
"GLV": "gv",
"MAY": "ms",
"MSA": "ms",
"BUR": "my",
"MYA": "my",
}
for tag, expected in cases.items():
assert _extract_mam_language(f"Book [{tag} / M4B]") == expected, tag
class _MultiIndexerClient:
"""Torznab client where each indexer entry returns its own result set.
Models one tracker configured in Prowlarr as several entries differing by a
server-side search filter, so the same guid comes back from more than one.
"""
def __init__(self, results_by_indexer: dict[int, list[dict]], priorities=None):
self.results_by_indexer = results_by_indexer
self.priorities = priorities or {}
self.indexer_timeout = 90
def get_enabled_indexers_detailed(self, *, raise_on_error=False):
del raise_on_error
return [
{
"id": indexer_id,
"enable": True,
"priority": self.priorities.get(indexer_id, 25),
"capabilities": {
"categories": [
{"id": 7000, "subCategories": []},
{"id": 3030, "subCategories": []},
]
},
}
for indexer_id in sorted(self.results_by_indexer)
]
def torznab_search(
self,
*,
indexer_id: int,
query: str,
categories=None,
search_type="book",
limit=100,
offset=0,
):
del query, categories, search_type, limit, offset
return self.results_by_indexer.get(indexer_id, [])
def get_enriched_indexer_ids(self, restrict_to=None, indexers=None):
del restrict_to, indexers
return []
def get_indexer_seed_settings(self, restrict_to=None):
del restrict_to
return {}
def _mam_result(indexer_id: int, indexer: str, guid: str, *, freeleech: bool = False) -> dict:
return {
"guid": guid,
"title": "Dune",
"indexerId": indexer_id,
"indexer": indexer,
"protocol": "torrent",
"size": 1048576,
"seeders": 10,
"leechers": 1,
"categories": [{"id": 7020}],
"downloadVolumeFactor": 0.0 if freeleech else 1.0,
"infoUrl": f"https://tracker.example/{guid}",
}
class TestIndexerAwareDeduplication:
"""One tracker as several Prowlarr entries must not collapse to one row (#1137).
These assert the split itself, so they turn PROWLARR_COLLAPSE_DUPLICATES off:
it ships on, which keeps the release list as it was before #1137 for everyone
who has not asked for the per-entry rows.
"""
SPLIT = {"PROWLARR_COLLAPSE_DUPLICATES": False}
ONLY_ACTIVE = 10
FREELEECH = 25
def _search(self, monkeypatch, results_by_indexer, config_values=None, priorities=None):
import shelfmark.release_sources.prowlarr.source as prowlarr_source
from shelfmark.core.search_plan import build_release_search_plan
values = {"PROWLARR_INDEXERS": "", "PROWLARR_AUTO_EXPAND": False}
values.update(config_values or {})
monkeypatch.setattr(
prowlarr_source.config, "get", lambda key, default=None: values.get(key, default)
)
source = ProwlarrSource()
monkeypatch.setattr(
source, "_get_client", lambda: _MultiIndexerClient(results_by_indexer, priorities)
)
book = BookMetadata(
provider="hardcover", provider_id="123", title="Dune", authors=["Frank Herbert"]
)
plan = build_release_search_plan(book, languages=["en"])
return source.search(book, plan)
def test_same_guid_from_two_indexer_entries_both_survive(self, monkeypatch):
shared_guid = "https://tracker.example/torrent/555"
releases = self._search(
monkeypatch,
{
self.ONLY_ACTIVE: [_mam_result(self.ONLY_ACTIVE, "MyAnonamouse", shared_guid)],
self.FREELEECH: [
_mam_result(
self.FREELEECH, "MyAnonamouse - Freeleech", shared_guid, freeleech=True
)
],
},
config_values=self.SPLIT,
)
assert len(releases) == 2
assert {r.indexer for r in releases} == {"MyAnonamouse", "MyAnonamouse - Freeleech"}
assert [r.extra["freeleech"] for r in releases].count(True) == 1
def test_surviving_rows_have_distinct_source_ids(self, monkeypatch):
shared_guid = "https://tracker.example/torrent/555"
releases = self._search(
monkeypatch,
{
self.ONLY_ACTIVE: [_mam_result(self.ONLY_ACTIVE, "MyAnonamouse", shared_guid)],
self.FREELEECH: [
_mam_result(self.FREELEECH, "MyAnonamouse - Freeleech", shared_guid)
],
},
config_values=self.SPLIT,
)
source_ids = [r.source_id for r in releases]
assert len(source_ids) == 2
assert len(set(source_ids)) == 2
def test_source_ids_resolve_to_their_own_indexer_entry(self, monkeypatch):
from shelfmark.release_sources.prowlarr.cache import get_release
shared_guid = "https://tracker.example/torrent/555"
releases = self._search(
monkeypatch,
{
self.ONLY_ACTIVE: [_mam_result(self.ONLY_ACTIVE, "MyAnonamouse", shared_guid)],
self.FREELEECH: [
_mam_result(self.FREELEECH, "MyAnonamouse - Freeleech", shared_guid)
],
},
config_values=self.SPLIT,
)
assert len(releases) == 2
cached_indexer_ids = []
for release in releases:
cached = get_release(release.source_id)
assert cached is not None
assert cached["indexerId"] == release.extra["indexer_id"]
cached_indexer_ids.append(cached["indexerId"])
assert sorted(cached_indexer_ids) == [self.ONLY_ACTIVE, self.FREELEECH]
def test_repeat_of_same_result_from_one_indexer_still_collapses(self, monkeypatch):
duplicate = _mam_result(self.ONLY_ACTIVE, "MyAnonamouse", "https://tracker.example/t/1")
releases = self._search(monkeypatch, {self.ONLY_ACTIVE: [duplicate, dict(duplicate)]})
assert len(releases) == 1
class TestBuildSourceId:
"""source_id keys the release cache, so it must survive shared guids."""
def test_same_guid_from_two_indexer_entries_gets_two_ids(self):
guid = "https://tracker.example/torrent/555"
assert build_source_id({"guid": guid, "indexerId": 10}) != build_source_id(
{"guid": guid, "indexerId": 25}
)
def test_same_indexer_and_guid_is_stable(self):
result = {"guid": "https://tracker.example/torrent/555", "indexerId": 10}
assert build_source_id(result) == build_source_id(dict(result))
def test_falls_back_to_the_bare_guid_without_an_indexer_id(self):
guid = "https://tracker.example/torrent/555"
assert build_source_id({"guid": guid}) == guid
def test_falls_back_to_indexer_and_title_without_a_guid(self):
source_id = build_source_id({"indexerId": 10, "indexer": "MAM", "title": "Dune"})
assert source_id.startswith("10:MAM:")
class TestCollapseDuplicatesSetting:
"""One-row-per-release collapse, on by default, resolved by Prowlarr's priority."""
ONLY_ACTIVE = 10
FREELEECH = 25
def _search_collapsed(self, monkeypatch, priorities):
shared_guid = "https://tracker.example/torrent/555"
return TestIndexerAwareDeduplication()._search(
monkeypatch,
{
self.ONLY_ACTIVE: [_mam_result(self.ONLY_ACTIVE, "MyAnonamouse", shared_guid)],
self.FREELEECH: [
_mam_result(self.FREELEECH, "MyAnonamouse - Freeleech", shared_guid)
],
},
config_values={"PROWLARR_COLLAPSE_DUPLICATES": True},
priorities=priorities,
)
def test_collapse_keeps_the_better_prowlarr_priority(self, monkeypatch):
releases = self._search_collapsed(monkeypatch, {self.FREELEECH: 20, self.ONLY_ACTIVE: 24})
assert len(releases) == 1
assert releases[0].indexer == "MyAnonamouse - Freeleech"
def test_collapse_honours_the_reverse_priority(self, monkeypatch):
releases = self._search_collapsed(monkeypatch, {self.FREELEECH: 30, self.ONLY_ACTIVE: 24})
assert len(releases) == 1
assert releases[0].indexer == "MyAnonamouse"
def test_equal_priority_keeps_the_first_queried(self, monkeypatch):
releases = self._search_collapsed(monkeypatch, {self.FREELEECH: 25, self.ONLY_ACTIVE: 25})
assert len(releases) == 1
assert releases[0].indexer == "MyAnonamouse"
def _search_shared_guid(self, monkeypatch, config_values):
return TestIndexerAwareDeduplication()._search(
monkeypatch,
{
self.ONLY_ACTIVE: [
_mam_result(self.ONLY_ACTIVE, "MyAnonamouse", "https://tracker.example/t/9")
],
self.FREELEECH: [
_mam_result(self.FREELEECH, "MAM - Freeleech", "https://tracker.example/t/9")
],
},
config_values=config_values,
priorities={self.FREELEECH: 20, self.ONLY_ACTIVE: 24},
)
def test_collapse_is_on_when_the_setting_is_untouched(self, monkeypatch):
"""An upgrading user who sets nothing keeps the single row they had before #1137."""
releases = self._search_shared_guid(monkeypatch, None)
assert len(releases) == 1
assert releases[0].indexer == "MAM - Freeleech"
def test_opting_out_keeps_every_indexer_entry(self, monkeypatch):
releases = self._search_shared_guid(monkeypatch, {"PROWLARR_COLLAPSE_DUPLICATES": False})
assert len(releases) == 2
def test_the_settings_field_and_the_search_fallback_agree(self):
"""The field default is what governs in production; the search fallback only
applies to an unregistered key. They have to say the same thing.
"""
from shelfmark.release_sources.prowlarr.settings import prowlarr_config_settings
field = next(
f for f in prowlarr_config_settings() if f.key == "PROWLARR_COLLAPSE_DUPLICATES"
)
assert field.default is True
class TestBuildIndexerPriority:
"""The priority NUMBER from Prowlarr is the rank; the id is only the key."""
def test_reads_the_priority_number_per_indexer(self):
priority = _build_indexer_priority([{"id": 25, "priority": 20}, {"id": 10, "priority": 24}])
assert priority == {25: 20, 10: 24}
def test_a_lower_number_is_preferred(self):
priority = _build_indexer_priority([{"id": 25, "priority": 20}, {"id": 10, "priority": 24}])
assert priority[25] < priority[10]
def test_coerces_string_ids_and_priorities(self):
assert _build_indexer_priority([{"id": "25", "priority": "20"}]) == {25: 20}
def test_skips_records_without_a_usable_id_or_priority(self):
assert (
_build_indexer_priority([{"priority": 20}, {"id": "abc", "priority": 20}, {"id": 7}])
== {}
)
def test_empty_input_yields_no_preferences(self):
assert _build_indexer_priority([]) == {}
class TestDedupEdgeCases:
"""Malformed and partial results must never silently lose a row."""
def test_unidentifiable_results_are_all_kept(self, monkeypatch):
blank = {"indexerId": 10, "indexer": "MAM", "protocol": "torrent"}
releases = TestIndexerAwareDeduplication()._search(
monkeypatch, {10: [dict(blank), dict(blank)]}
)
assert len(releases) == 2
def test_dedup_key_is_none_when_nothing_identifies_the_result(self):
assert _result_dedup_key({"indexerId": 10}) is None
assert _result_dedup_key({"guid": " ", "title": " "}) is None
def test_dedup_falls_back_to_title_within_one_indexer(self):
first = {"indexerId": 10, "title": "Dune"}
second = {"indexerId": 10, "title": "Dune"}
assert _result_dedup_key(first) == _result_dedup_key(second)
def test_same_title_from_two_indexers_is_not_deduped(self):
assert _result_dedup_key({"indexerId": 10, "title": "Dune"}) != _result_dedup_key(
{"indexerId": 25, "title": "Dune"}
)
def test_identity_ignores_whitespace_only_fields(self):
assert _release_identity({"guid": " ", "downloadUrl": "https://x/1"}) == "https://x/1"
def test_identity_accepts_a_numeric_guid(self):
assert _release_identity({"guid": 12345}) == "12345"
def test_identity_is_none_without_a_strong_identifier(self):
assert _release_identity({"title": "Dune", "indexerId": 10}) is None
class TestCollapseEdgeCases:
"""Collapse discards rows, so it must only ever merge on a strong identifier."""
def test_unidentifiable_results_are_never_collapsed(self):
blank = {"indexerId": 10, "title": "Dune"}
kept = _collapse_duplicate_indexer_results([dict(blank), dict(blank)], {})
assert len(kept) == 2
def test_same_title_different_torrents_are_not_collapsed(self):
results = [
{"indexerId": 10, "guid": "guid-a", "title": "Dune"},
{"indexerId": 25, "guid": "guid-b", "title": "Dune"},
]
assert len(_collapse_duplicate_indexer_results(results, {})) == 2
def test_collapse_preserves_the_position_of_the_row_it_replaces(self):
results = [
{"indexerId": 10, "guid": "shared"},
{"indexerId": 99, "guid": "other"},
{"indexerId": 25, "guid": "shared"},
]
kept = _collapse_duplicate_indexer_results(results, {25: 0, 10: 1})
assert [r["indexerId"] for r in kept] == [25, 99]
def test_results_without_an_indexer_id_still_collapse_on_guid(self):
results = [{"guid": "shared"}, {"guid": "shared"}]
assert len(_collapse_duplicate_indexer_results(results, {})) == 1
class TestIndexerPrioritySorting:
"""Results are listed by the indexer priority configured in Prowlarr."""
ONLY_ACTIVE = 10
FREELEECH = 25
OTHER = 99
def _search(self, monkeypatch, priorities):
return TestIndexerAwareDeduplication()._search(
monkeypatch,
{
self.ONLY_ACTIVE: [_mam_result(self.ONLY_ACTIVE, "MyAnonamouse", "https://t/a")],
self.FREELEECH: [
_mam_result(self.FREELEECH, "MyAnonamouse - Freeleech", "https://t/b")
],
self.OTHER: [_mam_result(self.OTHER, "Some Other Tracker", "https://t/c")],
},
priorities=priorities,
)
def test_results_follow_the_priority_numbers(self, monkeypatch):
releases = self._search(
monkeypatch, {self.FREELEECH: 20, self.ONLY_ACTIVE: 24, self.OTHER: 50}
)
assert [r.extra["indexer_id"] for r in releases] == [
self.FREELEECH,
self.ONLY_ACTIVE,
self.OTHER,
]
def test_reversing_the_numbers_reverses_the_list(self, monkeypatch):
releases = self._search(
monkeypatch, {self.FREELEECH: 50, self.ONLY_ACTIVE: 24, self.OTHER: 20}
)
assert [r.extra["indexer_id"] for r in releases] == [
self.OTHER,
self.ONLY_ACTIVE,
self.FREELEECH,
]
def test_equal_priorities_keep_query_order(self, monkeypatch):
releases = self._search(
monkeypatch, {self.FREELEECH: 25, self.ONLY_ACTIVE: 25, self.OTHER: 25}
)
assert [r.extra["indexer_id"] for r in releases] == [
self.ONLY_ACTIVE,
self.FREELEECH,
self.OTHER,
]
class TestIndexerPrioritySortOption:
"""The UI needs the priority on the row to offer an explicit sort on it."""
def test_releases_carry_their_prowlarr_priority(self, monkeypatch):
releases = TestIndexerPrioritySorting()._search(monkeypatch, {10: 24, 25: 20, 99: 50})
by_indexer = {r.extra["indexer_id"]: r.extra.get("indexer_priority") for r in releases}
assert by_indexer == {10: 24, 25: 20, 99: 50}
def test_priority_is_absent_when_prowlarr_reports_none(self, monkeypatch):
releases = TestIndexerPrioritySorting()._search(monkeypatch, {})
assert all(r.extra.get("indexer_priority") == 25 for r in releases)
def test_sort_option_is_offered_lowest_first(self):
from shelfmark.release_sources import serialize_column_config
source = ProwlarrSource()
serialized = serialize_column_config(source.get_column_config())
options = {o["label"]: o for o in serialized["extra_sort_options"]}
assert options["Indexer priority"]["sort_key"] == "extra.indexer_priority"
assert options["Indexer priority"]["default_direction"] == "asc"
assert options["Peers"]["default_direction"] == "desc"
class _FailingIndexerClient:
"""Torznab client where chosen indexers raise instead of answering.
Mirrors #1249: Prowlarr proxies the search to a Cloudflare-fronted tracker,
FlareSolverr is still solving the challenge when the HTTP read times out.
"""
def __init__(self, failing_indexers: set[int], results_by_indexer=None):
self.failing_indexers = failing_indexers
self.results_by_indexer = results_by_indexer or {}
self.indexer_timeout = 90
self.calls: list[tuple[int, object]] = []
def get_enabled_indexers_detailed(self, *, raise_on_error=False):
del raise_on_error
indexer_ids = sorted(self.failing_indexers | set(self.results_by_indexer))
return [
{
"id": indexer_id,
"enable": True,
"capabilities": {"categories": [{"id": 7000, "subCategories": []}]},
}
for indexer_id in indexer_ids
]
def torznab_search(
self,
*,
indexer_id: int,
query: str,
categories=None,
search_type="book",
limit=100,
offset=0,
):
del query, search_type, limit, offset
self.calls.append((indexer_id, categories))
if indexer_id in self.failing_indexers:
msg = f"indexer {indexer_id} did not respond within 90s"
raise ProwlarrSearchError(msg)
return self.results_by_indexer.get(indexer_id, [])
def get_enriched_indexer_ids(self, restrict_to=None, indexers=None):
del restrict_to, indexers
return []
def get_indexer_seed_settings(self, restrict_to=None):
del restrict_to
return {}
class TestFailedIndexerSearchIsNotNoResults:
"""A Torznab timeout must not read as "this book has no releases" (#1249)."""
def _search(self, monkeypatch, client, config_values=None):
import shelfmark.release_sources.prowlarr.source as prowlarr_source
from shelfmark.core.search_plan import build_release_search_plan
values = {"PROWLARR_INDEXERS": "", "PROWLARR_AUTO_EXPAND": False}
values.update(config_values or {})
monkeypatch.setattr(
prowlarr_source.config, "get", lambda key, default=None: values.get(key, default)
)
source = ProwlarrSource()
monkeypatch.setattr(source, "_get_client", lambda: client)
book = BookMetadata(
provider="hardcover", provider_id="123", title="Dune", authors=["Frank Herbert"]
)
plan = build_release_search_plan(book, languages=["en"])
return source.search(book, plan)
def test_sole_indexer_timing_out_raises_instead_of_returning_empty(self, monkeypatch):
client = _FailingIndexerClient({1})
with pytest.raises(SourceUnavailableError) as excinfo:
self._search(monkeypatch, client)
assert "1 of 1 indexer searches failed" in str(excinfo.value)
assert "did not respond within 90s" in str(excinfo.value)
def test_one_failure_does_not_sink_an_indexer_that_answered(self, monkeypatch):
client = _FailingIndexerClient(
{1},
{
2: [
{
"guid": "g2",
"title": "Dune",
"indexerId": 2,
"indexer": "working",
"protocol": "torrent",
"size": 1048576,
"seeders": 5,
}
]
},
)
releases = self._search(monkeypatch, client)
assert [r.indexer for r in releases] == ["working"]
def test_partial_failure_with_no_results_still_reports_the_failure(self, monkeypatch):
client = _FailingIndexerClient({1}, {2: []})
with pytest.raises(SourceUnavailableError) as excinfo:
self._search(monkeypatch, client)
assert "1 of 2 indexer searches failed" in str(excinfo.value)
def test_auto_expand_does_not_retry_on_top_of_a_failed_search(self, monkeypatch):
"""The second search is what crashes FlareSolverr's Chrome on a small host."""
client = _FailingIndexerClient({1})
with pytest.raises(SourceUnavailableError):
self._search(monkeypatch, client, {"PROWLARR_AUTO_EXPAND": True})
assert client.calls == [(1, [7000])]
def test_auto_expand_still_retries_when_the_indexer_answered_empty(self, monkeypatch):
client = _FailingIndexerClient(set(), {1: []})
assert self._search(monkeypatch, client, {"PROWLARR_AUTO_EXPAND": True}) == []
assert client.calls == [(1, [7000]), (1, None)]
class TestUnreachableProwlarrIsNotNoResults:
"""Prowlarr itself being down must not read as "this book has no releases" (#1249)."""
class _UnreachableClient:
indexer_timeout = 90
def get_enabled_indexers_detailed(self, *, raise_on_error=False):
del raise_on_error
raise requests.exceptions.ConnectionError("connection refused")
def test_search_reports_the_connection_failure(self, monkeypatch):
import shelfmark.release_sources.prowlarr.source as prowlarr_source
from shelfmark.core.search_plan import build_release_search_plan
monkeypatch.setattr(
prowlarr_source.config,
"get",
lambda key, default=None: {"PROWLARR_INDEXERS": ""}.get(key, default),
)
source = ProwlarrSource()
monkeypatch.setattr(source, "_get_client", lambda: self._UnreachableClient())
book = BookMetadata(
provider="hardcover", provider_id="123", title="Dune", authors=["Frank Herbert"]
)
plan = build_release_search_plan(book, languages=["en"])
with pytest.raises(SourceUnavailableError, match="could not reach Prowlarr"):
source.search(book, plan)
class TestSearchBudgetScalesWithIndexerTimeout:
def test_default_budget_is_unchanged(self):
assert _search_budget_seconds(30) == PROWLARR_SEARCH_TIMEOUT_SECONDS
def test_a_long_indexer_timeout_widens_the_budget(self):
assert _search_budget_seconds(120) == 240.0
def test_budget_stays_under_the_gunicorn_worker_timeout(self):
assert _search_budget_seconds(300) == 240.0
Reference in New Issue View Git Blame Copy Permalink