bb092c92d7
On the #650 full-text pass the cascade found 0 of 30 items that a one-off sweep recovered 4 of, all via Semantic Scholar; 7 more had PMCIDs Zotero lacked that NCBI idconv resolved in one call. - unpaywall_urls(): best_oa_location plus every oa_locations entry, direct PDF URLs before landing pages; fetch_one tries each. - fetch_semantic_scholar(): graph API openAccessPdf after Unpaywall, throttled to 1 req/s across workers; its externalIds.PubMedCentral fills a missing PMCID so the PMC tiers run. - pending_queue(client=, email=): PMIDs without a PMCID resolved via idconv in 200-id batches, persisted to extra as a PMCID: line. - run() reports an s2 counter; CLI help + docs page regenerated.
313 lines
12 KiB
Python
313 lines
12 KiB
Python
"""Tests for the #723 fetch tiers — Unpaywall all-locations, Semantic
|
|
Scholar openAccessPdf, and idconv PMCID resolution in the queue."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from prisma import fetch as fetch_mod
|
|
from prisma.fetch import (
|
|
PendingItem,
|
|
fetch_one,
|
|
fetch_semantic_scholar,
|
|
pending_queue,
|
|
resolve_pmcids,
|
|
unpaywall_urls,
|
|
)
|
|
from zot.db import TYPE_MAP, Db
|
|
from zot.schema import create_db
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _no_s2_throttle(monkeypatch):
|
|
monkeypatch.setattr(fetch_mod, "_S2_MIN_INTERVAL", 0.0)
|
|
|
|
|
|
def _json_client(routes: dict[str, object], seen: list[str] | None = None):
|
|
"""MockTransport client: substring of URL → JSON body (or status int)."""
|
|
|
|
def handler(request):
|
|
url = str(request.url)
|
|
if seen is not None:
|
|
seen.append(url)
|
|
for needle, body in routes.items():
|
|
if needle in url:
|
|
if isinstance(body, int):
|
|
return httpx.Response(body)
|
|
return httpx.Response(200, json=body)
|
|
return httpx.Response(404)
|
|
|
|
return httpx.Client(transport=httpx.MockTransport(handler))
|
|
|
|
|
|
# ── unpaywall_urls ────────────────────────────────────────────────
|
|
|
|
|
|
class TestUnpaywallUrls:
|
|
def test_all_locations_pdf_first(self):
|
|
body = {
|
|
"best_oa_location": {"url_for_pdf": None, "url": "https://a/landing"},
|
|
"oa_locations": [
|
|
{"url_for_pdf": None, "url": "https://a/landing"},
|
|
{"url_for_pdf": "https://b/paper.pdf", "url": "https://b/landing"},
|
|
{"url_for_pdf": "https://c/paper.pdf", "url": None},
|
|
],
|
|
}
|
|
urls = unpaywall_urls(_json_client({"unpaywall": body}), "10.1/x", "e@x")
|
|
# every PDF URL before any landing page; no duplicates
|
|
assert urls == [
|
|
"https://b/paper.pdf",
|
|
"https://c/paper.pdf",
|
|
"https://a/landing",
|
|
"https://b/landing",
|
|
]
|
|
|
|
def test_best_location_only_legacy_shape(self):
|
|
body = {"best_oa_location": {"url_for_pdf": "https://oa/p.pdf"}}
|
|
assert unpaywall_urls(_json_client({"unpaywall": body}), "10.1/x", "e@x") == [
|
|
"https://oa/p.pdf"
|
|
]
|
|
|
|
def test_no_doi(self):
|
|
assert unpaywall_urls(MagicMock(), "", "e@x") == []
|
|
|
|
def test_404(self):
|
|
assert unpaywall_urls(_json_client({"unpaywall": 404}), "10.1/x", "e@x") == []
|
|
|
|
def test_no_locations(self):
|
|
body = {"best_oa_location": None, "oa_locations": []}
|
|
assert unpaywall_urls(_json_client({"unpaywall": body}), "10.1/x", "e@x") == []
|
|
|
|
|
|
# ── fetch_semantic_scholar ────────────────────────────────────────
|
|
|
|
|
|
class TestFetchSemanticScholar:
|
|
def test_pdf_and_pmcid(self):
|
|
seen: list[str] = []
|
|
body = {
|
|
"openAccessPdf": {"url": "https://s2/p.pdf", "status": "GREEN"},
|
|
"externalIds": {"PubMedCentral": "5226373", "DOI": "10.1/x"},
|
|
}
|
|
hit = fetch_semantic_scholar(
|
|
_json_client({"semanticscholar": body}, seen), "10.1/x"
|
|
)
|
|
assert hit is not None
|
|
assert hit.pdf_url == "https://s2/p.pdf"
|
|
assert hit.pmcid == "PMC5226373"
|
|
assert "paper/DOI:10.1/x" in seen[0] and "openAccessPdf" in seen[0]
|
|
|
|
def test_pmcid_only(self):
|
|
body = {"openAccessPdf": None, "externalIds": {"PubMedCentral": "PMC42"}}
|
|
hit = fetch_semantic_scholar(_json_client({"semanticscholar": body}), "10.1/x")
|
|
assert hit is not None
|
|
assert hit.pdf_url == ""
|
|
assert hit.pmcid == "PMC42"
|
|
|
|
def test_nothing_useful(self):
|
|
body = {"openAccessPdf": None, "externalIds": {"DOI": "10.1/x"}}
|
|
assert (
|
|
fetch_semantic_scholar(_json_client({"semanticscholar": body}), "10.1/x")
|
|
is None
|
|
)
|
|
|
|
def test_404(self):
|
|
assert (
|
|
fetch_semantic_scholar(_json_client({"semanticscholar": 404}), "10.1/x")
|
|
is None
|
|
)
|
|
|
|
def test_no_doi(self):
|
|
assert fetch_semantic_scholar(MagicMock(), "") is None
|
|
|
|
def test_throttled(self, monkeypatch):
|
|
"""Calls are spaced by _S2_MIN_INTERVAL even from a cold start."""
|
|
monkeypatch.setattr(fetch_mod, "_S2_MIN_INTERVAL", 0.2)
|
|
monkeypatch.setattr(fetch_mod, "_S2_LAST", [0.0])
|
|
client = _json_client({"semanticscholar": 404})
|
|
import time
|
|
|
|
t0 = time.monotonic()
|
|
fetch_semantic_scholar(client, "10.1/a")
|
|
fetch_semantic_scholar(client, "10.1/b")
|
|
assert time.monotonic() - t0 >= 0.2
|
|
|
|
|
|
# ── resolve_pmcids ────────────────────────────────────────────────
|
|
|
|
|
|
class TestResolvePmcids:
|
|
_RECORDS = {
|
|
"status": "ok",
|
|
"records": [
|
|
{
|
|
"doi": "10.1/x",
|
|
"pmcid": "PMC5226373",
|
|
"pmid": 27893131,
|
|
"requested-id": "27893131",
|
|
},
|
|
{
|
|
"pmid": 28679817,
|
|
"requested-id": "28679817",
|
|
"status": "error",
|
|
"errmsg": "Identifier not found in PMC",
|
|
},
|
|
],
|
|
}
|
|
|
|
def test_maps_found_only(self):
|
|
seen: list[str] = []
|
|
out = resolve_pmcids(
|
|
_json_client({"idconv": self._RECORDS}, seen),
|
|
["27893131", "28679817"],
|
|
"e@x",
|
|
)
|
|
assert out == {"27893131": "PMC5226373"}
|
|
assert "idtype=pmid" in seen[0] and "ids=27893131%2C28679817" in seen[0]
|
|
assert "email=e%40x" in seen[0]
|
|
|
|
def test_batches_of_200(self):
|
|
seen: list[str] = []
|
|
client = _json_client({"idconv": {"status": "ok", "records": []}}, seen)
|
|
resolve_pmcids(client, [str(i) for i in range(1, 402)], "e@x")
|
|
assert len(seen) == 3
|
|
|
|
def test_skips_non_numeric_and_dupes(self):
|
|
seen: list[str] = []
|
|
client = _json_client({"idconv": {"status": "ok", "records": []}}, seen)
|
|
resolve_pmcids(client, ["", "abc", "12", "12"], "e@x")
|
|
assert len(seen) == 1 and "ids=12&" in seen[0]
|
|
|
|
def test_empty(self):
|
|
assert resolve_pmcids(MagicMock(), [], "e@x") == {}
|
|
|
|
def test_http_error(self):
|
|
assert resolve_pmcids(_json_client({"idconv": 500}), ["1"], "e@x") == {}
|
|
|
|
|
|
# ── pending_queue PMCID resolution ────────────────────────────────
|
|
|
|
|
|
def _db(tmp_path):
|
|
path = str(tmp_path / "z.sqlite")
|
|
create_db(path).close()
|
|
return path
|
|
|
|
|
|
def _item(db, project, extra, doi="10.1/x"):
|
|
iid = db.create_item(TYPE_MAP["journalArticle"])
|
|
db.set_fields(iid, {"title": "Test Item", "DOI": doi, "extra": extra})
|
|
db.sync_tags(iid, [f"project:{project}", "screen:include"])
|
|
db.commit()
|
|
return iid
|
|
|
|
|
|
class TestPendingQueueIdconv:
|
|
def test_resolves_and_persists(self, tmp_path):
|
|
path = _db(tmp_path)
|
|
recs = {
|
|
"status": "ok",
|
|
"records": [
|
|
{"pmcid": "PMC777", "pmid": 111, "requested-id": "111"},
|
|
],
|
|
}
|
|
seen: list[str] = []
|
|
client = _json_client({"idconv": recs}, seen)
|
|
with Db(path) as db:
|
|
a = _item(db, "p", "PMID: 111")
|
|
b = _item(db, "p", "PMID: 222\nPMCID: PMC222", doi="10.1/y")
|
|
c = _item(db, "p", "", doi="10.1/z")
|
|
items = pending_queue(db, "p", client=client, email="e@x")
|
|
by_id = {it.zot_id: it for it in items}
|
|
assert by_id[a].pmcid == "PMC777"
|
|
assert by_id[b].pmcid == "PMC222"
|
|
assert by_id[c].pmcid == ""
|
|
# only the one PMID lacking a PMCID was sent
|
|
assert len(seen) == 1 and "ids=111&" in seen[0]
|
|
# persisted into extra, existing lines kept
|
|
assert fetch_mod._field(db, a, "extra") == "PMID: 111\nPMCID: PMC777"
|
|
# second call: nothing left to resolve
|
|
pending_queue(db, "p", client=client, email="e@x")
|
|
assert len(seen) == 1
|
|
|
|
def test_no_client_no_network(self, tmp_path):
|
|
path = _db(tmp_path)
|
|
with Db(path) as db:
|
|
a = _item(db, "p", "PMID: 111")
|
|
items = pending_queue(db, "p")
|
|
assert items[0].zot_id == a and items[0].pmcid == ""
|
|
|
|
|
|
# ── fetch_one cascade order ───────────────────────────────────────
|
|
|
|
|
|
def _pdf_client(routes_json: dict[str, object], pdf_urls: set[str], seen: list[str]):
|
|
def handler(request):
|
|
url = str(request.url)
|
|
seen.append(url)
|
|
if url in pdf_urls:
|
|
return httpx.Response(
|
|
200,
|
|
content=b"%PDF-1.7\n" + b"x" * 2000,
|
|
headers={"content-type": "application/pdf"},
|
|
)
|
|
for needle, body in routes_json.items():
|
|
if needle in url:
|
|
return (
|
|
httpx.Response(200, json=body)
|
|
if not isinstance(body, int)
|
|
else httpx.Response(body)
|
|
)
|
|
return httpx.Response(404)
|
|
|
|
return httpx.Client(transport=httpx.MockTransport(handler))
|
|
|
|
|
|
class TestFetchOneTiers:
|
|
def test_second_unpaywall_location_wins(self, tmp_path):
|
|
seen: list[str] = []
|
|
up = {
|
|
"best_oa_location": {"url_for_pdf": "https://dead/p.pdf"},
|
|
"oa_locations": [
|
|
{"url_for_pdf": "https://dead/p.pdf"},
|
|
{"url_for_pdf": "https://alive/p.pdf"},
|
|
],
|
|
}
|
|
client = _pdf_client({"unpaywall": up}, {"https://alive/p.pdf"}, seen)
|
|
item = PendingItem(zot_id=1, doi="10.1/x", pmid="", pmcid="", title="T")
|
|
res = fetch_one(item, email="e@x", scratch=tmp_path, client=client)
|
|
assert res is not None and res[0] == "unpaywall"
|
|
assert "https://dead/p.pdf" in seen and "https://alive/p.pdf" in seen
|
|
assert not any("semanticscholar" in u for u in seen)
|
|
|
|
def test_s2_pdf_after_unpaywall_miss(self, tmp_path):
|
|
seen: list[str] = []
|
|
s2 = {"openAccessPdf": {"url": "https://s2/p.pdf"}, "externalIds": {}}
|
|
client = _pdf_client(
|
|
{"unpaywall": 404, "semanticscholar": s2}, {"https://s2/p.pdf"}, seen
|
|
)
|
|
item = PendingItem(zot_id=1, doi="10.1/x", pmid="", pmcid="", title="T")
|
|
res = fetch_one(item, email="e@x", scratch=tmp_path, client=client)
|
|
assert res is not None and res[0] == "s2"
|
|
|
|
def test_s2_pmcid_feeds_pmc_tiers(self, tmp_path):
|
|
seen: list[str] = []
|
|
s2 = {"openAccessPdf": None, "externalIds": {"PubMedCentral": "99"}}
|
|
client = _pdf_client({"unpaywall": 404, "semanticscholar": s2}, set(), seen)
|
|
item = PendingItem(zot_id=1, doi="10.1/x", pmid="", pmcid="", title="T")
|
|
assert fetch_one(item, email="e@x", scratch=tmp_path, client=client) is None
|
|
assert any("prefix=PMC99." in u for u in seen), seen # S3 listing tried
|
|
assert any("id=PMC99" in u for u in seen), seen # OA service tried
|
|
|
|
def test_item_pmcid_not_overridden(self, tmp_path):
|
|
seen: list[str] = []
|
|
s2 = {"openAccessPdf": None, "externalIds": {"PubMedCentral": "99"}}
|
|
client = _pdf_client({"unpaywall": 404, "semanticscholar": s2}, set(), seen)
|
|
item = PendingItem(zot_id=1, doi="10.1/x", pmid="", pmcid="PMC1", title="T")
|
|
fetch_one(item, email="e@x", scratch=tmp_path, client=client)
|
|
assert not any("PMC99" in u for u in seen)
|
|
assert any("prefix=PMC1." in u for u in seen)
|