user2595/codex-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(F-15): citation graph — PageRank, coupling, co-citation, CLI

Browse Source 
- codex/graph.py: build_citation_graph (DiGraph from DB), citation_pagerank
  (graceful uniform fallback < 5 nodes), find_related (bibliographic coupling),
  find_co_cited, dangling_citations
- codex/config.py: graph_cite_boost_alpha=0.3, graph_min_corpus_size=15
- codex/cli.py: graph report [--top-n] [--json], graph related <paper-id>
  [--min-shared]; search paper --cite-boost (PageRank score weighting)
- tests/graph/: 39 tests (graph functions + CLI) — 326 total green

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Tarik Moussa
2026-06-15 03:31:36 +02:00
parent 9647897173
commit fd51b70000
6 changed files with 747 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
tests/graph/__init__.py Normal file
View File

176
tests/graph/test_cli.py Normal file
View File

@@ -0,0 +1,176 @@
"""CLI tests for `codex graph` sub-commands (F-15)."""
from __future__ import annotations
from contextlib import contextmanager
from unittest.mock import MagicMock, patch
import networkx as nx
import numpy as np
from typer.testing import CliRunner
from codex.cli import app
runner = CliRunner()
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_conn_cm(conn: MagicMock):
@contextmanager
def _cm():
yield conn
return _cm
def _make_conn_with_paper_ids(*paper_ids: str) -> MagicMock:
"""Mock conn that returns paper rows for SELECT id FROM papers."""
conn = MagicMock()
conn.execute.return_value.fetchall.return_value = [{"id": pid} for pid in paper_ids]
return conn
def _sample_graph() -> nx.DiGraph:
"""A→B, A→C, B→C, B→D, C→D — 5 nodes, D is dangling (not in papers)."""
g = nx.DiGraph()
g.add_edges_from([("A", "B"), ("A", "C"), ("B", "C"), ("B", "D"), ("C", "D")])
return g
# ---------------------------------------------------------------------------
# codex graph report
# ---------------------------------------------------------------------------
class TestGraphReport:
def _run(self, graph=None, known_ids=("A", "B", "C"), *extra_args):
if graph is None:
graph = _sample_graph()
conn = _make_conn_with_paper_ids(*known_ids)
with (
patch("codex.db.get_conn", side_effect=_make_conn_cm(conn)),
patch("codex.graph.build_citation_graph", return_value=graph),
):
return runner.invoke(app, ["graph", "report"])
def test_exit_zero(self):
assert self._run().exit_code == 0
def test_shows_hub_section(self):
assert "HUB" in self._run().output.upper()
def test_shows_dangling_section(self):
assert "DANGLING" in self._run().output.upper()
def test_dangling_node_in_output(self):
# D is cited but not in known_ids
assert "D" in self._run().output
def test_json_output_valid(self):
import json
graph = _sample_graph()
conn = _make_conn_with_paper_ids("A", "B", "C")
with (
patch("codex.db.get_conn", side_effect=_make_conn_cm(conn)),
patch("codex.graph.build_citation_graph", return_value=graph),
):
result = runner.invoke(app, ["graph", "report", "--json"])
assert result.exit_code == 0
data = json.loads(result.output)
assert "hubs" in data
assert "dangling" in data
assert "nodes" in data
assert data["dangling_count"] == 1
def test_empty_graph_message(self):
conn = _make_conn_with_paper_ids()
with (
patch("codex.db.get_conn", side_effect=_make_conn_cm(conn)),
patch("codex.graph.build_citation_graph", return_value=nx.DiGraph()),
):
result = runner.invoke(app, ["graph", "report"])
assert result.exit_code == 0
assert "empty" in result.output.lower()
def test_top_n_respected(self):
# Build graph with 8 nodes, ask for top 3
g = nx.DiGraph()
for i in range(8):
g.add_edge(f"P{i}", "HUB")
conn = _make_conn_with_paper_ids(*[f"P{i}" for i in range(8)], "HUB")
with (
patch("codex.db.get_conn", side_effect=_make_conn_cm(conn)),
patch("codex.graph.build_citation_graph", return_value=g),
):
result = runner.invoke(app, ["graph", "report", "--top-n", "3"])
assert result.exit_code == 0
# ---------------------------------------------------------------------------
# codex graph related
# ---------------------------------------------------------------------------
class TestGraphRelated:
def _run(self, paper_id: str, graph=None, *extra_args):
if graph is None:
graph = _sample_graph()
conn = MagicMock()
with (
patch("codex.db.get_conn", side_effect=_make_conn_cm(conn)),
patch("codex.graph.build_citation_graph", return_value=graph),
):
return runner.invoke(app, ["graph", "related", paper_id, *extra_args])
def test_exit_zero_known_paper(self):
assert self._run("A").exit_code == 0
def test_unknown_paper_graceful(self):
result = self._run("UNKNOWN_ID")
assert result.exit_code == 0
assert "No papers" in result.output
def test_no_results_message_when_min_shared_high(self):
result = self._run("A", _sample_graph(), "--min-shared", "99")
assert "No papers" in result.output
def test_related_papers_shown(self):
# A cites {B,C}; B also cites C → shared=1 with A
result = self._run("A", _sample_graph(), "--min-shared", "1")
assert result.exit_code == 0
# B shares C with A; should appear
assert "B" in result.output
# ---------------------------------------------------------------------------
# codex search paper --cite-boost
# ---------------------------------------------------------------------------
class TestSearchCiteBoost:
def test_cite_boost_exits_zero(self):
conn = MagicMock()
paper_rows = [
{"id": "A", "title": "Paper A", "year": 2020, "distance": 0.2},
]
graph = _sample_graph()
conn.execute.return_value.fetchall.return_value = paper_rows
def _fake_encode(texts):
return np.zeros((len(texts), 1024), dtype="float32")
with (
patch("codex.db.get_conn", side_effect=_make_conn_cm(conn)),
patch("codex.embed.get_embedder") as mock_emb,
patch("codex.graph.build_citation_graph", return_value=graph),
):
mock_emb.return_value.encode_dense.side_effect = _fake_encode
result = runner.invoke(app, ["search", "paper", "hodge star", "--cite-boost"])
assert result.exit_code == 0

271
tests/graph/test_graph.py Normal file
View File

@@ -0,0 +1,271 @@
"""Tests for codex.graph — citation graph analytics (F-15)."""
from __future__ import annotations
from unittest.mock import MagicMock
import networkx as nx
import pytest
from codex.graph import (
build_citation_graph,
citation_pagerank,
dangling_citations,
find_co_cited,
find_related,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _make_conn(rows: list[dict]) -> MagicMock:
conn = MagicMock()
conn.execute.return_value.fetchall.return_value = rows
return conn
def _citation_rows(*pairs: tuple[str, str]) -> list[dict]:
return [{"citing_id": a, "cited_id": b} for a, b in pairs]
def _small_graph() -> nx.DiGraph:
"""
A→C, A→D, B→C, B→D, B→E, C→D
Nodes: A B C D E (5 nodes ingested)
"""
rows = _citation_rows(
("A", "C"),
("A", "D"),
("B", "C"),
("B", "D"),
("B", "E"),
("C", "D"),
)
return build_citation_graph(_make_conn(rows))
# ---------------------------------------------------------------------------
# build_citation_graph
# ---------------------------------------------------------------------------
class TestBuildCitationGraph:
def test_empty_db_returns_empty_graph(self):
g = build_citation_graph(_make_conn([]))
assert isinstance(g, nx.DiGraph)
assert g.number_of_nodes() == 0
assert g.number_of_edges() == 0
def test_nodes_and_edges_populated(self):
rows = _citation_rows(("P1", "P2"), ("P1", "P3"), ("P2", "P3"))
g = build_citation_graph(_make_conn(rows))
assert set(g.nodes()) == {"P1", "P2", "P3"}
assert g.number_of_edges() == 3
def test_edge_direction(self):
rows = _citation_rows(("citing", "cited"))
g = build_citation_graph(_make_conn(rows))
assert g.has_edge("citing", "cited")
assert not g.has_edge("cited", "citing")
def test_duplicate_edges_collapsed(self):
rows = _citation_rows(("A", "B"), ("A", "B"))
g = build_citation_graph(_make_conn(rows))
assert g.number_of_edges() == 1
def test_dangling_nodes_included(self):
rows = _citation_rows(("P1", "not_in_papers"))
g = build_citation_graph(_make_conn(rows))
assert "not_in_papers" in g.nodes()
# ---------------------------------------------------------------------------
# citation_pagerank
# ---------------------------------------------------------------------------
class TestCitationPagerank:
def test_empty_graph_returns_empty_dict(self):
g = nx.DiGraph()
assert citation_pagerank(g) == {}
def test_scores_sum_to_one(self):
g = _small_graph()
pr = citation_pagerank(g)
assert abs(sum(pr.values()) - 1.0) < 1e-6
def test_hub_node_has_highest_score(self):
# D is pointed to by A, B, C → should rank highest
g = _small_graph()
pr = citation_pagerank(g)
assert pr["D"] == max(pr.values())
def test_all_nodes_present(self):
g = _small_graph()
pr = citation_pagerank(g)
assert set(pr.keys()) == set(g.nodes())
def test_small_graph_returns_uniform_scores(self):
# 3 nodes < _MIN_PAGERANK_NODES (5) → uniform
rows = _citation_rows(("A", "B"), ("B", "C"))
g = build_citation_graph(_make_conn(rows))
pr = citation_pagerank(g)
assert len(pr) == 3
scores = list(pr.values())
assert all(abs(s - scores[0]) < 1e-9 for s in scores)
def test_small_graph_logs_warning(self, caplog: pytest.LogCaptureFixture):
import logging
rows = _citation_rows(("A", "B"), ("B", "C"))
g = build_citation_graph(_make_conn(rows))
with caplog.at_level(logging.WARNING, logger="codex.graph"):
citation_pagerank(g)
assert any("corpus too small" in rec.message for rec in caplog.records)
def test_custom_damping(self):
g = _small_graph()
pr_85 = citation_pagerank(g, damping=0.85)
pr_50 = citation_pagerank(g, damping=0.50)
# Different damping → different scores (not all equal)
assert pr_85 != pr_50
# ---------------------------------------------------------------------------
# find_related (bibliographic coupling)
# ---------------------------------------------------------------------------
class TestFindRelated:
def _coupling_graph(self) -> nx.DiGraph:
"""
A cites: X Y Z
B cites: X Y
C cites: X
D cites: W (no overlap with A)
"""
rows = _citation_rows(
("A", "X"),
("A", "Y"),
("A", "Z"),
("B", "X"),
("B", "Y"),
("C", "X"),
("D", "W"),
)
return build_citation_graph(_make_conn(rows))
def test_unknown_paper_returns_empty(self):
g = self._coupling_graph()
assert find_related("UNKNOWN", g) == []
def test_paper_with_no_refs_returns_empty(self):
# X has no outgoing edges → no references to couple on
g = self._coupling_graph()
assert find_related("X", g) == []
def test_related_by_two_shared_refs(self):
g = self._coupling_graph()
related = find_related("A", g, min_shared=2)
assert "B" in related
def test_not_related_below_threshold(self):
g = self._coupling_graph()
related = find_related("A", g, min_shared=2)
assert "C" not in related # only 1 shared ref
def test_ordered_by_shared_count_desc(self):
g = self._coupling_graph()
related = find_related("A", g, min_shared=1)
# B shares 2 refs, C shares 1 → B before C
assert related.index("B") < related.index("C")
def test_self_not_in_results(self):
g = self._coupling_graph()
related = find_related("A", g, min_shared=1)
assert "A" not in related
# ---------------------------------------------------------------------------
# find_co_cited
# ---------------------------------------------------------------------------
class TestFindCoCited:
def _co_citation_graph(self) -> nx.DiGraph:
"""
P1 cites: A, B, C
P2 cites: A, B
P3 cites: A
"""
rows = _citation_rows(
("P1", "A"),
("P1", "B"),
("P1", "C"),
("P2", "A"),
("P2", "B"),
("P3", "A"),
)
return build_citation_graph(_make_conn(rows))
def test_unknown_paper_returns_empty(self):
g = self._co_citation_graph()
assert find_co_cited("UNKNOWN", g) == []
def test_paper_with_no_citers_returns_empty(self):
# C has citers but let's test a node that has no predecessors
rows = _citation_rows(("P1", "X"))
g = build_citation_graph(_make_conn(rows))
# "X" is cited once by P1, but P1 has no other citations
# X is co-cited with nothing
result = find_co_cited("X", g)
assert result == []
def test_most_co_cited_first(self):
g = self._co_citation_graph()
result = find_co_cited("A", g)
# B is co-cited twice (by P1 and P2); C once (by P1 only)
ids = [r[0] for r in result]
assert ids[0] == "B"
def test_self_not_included(self):
g = self._co_citation_graph()
result = find_co_cited("A", g)
assert all(pid != "A" for pid, _ in result)
def test_counts_correct(self):
g = self._co_citation_graph()
result = dict(find_co_cited("A", g))
assert result["B"] == 2 # P1 and P2 both cite A and B
assert result["C"] == 1 # only P1 cites A and C
# ---------------------------------------------------------------------------
# dangling_citations
# ---------------------------------------------------------------------------
class TestDanglingCitations:
def test_empty_graph(self):
assert dangling_citations(nx.DiGraph(), set()) == []
def test_all_known_returns_empty(self):
rows = _citation_rows(("A", "B"), ("B", "C"))
g = build_citation_graph(_make_conn(rows))
assert dangling_citations(g, {"A", "B", "C"}) == []
def test_identifies_dangling(self):
rows = _citation_rows(("A", "B"), ("A", "external_doi"))
g = build_citation_graph(_make_conn(rows))
dangling = dangling_citations(g, {"A", "B"})
assert "external_doi" in dangling
assert "A" not in dangling
assert "B" not in dangling
def test_empty_known_set(self):
rows = _citation_rows(("A", "B"))
g = build_citation_graph(_make_conn(rows))
dangling = dangling_citations(g, set())
assert set(dangling) == {"A", "B"}