mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-06-11 23:04:41 -04:00
18a3f0bf44
I've added a plugin using the same architecture as the prowlarr plugin to enable Newznab as a source. I've tested locally with nzbhydra2 and it all seems to work as intended. I've added some unit tests for this feature, and found that a couple of other unit tests weren't behaving so fixed those up while I was at it. I also ran all of the linters in the makefile against it and fixed those up, too, so hopefully this should be as clean and as compatible as it can be.
340 lines
13 KiB
Python
340 lines
13 KiB
Python
"""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 == []
|