From fd51b7000038609a577a085e37e2c656497793b6 Mon Sep 17 00:00:00 2001 From: Tarik Moussa Date: Mon, 15 Jun 2026 03:31:36 +0200 Subject: [PATCH] =?UTF-8?q?feat(F-15):=20citation=20graph=20=E2=80=94=20Pa?= =?UTF-8?q?geRank,=20coupling,=20co-citation,=20CLI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 [--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 --- codex/cli.py | 107 +++++++++++++++ codex/config.py | 24 ++++ codex/graph.py | 169 ++++++++++++++++++++++++ tests/graph/__init__.py | 0 tests/graph/test_cli.py | 176 +++++++++++++++++++++++++ tests/graph/test_graph.py | 271 ++++++++++++++++++++++++++++++++++++++ 6 files changed, 747 insertions(+) create mode 100644 codex/graph.py create mode 100644 tests/graph/__init__.py create mode 100644 tests/graph/test_cli.py create mode 100644 tests/graph/test_graph.py diff --git a/codex/cli.py b/codex/cli.py index 0ddaefc..0816d6e 100644 --- a/codex/cli.py +++ b/codex/cli.py @@ -14,6 +14,7 @@ prov_app = typer.Typer(help="Provenance: @cite scan, code_links, bib export.") wiki_app = typer.Typer(help="Wiki-compile: grounded concept pages over the RAG substrate.") synth_app = typer.Typer(help="Synthesis: grounded leads and quarantined conjectures (F-13).") quality_app = typer.Typer(help="Chunk quality gate and section classification (F-16).") +graph_app = typer.Typer(help="Citation graph analytics: PageRank, coupling, dangling leads (F-15).") # F-09: search sub-app with paper (semantic) and formula (FTS) subcommands search_app = typer.Typer( help="Search: semantic paper search and formula full-text search.", @@ -24,6 +25,7 @@ app.add_typer(prov_app, name="provenance") app.add_typer(wiki_app, name="wiki") app.add_typer(synth_app, name="synthesis") app.add_typer(quality_app, name="quality") +app.add_typer(graph_app, name="graph") app.add_typer(search_app, name="search") @@ -66,8 +68,14 @@ def search_callback(ctx: typer.Context) -> None: def search_paper( query: str = typer.Argument(..., help="Natural-language search query"), limit: int = typer.Option(10, "--limit", "-n", help="Number of results"), + cite_boost: bool = typer.Option( + False, + "--cite-boost", + help="Weight results by citation PageRank (F-15). Graceful when corpus < 5 papers.", + ), ) -> None: """Semantic similarity search over paper abstracts.""" + from codex.config import get_settings from codex.db import get_conn from codex.embed import get_embedder @@ -84,6 +92,24 @@ def search_paper( """, {"emb": emb, "limit": limit}, ).fetchall() + + if cite_boost and rows: + from codex.graph import build_citation_graph, citation_pagerank + + settings = get_settings() + graph = build_citation_graph(conn) + pr = citation_pagerank(graph) + alpha = settings.graph_cite_boost_alpha + results = [] + for row in rows: + pr_score = pr.get(row["id"], 0.0) + boosted = row["distance"] * (1.0 + alpha * pr_score) + results.append((boosted, row)) + results.sort(key=lambda x: x[0]) + for boosted_dist, row in results: + typer.echo(f"[{boosted_dist:.3f}] {row['id']} ({row['year']}) — {row['title']}") + return + for row in rows: typer.echo(f"[{row['distance']:.3f}] {row['id']} ({row['year']}) — {row['title']}") @@ -496,3 +522,84 @@ def quality_run( f"{stats['removed']} removed, " f"{stats['tagged']} section-tagged" ) + + +# --------------------------------------------------------------------------- +# F-15 — graph sub-commands +# --------------------------------------------------------------------------- + + +@graph_app.command("report") +def graph_report( + top_n: int = typer.Option(10, "--top-n", "-n", help="Number of hub papers to show."), + output_json: bool = typer.Option(False, "--json", help="Output as JSON."), +) -> None: + """Show citation graph report: top hub papers, dangling citations, cluster summary.""" + from codex.db import get_conn + from codex.graph import build_citation_graph, citation_pagerank, dangling_citations + + with get_conn() as conn: + graph = build_citation_graph(conn) + known_ids = {row["id"] for row in conn.execute("SELECT id FROM papers").fetchall()} + + if graph.number_of_nodes() == 0: + typer.echo("Citation graph is empty — ingest some papers first.") + raise typer.Exit(0) + + pr = citation_pagerank(graph) + hubs = sorted(pr.items(), key=lambda x: -x[1])[:top_n] + dangling = dangling_citations(graph, known_ids) + + if output_json: + typer.echo( + json.dumps( + { + "nodes": graph.number_of_nodes(), + "edges": graph.number_of_edges(), + "hubs": [{"paper_id": pid, "pagerank": score} for pid, score in hubs], + "dangling_count": len(dangling), + "dangling": dangling, + }, + indent=2, + ) + ) + return + + typer.echo(f"Graph: {graph.number_of_nodes()} nodes, {graph.number_of_edges()} edges") + typer.echo("") + typer.echo(f"TOP-{top_n} HUB PAPERS (PageRank)") + typer.echo("-" * 48) + for pid, score in hubs: + typer.echo(f" {score:.4f} {pid}") + typer.echo("") + typer.echo(f"DANGLING CITATIONS ({len(dangling)} cited but not ingested)") + typer.echo("-" * 48) + for pid in dangling[:top_n]: + typer.echo(f" {pid}") + if len(dangling) > top_n: + typer.echo(f" … and {len(dangling) - top_n} more") + + +@graph_app.command("related") +def graph_related( + paper_id: str = typer.Argument(..., help="Paper ID to find related papers for."), + min_shared: int = typer.Option( + 2, "--min-shared", help="Minimum shared references for bibliographic coupling." + ), +) -> None: + """List bibliographically related papers (shared references ≥ min-shared).""" + from codex.db import get_conn + from codex.graph import build_citation_graph, find_related + + with get_conn() as conn: + graph = build_citation_graph(conn) + + related = find_related(paper_id, graph, min_shared=min_shared) + if not related: + typer.echo(f"No papers with ≥ {min_shared} shared references found for {paper_id}.") + return + typer.echo( + f"Papers related to {paper_id} (bibliographic coupling, ≥ {min_shared} shared refs):" + ) + for pid in related: + typer.echo(f" {pid}") diff --git a/codex/config.py b/codex/config.py index 91698d2..dfe27f2 100644 --- a/codex/config.py +++ b/codex/config.py @@ -286,6 +286,30 @@ class Settings(BaseSettings): ), ) + # ------------------------------------------------------------------ + # F-15 Literature Graph + # ------------------------------------------------------------------ + graph_cite_boost_alpha: float = Field( + default=0.3, + ge=0.0, + le=1.0, + description=( + "Weight of the PageRank citation-boost in hybrid search. " + "Final score = dense_score * (1 + alpha * pagerank_score). " + "Set to 0.0 to disable the boost without removing the flag." + ), + ) + + graph_min_corpus_size: int = Field( + default=15, + gt=0, + description=( + "Minimum number of ingested papers for citation_pagerank to yield " + "meaningful rankings. Below this threshold a warning is logged and " + "uniform scores are returned." + ), + ) + @lru_cache(maxsize=1) def get_settings() -> Settings: diff --git a/codex/graph.py b/codex/graph.py new file mode 100644 index 0000000..1d1e01d --- /dev/null +++ b/codex/graph.py @@ -0,0 +1,169 @@ +"""Citation graph analytics (F-15). + +Builds an in-memory NetworkX DiGraph from the ``citations`` table and +provides PageRank, bibliographic coupling, co-citation, and dangling- +citation queries. The graph is ephemeral — rebuilt per call, < 1 s for +≤ 100 papers. + +Graceful degradation +-------------------- +* ``citation_pagerank`` returns uniform scores when the graph has fewer + than 5 nodes, and logs a warning. +* All functions accept an empty graph without raising. +""" + +from __future__ import annotations + +import logging +from typing import Any, cast + +import networkx as nx + +logger = logging.getLogger(__name__) + +_MIN_PAGERANK_NODES = 5 + + +def build_citation_graph(conn: Any) -> nx.DiGraph: + """Load the ``citations`` table into a directed graph. + + Nodes are paper IDs (strings). An edge ``citing → cited`` means + the citing paper references the cited paper. Both ingested papers + and dangling targets (cited but not yet ingested) appear as nodes. + + Parameters + ---------- + conn: + Open psycopg connection (dict-row factory assumed). + + Returns + ------- + nx.DiGraph with every (citing_id, cited_id) pair as an edge. + """ + rows = conn.execute("SELECT citing_id, cited_id FROM citations").fetchall() + g: nx.DiGraph = nx.DiGraph() + for row in rows: + g.add_edge(row["citing_id"], row["cited_id"]) + logger.debug( + "Built citation graph: %d nodes, %d edges", g.number_of_nodes(), g.number_of_edges() + ) + return g + + +def citation_pagerank(graph: nx.DiGraph, *, damping: float = 0.85) -> dict[str, float]: + """Compute PageRank over the citation graph. + + Parameters + ---------- + graph: + DiGraph from :func:`build_citation_graph`. + damping: + PageRank damping factor (default 0.85). + + Returns + ------- + ``{paper_id: score}`` dict. Scores sum to approximately 1.0. + + When ``graph`` has fewer than :data:`_MIN_PAGERANK_NODES` nodes, + returns a uniform distribution and logs a warning — the signal is + too thin to be meaningful. + """ + n = graph.number_of_nodes() + if n == 0: + return {} + if n < _MIN_PAGERANK_NODES: + logger.warning( + "citation_pagerank: only %d nodes — corpus too small for meaningful ranking " + "(need ≥ %d). Returning uniform scores.", + n, + _MIN_PAGERANK_NODES, + ) + uniform = 1.0 / n + return {node: uniform for node in graph.nodes()} + return cast(dict[str, float], nx.pagerank(graph, alpha=damping)) + + +def find_related(paper_id: str, graph: nx.DiGraph, *, min_shared: int = 2) -> list[str]: + """Bibliographic coupling: papers sharing ≥ ``min_shared`` references with ``paper_id``. + + Two papers are bibliographically coupled when they cite the same sources. + The more shared references, the more likely they treat related topics. + + Parameters + ---------- + paper_id: + Source paper whose references form the seed set. + graph: + DiGraph from :func:`build_citation_graph`. + min_shared: + Minimum number of shared references to be included. + + Returns + ------- + List of paper IDs ordered by shared-reference count DESC. + """ + if paper_id not in graph: + return [] + # papers cited by paper_id + references: set[str] = set(graph.successors(paper_id)) + if not references: + return [] + + shared: dict[str, int] = {} + for ref in references: + # other papers that also cite this reference + for co_citer in graph.predecessors(ref): + if co_citer == paper_id: + continue + shared[co_citer] = shared.get(co_citer, 0) + 1 + + return [ + pid for pid, count in sorted(shared.items(), key=lambda x: -x[1]) if count >= min_shared + ] + + +def find_co_cited(paper_id: str, graph: nx.DiGraph) -> list[tuple[str, int]]: + """Co-citation: papers frequently cited alongside ``paper_id``. + + A paper X is co-cited with ``paper_id`` when some third paper cites both. + + Returns + ------- + ``[(paper_id, count)]`` ordered by count DESC. + """ + if paper_id not in graph: + return [] + # papers that cite paper_id + citers: set[str] = set(graph.predecessors(paper_id)) + if not citers: + return [] + + co_cited: dict[str, int] = {} + for citer in citers: + for other in graph.successors(citer): + if other == paper_id: + continue + co_cited[other] = co_cited.get(other, 0) + 1 + + return sorted(co_cited.items(), key=lambda x: -x[1]) + + +def dangling_citations(graph: nx.DiGraph, known_ids: set[str]) -> list[str]: + """Return graph nodes not present in ``known_ids`` (cited but not yet ingested). + + Complements :func:`codex.discover.discovery_leads` with graph context: + the returned IDs are reachable in the citation graph but have no paper + record in the DB. + + Parameters + ---------- + graph: + DiGraph from :func:`build_citation_graph`. + known_ids: + Set of paper IDs that exist in the ``papers`` table. + + Returns + ------- + List of dangling paper IDs (unordered). + """ + return [node for node in graph.nodes() if node not in known_ids] diff --git a/tests/graph/__init__.py b/tests/graph/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/graph/test_cli.py b/tests/graph/test_cli.py new file mode 100644 index 0000000..b2460c0 --- /dev/null +++ b/tests/graph/test_cli.py @@ -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 diff --git a/tests/graph/test_graph.py b/tests/graph/test_graph.py new file mode 100644 index 0000000..707507e --- /dev/null +++ b/tests/graph/test_graph.py @@ -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"}