"""re3data supplement: parsing, filtering, privacy, failure isolation and route wiring.
All upstream responses are mocked with httpx.MockTransport using fixtures in
re3data's published XML shape; no network access is required.
"""
from __future__ import annotations
import asyncio
from pathlib import Path
import httpx
import pytest
from fastapi.testclient import TestClient
import re3data
from repositories import REGISTRY
FIXTURES = Path(__file__).parent / "fixtures"
def _fixture(name: str) -> bytes:
return (FIXTURES / name).read_bytes()
class Recorder:
def __init__(self, fail: bool = False, status: int = 200):
self.requests: list[httpx.Request] = []
self.fail = fail
self.status = status
def __call__(self, request: httpx.Request) -> httpx.Response:
self.requests.append(request)
if self.fail:
raise httpx.ConnectError("offline", request=request)
if self.status != 200:
return httpx.Response(self.status, request=request)
path = request.url.path
if path == "/api/beta/repositories":
return httpx.Response(200, content=_fixture("re3data_search_materials.xml"), request=request)
rid = path.rsplit("/", 1)[-1]
target = FIXTURES / f"re3data_{rid}.xml"
if target.exists():
return httpx.Response(200, content=target.read_bytes(), request=request)
return httpx.Response(404, request=request)
@pytest.fixture
def mock_re3data(monkeypatch):
def install(recorder: Recorder):
re3data.clear_caches()
client = httpx.AsyncClient(
transport=httpx.MockTransport(recorder),
event_hooks={"request": [re3data._enforce_origin]},
)
monkeypatch.setattr(re3data, "_http_client", client)
return recorder
yield install
re3data.clear_caches()
monkeypatch.setattr(re3data, "_http_client", None)
def run(coro):
return asyncio.run(coro)
def test_parse_search_and_detail_are_namespace_agnostic():
hits = re3data.parse_search(_fixture("re3data_search_materials.xml"))
assert [h["id"] for h in hits][:2] == ["r3d100010134", "r3d100012345"]
assert all(h["id"].startswith("r3d") for h in hits) # malformed id dropped
record = re3data.parse_detail(_fixture("re3data_r3d100012345.xml"))
assert record["name"] == "Materials Data Facility"
assert record["url"] == "https://materialsdatafacility.org/"
assert "Materials Science" in record["subjects"] # DFG code stripped
assert record["pid_systems"] == ["DOI"]
assert record["data_upload"] == ["open"]
assert record["versioning"] == "Yes"
assert record["last_update"] == "2025-11-20"
def test_parser_rejects_doctype_and_garbage():
evil = b']>r3d100000001
'
assert re3data.parse_search(evil) == []
assert re3data.parse_search(b"not xml") == []
assert re3data.parse_detail(b"") is None
def test_supplement_filters_curated_closed_and_institutional(mock_re3data):
recorder = mock_re3data(Recorder())
result = run(re3data.find_supplementary("Materials science", {}, REGISTRY.values()))
assert result["status"] == "ok"
names = [r["name"] for r in result["repositories"]]
# PANGAEA is already curated; the institutional and retired ones are excluded.
assert "PANGAEA" not in names
assert "Example University Repository" not in names
assert "Retired Alloy Archive" not in names
assert names[0] == "Materials Data Facility"
top = result["repositories"][0]
assert top["re3data_url"] == "https://www.re3data.org/repository/r3d100012345"
assert top["doi_assignment"] == "Yes"
assert top["file_deposit_recommended"] is False
assert "not reviewed by KU Library" in top["review_notes"][0]
# Only the discipline is sent upstream.
search_requests = [r for r in recorder.requests if r.url.path.endswith("/repositories")]
assert search_requests and search_requests[0].url.params["query"] == "materials science"
def test_supplement_never_sends_title_or_description(mock_re3data):
recorder = mock_re3data(Recorder())
run(re3data.find_supplementary("Materials science", {
"title": "Confidential unpublished alloy dataset",
"description": "secret-description-token",
}, REGISTRY.values()))
sent = " ".join(str(r.url) for r in recorder.requests)
assert "secret" not in sent and "Confidential" not in sent
def test_sensitive_dataset_requires_recorded_controlled_access(mock_re3data):
mock_re3data(Recorder())
result = run(re3data.find_supplementary(
"Materials science", {"sensitivity": "Confidential", "contains_personal_data": "Yes"}, REGISTRY.values()))
# All fixture records list only open access, so none may be suggested.
assert result["repositories"] == []
assert result["status"] == "no_matches"
def test_unavailable_upstream_is_reported_not_raised(mock_re3data):
mock_re3data(Recorder(fail=True))
result = run(re3data.find_supplementary("Materials science", {}, REGISTRY.values()))
assert result["status"] == "unavailable"
assert result["repositories"] == []
mock_re3data(Recorder(status=503))
assert run(re3data.find_supplementary("Chemistry", {}, REGISTRY.values()))["status"] == "unavailable"
def test_results_are_cached(mock_re3data):
recorder = mock_re3data(Recorder())
run(re3data.find_supplementary("Materials science", {}, REGISTRY.values()))
first = len(recorder.requests)
run(re3data.find_supplementary("Materials science", {}, REGISTRY.values()))
assert len(recorder.requests) == first
def test_empty_discipline_and_disabled_switch(monkeypatch, mock_re3data):
recorder = mock_re3data(Recorder())
assert run(re3data.find_supplementary(" ", {}, []))["status"] == "skipped"
monkeypatch.setenv("RE3DATA_ENABLED", "false")
assert run(re3data.find_supplementary("Materials", {}, []))["status"] == "disabled"
assert recorder.requests == []
def test_client_refuses_other_origins(mock_re3data):
mock_re3data(Recorder())
with pytest.raises(re3data.Re3dataUnavailable):
run(re3data._get("https://evil.example.org/x"))
# ── Route integration ────────────────────────────────────────────────────
import app # noqa: E402
DATASET = {
"title": "Alloy fatigue measurements", "description": "Strain-life data.",
"discipline": "Materials science", "data_types": "Tabular",
"sensitivity": "Open", "contains_personal_data": "No",
}
def _model_answer(ids):
return {
"dataset_understanding": {"summary": "s"},
"repositories": [{"id": rid, "fit_reason": "fits", "fit_confidence": "High"} for rid in ids],
"deposit_strategy": {"primary_repository_id": ids[0] if ids else ""},
"next_actions": [], "global_notes": "",
}
def _patch_route(monkeypatch, ids, supplement):
async def fake_openai(*_a, **_k):
return _model_answer(ids)
async def fake_supplement(**_k):
return supplement
async def fake_find(*_a, **_k):
return dict(supplement)
monkeypatch.setattr(app, "call_openai", fake_openai)
monkeypatch.setattr(app, "get_client", lambda: None)
monkeypatch.setattr(app, "check_rate_limit", lambda *a: True)
monkeypatch.setattr(app.re3data, "find_supplementary", fake_find)
def test_route_keeps_curated_results_and_adds_supplement(monkeypatch):
supplement = {"status": "ok", "query": "materials science", "repositories": [{"id": "r3d1", "name": "X"}]}
_patch_route(monkeypatch, ["materials_cloud", "zenodo"], supplement)
response = TestClient(app.app).post("/api/find-repository", json={"dataset": DATASET})
assert response.status_code == 200
result = response.json()["result"]
assert [r["id"] for r in result["repositories"]] == ["khazna", "materials_cloud", "zenodo"]
assert result["deposit_strategy"]["primary_repository_id"] == "materials_cloud"
assert result["re3data_supplement"]["repositories"][0]["name"] == "X"
assert result["re3data_supplement"]["recommended_view"] == "collapsed"
# Verify links are URL-encoded.
assert "query=Materials+Cloud" in result["repositories"][1]["verify_links"]["re3data"]
def test_route_expands_supplement_when_curated_is_thin(monkeypatch):
_patch_route(monkeypatch, [], {"status": "ok", "query": "q", "repositories": []})
result = TestClient(app.app).post("/api/find-repository", json={"dataset": DATASET}).json()["result"]
assert [r["id"] for r in result["repositories"]] == ["khazna"]
assert result["re3data_supplement"]["recommended_view"] == "expanded"
def test_route_survives_re3data_exception_and_timeout(monkeypatch):
_patch_route(monkeypatch, ["zenodo"], {})
async def boom(*_a, **_k):
raise RuntimeError("unexpected")
monkeypatch.setattr(app.re3data, "find_supplementary", boom)
result = TestClient(app.app).post("/api/find-repository", json={"dataset": DATASET}).json()["result"]
assert result["re3data_supplement"]["status"] == "unavailable"
assert [r["id"] for r in result["repositories"]] == ["khazna", "zenodo"]
async def slow(*_a, **_k):
await asyncio.sleep(5)
monkeypatch.setattr(app.re3data, "find_supplementary", slow)
monkeypatch.setenv("RE3DATA_TIMEOUT_SECONDS", "0.05")
result = TestClient(app.app).post("/api/find-repository", json={"dataset": DATASET}).json()["result"]
assert result["re3data_supplement"]["status"] == "unavailable"
def test_route_model_failure_cancels_lookup(monkeypatch):
cancelled = []
async def fake_openai(*_a, **_k):
await asyncio.sleep(0.05) # let the lookup start running
return {"error": "The AI service returned an unreadable answer. Please try again."}
async def slow(*_a, **_k):
try:
await asyncio.sleep(5)
except asyncio.CancelledError:
cancelled.append(True)
raise
monkeypatch.setattr(app, "call_openai", fake_openai)
monkeypatch.setattr(app, "get_client", lambda: None)
monkeypatch.setattr(app, "check_rate_limit", lambda *a: True)
monkeypatch.setattr(app.re3data, "find_supplementary", slow)
response = TestClient(app.app).post("/api/find-repository", json={"dataset": DATASET})
assert response.status_code == 500
assert cancelled == [True]
def test_health_reports_re3data():
body = TestClient(app.app).get("/api/health").json()
assert "re3data" in body["sources"]