264 lines
8.8 KiB
Python
264 lines
8.8 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
from zot.fetch.resolve import (
|
||
BROWSER_UA,
|
||
LIBGEN,
|
||
Candidate,
|
||
Pacer,
|
||
candidates,
|
||
libgen_candidates,
|
||
libgen_editions,
|
||
libgen_files,
|
||
title_candidates,
|
||
title_matches,
|
||
)
|
||
from zot.fetch.select import ItemRef
|
||
|
||
FX = Path(__file__).parent / "fixtures"
|
||
TEMEL = "Early Palliative Care for Patients with Metastatic Non–Small-Cell Lung Cancer"
|
||
|
||
|
||
def _fx(name: str) -> str:
|
||
return (FX / name).read_text()
|
||
|
||
|
||
def _pacer() -> Pacer:
|
||
return Pacer(3.0, clock=lambda: 0.0, sleep=lambda s: None)
|
||
|
||
|
||
def _client(handler) -> httpx.Client:
|
||
return httpx.Client(
|
||
transport=httpx.MockTransport(handler), headers={"User-Agent": BROWSER_UA}
|
||
)
|
||
|
||
|
||
def _libgen_handler(seen: list[str]):
|
||
def handler(req: httpx.Request) -> httpx.Response:
|
||
seen.append(str(req.url))
|
||
assert req.headers["user-agent"] == BROWSER_UA
|
||
q = req.url.params
|
||
if req.url.path == "/json.php" and q.get("object") == "e":
|
||
if q.get("doi") == "10.1056/NEJMoa1000678":
|
||
return httpx.Response(200, text=_fx("libgen_json_doi.json"))
|
||
if q.get("isbn") == "9781640162280" or q.get("ids") == "146962484":
|
||
return httpx.Response(200, text=_fx("libgen_json_isbn.json"))
|
||
return httpx.Response(200, text="[]")
|
||
if req.url.path == "/json.php" and q.get("object") == "f":
|
||
want = set(q.get("ids", "").split(","))
|
||
allf = json.loads(_fx("libgen_json_file.json"))
|
||
return httpx.Response(
|
||
200, json={k: v for k, v in allf.items() if k in want}
|
||
)
|
||
if req.url.path == "/index.php":
|
||
return httpx.Response(200, text=_fx("libgen_index.html"))
|
||
return httpx.Response(404)
|
||
|
||
return handler
|
||
|
||
|
||
def test_pacer_sleeps_only_the_remaining_interval():
|
||
now = [100.0]
|
||
slept: list[float] = []
|
||
p = Pacer(3.0, clock=lambda: now[0], sleep=slept.append)
|
||
p.wait() # first call never sleeps
|
||
now[0] += 1.0
|
||
p.wait()
|
||
assert slept == [2.0]
|
||
now[0] += 10.0
|
||
p.wait()
|
||
assert slept == [2.0]
|
||
|
||
|
||
def test_libgen_editions_doi_and_isbn_and_ids():
|
||
seen: list[str] = []
|
||
c = _client(_libgen_handler(seen))
|
||
e = libgen_editions(c, doi="10.1056/NEJMoa1000678", pacer=_pacer())
|
||
assert list(e) == ["27703795"]
|
||
assert list(libgen_editions(c, isbn="9781640162280", pacer=_pacer())) == [
|
||
"146962484"
|
||
]
|
||
assert list(libgen_editions(c, ids=("146962484",), pacer=_pacer())) == ["146962484"]
|
||
assert seen[0].startswith(f"{LIBGEN}/json.php?object=e&addkeys=%2A&doi=")
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"status, body",
|
||
[(500, "x"), (200, "<html>nginx</html>"), (200, "[]"), (200, "null")],
|
||
)
|
||
def test_libgen_editions_bad_answers_are_empty(status, body):
|
||
c = _client(lambda req: httpx.Response(status, text=body))
|
||
assert libgen_editions(c, doi="10.1/x", pacer=_pacer()) == {}
|
||
|
||
|
||
def test_libgen_editions_transport_error_is_empty():
|
||
def boom(req):
|
||
raise httpx.ConnectError("down")
|
||
|
||
assert libgen_editions(_client(boom), doi="10.1/x", pacer=_pacer()) == {}
|
||
|
||
|
||
def test_libgen_files_batches_ids_in_one_call():
|
||
seen: list[str] = []
|
||
c = _client(_libgen_handler(seen))
|
||
f = libgen_files(c, ["27858481", "104009164"], pacer=_pacer())
|
||
assert set(f) == {"27858481", "104009164"}
|
||
assert len(seen) == 1 and "ids=27858481%2C104009164" in seen[0]
|
||
assert libgen_files(c, [], pacer=_pacer()) == {}
|
||
|
||
|
||
def test_libgen_candidates_paper_pdf_only_with_size():
|
||
c = _client(_libgen_handler([]))
|
||
got = libgen_candidates(
|
||
c, kind="paper", doi="10.1056/NEJMoa1000678", pacer=_pacer()
|
||
)
|
||
assert got == [
|
||
Candidate(
|
||
"libgen", "a78add1bd62047ead5945134488f470f", None, "pdf", 401662, TEMEL
|
||
),
|
||
Candidate(
|
||
"libgen", "b78add1bd62047ead5945134488f470f", None, "pdf", None, TEMEL
|
||
),
|
||
]
|
||
|
||
|
||
def test_libgen_candidates_book_epub_before_pdf():
|
||
c = _client(_libgen_handler([]))
|
||
got = libgen_candidates(c, kind="book", isbn="9781640162280", pacer=_pacer())
|
||
assert [(g.ext, g.md5[:3]) for g in got] == [("epub", "220"), ("pdf", "120")]
|
||
|
||
|
||
def test_libgen_file_lookup_missing_defaults_paper_to_pdf_drops_book():
|
||
def handler(req: httpx.Request) -> httpx.Response:
|
||
q = req.url.params
|
||
if q.get("object") == "f":
|
||
return httpx.Response(503)
|
||
if q.get("doi"):
|
||
return httpx.Response(200, text=_fx("libgen_json_doi.json"))
|
||
return httpx.Response(200, text=_fx("libgen_json_isbn.json"))
|
||
|
||
c = _client(handler)
|
||
paper = libgen_candidates(
|
||
c, kind="paper", doi="10.1056/NEJMoa1000678", pacer=_pacer()
|
||
)
|
||
assert [p.ext for p in paper] == ["pdf", "pdf"] and paper[0].size_bytes is None
|
||
book = libgen_candidates(c, kind="book", isbn="9781640162280", pacer=_pacer())
|
||
assert book == []
|
||
|
||
|
||
def test_libgen_candidates_skips_files_without_md5():
|
||
body = json.dumps(
|
||
{"1": {"title": "T", "files": {"9": {"f_id": "9"}}, "year": "2000"}}
|
||
)
|
||
c = _client(lambda req: httpx.Response(200, text=body))
|
||
assert libgen_candidates(c, kind="paper", doi="10.1/x", pacer=_pacer()) == []
|
||
|
||
|
||
def test_title_matches_exact_then_containment_with_year():
|
||
entries = [
|
||
{"id": "1", "title": "HCPCS Level II Professional 2023", "year": "2023"},
|
||
{"id": "2", "title": "HCPCS Level II Professional 2019", "year": "2019"},
|
||
{"id": "3", "title": "CPT 2023 Professional", "year": "2023"},
|
||
]
|
||
exact = title_matches("HCPCS Level II Professional 2023", "", entries)
|
||
assert [e["id"] for e in exact] == ["1"]
|
||
by_year = title_matches("HCPCS Level II Professional", "2019", entries)
|
||
assert [e["id"] for e in by_year] == ["2"]
|
||
both = title_matches("HCPCS Level II Professional", "", entries)
|
||
assert [e["id"] for e in both] == ["1", "2"]
|
||
assert title_matches("Netter's Atlas", "", entries) == []
|
||
|
||
|
||
def test_title_candidates_parses_index_then_fetches_editions():
|
||
seen: list[str] = []
|
||
c = _client(_libgen_handler(seen))
|
||
got = title_candidates(
|
||
c, title="HCPCS Level II Professional", year="2023", pacer=_pacer()
|
||
)
|
||
assert [g.ext for g in got] == ["epub", "pdf"]
|
||
assert any(
|
||
"/index.php?" in s and "req=HCPCS+Level+II+Professional" in s for s in seen
|
||
)
|
||
assert any("object=e" in s and "ids=146962484" in s for s in seen)
|
||
|
||
|
||
def test_title_candidates_no_rows_or_error_is_empty():
|
||
empty = _client(lambda r: httpx.Response(200, text="<html></html>"))
|
||
assert title_candidates(empty, title="X", year="", pacer=_pacer()) == []
|
||
err = _client(lambda r: httpx.Response(500))
|
||
assert title_candidates(err, title="X", year="", pacer=_pacer()) == []
|
||
ok = _client(lambda r: httpx.Response(200))
|
||
assert title_candidates(ok, title="", year="", pacer=_pacer()) == []
|
||
|
||
|
||
def test_title_candidates_transport_error_is_empty():
|
||
def boom(req):
|
||
raise httpx.ConnectError("down")
|
||
|
||
assert title_candidates(_client(boom), title="X", year="", pacer=_pacer()) == []
|
||
|
||
|
||
def _direct(oa: bool):
|
||
def handler(req: httpx.Request) -> httpx.Response:
|
||
assert req.url.host == "api.unpaywall.org"
|
||
assert req.url.params["email"] == "dev@fhirworx.io"
|
||
if not oa:
|
||
return httpx.Response(404)
|
||
return httpx.Response(200, text=_fx("unpaywall_doi.json"))
|
||
|
||
return httpx.Client(transport=httpx.MockTransport(handler))
|
||
|
||
|
||
def _ref(**kw) -> ItemRef:
|
||
base = dict(
|
||
item_id=1,
|
||
key="ABCD2345",
|
||
item_type="journalArticle",
|
||
title="T",
|
||
doi="",
|
||
isbn="",
|
||
year="",
|
||
)
|
||
base.update(kw)
|
||
return ItemRef(**base)
|
||
|
||
|
||
def test_candidates_doi_unpaywall_first_then_libgen():
|
||
got = candidates(
|
||
_ref(doi="10.1056/NEJMoa1000678"),
|
||
direct=_direct(True),
|
||
proxied=_client(_libgen_handler([])),
|
||
email="dev@fhirworx.io",
|
||
up_pacer=_pacer(),
|
||
lg_pacer=_pacer(),
|
||
)
|
||
assert [g.source for g in got] == ["unpaywall"] * 3 + ["libgen"] * 2
|
||
assert got[0] == Candidate(
|
||
"unpaywall", None, "https://example.org/best.pdf", "pdf", None, "T"
|
||
)
|
||
assert got[3].md5 == "a78add1bd62047ead5945134488f470f"
|
||
|
||
|
||
def test_candidates_isbn_book_and_title_fallback_and_nothing():
|
||
proxied = _client(_libgen_handler([]))
|
||
kw = dict(
|
||
direct=_direct(False),
|
||
proxied=proxied,
|
||
email="dev@fhirworx.io",
|
||
up_pacer=_pacer(),
|
||
lg_pacer=_pacer(),
|
||
)
|
||
book = candidates(_ref(item_type="book", isbn="9781640162280"), **kw)
|
||
assert [b.source for b in book] == ["libgen", "libgen"]
|
||
by_title = candidates(
|
||
_ref(item_type="book", title="HCPCS Level II Professional", year="2023"), **kw
|
||
)
|
||
assert [b.ext for b in by_title] == ["epub", "pdf"]
|
||
assert candidates(_ref(title="paper without doi"), **kw) == []
|
||
assert candidates(_ref(item_type="book", title=""), **kw) == []
|