53 lines
2.0 KiB
Python
53 lines
2.0 KiB
Python
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from wordarr.metadata import audible
|
|
|
|
|
|
@pytest.fixture
|
|
def calls(monkeypatch):
|
|
"""Capture the query params audible.py sends, answering from a fake catalog."""
|
|
seen = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
params = dict(request.url.params)
|
|
seen.append(params)
|
|
# only the first page carries results, like a query with few hits
|
|
if int(params.get("page", 0)) > 0:
|
|
return httpx.Response(200, json={"products": []})
|
|
return httpx.Response(200, json={"products": [{
|
|
"asin": "ASIN1",
|
|
"title": "Die drei ??? und der Super-Papagei",
|
|
"authors": [{"name": "H.G. Francis"}],
|
|
"series": [{"asin": "SERIES1", "title": "Die drei ???", "sequence": "1"}],
|
|
}]})
|
|
|
|
real_client = httpx.AsyncClient
|
|
monkeypatch.setattr(audible.httpx, "AsyncClient",
|
|
lambda **kw: real_client(transport=httpx.MockTransport(handler), **kw))
|
|
return seen
|
|
|
|
|
|
def test_search_asks_for_the_first_page(calls):
|
|
"""Audible's page parameter is 0-based, page=1 skips the only page of hits."""
|
|
results = asyncio.run(audible.search("Der Superpapagei"))
|
|
assert [r.title for r in results] == ["Die drei ??? und der Super-Papagei"]
|
|
assert calls and all(int(c["page"]) == 0 for c in calls)
|
|
assert all(c["products_sort_by"] == "Relevance" for c in calls)
|
|
|
|
|
|
def test_series_crawl_starts_at_page_zero(calls):
|
|
episodes = asyncio.run(audible.series_episodes("SERIES1", "Die drei ???"))
|
|
assert [e.volume for e in episodes] == [1]
|
|
assert episodes[0].series_id == "SERIES1"
|
|
pages = {int(c["page"]) for c in calls}
|
|
assert 0 in pages and max(pages) > 0
|
|
# relevance sorting shrinks a deep sweep, so the crawl must not send it
|
|
assert all("products_sort_by" not in c for c in calls)
|
|
|
|
|
|
def test_series_crawl_ignores_other_series(calls):
|
|
assert asyncio.run(audible.series_episodes("OTHER", "Die drei ???")) == []
|