""" Tests for the Prowlarr source module. Tests the utility functions for parsing release metadata. """ # Import the functions to test import pytest from shelfmark.metadata_providers import BookMetadata from shelfmark.release_sources.prowlarr.api import ProwlarrClient from shelfmark.release_sources.prowlarr.source import ( ProwlarrSource, _build_indexer_priority, _collapse_duplicate_indexer_results, _detect_content_type_from_categories, _extract_format, _fetch_indexer_seed_settings, _last_known_seed_settings, _parse_size, _release_identity, _result_dedup_key, ) 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 {} 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 _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 {} 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).""" 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 ) ], }, ) 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) ], }, ) 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) ], }, ) 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: """Opt-in one-row-per-release collapse, 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 test_collapse_off_by_default_keeps_both_rows(self, monkeypatch): releases = 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") ], }, priorities={self.FREELEECH: 20, self.ONLY_ACTIVE: 24}, ) assert len(releases) == 2 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"