206 lines
8.0 KiB
Python
206 lines
8.0 KiB
Python
"""OpenAlex collector: abstract reconstruction, work mapping, filters, pagination, failure modes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
|
|
import pytest
|
|
|
|
from sebrief.collectors import openalex
|
|
from sebrief.collectors.openalex import (
|
|
fetch_openalex_papers,
|
|
normalize_doi,
|
|
paper_from_work,
|
|
reconstruct_abstract,
|
|
)
|
|
from sebrief.config import Settings
|
|
|
|
SETTINGS = Settings(_env_file=None)
|
|
START, END = date(2026, 8, 24), date(2026, 9, 6)
|
|
|
|
|
|
# ------------------------------------------------------------------------------- pure parts
|
|
@pytest.mark.parametrize(
|
|
("raw", "expected"),
|
|
[
|
|
("https://doi.org/10.1145/1234567", "10.1145/1234567"),
|
|
("http://dx.doi.org/10.1109/TSE.2026.1", "10.1109/tse.2026.1"),
|
|
("doi:10.1000/ABC", "10.1000/abc"),
|
|
("10.1000/abc", "10.1000/abc"),
|
|
("https://example.org/paper", None),
|
|
("", None),
|
|
(None, None),
|
|
],
|
|
)
|
|
def test_normalize_doi(raw, expected):
|
|
assert normalize_doi(raw) == expected
|
|
|
|
|
|
def test_reconstruct_abstract():
|
|
inv = {"study": [2], "We": [0], "digests.": [3], "weekly": [1]}
|
|
assert reconstruct_abstract(inv) == "We weekly study digests."
|
|
assert reconstruct_abstract(None) == ""
|
|
assert reconstruct_abstract({}) == ""
|
|
|
|
|
|
def make_work(**overrides) -> dict:
|
|
work = {
|
|
"doi": "https://doi.org/10.1145/1234567",
|
|
"title": "Test-Driven Digest Generation",
|
|
"abstract_inverted_index": {"We": [0], "test.": [1]},
|
|
"publication_date": "2026-09-02",
|
|
"authorships": [
|
|
{"author": {"display_name": "Alice Kim"}},
|
|
{"author": {"display_name": "Bob Lee"}},
|
|
],
|
|
"primary_location": {
|
|
"source": {
|
|
"display_name": "ACM Transactions on Software Engineering and Methodology",
|
|
"issn_l": "1049-331X",
|
|
"issn": ["1049-331X", "1557-7392"],
|
|
"host_organization_name": "Association for Computing Machinery",
|
|
}
|
|
},
|
|
"best_oa_location": {
|
|
"landing_page_url": "https://dl.acm.org/doi/10.1145/1234567",
|
|
"pdf_url": "https://dl.acm.org/doi/pdf/10.1145/1234567",
|
|
},
|
|
"open_access": {"oa_url": "https://dl.acm.org/doi/10.1145/1234567"},
|
|
"topics": [{"display_name": "Software Engineering Research"}],
|
|
}
|
|
work.update(overrides)
|
|
return work
|
|
|
|
|
|
def test_paper_from_work_maps_all_fields():
|
|
paper = paper_from_work(make_work())
|
|
assert paper is not None
|
|
assert paper.doi == "10.1145/1234567"
|
|
assert paper.title == "Test-Driven Digest Generation"
|
|
assert paper.abstract == "We test."
|
|
assert paper.authors == ["Alice Kim", "Bob Lee"]
|
|
assert paper.venue.startswith("ACM Transactions")
|
|
assert paper.published_date == date(2026, 9, 2)
|
|
assert paper.oa_url == "https://dl.acm.org/doi/10.1145/1234567"
|
|
assert paper.pdf_url == "https://dl.acm.org/doi/pdf/10.1145/1234567"
|
|
assert paper.topics == ["Software Engineering Research"]
|
|
|
|
|
|
def test_paper_from_work_rejects_missing_doi_or_title():
|
|
assert paper_from_work(make_work(doi=None)) is None
|
|
assert paper_from_work(make_work(doi="https://example.org/x")) is None
|
|
assert paper_from_work(make_work(title="", display_name="")) is None
|
|
|
|
|
|
def test_paper_from_work_tolerates_sparse_work():
|
|
paper = paper_from_work({"doi": "https://doi.org/10.1/x", "title": "T"})
|
|
assert paper is not None
|
|
assert paper.abstract == "" and paper.authors == [] and paper.published_date is None
|
|
|
|
|
|
# -------------------------------------------------------------------------------- fetching
|
|
class FakeResponse:
|
|
def __init__(self, body: dict, status: int = 200):
|
|
self._body = body
|
|
self.status_code = status
|
|
|
|
def raise_for_status(self) -> None:
|
|
if self.status_code >= 400:
|
|
raise RuntimeError(f"HTTP {self.status_code}")
|
|
|
|
def json(self) -> dict:
|
|
return self._body
|
|
|
|
|
|
def install_fetch(monkeypatch, pages_by_query: dict[str, list[dict]]):
|
|
"""Route requests.get by whether the filter string contains the subfield filter (mega query)."""
|
|
calls: list[dict] = []
|
|
|
|
def fake_get(url, params=None, timeout=None, headers=None):
|
|
calls.append({"url": url, "params": params})
|
|
key = "mega" if "topics.subfield.id" in params["filter"] else "whitelist"
|
|
pages = pages_by_query.get(key, [])
|
|
idx = sum(1 for c in calls[:-1] if ("topics.subfield.id" in c["params"]["filter"]) == (key == "mega"))
|
|
if isinstance(pages, Exception):
|
|
raise pages
|
|
body = pages[idx] if idx < len(pages) else {"results": [], "meta": {"next_cursor": None}}
|
|
return FakeResponse(body)
|
|
|
|
monkeypatch.setattr(openalex.requests, "get", fake_get)
|
|
return calls
|
|
|
|
|
|
def test_fetch_merges_queries_applies_filters_and_dedups(monkeypatch):
|
|
mdpi = make_work(
|
|
doi="https://doi.org/10.3390/xyz",
|
|
primary_location={"source": {"display_name": "Electronics", "issn_l": "2079-9292", "issn": [], "host_organization_name": "MDPI AG"}},
|
|
)
|
|
off_topic = make_work(
|
|
doi="https://doi.org/10.1109/access.1",
|
|
title="Optimal Beamforming for 6G Networks",
|
|
abstract_inverted_index={"MIMO": [0]},
|
|
topics=[{"display_name": "Wireless Communication"}],
|
|
)
|
|
on_topic = make_work(
|
|
doi="https://doi.org/10.1109/access.2",
|
|
title="LLM-based Code Review at Scale",
|
|
topics=[{"display_name": "Software Engineering Research"}],
|
|
)
|
|
dup = make_work() # same DOI as the whitelist hit
|
|
pages = {
|
|
"whitelist": [{"results": [make_work(), mdpi], "meta": {"next_cursor": None}}],
|
|
"mega": [{"results": [off_topic, on_topic, dup], "meta": {"next_cursor": None}}],
|
|
}
|
|
calls = install_fetch(monkeypatch, pages)
|
|
|
|
papers = fetch_openalex_papers(START, END, SETTINGS)
|
|
|
|
assert [p.doi for p in papers] == ["10.1145/1234567", "10.1109/access.2"]
|
|
filters = [c["params"]["filter"] for c in calls]
|
|
assert any("primary_location.source.issn:" in f and "is_oa:true" in f for f in filters)
|
|
assert any(f"topics.subfield.id:{SETTINGS.se_subfield_id}" in f for f in filters)
|
|
assert all(f"from_publication_date:{START}" in f and f"to_publication_date:{END}" in f for f in filters)
|
|
|
|
|
|
def test_fetch_follows_cursor_pagination(monkeypatch):
|
|
page1 = {"results": [make_work()], "meta": {"next_cursor": "c2"}}
|
|
page2 = {"results": [make_work(doi="https://doi.org/10.1/second")], "meta": {"next_cursor": None}}
|
|
calls = install_fetch(monkeypatch, {"whitelist": [page1, page2], "mega": []})
|
|
|
|
papers = fetch_openalex_papers(START, END, SETTINGS)
|
|
|
|
assert {p.doi for p in papers} == {"10.1145/1234567", "10.1/second"}
|
|
whitelist_calls = [c for c in calls if "subfield" not in c["params"]["filter"]]
|
|
assert [c["params"]["cursor"] for c in whitelist_calls] == ["*", "c2"]
|
|
|
|
|
|
def test_one_failed_query_keeps_the_other(monkeypatch):
|
|
def fake_get(url, params=None, timeout=None, headers=None):
|
|
if "topics.subfield.id" in params["filter"]:
|
|
raise RuntimeError("mega query down")
|
|
return FakeResponse({"results": [make_work()], "meta": {"next_cursor": None}})
|
|
|
|
monkeypatch.setattr(openalex.requests, "get", fake_get)
|
|
papers = fetch_openalex_papers(START, END, SETTINGS)
|
|
assert [p.doi for p in papers] == ["10.1145/1234567"]
|
|
|
|
|
|
def test_all_queries_failed_raises(monkeypatch):
|
|
def fake_get(*a, **k):
|
|
raise RuntimeError("openalex down")
|
|
|
|
monkeypatch.setattr(openalex.requests, "get", fake_get)
|
|
with pytest.raises(RuntimeError, match="all openalex queries failed"):
|
|
fetch_openalex_papers(START, END, SETTINGS)
|
|
|
|
|
|
def test_mailto_param_only_when_configured(monkeypatch):
|
|
calls = install_fetch(monkeypatch, {"whitelist": [], "mega": []})
|
|
fetch_openalex_papers(START, END, SETTINGS)
|
|
assert all("mailto" not in c["params"] for c in calls)
|
|
|
|
calls.clear()
|
|
settings = Settings(_env_file=None, SEBRIEF_OPENALEX_MAILTO="ops@example.org")
|
|
fetch_openalex_papers(START, END, settings)
|
|
assert all(c["params"]["mailto"] == "ops@example.org" for c in calls)
|