fix(prisma): repair tests stale since 81cbe53 — OA-service fetch_pmc + extraction-template guard (closes #658, refs #657)

This commit is contained in:
kert
2026-08-19 11:20:13 -04:00
parent 6a399bece9
commit 19e24cd0bd
2 changed files with 47 additions and 10 deletions

View File

@@ -5,6 +5,8 @@ from __future__ import annotations
import hashlib import hashlib
from unittest.mock import MagicMock from unittest.mock import MagicMock
import httpx
from prisma.fetch import ( from prisma.fetch import (
PendingItem, PendingItem,
_altcha_bootstrap, _altcha_bootstrap,
@@ -140,17 +142,39 @@ class TestFetchUnpaywall:
class TestFetchPmc: 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): def test_with_prefix(self):
assert ( seen: list[str] = []
fetch_pmc(MagicMock(), "PMC12345") url = fetch_pmc(self._client(self._OA_HIT, seen), "PMC12345")
== "https://www.ncbi.nlm.nih.gov/pmc/articles/PMC12345/pdf/" assert url == (
"https://ftp.ncbi.nlm.nih.gov/pub/pmc/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): def test_without_prefix(self):
assert ( seen: list[str] = []
fetch_pmc(MagicMock(), "12345") url = fetch_pmc(self._client(self._OA_HIT, seen), "12345")
== "https://www.ncbi.nlm.nih.gov/pmc/articles/PMC12345/pdf/" assert url == (
"https://ftp.ncbi.nlm.nih.gov/pub/pmc/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): def test_empty(self):
assert fetch_pmc(MagicMock(), "") is None assert fetch_pmc(MagicMock(), "") is None

View File

@@ -117,9 +117,15 @@ class TestEligRun:
class TestExtBuildCall: class TestExtBuildCall:
def test_returns_call(self): def test_returns_call(self):
from prisma.tools import build_extract_tool
project = MagicMock() project = MagicMock()
project.extraction_template = "template" project.name = "test"
call = ext_build(project, "# Item") project.extraction_template = (
"- **population** — study population\n- **outcome** — primary outcome"
)
tool = build_extract_tool(project.extraction_template, project=project.name)
call = ext_build(project, "# Item", tool)
assert len(call.messages) == 3 assert len(call.messages) == 3
@@ -136,7 +142,9 @@ class TestExtRun:
provider = MagicMock() provider = MagicMock()
project = MagicMock() project = MagicMock()
project.name = "test" project.name = "test"
project.extraction_template = "t" project.extraction_template = (
"- **population** — study population\n- **outcome** — primary outcome"
)
with Db(path) as db: with Db(path) as db:
stats = ext_run(db, provider, project, storage_dir=Path(tmp_path)) stats = ext_run(db, provider, project, storage_dir=Path(tmp_path))
assert stats["extracted"] == 0 assert stats["extracted"] == 0
@@ -158,7 +166,9 @@ class TestExtRun:
project = MagicMock() project = MagicMock()
project.name = "test" project.name = "test"
project.extraction_template = "t" project.extraction_template = (
"- **population** — study population\n- **outcome** — primary outcome"
)
stats = ext_run(db, provider, project, storage_dir=Path(tmp_path)) stats = ext_run(db, provider, project, storage_dir=Path(tmp_path))
assert stats["extracted"] == 1 assert stats["extracted"] == 1
@@ -171,6 +181,9 @@ class TestExtRun:
provider = MagicMock() provider = MagicMock()
project = MagicMock() project = MagicMock()
project.name = "test" project.name = "test"
project.extraction_template = (
"- **population** — study population\n- **outcome** — primary outcome"
)
stats = ext_run(db, provider, project, storage_dir=Path(tmp_path)) stats = ext_run(db, provider, project, storage_dir=Path(tmp_path))
assert stats["errors"] == 1 assert stats["errors"] == 1