diff --git a/docker-compose.test-clients.yml b/docker-compose.test-clients.yml index 6a1f50f7..1562a88e 100644 --- a/docker-compose.test-clients.yml +++ b/docker-compose.test-clients.yml @@ -13,6 +13,7 @@ # - Transmission: http://localhost:9091 (admin / admin) # - Deluge: http://localhost:8112 (password: deluge) # - NZBGet: http://localhost:6789 (nzbget / tegbzn6789) +# - NZBHydra: http://localhost:5076 (no auth by default) # - SABnzbd: http://localhost:8085 (complete setup wizard for API key) # - rTorrent: http://localhost:8000 (admin / admin - if auth enabled) # @@ -35,6 +36,7 @@ services: # - Transmission: http://transmission:9091 # - Deluge Web UI: http://deluge:8112 # - NZBGet: http://nzbget:6789 + # - NZBHydra: http://nzbhydra:5076 # - SABnzbd: http://sabnzbd:8080 # - rTorrent: http://rtorrent:80 (XMLRPC via HTTP) or rtorrent (port 5000 for SCGI) ports: @@ -60,6 +62,7 @@ services: - ./.local/test-clients/sabnzbd/config:/sabnzbd-config:ro depends_on: - nzbget + - nzbhydra - sabnzbd - qbittorrent - transmission @@ -180,3 +183,18 @@ services: - "50000:50000" # Incoming connections - "6881:6881/udp" restart: unless-stopped + + nzbhydra: + image: lscr.io/linuxserver/nzbhydra2:latest + container_name: nzbhydra + environment: + - PUID=1000 + - PGID=1000 + - TZ=Europe/London + volumes: + - ./.local/test-clients/nzbhydra/config:/config + - ./.local/test-clients/downloads:/downloads + ports: + - 5076:5076 + restart: unless-stopped + diff --git a/shelfmark/release_sources/__init__.py b/shelfmark/release_sources/__init__.py index 379cc003..e2ff355e 100644 --- a/shelfmark/release_sources/__init__.py +++ b/shelfmark/release_sources/__init__.py @@ -403,6 +403,7 @@ _BUILTIN_SOURCE_MODULES = ( "shelfmark.release_sources.audiobookbay", "shelfmark.release_sources.direct_download", "shelfmark.release_sources.irc", + "shelfmark.release_sources.newznab", "shelfmark.release_sources.prowlarr", ) _builtin_source_state = {"loaded": False} diff --git a/shelfmark/release_sources/newznab/__init__.py b/shelfmark/release_sources/newznab/__init__.py new file mode 100644 index 00000000..4723904d --- /dev/null +++ b/shelfmark/release_sources/newznab/__init__.py @@ -0,0 +1,32 @@ +""" +Newznab release source plugin. + +Integrates with any Newznab-compatible indexer or aggregator (e.g. NZBHydra2, +NZBGeek, Drunkenslug) to search for book releases via the standard Newznab API. + +Includes: +- NewznabSource: Search integration +- NewznabHandler: Download handling via configured usenet/torrent client +""" + +from importlib import import_module + +# Import submodules to trigger decorator registration +from shelfmark.release_sources.newznab import ( + handler as handler, +) +from shelfmark.release_sources.newznab import ( + settings as settings, +) +from shelfmark.release_sources.newznab import ( + source as source, +) + +# Import shared download clients/settings to trigger registration. +try: + import_module("shelfmark.download.clients") + import_module("shelfmark.download.clients.settings") +except ImportError as e: + import logging + + logging.getLogger(__name__).debug("Download clients not loaded: %s", e) diff --git a/shelfmark/release_sources/newznab/api.py b/shelfmark/release_sources/newznab/api.py new file mode 100644 index 00000000..0a02bf3b --- /dev/null +++ b/shelfmark/release_sources/newznab/api.py @@ -0,0 +1,143 @@ +"""Newznab API client - connects to any Newznab-compatible indexer or aggregator.""" + +from __future__ import annotations + +import re +from typing import Any + +import requests + +from shelfmark.core.logger import setup_logger +from shelfmark.core.utils import normalize_http_url +from shelfmark.download.network import get_ssl_verify +from shelfmark.release_sources.prowlarr.torznab import parse_torznab_xml + +logger = setup_logger(__name__) + +# Newznab standard book category IDs +NEWZNAB_BOOKS = 7000 +NEWZNAB_AUDIOBOOKS = 3030 + + +class NewznabClient: + """Client for any Newznab-compatible indexer API.""" + + def __init__(self, url: str, api_key: str, timeout: int = 30) -> None: + self.base_url = normalize_http_url(url) + self.api_key = api_key + self.timeout = timeout + self._session = requests.Session() + + def _api_url(self) -> str: + """Return the Newznab API endpoint URL.""" + base = self.base_url.rstrip("/") + # Many indexers expose the API at /api; others at the root with ?page=rss. + # Prefer /api if the base URL doesn't already end with it. + if not base.endswith("/api"): + return base + "/api" + return base + + def _get( + self, + params: dict[str, Any], + *, + accept_xml: bool = False, + ) -> requests.Response: + """Make a GET request to the Newznab API endpoint.""" + params = {k: v for k, v in params.items() if v is not None} + if self.api_key: + params["apikey"] = self.api_key + + url = self._api_url() + logger.debug("Newznab API: GET %s params=%s", url, *params) + + headers = {} + if accept_xml: + headers["Accept"] = "application/rss+xml, application/xml;q=0.9, */*;q=0.8" + + response = self._session.get( + url=url, + params=params, + headers=headers, + timeout=self.timeout, + verify=get_ssl_verify(url), + ) + response.raise_for_status() + return response + + def test_connection(self) -> tuple[bool, str]: + """Test connection via the capabilities endpoint. Returns (success, message).""" + logger.info("Testing Newznab connection to: %s", self.base_url) + try: + response = self._get({"t": "caps"}) + # Caps endpoint returns XML; a 200 is sufficient to confirm connectivity. + # Try to extract the server title from the XML for a friendly message. + text = response.text or "" + title = "Newznab indexer" + # Try attribute (NZBHydra2 style), then element + m = re.search(r'<server[^>]+title="([^"]+)"', text, re.IGNORECASE) + if not m: + m = re.search(r"<title>([^<]+)", text, re.IGNORECASE) + if m: + title = m.group(1).strip() + logger.info("Newznab connection successful: %s", title) + except requests.exceptions.ConnectionError: + return False, "Could not connect. Check the URL." + except requests.exceptions.HTTPError as e: + status = e.response.status_code if e.response is not None else "unknown" + if e.response is not None and e.response.status_code == 401: + return False, "Invalid API key" + return False, f"HTTP error {status}" + except requests.exceptions.RequestException as e: + return False, f"Connection failed: {e!s}" + else: + return True, f"Connected to {title}" + + def search( + self, + query: str, + categories: list[int] | None = None, + search_type: str = "search", + limit: int = 100, + offset: int = 0, + ) -> list[dict[str, Any]]: + """Search the Newznab indexer and return parsed results. + + Args: + query: Search string. + categories: Optional list of Newznab category IDs (e.g. [7000, 3030]). + search_type: Newznab search type ("search", "book", "audio"). + limit: Max results to return. + offset: Result page offset. + + Returns: + List of result dicts shaped like Prowlarr JSON search results so that + the shared ``_prowlarr_result_to_release`` converter can process them. + """ + if not query: + return [] + + params: dict[str, Any] = { + "t": search_type, + "q": query, + "limit": limit, + "offset": offset, + } + if categories: + params["cat"] = ",".join(str(c) for c in categories) + + try: + response = self._get(params, accept_xml=True) + results = parse_torznab_xml(response.text) + logger.debug("Newznab search '%s': %d results", query, len(results)) + if not results: + preview = response.text[:300].strip() if response.text else "" + logger.debug("Newznab empty response body: %s", preview) + except requests.exceptions.RequestException: + logger.exception("Newznab search request failed") + return [] + except Exception: + logger.exception("Newznab search failed") + return [] + else: + return results diff --git a/shelfmark/release_sources/newznab/cache.py b/shelfmark/release_sources/newznab/cache.py new file mode 100644 index 00000000..6506542c --- /dev/null +++ b/shelfmark/release_sources/newznab/cache.py @@ -0,0 +1,61 @@ +""" +Newznab release cache. + +Stores search results so the handler can look up releases by source_id. +""" + +import time +from threading import Lock + +from shelfmark.core.logger import setup_logger + +logger = setup_logger(__name__) + +RELEASE_CACHE_TTL = 3600 # 1 hour + +_cache: dict[str, tuple] = {} +_cache_lock = Lock() + + +def cache_release(source_id: str, release_data: dict) -> None: + with _cache_lock: + _cache[source_id] = (release_data, time.time()) + + +def get_release(source_id: str) -> dict | None: + with _cache_lock: + if source_id not in _cache: + logger.debug("Newznab release not in cache: %s", source_id) + return None + + release_data, cached_at = _cache[source_id] + if time.time() - cached_at > RELEASE_CACHE_TTL: + del _cache[source_id] + logger.debug("Newznab release expired: %s", source_id) + return None + + return release_data + + +def remove_release(source_id: str) -> None: + with _cache_lock: + if source_id in _cache: + del _cache[source_id] + logger.debug("Removed Newznab release from cache: %s", source_id) + + +def cleanup_expired() -> int: + current_time = time.time() + removed = 0 + with _cache_lock: + expired_ids = [ + sid + for sid, (_, cached_at) in _cache.items() + if current_time - cached_at > RELEASE_CACHE_TTL + ] + for sid in expired_ids: + del _cache[sid] + removed += 1 + if removed: + logger.debug("Cleaned up %d expired Newznab cache entries", removed) + return removed diff --git a/shelfmark/release_sources/newznab/handler.py b/shelfmark/release_sources/newznab/handler.py new file mode 100644 index 00000000..9665493a --- /dev/null +++ b/shelfmark/release_sources/newznab/handler.py @@ -0,0 +1,123 @@ +"""Newznab download handler - resolves releases and delegates to shared clients.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Callable + +if TYPE_CHECKING: + from shelfmark.core.models import DownloadTask + +from shelfmark.core.logger import setup_logger +from shelfmark.download.clients import DownloadClient, get_client, list_configured_clients +from shelfmark.download.clients.base_handler import ( + COMPLETED_PATH_MAX_ATTEMPTS as _DEFAULT_COMPLETED_PATH_MAX_ATTEMPTS, +) +from shelfmark.download.clients.base_handler import ( + COMPLETED_PATH_RETRY_INTERVAL as _DEFAULT_COMPLETED_PATH_RETRY_INTERVAL, +) +from shelfmark.download.clients.base_handler import ( + POLL_INTERVAL as _DEFAULT_POLL_INTERVAL, +) +from shelfmark.download.clients.base_handler import ( + DownloadRequest, + ExternalClientHandler, +) +from shelfmark.release_sources import register_handler +from shelfmark.release_sources.newznab.cache import get_release, remove_release + +logger = setup_logger(__name__) + +# Backwards-compat constants for tests patching this module. +POLL_INTERVAL = _DEFAULT_POLL_INTERVAL +COMPLETED_PATH_RETRY_INTERVAL = _DEFAULT_COMPLETED_PATH_RETRY_INTERVAL +COMPLETED_PATH_MAX_ATTEMPTS = _DEFAULT_COMPLETED_PATH_MAX_ATTEMPTS + + +def _get_protocol(result: dict) -> str: + """Infer download protocol from a Newznab result dict.""" + protocol = str(result.get("protocol", "")).lower() + if protocol in ("torrent", "usenet"): + return protocol + + download_url = str(result.get("downloadUrl") or "").lower() + magnet_url = str(result.get("magnetUrl") or "").lower() + + if magnet_url.startswith("magnet:"): + return "torrent" + if download_url.startswith("magnet:") or ".torrent" in download_url: + return "torrent" + if ".nzb" in download_url: + return "usenet" + + # Newznab indexers are usenet-native; default to usenet when ambiguous. + return "usenet" + + +def _get_download_url(result: dict) -> str: + """Pick the best URL to hand to a download client.""" + protocol = _get_protocol(result) + magnet_url = str(result.get("magnetUrl") or "").strip() + download_url = str(result.get("downloadUrl") or "").strip() + + if protocol == "torrent": + return magnet_url or download_url + return download_url or magnet_url + + +@register_handler("newznab") +class NewznabHandler(ExternalClientHandler): + """Handler for Newznab downloads via configured usenet/torrent client.""" + + def _get_client(self, protocol: str) -> DownloadClient | None: + return get_client(protocol) + + def _list_configured_clients(self) -> list[str]: + return list_configured_clients() + + def _poll_interval(self) -> float: + return POLL_INTERVAL + + def _completed_path_retry_interval(self) -> float: + return COMPLETED_PATH_RETRY_INTERVAL + + def _completed_path_max_attempts(self) -> int: + return COMPLETED_PATH_MAX_ATTEMPTS + + def _resolve_download( + self, + task: DownloadTask, + status_callback: Callable[[str, str | None], None], + ) -> DownloadRequest | None: + result = get_release(task.task_id) + if not result: + logger.warning("Newznab release cache miss: %s", task.task_id) + status_callback("error", "Release not found in cache (may have expired)") + return None + + download_url = _get_download_url(result) + if not download_url: + status_callback("error", "No download URL available") + return None + + protocol = _get_protocol(result) + if protocol not in ("torrent", "usenet"): + status_callback("error", "Could not determine download protocol") + return None + + release_name = result.get("title") or task.title or "Unknown" + expected_hash = str(result.get("infoHash") or "").strip() or None + + return DownloadRequest( + url=download_url, + protocol=protocol, + release_name=release_name, + expected_hash=expected_hash, + ) + + def _on_download_complete(self, task: DownloadTask) -> None: + remove_release(task.task_id) + + def cancel(self, task_id: str) -> bool: + logger.debug("Cancel requested for Newznab task: %s", task_id) + remove_release(task_id) + return super().cancel(task_id) diff --git a/shelfmark/release_sources/newznab/settings.py b/shelfmark/release_sources/newznab/settings.py new file mode 100644 index 00000000..1d580076 --- /dev/null +++ b/shelfmark/release_sources/newznab/settings.py @@ -0,0 +1,96 @@ +"""Newznab settings registration.""" + +from typing import Any + +from shelfmark.core.settings_registry import ( + ActionButton, + CheckboxField, + HeadingField, + PasswordField, + SettingsField, + TextField, + register_settings, +) +from shelfmark.core.utils import normalize_http_url + + +def _test_newznab_connection(current_values: dict[str, Any] | None = None) -> dict[str, Any]: + """Test the Newznab connection using current form values.""" + from shelfmark.core.config import config + from shelfmark.release_sources.newznab.api import NewznabClient + + current_values = current_values or {} + + raw_url = str(current_values.get("NEWZNAB_URL") or config.get("NEWZNAB_URL", "") or "") + api_key = str(current_values.get("NEWZNAB_API_KEY") or config.get("NEWZNAB_API_KEY", "") or "") + + if not raw_url: + return {"success": False, "message": "Newznab URL is required"} + + url = normalize_http_url(raw_url) + if not url: + return {"success": False, "message": "Newznab URL is invalid"} + + try: + client = NewznabClient(url, api_key) + success, message = client.test_connection() + except Exception as e: # noqa: BLE001 — surface any unexpected error to the UI + return {"success": False, "message": f"Connection failed: {e!s}"} + else: + return {"success": success, "message": message} + + +@register_settings( + name="newznab_config", + display_name="Newznab", + icon="download", + order=42, +) +def newznab_config_settings() -> list[SettingsField]: + """Newznab connection settings.""" + return [ + HeadingField( + key="newznab_heading", + title="Newznab Integration", + description=( + "Search for books via any Newznab-compatible indexer or aggregator " + "(e.g. NZBHydra2, NZBGeek, Drunkenslug)." + ), + ), + CheckboxField( + key="NEWZNAB_ENABLED", + label="Enable Newznab source", + default=False, + description="Enable searching for books via a Newznab-compatible indexer", + ), + TextField( + key="NEWZNAB_URL", + label="Newznab URL", + description="Base URL of your Newznab indexer or aggregator", + placeholder="http://nzbhydra:5076", + required=True, + show_when={"field": "NEWZNAB_ENABLED", "value": True}, + ), + PasswordField( + key="NEWZNAB_API_KEY", + label="API Key", + description="Your Newznab API key (leave blank if not required)", + required=False, + show_when={"field": "NEWZNAB_ENABLED", "value": True}, + ), + ActionButton( + key="test_newznab", + label="Test Connection", + description="Verify your Newznab configuration", + style="primary", + callback=_test_newznab_connection, + show_when={"field": "NEWZNAB_ENABLED", "value": True}, + ), + CheckboxField( + key="NEWZNAB_AUTO_EXPAND", + label="Auto-expand search on no results", + default=False, + description="Automatically retry search without category filtering if no results are found", + show_when={"field": "NEWZNAB_ENABLED", "value": True}, + ), + ] diff --git a/shelfmark/release_sources/newznab/source.py b/shelfmark/release_sources/newznab/source.py new file mode 100644 index 00000000..0a1a0cb0 --- /dev/null +++ b/shelfmark/release_sources/newznab/source.py @@ -0,0 +1,310 @@ +"""Newznab release source - searches a Newznab-compatible indexer for book releases.""" + +from __future__ import annotations + +import time +from typing import TYPE_CHECKING, ClassVar + +if TYPE_CHECKING: + from shelfmark.core.search_plan import ReleaseSearchPlan + from shelfmark.metadata_providers import BookMetadata + +from shelfmark.core.config import config +from shelfmark.core.logger import setup_logger +from shelfmark.core.utils import normalize_http_url +from shelfmark.release_sources import ( + ColumnAlign, + ColumnColorHint, + ColumnRenderType, + ColumnSchema, + LeadingCellConfig, + LeadingCellType, + Release, + ReleaseColumnConfig, + ReleaseProtocol, + ReleaseSource, + register_source, +) +from shelfmark.release_sources.newznab.api import NewznabClient +from shelfmark.release_sources.newznab.cache import cache_release +from shelfmark.release_sources.prowlarr.source import ( + PROWLARR_SEARCH_TIMEOUT_SECONDS as _SEARCH_TIMEOUT, +) + +# Re-use the Prowlarr source helpers — they operate on generic result dicts. +from shelfmark.release_sources.prowlarr.source import ( + _detect_content_type_from_categories, + _parse_size, +) + +logger = setup_logger(__name__) + +# Newznab category IDs +_AUDIOBOOK_CATS = [3030] +_BOOK_CATS = [7000] + +# Reuse the same timeout constant as Prowlarr. +NEWZNAB_SEARCH_TIMEOUT_SECONDS = _SEARCH_TIMEOUT + + +def _newznab_result_to_release(result: dict, content_type: str = "ebook") -> Release: + """Convert a parsed Newznab XML result dict to a Release object.""" + raw_title = result.get("title", "Unknown") + size_bytes = result.get("size") + indexer = result.get("indexer") or "Newznab" + categories = result.get("categories", []) + + protocol_str = str(result.get("protocol", "usenet")).lower() + protocol = ReleaseProtocol.TORRENT if protocol_str == "torrent" else ReleaseProtocol.NZB + + seeders = result.get("seeders") + leechers = result.get("leechers") + is_torrent = protocol == ReleaseProtocol.TORRENT + + peers_display = ( + f"{seeders} / {leechers}" + if is_torrent and seeders is not None and leechers is not None + else None + ) + + # Build source_id from GUID + source_id = result.get("guid") or f"newznab:{hash(raw_title)}" + + # Cache the raw result for the handler + cache_release(source_id, result) + + # Freeleech / VIP detection + raw_indexer_flags = result.get("indexerFlags") or [] + indexer_flags: list[str] = [] + seen: set = set() + + def add_flag(flag: object) -> None: + if flag is None: + return + s = str(flag).strip() + if s and s.lower() not in seen: + seen.add(s.lower()) + indexer_flags.append(s) + + if isinstance(raw_indexer_flags, list): + for f in raw_indexer_flags: + add_flag(f) + elif raw_indexer_flags: + add_flag(raw_indexer_flags) + + download_volume_factor = result.get("downloadVolumeFactor") + is_freeleech = False + try: + if download_volume_factor is not None and float(download_volume_factor) == 0.0: + is_freeleech = True + except TypeError, ValueError: + pass + + if any(f.lower() in {"freeleech", "fl"} for f in indexer_flags): + is_freeleech = True + + is_vip = "[vip]" in str(raw_title).lower() + if is_vip: + add_flag("VIP") + if is_freeleech: + add_flag("FreeLeech") + + download_url = str(result.get("downloadUrl") or "").strip() + info_url = result.get("infoUrl") or result.get("guid") + + return Release( + source="newznab", + source_id=source_id, + title=raw_title, + format=None, + language=None, + size=_parse_size(size_bytes), + size_bytes=size_bytes, + download_url=download_url or None, + info_url=info_url, + protocol=protocol, + indexer=indexer, + seeders=seeders if is_torrent else None, + peers=peers_display, + content_type=_detect_content_type_from_categories(categories, content_type), + extra={ + "publish_date": result.get("publishDate"), + "categories": categories, + "indexer_flags": indexer_flags, + "vip": is_vip, + "freeleech": is_freeleech, + "download_volume_factor": download_volume_factor, + "upload_volume_factor": result.get("uploadVolumeFactor"), + "minimum_ratio": result.get("minimumRatio"), + "minimum_seed_time": result.get("minimumSeedTime"), + "info_hash": result.get("infoHash"), + "files": result.get("files"), + "grabs": result.get("grabs"), + "author": result.get("author"), + "book_title": result.get("bookTitle"), + }, + ) + + +@register_source("newznab") +class NewznabSource(ReleaseSource): + """Release source for any Newznab-compatible indexer or aggregator.""" + + name = "newznab" + display_name = "Newznab" + supported_content_types: ClassVar[list[str]] = ["ebook", "audiobook"] + + def get_column_config(self) -> ReleaseColumnConfig: + return ReleaseColumnConfig( + columns=[ + ColumnSchema( + key="indexer", + label="Indexer", + render_type=ColumnRenderType.INDEXER_PROTOCOL, + align=ColumnAlign.LEFT, + width="minmax(140px, 1fr)", + hide_mobile=False, + sortable=True, + ), + ColumnSchema( + key="extra.indexer_flags", + label="Flags", + render_type=ColumnRenderType.TAGS, + align=ColumnAlign.CENTER, + width="50px", + hide_mobile=False, + color_hint=ColumnColorHint(type="map", value="flags"), + fallback="", + uppercase=True, + ), + ColumnSchema( + key="size", + label="Size", + render_type=ColumnRenderType.SIZE, + align=ColumnAlign.CENTER, + width="80px", + hide_mobile=False, + sortable=True, + sort_key="size_bytes", + ), + ], + grid_template="minmax(0,2fr) minmax(140px,1fr) 50px 80px", + leading_cell=LeadingCellConfig(type=LeadingCellType.NONE), + supported_filters=["indexer"], + ) + + def _get_client(self) -> NewznabClient | None: + raw_url = str(config.get("NEWZNAB_URL", "") or "") + api_key = str(config.get("NEWZNAB_API_KEY", "") or "") + + if not raw_url: + return None + + url = normalize_http_url(raw_url) + if not url: + return None + + return NewznabClient(url, api_key or "") + + def search( + self, + book: BookMetadata, + plan: ReleaseSearchPlan, + *, + expand_search: bool = False, + content_type: str = "ebook", + ) -> list[Release]: + """Search the Newznab indexer for releases matching the book.""" + client = self._get_client() + if not client: + logger.warning("Newznab not configured - skipping search") + return [] + + queries = [v.title for v in plan.title_variants if v.title] + queries = [q for q in queries if q] + + if not queries and plan.isbn_candidates: + queries = list(plan.isbn_candidates) + + if not queries: + logger.warning("Newznab: no search query available") + return [] + + # Category selection — omit categories when expanding search + if expand_search: + categories = None + elif content_type == "audiobook": + categories = [3030] + else: + categories = [7000] + + auto_expand = config.get("NEWZNAB_AUTO_EXPAND", False) + deadline = time.monotonic() + NEWZNAB_SEARCH_TIMEOUT_SECONDS + + def _check_timeout() -> None: + if time.monotonic() > deadline: + raise TimeoutError( + f"Newznab search timed out after {int(NEWZNAB_SEARCH_TIMEOUT_SECONDS)}s" + ) + + seen_keys: set = set() + all_results: list[dict] = [] + + try: + for idx, query in enumerate(queries, start=1): + _check_timeout() + if len(queries) > 1: + logger.debug("Newznab query %d/%d: '%s'", idx, len(queries), query) + + raw = client.search(query=query, categories=categories) + + # Auto-expand: retry without category filter if no results + if not raw and categories and auto_expand: + _check_timeout() + logger.info( + "Newznab: no results for '%s' with category filter, auto-expanding", + query, + ) + raw = client.search(query=query, categories=None) + + for r in raw: + key = ( + r.get("guid") + or r.get("downloadUrl") + or f"{r.get('indexer')}:{r.get('title')}" + ) + if key in seen_keys: + continue + seen_keys.add(key) + all_results.append(r) + + except TimeoutError as e: + logger.warning("Newznab search timed out: %s", e) + except Exception: + logger.exception("Newznab search failed") + return [] + + results = [_newznab_result_to_release(r, content_type) for r in all_results] + + if results: + nzb_count = sum(1 for r in results if r.protocol == ReleaseProtocol.NZB) + torrent_count = sum(1 for r in results if r.protocol == ReleaseProtocol.TORRENT) + indexers = sorted({r.indexer for r in results if r.indexer}) + indexer_str = ", ".join(indexers) if indexers else "unknown" + logger.info( + "Newznab: %d results (%d nzb, %d torrent) from %s", + len(results), + nzb_count, + torrent_count, + indexer_str, + ) + else: + logger.debug("Newznab: no results found") + + return results + + def is_available(self) -> bool: + if not config.get("NEWZNAB_ENABLED", False): + return False + url = normalize_http_url(str(config.get("NEWZNAB_URL", "") or "")) + return bool(url) diff --git a/tests/newznab/__init__.py b/tests/newznab/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/newznab/conftest.py b/tests/newznab/conftest.py new file mode 100644 index 00000000..0adcd13a --- /dev/null +++ b/tests/newznab/conftest.py @@ -0,0 +1,24 @@ +""" +Newznab test configuration. + +Stubs out optional heavy dependencies (flask_socketio) that are not available +in lightweight dev/CI environments. These are only needed by the IRC source, +which is unrelated to the newznab plugin under test. +""" + +import sys +import types + + +def _stub_module(name: str) -> None: + """Insert a minimal stub module into sys.modules if not already present.""" + if name not in sys.modules: + stub = types.ModuleType(name) + # Provide no-op stand-ins for the symbols IRC source imports. + stub.SocketIO = object + stub.join_room = lambda *a, **kw: None + stub.leave_room = lambda *a, **kw: None + sys.modules[name] = stub + + +_stub_module("flask_socketio") diff --git a/tests/newznab/test_api.py b/tests/newznab/test_api.py new file mode 100644 index 00000000..87c4552c --- /dev/null +++ b/tests/newznab/test_api.py @@ -0,0 +1,208 @@ +"""Unit tests for the Newznab API client.""" + +from unittest.mock import MagicMock, patch + +import requests + +from shelfmark.release_sources.newznab.api import NewznabClient + +# ── helpers ──────────────────────────────────────────────────────────────────── + +NZB_XML = """ + + + My Indexer + + Some Book (2024) [EPUB] + https://indexer.example.com/nzb/1 + https://indexer.example.com/nzb/1?apikey=secret + Sun, 01 Jan 2024 00:00:00 +0000 + 2097152 + + 7000 + + + +""" + +CAPS_XML = """ + + +""" + + +def _make_response(text: str, status: int = 200) -> MagicMock: + r = MagicMock(spec=requests.Response) + r.status_code = status + r.text = text + r.ok = status < 400 + r.raise_for_status = MagicMock() + if status >= 400: + r.raise_for_status.side_effect = requests.exceptions.HTTPError(response=r) + return r + + +# ── URL construction ──────────────────────────────────────────────────────────── + + +class TestApiUrl: + def test_appends_api_path(self): + client = NewznabClient("http://nzbhydra:5076", "key") + assert client._api_url() == "http://nzbhydra:5076/api" + + def test_does_not_double_append(self): + client = NewznabClient("http://nzbhydra:5076/api", "key") + assert client._api_url() == "http://nzbhydra:5076/api" + + def test_strips_trailing_slash(self): + client = NewznabClient("http://nzbhydra:5076/", "key") + assert client._api_url() == "http://nzbhydra:5076/api" + + +# ── test_connection ───────────────────────────────────────────────────────────── + + +class TestTestConnection: + def test_success_returns_true_with_title(self): + client = NewznabClient("http://nzbhydra:5076", "key") + with patch.object(client, "_get", return_value=_make_response(CAPS_XML)): + ok, msg = client.test_connection() + assert ok is True + assert "NZBHydra2" in msg + + def test_connection_error_returns_false(self): + client = NewznabClient("http://nzbhydra:5076", "key") + with patch.object( + client, + "_get", + side_effect=requests.exceptions.ConnectionError("refused"), + ): + ok, msg = client.test_connection() + assert ok is False + assert "connect" in msg.lower() + + def test_401_returns_api_key_error(self): + client = NewznabClient("http://nzbhydra:5076", "key") + fake_resp = _make_response("", status=401) + with patch.object( + client, + "_get", + side_effect=requests.exceptions.HTTPError(response=fake_resp), + ): + ok, msg = client.test_connection() + assert ok is False + assert "api key" in msg.lower() + + def test_generic_exception_returns_false(self): + client = NewznabClient("http://nzbhydra:5076", "key") + with patch.object( + client, + "_get", + side_effect=requests.exceptions.Timeout("oops"), + ): + ok, msg = client.test_connection() + assert ok is False + assert "oops" in msg.lower() + + def test_caps_without_title_still_succeeds(self): + client = NewznabClient("http://nzbhydra:5076", "key") + caps_no_title = "" + with patch.object(client, "_get", return_value=_make_response(caps_no_title)): + ok, msg = client.test_connection() + assert ok is True + assert msg # some non-empty message + + +# ── search ────────────────────────────────────────────────────────────────────── + + +class TestSearch: + def test_empty_query_returns_empty(self): + client = NewznabClient("http://nzbhydra:5076", "key") + results = client.search(query="") + assert results == [] + + def test_parses_nzb_xml(self): + client = NewznabClient("http://nzbhydra:5076", "key") + with patch.object(client, "_get", return_value=_make_response(NZB_XML)): + results = client.search(query="Some Book") + assert len(results) == 1 + r = results[0] + assert r["title"] == "Some Book (2024) [EPUB]" + assert r["protocol"] == "usenet" + assert r["size"] == 2097152 + assert r["downloadUrl"] == "https://indexer.example.com/nzb/1?apikey=secret" + + def test_sends_category_param(self): + client = NewznabClient("http://nzbhydra:5076", "key") + captured: list = [] + + def fake_get(params, accept_xml=False): + captured.append(params.copy()) + return _make_response(NZB_XML) + + with patch.object(client, "_get", side_effect=fake_get): + client.search(query="book", categories=[7000, 3030]) + + assert len(captured) == 1 + assert captured[0]["cat"] == "7000,3030" + + def test_omits_category_when_none(self): + client = NewznabClient("http://nzbhydra:5076", "key") + captured: list = [] + + def fake_get(params, accept_xml=False): + captured.append(params.copy()) + return _make_response(NZB_XML) + + with patch.object(client, "_get", side_effect=fake_get): + client.search(query="book", categories=None) + + assert "cat" not in captured[0] + + def test_returns_empty_on_request_error(self): + client = NewznabClient("http://nzbhydra:5076", "key") + with patch.object( + client, + "_get", + side_effect=requests.exceptions.ConnectionError("down"), + ): + results = client.search(query="book") + assert results == [] + + def test_returns_empty_on_malformed_xml(self): + client = NewznabClient("http://nzbhydra:5076", "key") + with patch.object(client, "_get", return_value=_make_response("not xml at all")): + results = client.search(query="book") + assert results == [] + + def test_includes_apikey_in_request(self): + """The apikey is injected by _get() into the outgoing HTTP request.""" + client = NewznabClient("http://nzbhydra:5076", "mykey") + captured_params: list = [] + + def fake_session_get(url, params=None, **kwargs): + captured_params.append(dict(params or {})) + r = _make_response(NZB_XML) + return r + + with patch.object(client._session, "get", side_effect=fake_session_get): + client.search(query="test") + + assert len(captured_params) == 1 + assert captured_params[0].get("apikey") == "mykey" + + def test_uses_book_search_type_when_specified(self): + client = NewznabClient("http://nzbhydra:5076", "key") + captured: list = [] + + def fake_get(params, accept_xml=False): + captured.append(params.copy()) + return _make_response(NZB_XML) + + with patch.object(client, "_get", side_effect=fake_get): + client.search(query="book", search_type="book") + + assert captured[0]["t"] == "book" diff --git a/tests/newznab/test_cache.py b/tests/newznab/test_cache.py new file mode 100644 index 00000000..fd294aee --- /dev/null +++ b/tests/newznab/test_cache.py @@ -0,0 +1,93 @@ +"""Unit tests for the Newznab release cache.""" + +import time +from unittest.mock import patch + +import pytest + +from shelfmark.release_sources.newznab import cache as cache_module +from shelfmark.release_sources.newznab.cache import ( + cache_release, + cleanup_expired, + get_release, + remove_release, +) + + +@pytest.fixture(autouse=True) +def clear_cache(): + """Ensure the cache is empty before and after each test.""" + cache_module._cache.clear() + yield + cache_module._cache.clear() + + +class TestCacheRelease: + def test_stores_and_retrieves_release(self): + data = {"title": "My Book", "downloadUrl": "https://example.com/nzb/1"} + cache_release("id-1", data) + assert get_release("id-1") == data + + def test_overwrites_existing_entry(self): + cache_release("id-1", {"title": "Old"}) + cache_release("id-1", {"title": "New"}) + assert get_release("id-1")["title"] == "New" + + +class TestGetRelease: + def test_returns_none_for_unknown_id(self): + assert get_release("no-such-id") is None + + def test_returns_none_after_ttl_expires(self): + cache_release("id-ttl", {"title": "Expiring"}) + past = time.time() - cache_module.RELEASE_CACHE_TTL - 1 + cache_module._cache["id-ttl"] = (cache_module._cache["id-ttl"][0], past) + assert get_release("id-ttl") is None + + def test_expired_entry_is_removed(self): + cache_release("id-evict", {"title": "Gone"}) + past = time.time() - cache_module.RELEASE_CACHE_TTL - 1 + cache_module._cache["id-evict"] = (cache_module._cache["id-evict"][0], past) + get_release("id-evict") + assert "id-evict" not in cache_module._cache + + def test_fresh_entry_not_expired(self): + cache_release("id-fresh", {"title": "Fresh"}) + # Advance time by less than TTL + future = time.time() + cache_module.RELEASE_CACHE_TTL - 60 + with patch("shelfmark.release_sources.newznab.cache.time") as mock_time: + mock_time.time.return_value = future + result = get_release("id-fresh") + assert result is not None + + +class TestRemoveRelease: + def test_removes_existing_entry(self): + cache_release("id-remove", {"title": "To Remove"}) + remove_release("id-remove") + assert get_release("id-remove") is None + + def test_no_error_when_removing_absent_entry(self): + remove_release("no-such-id") # should not raise + + +class TestCleanupExpired: + def test_removes_only_expired_entries(self): + cache_release("fresh", {"title": "Fresh"}) + cache_release("stale", {"title": "Stale"}) + + past = time.time() - cache_module.RELEASE_CACHE_TTL - 1 + cache_module._cache["stale"] = (cache_module._cache["stale"][0], past) + + removed = cleanup_expired() + assert removed == 1 + assert get_release("fresh") is not None + assert "stale" not in cache_module._cache + + def test_returns_zero_when_nothing_expired(self): + cache_release("a", {}) + cache_release("b", {}) + assert cleanup_expired() == 0 + + def test_returns_zero_when_cache_empty(self): + assert cleanup_expired() == 0 diff --git a/tests/newznab/test_handler.py b/tests/newznab/test_handler.py new file mode 100644 index 00000000..725376be --- /dev/null +++ b/tests/newznab/test_handler.py @@ -0,0 +1,174 @@ +"""Unit tests for the Newznab download handler.""" + +from threading import Event +from unittest.mock import patch + +from shelfmark.core.models import DownloadTask +from shelfmark.release_sources.newznab.handler import ( + NewznabHandler, + _get_download_url, + _get_protocol, +) + +# ── helpers ──────────────────────────────────────────────────────────────────── + + +class ProgressRecorder: + def __init__(self): + self.progress_values: list[float] = [] + self.status_updates: list[tuple[str, str | None]] = [] + + def progress_callback(self, v: float): + self.progress_values.append(v) + + def status_callback(self, status: str, message: str | None): + self.status_updates.append((status, message)) + + @property + def last_status(self) -> str | None: + return self.status_updates[-1][0] if self.status_updates else None + + @property + def last_message(self) -> str | None: + return self.status_updates[-1][1] if self.status_updates else None + + @property + def statuses(self) -> list[str]: + return [s[0] for s in self.status_updates] + + +# ── _get_protocol ──────────────────────────────────────────────────────────────── + + +class TestGetProtocol: + def test_explicit_usenet(self): + assert _get_protocol({"protocol": "usenet"}) == "usenet" + + def test_explicit_torrent(self): + assert _get_protocol({"protocol": "torrent"}) == "torrent" + + def test_magnet_url_infers_torrent(self): + assert _get_protocol({"magnetUrl": "magnet:?xt=urn:btih:abc"}) == "torrent" + + def test_torrent_extension_infers_torrent(self): + assert _get_protocol({"downloadUrl": "https://example.com/file.torrent"}) == "torrent" + + def test_nzb_extension_infers_usenet(self): + assert _get_protocol({"downloadUrl": "https://example.com/file.nzb"}) == "usenet" + + def test_defaults_to_usenet_when_ambiguous(self): + # Newznab is usenet-native; ambiguous URLs default to usenet. + assert _get_protocol({"downloadUrl": "https://example.com/download/123"}) == "usenet" + + def test_empty_dict_defaults_to_usenet(self): + assert _get_protocol({}) == "usenet" + + +# ── _get_download_url ────────────────────────────────────────────────────────── + + +class TestGetDownloadUrl: + def test_usenet_prefers_download_url(self): + result = { + "protocol": "usenet", + "downloadUrl": "https://example.com/nzb", + "magnetUrl": "magnet:?xt=urn:btih:abc", + } + assert _get_download_url(result) == "https://example.com/nzb" + + def test_torrent_prefers_magnet(self): + result = { + "protocol": "torrent", + "downloadUrl": "https://example.com/file.torrent", + "magnetUrl": "magnet:?xt=urn:btih:abc", + } + assert _get_download_url(result) == "magnet:?xt=urn:btih:abc" + + def test_falls_back_to_download_url_when_no_magnet(self): + result = { + "protocol": "torrent", + "downloadUrl": "https://example.com/file.torrent", + } + assert _get_download_url(result) == "https://example.com/file.torrent" + + +# ── error paths ──────────────────────────────────────────────────────────────── + + +class TestHandlerErrors: + def test_cache_miss_returns_error(self): + with patch("shelfmark.release_sources.newznab.handler.get_release", return_value=None): + handler = NewznabHandler() + task = DownloadTask(task_id="missing", source="newznab", title="Book") + recorder = ProgressRecorder() + result = handler.download( + task=task, + cancel_flag=Event(), + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + assert result is None + assert recorder.last_status == "error" + assert "cache" in (recorder.last_message or "").lower() + + def test_no_download_url_returns_error(self): + with patch( + "shelfmark.release_sources.newznab.handler.get_release", + return_value={"protocol": "usenet", "title": "Book"}, + ): + handler = NewznabHandler() + task = DownloadTask(task_id="no-url", source="newznab", title="Book") + recorder = ProgressRecorder() + result = handler.download( + task=task, + cancel_flag=Event(), + progress_callback=recorder.progress_callback, + status_callback=recorder.status_callback, + ) + assert result is None + assert recorder.last_status == "error" + assert "url" in (recorder.last_message or "").lower() + + def test_no_client_configured_returns_error(self): + with ( + patch( + "shelfmark.release_sources.newznab.handler.get_release", + return_value={ + "protocol": "usenet", + "downloadUrl": "https://example.com/nzb/1", + }, + ), + patch("shelfmark.release_sources.newznab.handler.get_client", return_value=None), + patch( + "shelfmark.release_sources.newznab.handler.list_configured_clients", + return_value=[], + ), + ): + handler = NewznabHandler() + task = DownloadTask(task_id="no-client", source="newznab", title="Book") + recorder = ProgressRecorder() + result = handler.download( + task=task, + cancel_flag=Event(), + 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 or "").lower() + + +# ── cancel ───────────────────────────────────────────────────────────────────── + + +class TestHandlerCancel: + def test_cancel_removes_from_cache(self): + with patch("shelfmark.release_sources.newznab.handler.remove_release") as mock_remove: + result = NewznabHandler().cancel("task-123") + assert result is True + mock_remove.assert_called_once_with("task-123") + + def test_cancel_handles_absent_task(self): + with patch("shelfmark.release_sources.newznab.handler.remove_release"): + result = NewznabHandler().cancel("no-such-task") + assert result is True diff --git a/tests/newznab/test_source.py b/tests/newznab/test_source.py new file mode 100644 index 00000000..c4efa264 --- /dev/null +++ b/tests/newznab/test_source.py @@ -0,0 +1,339 @@ +"""Unit tests for the Newznab release source.""" + +from unittest.mock import MagicMock + +from shelfmark.core.search_plan import ReleaseSearchPlan, ReleaseSearchVariant +from shelfmark.metadata_providers import BookMetadata +from shelfmark.release_sources import ReleaseProtocol +from shelfmark.release_sources.newznab.source import ( + NewznabSource, + _newznab_result_to_release, +) + +# ── fixtures / helpers ───────────────────────────────────────────────────────── + + +def _make_book(**kwargs) -> BookMetadata: + defaults = { + "provider": "hardcover", + "provider_id": "1", + "title": "Dune", + "authors": ["Frank Herbert"], + } + defaults.update(kwargs) + return BookMetadata(**defaults) + + +def _make_result(**kwargs) -> dict: + """Minimal Newznab-like result dict.""" + base = { + "title": "Dune (2024) [EPUB]", + "guid": "https://indexer.example.com/nzb/42", + "downloadUrl": "https://indexer.example.com/nzb/42?apikey=secret", + "protocol": "usenet", + "size": 2097152, + "indexer": "MyIndexer", + "categories": [7000], + "indexerFlags": [], + "publishDate": "2024-01-01", + } + base.update(kwargs) + return base + + +def _make_plan(book: BookMetadata, *, manual_query: str | None = None): + from shelfmark.core.search_plan import build_release_search_plan + + return build_release_search_plan(book, languages=["en"], manual_query=manual_query) + + +# ── _newznab_result_to_release ───────────────────────────────────────────────── + + +class TestResultToRelease: + def test_basic_usenet_result(self): + r = _newznab_result_to_release(_make_result()) + assert r.source == "newznab" + assert r.title == "Dune (2024) [EPUB]" + assert r.protocol == ReleaseProtocol.NZB + assert r.size == "2.0 MB" + assert r.size_bytes == 2097152 + assert r.indexer == "MyIndexer" + assert r.source_id == "https://indexer.example.com/nzb/42" + assert r.download_url == "https://indexer.example.com/nzb/42?apikey=secret" + + def test_torrent_result_has_torrent_protocol(self): + r = _newznab_result_to_release( + _make_result( + protocol="torrent", + magnetUrl="magnet:?xt=urn:btih:abc", + categories=[3030], + ) + ) + assert r.protocol == ReleaseProtocol.TORRENT + + def test_audiobook_category_detected(self): + r = _newznab_result_to_release(_make_result(categories=[3030]), "ebook") + assert r.content_type == "audiobook" + + def test_book_category_detected(self): + r = _newznab_result_to_release(_make_result(categories=[7000])) + assert r.content_type == "book" + + def test_no_categories_uses_content_type_fallback(self): + r = _newznab_result_to_release(_make_result(categories=[]), "audiobook") + assert r.content_type == "audiobook" + + def test_freeleech_flag_detected_via_download_volume(self): + r = _newznab_result_to_release(_make_result(downloadVolumeFactor=0.0)) + assert r.extra["freeleech"] is True + assert "FreeLeech" in r.extra["indexer_flags"] + + def test_freeleech_flag_detected_via_indexer_flags(self): + r = _newznab_result_to_release(_make_result(indexerFlags=["freeleech"])) + assert r.extra["freeleech"] is True + + def test_vip_detected_from_title(self): + r = _newznab_result_to_release(_make_result(title="Dune [VIP] [EPUB]")) + assert r.extra["vip"] is True + assert "VIP" in r.extra["indexer_flags"] + + def test_duplicate_flags_deduplicated(self): + r = _newznab_result_to_release(_make_result(indexerFlags=["FreeLeech", "freeleech", "FL"])) + lower_flags = [f.lower() for f in r.extra["indexer_flags"]] + assert lower_flags.count("freeleech") == 1 + + def test_seeders_only_set_for_torrents(self): + usenet = _newznab_result_to_release(_make_result(protocol="usenet", seeders=10)) + assert usenet.seeders is None + + torrent = _newznab_result_to_release( + _make_result(protocol="torrent", seeders=10, leechers=2) + ) + assert torrent.seeders == 10 + assert torrent.peers == "10 / 2" + + def test_fallback_source_id_when_no_guid(self): + r = _newznab_result_to_release(_make_result(guid=None)) + assert r.source_id.startswith("newznab:") + + def test_none_size_returns_none(self): + r = _newznab_result_to_release(_make_result(size=None)) + assert r.size is None + assert r.size_bytes is None + + def test_extra_fields_preserved(self): + r = _newznab_result_to_release( + _make_result( + author="Frank Herbert", + bookTitle="Dune", + infoHash="abc123", + ) + ) + assert r.extra["author"] == "Frank Herbert" + assert r.extra["book_title"] == "Dune" + assert r.extra["info_hash"] == "abc123" + + +# ── NewznabSource.is_available ───────────────────────────────────────────────── + + +class TestIsAvailable: + def _config(self, **overrides): + values = { + "NEWZNAB_ENABLED": True, + "NEWZNAB_URL": "http://nzbhydra:5076", + } + values.update(overrides) + return lambda k, default=None: values.get(k, default) + + def test_available_when_enabled_and_url_set(self, monkeypatch): + import shelfmark.release_sources.newznab.source as mod + + monkeypatch.setattr(mod.config, "get", self._config()) + assert NewznabSource().is_available() is True + + def test_unavailable_when_disabled(self, monkeypatch): + import shelfmark.release_sources.newznab.source as mod + + monkeypatch.setattr(mod.config, "get", self._config(NEWZNAB_ENABLED=False)) + assert NewznabSource().is_available() is False + + def test_unavailable_when_no_url(self, monkeypatch): + import shelfmark.release_sources.newznab.source as mod + + monkeypatch.setattr(mod.config, "get", self._config(NEWZNAB_URL="")) + assert NewznabSource().is_available() is False + + +# ── NewznabSource.search ─────────────────────────────────────────────────────── + + +class TestSearch: + def _fake_config(self, **overrides): + values = { + "NEWZNAB_AUTO_EXPAND": False, + } + values.update(overrides) + return lambda k, default=None: values.get(k, default) + + def _patched_source(self, monkeypatch, client, config_overrides=None): + import shelfmark.release_sources.newznab.source as mod + + monkeypatch.setattr(mod.config, "get", self._fake_config(**(config_overrides or {}))) + src = NewznabSource() + monkeypatch.setattr(src, "_get_client", lambda: client) + return src + + def test_returns_empty_when_no_client(self, monkeypatch): + import shelfmark.release_sources.newznab.source as mod + + monkeypatch.setattr(mod.config, "get", self._fake_config()) + src = NewznabSource() + monkeypatch.setattr(src, "_get_client", lambda: None) + book = _make_book() + results = src.search(book, _make_plan(book)) + assert results == [] + + def test_returns_empty_with_no_queries(self, monkeypatch): + client = MagicMock() + client.search.return_value = [] + src = self._patched_source(monkeypatch, client) + book = BookMetadata(provider="test", provider_id="1", title="", authors=[]) + from shelfmark.core.search_plan import ReleaseSearchPlan + + empty_plan = ReleaseSearchPlan( + languages=["en"], + isbn_candidates=[], + author="", + title_variants=[], + grouped_title_variants=[], + manual_query=None, + indexers=[], + ) + results = src.search(book, empty_plan) + assert results == [] + client.search.assert_not_called() + + def test_searches_with_ebook_category(self, monkeypatch): + client = MagicMock() + client.search.return_value = [] + src = self._patched_source(monkeypatch, client) + book = _make_book() + src.search(book, _make_plan(book), content_type="ebook") + _, kwargs = client.search.call_args + assert kwargs["categories"] == [7000] + + def test_searches_with_audiobook_category(self, monkeypatch): + client = MagicMock() + client.search.return_value = [] + src = self._patched_source(monkeypatch, client) + book = _make_book() + src.search(book, _make_plan(book), content_type="audiobook") + _, kwargs = client.search.call_args + assert kwargs["categories"] == [3030] + + def test_expand_search_removes_categories(self, monkeypatch): + client = MagicMock() + client.search.return_value = [] + src = self._patched_source(monkeypatch, client) + book = _make_book() + src.search(book, _make_plan(book), expand_search=True, content_type="ebook") + _, kwargs = client.search.call_args + assert kwargs["categories"] is None + + def test_deduplicates_results_by_guid(self, monkeypatch): + dup = _make_result(guid="same-guid") + client = MagicMock() + client.search.side_effect = [[dup], [dup]] # two queries, same result each + book = _make_book() + src = self._patched_source(monkeypatch, client) + plan_two = ReleaseSearchPlan( + languages=["en"], + isbn_candidates=[], + author="Frank Herbert", + title_variants=[ + ReleaseSearchVariant("Dune", "Frank Herbert"), + ReleaseSearchVariant("Düne", "Frank Herbert"), + ], + grouped_title_variants=[], + manual_query=None, + indexers=[], + ) + results = src.search(book, plan_two) + assert len(results) == 1 + + def test_auto_expand_retries_without_categories(self, monkeypatch): + calls: list = [] + + def fake_search(query, categories=None): + calls.append(categories) + return [] if categories else [_make_result()] + + client = MagicMock() + client.search.side_effect = fake_search + src = self._patched_source(monkeypatch, client, {"NEWZNAB_AUTO_EXPAND": True}) + book = _make_book() + results = src.search(book, _make_plan(book), content_type="ebook") + + assert [7000] in calls # first call with category + assert None in calls # auto-expanded call without + assert len(results) == 1 + + def test_no_auto_expand_when_disabled(self, monkeypatch): + client = MagicMock() + client.search.return_value = [] + src = self._patched_source(monkeypatch, client, {"NEWZNAB_AUTO_EXPAND": False}) + book = _make_book() + src.search(book, _make_plan(book)) + # Only one call per query, no retry + assert client.search.call_count == 1 + + def test_converts_results_to_releases(self, monkeypatch): + client = MagicMock() + client.search.return_value = [_make_result()] + src = self._patched_source(monkeypatch, client) + book = _make_book() + results = src.search(book, _make_plan(book)) + assert len(results) == 1 + r = results[0] + assert r.source == "newznab" + assert r.protocol == ReleaseProtocol.NZB + + def test_manual_query_overrides_title_variants(self, monkeypatch): + client = MagicMock() + client.search.return_value = [] + src = self._patched_source(monkeypatch, client) + book = _make_book() + plan = _make_plan(book, manual_query="custom search term") + src.search(book, plan) + call_kwargs = client.search.call_args + assert call_kwargs[1]["query"] == "custom search term" + + def test_isbn_used_when_no_title_variants(self, monkeypatch): + client = MagicMock() + client.search.return_value = [] + src = self._patched_source(monkeypatch, client) + book = _make_book() + from shelfmark.core.search_plan import ReleaseSearchPlan + + isbn_plan = ReleaseSearchPlan( + languages=["en"], + isbn_candidates=["9780441013593"], + author="Frank Herbert", + title_variants=[], + grouped_title_variants=[], + manual_query=None, + indexers=[], + ) + src.search(book, isbn_plan) + call_kwargs = client.search.call_args + assert call_kwargs[1]["query"] == "9780441013593" + + def test_exception_in_client_returns_empty(self, monkeypatch): + client = MagicMock() + client.search.side_effect = RuntimeError("boom") + src = self._patched_source(monkeypatch, client) + book = _make_book() + results = src.search(book, _make_plan(book)) + assert results == []