664600fdc7
build_citation_graph now does: SELECT c.citing_id, COALESCE(p.id, c.cited_id) AS cited_id FROM citations c LEFT JOIN papers p ON p.openalex_id = c.cited_id Before: cited_id (OpenAlex IDs) never matched papers.id (DOIs) → 13 cross-citations between ingested papers were invisible in graph. After: in-KB papers use their canonical DOI key; dangling references keep their raw OpenAlex ID unchanged. Test: test_cross_citation_uses_doi_when_in_kb (42 passed). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
290 lines
9.4 KiB
Python
290 lines
9.4 KiB
Python
"""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()
|
|
|
|
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"}
|