"""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, title FROM papers.""" conn = MagicMock() conn.execute.return_value.fetchall.return_value = [ {"id": pid, "title": f"Title of {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", *extra_args]) 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 def test_small_corpus_warning_shown(self): # known_ids has only 3 papers (< graph_min_corpus_size default 15) result = self._run(known_ids=("A", "B", "C")) assert result.exit_code == 0 assert "Warning" in result.output or "warning" in result.output # --------------------------------------------------------------------------- # 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(), extra_args=("--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(), extra_args=("--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 _make_two_paper_conn(self, pr_winner: str) -> MagicMock: """Two papers A and B at equal distance; pr_winner has a high PageRank score.""" conn = MagicMock() conn.execute.return_value.fetchall.return_value = [ {"id": "A", "title": "Paper A", "year": 2020, "distance": 0.5}, {"id": "B", "title": "Paper B", "year": 2021, "distance": 0.5}, ] return conn def test_cite_boost_exits_zero(self): conn = MagicMock() conn.execute.return_value.fetchall.return_value = [ {"id": "A", "title": "Paper A", "year": 2020, "distance": 0.2}, ] graph = _sample_graph() 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 def test_cite_boost_promotes_high_pr_paper(self): """High-PR paper should appear first (lower boosted distance) via divide formula.""" import networkx as nx # Build a graph where A has high PageRank (many papers cite A) g = nx.DiGraph() for i in range(10): g.add_edge(f"P{i}", "A") # A is heavily cited → high PR g.add_edge("P0", "B") # B is barely cited → low PR conn = MagicMock() conn.execute.return_value.fetchall.return_value = [ {"id": "A", "title": "Hub", "year": 2020, "distance": 0.5}, {"id": "B", "title": "Niche", "year": 2021, "distance": 0.5}, ] 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=g), ): mock_emb.return_value.encode_dense.side_effect = _fake_encode result = runner.invoke(app, ["search", "paper", "query", "--cite-boost"]) assert result.exit_code == 0 lines = [ln for ln in result.output.strip().splitlines() if "Hub" in ln or "Niche" in ln] assert len(lines) == 2 # A (Hub) should appear before B (Niche) — lower boosted distance hub_line = next(line for line in lines if "Hub" in line) niche_line = next(line for line in lines if "Niche" in line) assert lines.index(hub_line) < lines.index(niche_line)