stack/tests/zot/fetch/test_resolve.py

from __future__ import annotations

import json
from pathlib import Path

import httpx
import pytest

from zot.fetch.resolve import (
    BROWSER_UA,
    LIBGEN,
    Candidate,
    Notes,
    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) == []


def test_libgen_candidates_skips_non_dict_editions():
    body = json.dumps({"1": None, "2": [1, 2], "3": {"title": "T", "files": []}})
    c = _client(lambda req: httpx.Response(200, text=body))
    assert libgen_candidates(c, kind="paper", doi="10.1/x", pacer=_pacer()) == []


# ── fix pass (review) ─────────────────────────────────────────────


def test_notes_error_for_http_and_transport_miss_for_empty():
    n = Notes()
    libgen_editions(
        _client(lambda r: httpx.Response(503)), doi="10.1/x", pacer=_pacer(), notes=n
    )
    assert n.had_error and "503" in str(n)

    def boom(req):
        raise httpx.ConnectError("down")

    n2 = Notes()
    libgen_editions(_client(boom), doi="10.1/x", pacer=_pacer(), notes=n2)
    assert n2.had_error and "ConnectError" in str(n2)

    n3 = Notes()
    libgen_candidates(
        _client(lambda r: httpx.Response(200, text="[]")),
        kind="paper",
        doi="10.1/x",
        pacer=_pacer(),
        notes=n3,
    )
    assert not n3.had_error and "no edition" in str(n3)


def test_candidates_note_unpaywall_miss():
    n = Notes()
    got = candidates(
        _ref(doi="10.1/x"),
        direct=_direct(False),
        proxied=_client(lambda r: httpx.Response(200, text="[]")),
        email="dev@fhirworx.io",
        up_pacer=_pacer(),
        lg_pacer=_pacer(),
        notes=n,
    )
    assert got == [] and "unpaywall" in str(n) and not n.had_error


def test_libgen_candidates_tolerates_bad_file_entries():
    body = json.dumps(
        {
            "1": {
                "title": "T",
                "files": {"a": None, "b": 3, "c": {"f_id": "9", "md5": "c" * 32}},
            }
        }
    )

    def h(r):
        if r.url.params.get("object") == "f":
            return httpx.Response(
                200, json={"9": {"extension": "pdf", "filesize": "n/a"}}
            )
        return httpx.Response(200, text=body)

    got = libgen_candidates(_client(h), kind="paper", doi="10.1/x", pacer=_pacer())
    assert [(g.md5, g.ext, g.size_bytes) for g in got] == [("c" * 32, "pdf", None)]