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