Spaces:
Running
Running
| """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) | |
| 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'<?xml version="1.0"?><!DOCTYPE x [<!ENTITY a "aaaa">]><list><repository><id>r3d100000001</id></repository></list>' | |
| assert re3data.parse_search(evil) == [] | |
| assert re3data.parse_search(b"not xml") == [] | |
| assert re3data.parse_detail(b"<r3d:re3data xmlns:r3d='x'/>") 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"] | |