101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
"""End-to-end smoke test: ingest → discover → provenance flow (all mocked)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import contextmanager
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import numpy as np
|
|
|
|
from codex.discover import discovery_leads
|
|
from codex.ingest import ingest_paper
|
|
from codex.models import Citation, Paper
|
|
from codex.provenance import export_bib
|
|
|
|
|
|
def _make_conn_cm(mock_conn: MagicMock) -> Any:
|
|
"""Return a context-manager factory that yields *mock_conn*."""
|
|
|
|
@contextmanager
|
|
def _cm() -> Any:
|
|
yield mock_conn
|
|
|
|
return _cm
|
|
|
|
|
|
def test_full_pipeline_smoke() -> None:
|
|
"""ingest_paper → discovery_leads → export_bib all execute without errors (all mocked)."""
|
|
paper = Paper(id="2301.07041", title="T", openalex_id="W123")
|
|
citations = [Citation(citing_id="2301.07041", cited_id="2301.99999")]
|
|
|
|
# --- mock connection setup ---
|
|
mock_conn = MagicMock()
|
|
mock_conn.execute = MagicMock()
|
|
mock_conn.commit = MagicMock()
|
|
|
|
# discovery_leads: conn.execute(...).fetchall() → empty list (no leads)
|
|
discover_execute_result = MagicMock()
|
|
discover_execute_result.fetchall.return_value = []
|
|
|
|
# export_bib: conn.execute(...).fetchall() → one paper row with bibkey
|
|
bib_row: dict[str, Any] = {
|
|
"id": "2301.07041",
|
|
"bibkey": "paper2023",
|
|
"title": "T",
|
|
"authors": ["Author One"],
|
|
"year": 2023,
|
|
}
|
|
bib_execute_result = MagicMock()
|
|
bib_execute_result.fetchall.return_value = [bib_row]
|
|
|
|
# The ingest path calls conn.execute multiple times (paper upsert, etc.).
|
|
# discovery_leads and export_bib each call conn.execute once and use fetchall.
|
|
# We use side_effect to route calls: first N calls from ingest return a generic
|
|
# mock; then we supply discover and bib results in order.
|
|
_call_count: list[int] = [0]
|
|
|
|
def _execute_side_effect(*args: Any, **kwargs: Any) -> MagicMock:
|
|
_call_count[0] += 1
|
|
sql: str = str(args[0]) if args else ""
|
|
if "cited_id NOT IN" in sql:
|
|
return discover_execute_result
|
|
if "FROM papers" in sql and "bibkey IS NOT NULL" in sql:
|
|
return bib_execute_result
|
|
return MagicMock()
|
|
|
|
mock_conn.execute.side_effect = _execute_side_effect
|
|
|
|
# cursor() context manager for executemany calls inside ingest
|
|
mock_cursor = MagicMock()
|
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
|
|
|
|
embedder = MagicMock()
|
|
embedder.dim = 1024
|
|
embedder.encode_dense.return_value = np.zeros((1, 1024), dtype=np.float32)
|
|
|
|
conn_cm = _make_conn_cm(mock_conn)
|
|
|
|
with (
|
|
patch("codex.ingest.openalex.fetch_paper", return_value=paper),
|
|
patch("codex.ingest.openalex.fetch_citations", return_value=citations),
|
|
patch("codex.ingest.get_embedder", return_value=embedder),
|
|
patch("codex.ingest.get_conn", side_effect=conn_cm),
|
|
patch("codex.discover.get_conn", side_effect=conn_cm),
|
|
patch("codex.provenance.get_conn", side_effect=conn_cm),
|
|
):
|
|
# Step 1: ingest
|
|
result = ingest_paper("2301.07041")
|
|
|
|
# Step 2: discover
|
|
leads = discovery_leads()
|
|
|
|
# Step 3: export bib
|
|
bib = export_bib(["2301.07041"])
|
|
|
|
# --- assertions ---
|
|
assert result.paper_id == "2301.07041"
|
|
assert isinstance(leads, list)
|
|
assert "@article" in bib
|