407 lines
14 KiB
Python
407 lines
14 KiB
Python
"""Tests for prisma.fetch — PDF retrieval with altcha solver."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from unittest.mock import MagicMock
|
|
|
|
import httpx
|
|
|
|
from prisma import fetch as fetch_mod
|
|
from prisma.fetch import (
|
|
PendingItem,
|
|
_altcha_bootstrap,
|
|
_extract_pdf_url,
|
|
_solve_altcha_pow,
|
|
download,
|
|
fetch_europepmc_download,
|
|
fetch_one,
|
|
fetch_pmc,
|
|
fetch_pmc_s3,
|
|
fetch_unpaywall,
|
|
resolve_doi_from_title,
|
|
)
|
|
|
|
# ── Altcha solver (pure, deterministic) ───────────────────────────
|
|
|
|
|
|
class TestAltchaPow:
|
|
def test_known_vector(self):
|
|
salt = "testsalt"
|
|
target_n = 42
|
|
challenge = hashlib.sha256(f"{salt}{target_n}".encode()).hexdigest()
|
|
result = _solve_altcha_pow(salt, challenge, 1000)
|
|
assert result is not None
|
|
nonce, elapsed = result
|
|
assert nonce == 42
|
|
assert elapsed >= 0
|
|
|
|
def test_not_found(self):
|
|
result = _solve_altcha_pow("salt", "impossible" * 4, 10)
|
|
assert result is None
|
|
|
|
def test_zero_nonce(self):
|
|
salt = "zero"
|
|
challenge = hashlib.sha256(f"{salt}0".encode()).hexdigest()
|
|
result = _solve_altcha_pow(salt, challenge, 100)
|
|
assert result is not None
|
|
assert result[0] == 0
|
|
|
|
|
|
# ── _extract_pdf_url ──────────────────────────────────────────────
|
|
|
|
|
|
class TestExtractPdfUrl:
|
|
def test_citation_meta(self):
|
|
html = '<meta name="citation_pdf_url" content="/storage/test.pdf">'
|
|
assert (
|
|
_extract_pdf_url(html, "https://example.com")
|
|
== "https://example.com/storage/test.pdf"
|
|
)
|
|
|
|
def test_embed_src(self):
|
|
html = '<embed src="//cdn.example.com/paper.pdf" type="application/pdf">'
|
|
assert (
|
|
_extract_pdf_url(html, "https://x.com")
|
|
== "https://cdn.example.com/paper.pdf"
|
|
)
|
|
|
|
def test_no_match(self):
|
|
assert _extract_pdf_url("<html>no pdf here</html>", "https://x.com") is None
|
|
|
|
def test_absolute_url(self):
|
|
html = '<meta name="citation_pdf_url" content="https://full.url/doc.pdf">'
|
|
assert _extract_pdf_url(html, "https://x.com") == "https://full.url/doc.pdf"
|
|
|
|
|
|
# ── _altcha_bootstrap ─────────────────────────────────────────────
|
|
|
|
|
|
class TestAltchaBootstrap:
|
|
def test_solves_challenge(self):
|
|
salt = "bootsalt"
|
|
nonce = 77
|
|
challenge = hashlib.sha256(f"{salt}{nonce}".encode()).hexdigest()
|
|
|
|
client = MagicMock()
|
|
challenge_resp = MagicMock()
|
|
challenge_resp.json.return_value = {
|
|
"algorithm": "SHA-256",
|
|
"salt": salt,
|
|
"challenge": challenge,
|
|
"maxNumber": 1000,
|
|
"signature": "sig",
|
|
}
|
|
solve_resp = MagicMock()
|
|
solve_resp.status_code = 200
|
|
solve_resp.json.return_value = {"success": True}
|
|
|
|
client.get.return_value = challenge_resp
|
|
client.post.return_value = solve_resp
|
|
|
|
html = (
|
|
'<altcha-widget challengeurl="/captcha/challenge/123">'
|
|
"fetch('/captcha/solution/123'"
|
|
)
|
|
result = _altcha_bootstrap(client, "https://mirror.test/doi", html)
|
|
assert result is True
|
|
|
|
def test_no_challenge(self):
|
|
assert (
|
|
_altcha_bootstrap(MagicMock(), "https://x.com/doi", "<html></html>")
|
|
is False
|
|
)
|
|
|
|
|
|
# ── fetch_unpaywall ───────────────────────────────────────────────
|
|
|
|
|
|
class TestFetchUnpaywall:
|
|
def test_returns_pdf_url(self):
|
|
client = MagicMock()
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
resp.json.return_value = {
|
|
"best_oa_location": {"url_for_pdf": "https://oa.org/paper.pdf"}
|
|
}
|
|
client.get.return_value = resp
|
|
assert (
|
|
fetch_unpaywall(client, "10.1234/test", "e@x.com")
|
|
== "https://oa.org/paper.pdf"
|
|
)
|
|
|
|
def test_no_doi(self):
|
|
assert fetch_unpaywall(MagicMock(), "", "e@x.com") is None
|
|
|
|
def test_404(self):
|
|
client = MagicMock()
|
|
resp = MagicMock()
|
|
resp.status_code = 404
|
|
client.get.return_value = resp
|
|
assert fetch_unpaywall(client, "10.1234/x", "e@x.com") is None
|
|
|
|
|
|
# ── fetch_pmc ─────────────────────────────────────────────────────
|
|
|
|
|
|
class TestFetchPmc:
|
|
_OA_HIT = (
|
|
'<OA><records><record id="PMC12345">'
|
|
'<link format="pdf" href="ftp://ftp.ncbi.nlm.nih.gov/pub/pmc/oa_pdf/aa/bb/main.PMC12345.pdf" />'
|
|
"</record></records></OA>"
|
|
)
|
|
_OA_MISS = '<OA><error code="idIsNotOpenAccess">identifier not OA</error></OA>'
|
|
|
|
def _client(self, body, seen):
|
|
def handler(request):
|
|
seen.append(str(request.url))
|
|
return httpx.Response(200, text=body)
|
|
|
|
return httpx.Client(transport=httpx.MockTransport(handler))
|
|
|
|
def test_with_prefix(self):
|
|
seen: list[str] = []
|
|
url = fetch_pmc(self._client(self._OA_HIT, seen), "PMC12345")
|
|
assert url == (
|
|
"https://ftp.ncbi.nlm.nih.gov/pub/pmc/deprecated/oa_pdf/aa/bb/main.PMC12345.pdf"
|
|
)
|
|
assert "oa/oa.fcgi" in seen[0] and "id=PMC12345" in seen[0]
|
|
|
|
def test_without_prefix(self):
|
|
seen: list[str] = []
|
|
url = fetch_pmc(self._client(self._OA_HIT, seen), "12345")
|
|
assert url == (
|
|
"https://ftp.ncbi.nlm.nih.gov/pub/pmc/deprecated/oa_pdf/aa/bb/main.PMC12345.pdf"
|
|
)
|
|
assert "id=PMC12345" in seen[0]
|
|
|
|
def test_not_open_access(self):
|
|
seen: list[str] = []
|
|
assert fetch_pmc(self._client(self._OA_MISS, seen), "PMC12345") is None
|
|
|
|
def test_empty(self):
|
|
assert fetch_pmc(MagicMock(), "") is None
|
|
|
|
|
|
# ── fetch_pmc_s3 / deprecated path / Europe PMC breaker ──────────
|
|
|
|
|
|
class TestFetchPmcS3:
|
|
_LIST = (
|
|
'<?xml version="1.0"?><ListBucketResult>'
|
|
"<Contents><Key>PMC7975862.1/img1.jpg</Key></Contents>"
|
|
"<Contents><Key>PMC7975862.1/PMC7975862.1.pdf</Key></Contents>"
|
|
"<Contents><Key>PMC7975862.2/PMC7975862.2.pdf</Key></Contents>"
|
|
"</ListBucketResult>"
|
|
)
|
|
|
|
def _client(self, body, status=200):
|
|
return httpx.Client(
|
|
transport=httpx.MockTransport(lambda r: httpx.Response(status, text=body))
|
|
)
|
|
|
|
def test_picks_highest_version(self):
|
|
url = fetch_pmc_s3(self._client(self._LIST), "PMC7975862")
|
|
assert url == (
|
|
"https://pmc-oa-opendata.s3.amazonaws.com/PMC7975862.2/PMC7975862.2.pdf"
|
|
)
|
|
|
|
def test_without_prefix(self):
|
|
url = fetch_pmc_s3(self._client(self._LIST), "7975862")
|
|
assert url and "PMC7975862.2.pdf" in url
|
|
|
|
def test_not_in_bucket(self):
|
|
empty = '<?xml version="1.0"?><ListBucketResult></ListBucketResult>'
|
|
assert fetch_pmc_s3(self._client(empty), "PMC4781080") is None
|
|
|
|
def test_empty(self):
|
|
assert fetch_pmc_s3(MagicMock(), "") is None
|
|
|
|
|
|
class TestFetchPmcDeprecatedPath:
|
|
def test_ftp_rewrite_targets_deprecated_tree(self):
|
|
body = (
|
|
'<OA><records><record id="PMC1"><link format="pdf" '
|
|
'href="ftp://ftp.ncbi.nlm.nih.gov/pub/pmc/oa_pdf/aa/bb/x.pdf" />'
|
|
"</record></records></OA>"
|
|
)
|
|
client = httpx.Client(
|
|
transport=httpx.MockTransport(lambda r: httpx.Response(200, text=body))
|
|
)
|
|
url = fetch_pmc(client, "PMC1")
|
|
assert url == (
|
|
"https://ftp.ncbi.nlm.nih.gov/pub/pmc/deprecated/oa_pdf/aa/bb/x.pdf"
|
|
)
|
|
|
|
|
|
class TestEuropePmcBreaker:
|
|
def test_breaker_opens_after_three_connect_errors(self, tmp_path):
|
|
calls = {"n": 0}
|
|
|
|
def boom(request):
|
|
calls["n"] += 1
|
|
raise httpx.ConnectError("down")
|
|
|
|
client = httpx.Client(transport=httpx.MockTransport(boom))
|
|
fetch_mod._EPMC_BREAKER["fails"] = 0
|
|
try:
|
|
for _ in range(5):
|
|
fetch_europepmc_download(client, "PMC1", tmp_path / "x.pdf")
|
|
assert calls["n"] == 3 # 4th and 5th short-circuited
|
|
finally:
|
|
fetch_mod._EPMC_BREAKER["fails"] = 0
|
|
|
|
def test_success_resets_breaker(self, tmp_path):
|
|
pdf = b"%PDF-1.4 " + b"x" * 2000
|
|
|
|
def ok(request):
|
|
return httpx.Response(
|
|
200, content=pdf, headers={"content-type": "application/pdf"}
|
|
)
|
|
|
|
client = httpx.Client(transport=httpx.MockTransport(ok))
|
|
fetch_mod._EPMC_BREAKER["fails"] = 2
|
|
p = fetch_europepmc_download(client, "PMC1", tmp_path / "y.pdf")
|
|
assert p is not None
|
|
assert fetch_mod._EPMC_BREAKER["fails"] == 0
|
|
|
|
def test_non_pdf_response_rejected(self, tmp_path):
|
|
def html(request):
|
|
return httpx.Response(
|
|
200,
|
|
text="<html>challenge</html>" + "x" * 2000,
|
|
headers={"content-type": "text/html"},
|
|
)
|
|
|
|
client = httpx.Client(transport=httpx.MockTransport(html))
|
|
fetch_mod._EPMC_BREAKER["fails"] = 0
|
|
assert fetch_europepmc_download(client, "PMC1", tmp_path / "z.pdf") is None
|
|
|
|
|
|
# ── resolve_doi_from_title ────────────────────────────────────────
|
|
|
|
|
|
class TestResolveDoi:
|
|
def test_match(self):
|
|
client = MagicMock()
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
resp.json.return_value = {
|
|
"message": {
|
|
"items": [
|
|
{
|
|
"DOI": "10.1234/found",
|
|
"title": ["Skin substitutes in wound care"],
|
|
"score": 80,
|
|
}
|
|
]
|
|
}
|
|
}
|
|
client.get.return_value = resp
|
|
assert (
|
|
resolve_doi_from_title(client, "Skin substitutes in wound care")
|
|
== "10.1234/found"
|
|
)
|
|
|
|
def test_low_score(self):
|
|
client = MagicMock()
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
resp.json.return_value = {
|
|
"message": {
|
|
"items": [{"DOI": "10.1234/x", "title": ["Unrelated"], "score": 10}]
|
|
}
|
|
}
|
|
client.get.return_value = resp
|
|
assert resolve_doi_from_title(client, "Skin substitutes in wound care") is None
|
|
|
|
def test_short_title(self):
|
|
assert resolve_doi_from_title(MagicMock(), "Short") is None
|
|
|
|
|
|
# ── download ──────────────────────────────────────────────────────
|
|
|
|
|
|
class TestDownload:
|
|
def test_valid_pdf(self, tmp_path):
|
|
client = MagicMock()
|
|
stream_ctx = MagicMock()
|
|
stream_ctx.__enter__ = MagicMock(return_value=stream_ctx)
|
|
stream_ctx.__exit__ = MagicMock(return_value=False)
|
|
stream_ctx.status_code = 200
|
|
stream_ctx.headers = {"content-type": "application/pdf"}
|
|
stream_ctx.iter_bytes.return_value = [b"%PDF-1.7\n" + b"x" * 2000]
|
|
client.stream.return_value = stream_ctx
|
|
|
|
dest = tmp_path / "test.pdf"
|
|
result = download(client, "https://x.com/paper.pdf", dest)
|
|
assert result == dest
|
|
assert dest.read_bytes().startswith(b"%PDF")
|
|
|
|
def test_not_pdf(self, tmp_path):
|
|
client = MagicMock()
|
|
stream_ctx = MagicMock()
|
|
stream_ctx.__enter__ = MagicMock(return_value=stream_ctx)
|
|
stream_ctx.__exit__ = MagicMock(return_value=False)
|
|
stream_ctx.status_code = 200
|
|
stream_ctx.headers = {"content-type": "text/html"}
|
|
stream_ctx.iter_bytes.return_value = [b"<html>Not a PDF</html>" + b"x" * 2000]
|
|
client.stream.return_value = stream_ctx
|
|
|
|
dest = tmp_path / "bad.pdf"
|
|
result = download(client, "https://x.com/bad", dest)
|
|
assert result is None
|
|
|
|
def test_too_small(self, tmp_path):
|
|
client = MagicMock()
|
|
stream_ctx = MagicMock()
|
|
stream_ctx.__enter__ = MagicMock(return_value=stream_ctx)
|
|
stream_ctx.__exit__ = MagicMock(return_value=False)
|
|
stream_ctx.status_code = 200
|
|
stream_ctx.headers = {}
|
|
stream_ctx.iter_bytes.return_value = [b"%PDF"]
|
|
client.stream.return_value = stream_ctx
|
|
|
|
dest = tmp_path / "tiny.pdf"
|
|
result = download(client, "https://x.com/tiny", dest)
|
|
assert result is None
|
|
|
|
|
|
# ── fetch_one cascade ─────────────────────────────────────────────
|
|
|
|
|
|
class TestFetchOne:
|
|
def test_unpaywall_hit(self, tmp_path):
|
|
item = PendingItem(
|
|
zot_id=1, doi="10.1234/test", pmid="", pmcid="", title="Test"
|
|
)
|
|
client = MagicMock()
|
|
resp = MagicMock()
|
|
resp.status_code = 200
|
|
resp.json.return_value = {
|
|
"best_oa_location": {"url_for_pdf": "https://oa.org/p.pdf"}
|
|
}
|
|
client.get.return_value = resp
|
|
|
|
stream_ctx = MagicMock()
|
|
stream_ctx.__enter__ = MagicMock(return_value=stream_ctx)
|
|
stream_ctx.__exit__ = MagicMock(return_value=False)
|
|
stream_ctx.status_code = 200
|
|
stream_ctx.headers = {"content-type": "application/pdf"}
|
|
stream_ctx.iter_bytes.return_value = [b"%PDF-1.7\n" + b"x" * 2000]
|
|
client.stream.return_value = stream_ctx
|
|
|
|
result = fetch_one(item, email="e@x.com", scratch=tmp_path, client=client)
|
|
assert result is not None
|
|
source, path = result
|
|
assert source == "unpaywall"
|
|
|
|
def test_all_miss(self, tmp_path):
|
|
item = PendingItem(zot_id=1, doi="", pmid="", pmcid="", title="X")
|
|
client = MagicMock()
|
|
resp = MagicMock()
|
|
resp.status_code = 404
|
|
client.get.return_value = resp
|
|
result = fetch_one(item, email="e@x.com", scratch=tmp_path, client=client)
|
|
assert result is None
|