"""The three ebook providers and the search that merges them.""" import asyncio import httpx import pytest from wordarr import metadata from wordarr.metadata import googlebooks, hardcover, openlibrary HARDCOVER_HIT = { "document": { "id": "42", "title": "The Hobbit", "author_names": ["J. R. R. Tolkien"], "release_year": 1937, "isbns": ["9780261102217"], "image": {"url": "https://example.invalid/hobbit.jpg"}, "series_names": [], } } GERMAN_EDITION = { "book_id": 42, "title": "Der Hobbit", "isbn_13": "9783608939828", "isbn_10": None, "release_date": "1957-03-01", "cached_image": "https://example.invalid/hobbit-de.jpg", } def mock_client(monkeypatch, module, handler): real = httpx.AsyncClient monkeypatch.setattr(module.httpx, "AsyncClient", lambda **kw: real(transport=httpx.MockTransport(handler), **kw)) @pytest.fixture def hardcover_calls(monkeypatch): seen = [] def handler(request: httpx.Request) -> httpx.Response: import json body = json.loads(request.content) seen.append((body["variables"], request.headers.get("authorization"))) if "editions" in body["query"]: return httpx.Response(200, json={"data": {"editions": [GERMAN_EDITION]}}) return httpx.Response(200, json={"data": {"search": {"results": {"hits": [HARDCOVER_HIT]}}}}) monkeypatch.setattr(hardcover, "TOKEN", "test-token") mock_client(monkeypatch, hardcover, handler) return seen def test_hardcover_needs_no_network_without_a_token(monkeypatch): monkeypatch.setattr(hardcover, "TOKEN", "") assert asyncio.run(hardcover.search("hobbit")) == [] def test_hardcover_parses_a_hit(hardcover_calls): hit = asyncio.run(hardcover.search("hobbit"))[0] assert (hit.title, hit.authors, hit.year) == ("The Hobbit", "J. R. R. Tolkien", 1937) assert hit.external_id == "9780261102217" assert hit.source == "hardcover" assert hardcover_calls[0][1] == "Bearer test-token" def test_hardcover_swaps_in_the_german_edition(hardcover_calls): hit = asyncio.run(hardcover.search("hobbit", "german"))[0] assert (hit.title, hit.year, hit.language) == ("Der Hobbit", 1957, "german") assert hit.external_id == "9783608939828" assert hardcover_calls[1][0] == {"ids": [42], "lang": "de"} def test_hardcover_drops_books_without_an_edition_in_that_language(monkeypatch): def handler(request: httpx.Request) -> httpx.Response: import json if "editions" in json.loads(request.content)["query"]: return httpx.Response(200, json={"data": {"editions": []}}) return httpx.Response(200, json={"data": {"search": {"results": {"hits": [HARDCOVER_HIT]}}}}) monkeypatch.setattr(hardcover, "TOKEN", "test-token") mock_client(monkeypatch, hardcover, handler) assert asyncio.run(hardcover.search("hobbit", "german")) == [] def test_google_books_restricts_the_language(monkeypatch): seen = [] def handler(request: httpx.Request) -> httpx.Response: seen.append(dict(request.url.params)) return httpx.Response(200, json={"items": [{"volumeInfo": { "title": "Der Hobbit", "authors": ["J. R. R. Tolkien"], "publishedDate": "1957", "language": "de", "industryIdentifiers": [{"type": "ISBN_13", "identifier": "9783608939828"}], "imageLinks": {"thumbnail": "http://example.invalid/de.jpg"}, }}]}) mock_client(monkeypatch, googlebooks, handler) hit = asyncio.run(googlebooks.search("hobbit", "german"))[0] assert seen[0]["langRestrict"] == "de" assert (hit.title, hit.language, hit.year) == ("Der Hobbit", "german", 1957) assert hit.cover_url.startswith("https://") # the API still hands out http def test_google_books_searches_an_isbn_as_an_isbn(monkeypatch): seen = [] def handler(request: httpx.Request) -> httpx.Response: seen.append(dict(request.url.params)) return httpx.Response(200, json={"items": []}) mock_client(monkeypatch, googlebooks, handler) asyncio.run(googlebooks.search("978-3-608-93982-8")) assert seen[0]["q"] == "isbn:9783608939828" def test_open_library_filters_and_labels_the_language(monkeypatch): seen = [] def handler(request: httpx.Request) -> httpx.Response: seen.append(dict(request.url.params)) return httpx.Response(200, json={"docs": [{ "title": "Der Hobbit", "author_name": ["J. R. R. Tolkien"], "first_publish_year": 1957, "language": ["ger", "eng"], "isbn": ["9783608939828"], }]}) mock_client(monkeypatch, openlibrary, handler) hit = asyncio.run(openlibrary.search("hobbit", "german"))[0] assert seen[0]["q"] == "hobbit language:ger" assert hit.language == "german" assert hit.source == "openlibrary" def test_ebook_search_merges_the_providers_and_survives_a_broken_one(monkeypatch): async def hits(source, **fields): return [metadata.MetadataResult(media_type="ebook", source=source, **fields)] monkeypatch.setattr(hardcover, "TOKEN", "test-token") monkeypatch.setattr(hardcover, "search", lambda q, lang="", fb=True: hits( "hardcover", title="Der Hobbit", external_id="9783608939828")) monkeypatch.setattr(googlebooks, "search", lambda q, lang="", fb=True: hits( "googlebooks", title="Der Hobbit", external_id="9783608939828")) # same edition async def boom(q, lang="", fb=True): raise httpx.ConnectError("open library is down") monkeypatch.setattr(openlibrary, "search", boom) results = asyncio.run(metadata.ebook_search("hobbit", "german")) assert [r.source for r in results] == ["hardcover"] def test_automatic_search_asks_only_the_best_provider(monkeypatch): called = [] async def record(name, q, lang): called.append(name) return [] monkeypatch.setattr(hardcover, "TOKEN", "") monkeypatch.setattr(googlebooks, "search", lambda q, lang="", fb=True: record("gb", q, lang)) monkeypatch.setattr(openlibrary, "search", lambda q, lang="", fb=True: record("ol", q, lang)) asyncio.run(metadata.ebook_search("hobbit", "german", fallback=False)) assert called == ["gb"] # no token, so Google Books stands in for Hardcover