codex-py/tests/graph/test_graph.py

"""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,
    citing_coverage,
    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()

    def test_cross_citation_uses_doi_when_in_kb(self):
        # Simulate DB COALESCE: cited_id already resolved to papers.id (DOI)
        # for in-KB papers; dangling references keep their OpenAlex ID.
        rows = [
            {
                "citing_id": "https://doi.org/10.1234/paper-a",
                "cited_id": "https://doi.org/10.5678/paper-b",
            },
            {
                "citing_id": "https://doi.org/10.5678/paper-b",
                "cited_id": "https://openalex.org/W999",
            },
        ]
        g = build_citation_graph(_make_conn(rows))
        assert g.has_edge("https://doi.org/10.1234/paper-a", "https://doi.org/10.5678/paper-b")
        assert g.has_edge("https://doi.org/10.5678/paper-b", "https://openalex.org/W999")
        assert g.number_of_nodes() == 3


# ---------------------------------------------------------------------------
# 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"}


# ---------------------------------------------------------------------------
# citing_coverage (R-D)
# ---------------------------------------------------------------------------


class TestCitingCoverage:
    def test_counts_papers_with_out_edges(self):
        # _small_graph: A→.., B→.., C→D have out-edges; D, E have none.
        g = _small_graph()
        n_citing, n_papers = citing_coverage(g, {"A", "B", "C", "D", "E"})
        assert (n_citing, n_papers) == (3, 5)

    def test_paper_absent_from_graph_counts_as_no_out_edges(self):
        # A cites; Z was ingested but has no chunks/edges, so it isn't a graph node.
        g = _small_graph()
        assert citing_coverage(g, {"A", "Z"}) == (1, 2)

    def test_empty(self):
        assert citing_coverage(nx.DiGraph(), set()) == (0, 0)