146 lines
4.7 KiB
Python
146 lines
4.7 KiB
Python
"""Tests for codex.discover — graph-based discovery queries.
|
|
|
|
All DB access is mocked; no real PostgreSQL server required.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import contextmanager
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_conn_cm(mock_conn: MagicMock) -> Any:
|
|
"""Return a context-manager factory that yields *mock_conn*."""
|
|
|
|
@contextmanager # type: ignore[arg-type]
|
|
def _cm() -> Any:
|
|
yield mock_conn
|
|
|
|
return _cm
|
|
|
|
|
|
def _dict_row(**kwargs: Any) -> dict[str, Any]:
|
|
return dict(kwargs)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. test_discovery_leads_returns_correct_format
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_discovery_leads_returns_correct_format() -> None:
|
|
"""Mock conn returns rows; verify output has correct shape and ordering."""
|
|
from codex.discover import discovery_leads
|
|
|
|
mock_conn = MagicMock()
|
|
mock_conn.execute.return_value.fetchall.return_value = [
|
|
_dict_row(cited_id="W111", pull=5),
|
|
_dict_row(cited_id="W222", pull=3),
|
|
_dict_row(cited_id="W333", pull=1),
|
|
]
|
|
|
|
with patch("codex.discover.get_conn", side_effect=_make_conn_cm(mock_conn)):
|
|
result = discovery_leads(limit=10)
|
|
|
|
assert len(result) == 3
|
|
assert result[0] == {"cited_id": "W111", "pull": 5}
|
|
assert result[1] == {"cited_id": "W222", "pull": 3}
|
|
assert result[2] == {"cited_id": "W333", "pull": 1}
|
|
|
|
# Verify the SQL and limit param were passed
|
|
call_args = mock_conn.execute.call_args
|
|
sql: str = call_args[0][0]
|
|
params: dict[str, Any] = call_args[0][1]
|
|
assert "cited_id NOT IN" in sql
|
|
assert "pull DESC" in sql
|
|
assert params["limit"] == 10
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. test_citing_papers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_citing_papers() -> None:
|
|
"""Mock returns citing_id list; verify flat list of IDs is returned."""
|
|
from codex.discover import citing_papers
|
|
|
|
mock_conn = MagicMock()
|
|
mock_conn.execute.return_value.fetchall.return_value = [
|
|
_dict_row(citing_id="2301.00001"),
|
|
_dict_row(citing_id="2302.00002"),
|
|
]
|
|
|
|
with patch("codex.discover.get_conn", side_effect=_make_conn_cm(mock_conn)):
|
|
result = citing_papers("W999")
|
|
|
|
assert result == ["2301.00001", "2302.00002"]
|
|
|
|
call_args = mock_conn.execute.call_args
|
|
sql: str = call_args[0][0]
|
|
params: dict[str, Any] = call_args[0][1]
|
|
assert "cited_id" in sql
|
|
assert params["paper_id"] == "W999"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. test_cited_by
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_cited_by() -> None:
|
|
"""Mock returns cited_id list; verify flat list of IDs is returned."""
|
|
from codex.discover import cited_by
|
|
|
|
mock_conn = MagicMock()
|
|
mock_conn.execute.return_value.fetchall.return_value = [
|
|
_dict_row(cited_id="10.1000/ref_a"),
|
|
_dict_row(cited_id="10.1000/ref_b"),
|
|
]
|
|
|
|
with patch("codex.discover.get_conn", side_effect=_make_conn_cm(mock_conn)):
|
|
result = cited_by("2301.00001")
|
|
|
|
assert result == ["10.1000/ref_a", "10.1000/ref_b"]
|
|
|
|
call_args = mock_conn.execute.call_args
|
|
sql: str = call_args[0][0]
|
|
params: dict[str, Any] = call_args[0][1]
|
|
assert "citing_id" in sql
|
|
assert params["paper_id"] == "2301.00001"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 4. test_cocited_papers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_cocited_papers() -> None:
|
|
"""Mock returns co-citation list; verify shape and ordering."""
|
|
from codex.discover import cocited_papers
|
|
|
|
mock_conn = MagicMock()
|
|
mock_conn.execute.return_value.fetchall.return_value = [
|
|
_dict_row(paper_id="W444", co_citations=7),
|
|
_dict_row(paper_id="W555", co_citations=2),
|
|
]
|
|
|
|
with patch("codex.discover.get_conn", side_effect=_make_conn_cm(mock_conn)):
|
|
result = cocited_papers("W111", limit=5)
|
|
|
|
assert len(result) == 2
|
|
assert result[0] == {"paper_id": "W444", "co_citations": 7}
|
|
assert result[1] == {"paper_id": "W555", "co_citations": 2}
|
|
|
|
call_args = mock_conn.execute.call_args
|
|
sql: str = call_args[0][0]
|
|
params: dict[str, Any] = call_args[0][1]
|
|
assert "co_citations DESC" in sql
|
|
assert params["paper_id"] == "W111"
|
|
assert params["limit"] == 5
|