codex-py/tests/graph/test_cli.py

"""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", *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

    def test_shows_citing_coverage_line(self):
        # _sample_graph: A, B, C all cite → 3/3 = 100%, so the line shows but no warning.
        out = self._run(known_ids=("A", "B", "C")).output
        assert "Citing coverage" in out
        assert "weakens" not in out  # 100% coverage → no R-D warning

    def test_low_citing_coverage_warning(self):
        # Only A has an out-edge; B and C are ingested but cite nothing → 1/3 = 33%.
        g = nx.DiGraph()
        g.add_edge("A", "X")
        g.add_nodes_from(["B", "C"])
        result = self._run(graph=g, known_ids=("A", "B", "C"))
        assert result.exit_code == 0
        assert "weakens PageRank" in result.output  # the R-D coverage warning

    def test_json_includes_citing_coverage(self):
        import json

        g = nx.DiGraph()
        g.add_edge("A", "X")
        g.add_nodes_from(["B", "C"])
        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=g),
        ):
            result = runner.invoke(app, ["graph", "report", "--json"])
        data = json.loads(result.output)
        assert data["citing_coverage"] == {"citing": 1, "papers": 3, "fraction": 0.3333}


# ---------------------------------------------------------------------------
# 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)