82cea2a33b
Add crossref.fetch_references(doi), which parses message.reference[].DOI into Citations (DOI-less book/unstructured refs are skipped — they cannot join the graph). Wire it as the third leg of the ingest citation fallback chain: OpenAlex-empty -> S2 -> Crossref, for DOI papers only (Crossref's works endpoint is DOI-keyed). Cited DOIs are normalized to the canonical bare lower-case form via _norm_cited_id. As a fallback it fires only when OpenAlex and S2 both return no references, so it adds a forward-looking third source without changing the already-covered corpus. Tests: fetch_references unit tests (DOI edges / unstructured skipped / no refs / 404) and ingest tests for both fallback directions. Full suite 361 passed; ruff + mypy clean. Live-validated: 33/40/24 refs for Springer/ACM/Wiley DOIs. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
105 lines
3.9 KiB
Python
105 lines
3.9 KiB
Python
"""Tests for codex.sources.crossref."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from codex.models import Citation
|
|
from codex.sources import crossref
|
|
|
|
|
|
def test_fetch_abstract_strips_jats(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""A JATS-tagged abstract is returned as plain text with tags removed."""
|
|
body = {
|
|
"message": {
|
|
"abstract": "<jats:p>We prove a <jats:italic>theorem</jats:italic> about "
|
|
"polyhedra.</jats:p>"
|
|
}
|
|
}
|
|
|
|
def mock_get(url: str, params: dict[str, object] | None = None) -> httpx.Response:
|
|
return httpx.Response(200, json=body)
|
|
|
|
monkeypatch.setattr(crossref, "_get", mock_get)
|
|
|
|
result = crossref.fetch_abstract("10.1007/s00454-019-00132-8")
|
|
assert result == "We prove a theorem about polyhedra."
|
|
|
|
|
|
def test_fetch_abstract_absent_returns_none(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""A work with no abstract field (e.g. a book chapter) returns None."""
|
|
|
|
def mock_get(url: str, params: dict[str, object] | None = None) -> httpx.Response:
|
|
return httpx.Response(200, json={"message": {"title": ["Some Chapter"]}})
|
|
|
|
monkeypatch.setattr(crossref, "_get", mock_get)
|
|
assert crossref.fetch_abstract("10.1007/978-3-642-17413-1_7") is None
|
|
|
|
|
|
def test_fetch_abstract_404_returns_none(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""A 404 returns None rather than raising."""
|
|
|
|
def mock_get(url: str, params: dict[str, object] | None = None) -> httpx.Response:
|
|
raise httpx.HTTPStatusError(
|
|
"404",
|
|
request=httpx.Request("GET", url),
|
|
response=httpx.Response(404, request=httpx.Request("GET", url)),
|
|
)
|
|
|
|
monkeypatch.setattr(crossref, "_get", mock_get)
|
|
assert crossref.fetch_abstract("10.0000/nonexistent") is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# fetch_references (roadmap R-B)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_fetch_references_returns_doi_edges(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Only references carrying a DOI become Citations; citing_id is the queried DOI."""
|
|
body = {
|
|
"message": {
|
|
"reference": [
|
|
{"key": "ref1", "DOI": "10.1090/s0002-9947-04-03545-7"},
|
|
{"key": "ref2", "unstructured": "A book with no DOI, 1992."}, # skipped
|
|
{"key": "ref3", "DOI": "10.1007/bf02392449", "article-title": "X"},
|
|
]
|
|
}
|
|
}
|
|
|
|
def mock_get(url: str, params: dict[str, object] | None = None) -> httpx.Response:
|
|
return httpx.Response(200, json=body)
|
|
|
|
monkeypatch.setattr(crossref, "_get", mock_get)
|
|
refs = crossref.fetch_references("10.1007/s00454-019-00132-8")
|
|
|
|
assert refs == [
|
|
Citation(citing_id="10.1007/s00454-019-00132-8", cited_id="10.1090/s0002-9947-04-03545-7"),
|
|
Citation(citing_id="10.1007/s00454-019-00132-8", cited_id="10.1007/bf02392449"),
|
|
]
|
|
|
|
|
|
def test_fetch_references_no_reference_key_returns_empty(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""A work with no deposited reference list returns []."""
|
|
|
|
def mock_get(url: str, params: dict[str, object] | None = None) -> httpx.Response:
|
|
return httpx.Response(200, json={"message": {"title": ["No refs deposited"]}})
|
|
|
|
monkeypatch.setattr(crossref, "_get", mock_get)
|
|
assert crossref.fetch_references("10.1007/978-3-642-17413-1_7") == []
|
|
|
|
|
|
def test_fetch_references_404_returns_empty(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""A 404 returns [] rather than raising."""
|
|
|
|
def mock_get(url: str, params: dict[str, object] | None = None) -> httpx.Response:
|
|
raise httpx.HTTPStatusError(
|
|
"404",
|
|
request=httpx.Request("GET", url),
|
|
response=httpx.Response(404, request=httpx.Request("GET", url)),
|
|
)
|
|
|
|
monkeypatch.setattr(crossref, "_get", mock_get)
|
|
assert crossref.fetch_references("10.0000/nonexistent") == []
|