Safi
d14e8a72d1
In Python+TypeScript monorepos, the call and import resolvers match by label across language boundaries (AuthError -> Member), producing false positives that dominate 'Surprising Connections' due to cross-dir (+2) and cross-community (+1) bonuses. Expand the existing calls guard to also cover uses edges, and zero the cross-dir and cross-community bonuses for these pairs — not just conf_bonus. Leaves semantically_similar_to, EXTRACTED, and AMBIGUOUS edges unaffected. 5 new tests: calls suppressed, uses suppressed, semantically_similar_to preserved, same-language INFERRED preserved, cross-language EXTRACTED preserved.
325 lines
14 KiB
Python
325 lines
14 KiB
Python
"""Tests for analyze.py."""
|
|
import json
|
|
import networkx as nx
|
|
from pathlib import Path
|
|
from graphify.build import build_from_json
|
|
from graphify.cluster import cluster
|
|
from graphify.analyze import god_nodes, surprising_connections, _is_concept_node, graph_diff, _surprise_score, _file_category
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
def make_graph():
|
|
return build_from_json(json.loads((FIXTURES / "extraction.json").read_text()))
|
|
|
|
|
|
def test_god_nodes_returns_list():
|
|
G = make_graph()
|
|
result = god_nodes(G, top_n=3)
|
|
assert isinstance(result, list)
|
|
assert len(result) <= 3
|
|
|
|
|
|
def test_god_nodes_sorted_by_degree():
|
|
G = make_graph()
|
|
result = god_nodes(G, top_n=10)
|
|
degrees = [r["degree"] for r in result]
|
|
assert degrees == sorted(degrees, reverse=True)
|
|
|
|
|
|
def test_god_nodes_have_required_keys():
|
|
G = make_graph()
|
|
result = god_nodes(G, top_n=1)
|
|
assert "id" in result[0]
|
|
assert "label" in result[0]
|
|
assert "degree" in result[0]
|
|
|
|
|
|
def test_surprising_connections_cross_source_multi_file():
|
|
"""Multi-file graph: should find cross-file edges between real entities."""
|
|
G = make_graph()
|
|
communities = cluster(G)
|
|
surprises = surprising_connections(G, communities)
|
|
assert len(surprises) > 0
|
|
for s in surprises:
|
|
assert s["source_files"][0] != s["source_files"][1]
|
|
|
|
|
|
def test_surprising_connections_excludes_concept_nodes():
|
|
"""Concept nodes (empty source_file) must not appear in surprises."""
|
|
G = make_graph()
|
|
# Add a concept node with empty source_file
|
|
G.add_node("concept_x", label="Abstract Concept", file_type="document", source_file="")
|
|
G.add_edge("n_transformer", "concept_x", relation="relates_to",
|
|
confidence="INFERRED", source_file="", weight=0.5)
|
|
communities = cluster(G)
|
|
surprises = surprising_connections(G, communities)
|
|
labels = [s["source"] for s in surprises] + [s["target"] for s in surprises]
|
|
assert "Abstract Concept" not in labels
|
|
|
|
|
|
def test_surprising_connections_single_file_uses_community_bridges():
|
|
"""Single-file graph: should return cross-community edges, not empty list."""
|
|
G = nx.Graph()
|
|
# Build a graph with 2 clear communities + 1 bridge edge
|
|
for i in range(5):
|
|
G.add_node(f"a{i}", label=f"A{i}", file_type="code", source_file="single.py",
|
|
source_location=f"L{i}")
|
|
for i in range(5):
|
|
G.add_node(f"b{i}", label=f"B{i}", file_type="code", source_file="single.py",
|
|
source_location=f"L{i+10}")
|
|
# Dense intra-community edges
|
|
for i in range(4):
|
|
G.add_edge(f"a{i}", f"a{i+1}", relation="calls", confidence="EXTRACTED",
|
|
source_file="single.py", weight=1.0)
|
|
for i in range(4):
|
|
G.add_edge(f"b{i}", f"b{i+1}", relation="calls", confidence="EXTRACTED",
|
|
source_file="single.py", weight=1.0)
|
|
# One cross-community bridge
|
|
G.add_edge("a4", "b0", relation="references", confidence="INFERRED",
|
|
source_file="single.py", weight=0.5)
|
|
|
|
communities = cluster(G)
|
|
surprises = surprising_connections(G, communities)
|
|
# Should find at least the bridge edge
|
|
assert len(surprises) > 0
|
|
|
|
|
|
def test_surprising_connections_ambiguous_scores_higher_than_extracted():
|
|
"""AMBIGUOUS edge should score higher than an otherwise identical EXTRACTED edge."""
|
|
G = nx.Graph()
|
|
for nid, label, src in [
|
|
("a", "Alpha", "repo1/model.py"),
|
|
("b", "Beta", "repo2/train.py"),
|
|
("c", "Gamma", "repo1/data.py"),
|
|
("d", "Delta", "repo2/eval.py"),
|
|
]:
|
|
G.add_node(nid, label=label, source_file=src, file_type="code")
|
|
G.add_edge("a", "b", relation="calls", confidence="AMBIGUOUS", weight=1.0, source_file="repo1/model.py")
|
|
G.add_edge("c", "d", relation="calls", confidence="EXTRACTED", weight=1.0, source_file="repo1/data.py")
|
|
communities = {0: ["a", "c"], 1: ["b", "d"]}
|
|
nc = {"a": 0, "c": 0, "b": 1, "d": 1}
|
|
score_amb, _ = _surprise_score(G, "a", "b", G.edges["a", "b"], nc, "repo1/model.py", "repo2/train.py")
|
|
score_ext, _ = _surprise_score(G, "c", "d", G.edges["c", "d"], nc, "repo1/data.py", "repo2/eval.py")
|
|
assert score_amb > score_ext
|
|
|
|
|
|
def test_surprising_connections_cross_type_scores_higher():
|
|
"""Code↔paper edge should score higher than code↔code edge."""
|
|
G = nx.Graph()
|
|
for nid, label, src in [
|
|
("a", "Transformer", "code/model.py"),
|
|
("b", "FlashAttn", "papers/flash.pdf"),
|
|
("c", "Trainer", "code/train.py"),
|
|
("d", "Dataset", "code/data.py"),
|
|
]:
|
|
G.add_node(nid, label=label, source_file=src, file_type="code")
|
|
G.add_edge("a", "b", relation="references", confidence="EXTRACTED", weight=1.0, source_file="code/model.py")
|
|
G.add_edge("c", "d", relation="calls", confidence="EXTRACTED", weight=1.0, source_file="code/train.py")
|
|
nc = {"a": 0, "b": 1, "c": 0, "d": 0}
|
|
score_cross, reasons_cross = _surprise_score(G, "a", "b", G.edges["a", "b"], nc, "code/model.py", "papers/flash.pdf")
|
|
score_same, _ = _surprise_score(G, "c", "d", G.edges["c", "d"], nc, "code/train.py", "code/data.py")
|
|
assert score_cross > score_same
|
|
assert any("code" in r and "paper" in r for r in reasons_cross)
|
|
|
|
|
|
def _make_cross_lang_graph():
|
|
"""Helper: Python node in backend/, TypeScript node in frontend/, different communities."""
|
|
G = nx.Graph()
|
|
G.add_node("py_auth", label="AuthError", source_file="backend/auth.py", file_type="code")
|
|
G.add_node("ts_member", label="Member", source_file="frontend/types.ts", file_type="code")
|
|
G.add_node("py_a", label="ServiceA", source_file="backend/service.py", file_type="code")
|
|
G.add_node("py_b", label="ServiceB", source_file="backend/utils.py", file_type="code")
|
|
return G
|
|
|
|
|
|
def test_cross_language_inferred_calls_suppressed():
|
|
"""Cross-language INFERRED calls edge should score lower than same-language EXTRACTED."""
|
|
G = _make_cross_lang_graph()
|
|
G.add_edge("py_auth", "ts_member", relation="calls", confidence="INFERRED",
|
|
weight=0.8, source_file="backend/auth.py")
|
|
G.add_edge("py_a", "py_b", relation="calls", confidence="EXTRACTED",
|
|
weight=1.0, source_file="backend/service.py")
|
|
nc = {"py_auth": 0, "ts_member": 1, "py_a": 0, "py_b": 0}
|
|
score_cross, _ = _surprise_score(G, "py_auth", "ts_member",
|
|
G.edges["py_auth", "ts_member"], nc,
|
|
"backend/auth.py", "frontend/types.ts")
|
|
score_same, _ = _surprise_score(G, "py_a", "py_b",
|
|
G.edges["py_a", "py_b"], nc,
|
|
"backend/service.py", "backend/utils.py")
|
|
assert score_cross <= score_same
|
|
|
|
|
|
def test_cross_language_inferred_uses_suppressed():
|
|
"""Cross-language INFERRED uses edge (the exact rsl-siege-manager false positive) should be suppressed."""
|
|
G = _make_cross_lang_graph()
|
|
G.add_edge("py_auth", "ts_member", relation="uses", confidence="INFERRED",
|
|
weight=0.8, source_file="backend/auth.py")
|
|
G.add_edge("py_a", "py_b", relation="calls", confidence="EXTRACTED",
|
|
weight=1.0, source_file="backend/service.py")
|
|
nc = {"py_auth": 0, "ts_member": 1, "py_a": 0, "py_b": 0}
|
|
score_cross, _ = _surprise_score(G, "py_auth", "ts_member",
|
|
G.edges["py_auth", "ts_member"], nc,
|
|
"backend/auth.py", "frontend/types.ts")
|
|
score_same, _ = _surprise_score(G, "py_a", "py_b",
|
|
G.edges["py_a", "py_b"], nc,
|
|
"backend/service.py", "backend/utils.py")
|
|
assert score_cross <= score_same
|
|
|
|
|
|
def test_cross_language_semantically_similar_not_suppressed():
|
|
"""`semantically_similar_to` across languages is a genuine insight — must not be suppressed."""
|
|
G = _make_cross_lang_graph()
|
|
G.add_edge("py_auth", "ts_member", relation="semantically_similar_to",
|
|
confidence="INFERRED", weight=0.85, source_file="backend/auth.py")
|
|
G.add_edge("py_a", "py_b", relation="calls", confidence="EXTRACTED",
|
|
weight=1.0, source_file="backend/service.py")
|
|
nc = {"py_auth": 0, "ts_member": 1, "py_a": 0, "py_b": 0}
|
|
score_sem, _ = _surprise_score(G, "py_auth", "ts_member",
|
|
G.edges["py_auth", "ts_member"], nc,
|
|
"backend/auth.py", "frontend/types.ts")
|
|
score_same, _ = _surprise_score(G, "py_a", "py_b",
|
|
G.edges["py_a", "py_b"], nc,
|
|
"backend/service.py", "backend/utils.py")
|
|
assert score_sem > score_same
|
|
|
|
|
|
def test_same_language_inferred_calls_not_suppressed():
|
|
"""INFERRED calls within the same language family must not be affected."""
|
|
G = nx.Graph()
|
|
G.add_node("py_a", label="ModuleA", source_file="src/a.py", file_type="code")
|
|
G.add_node("py_b", label="ModuleB", source_file="src/b.py", file_type="code")
|
|
G.add_node("py_c", label="ModuleC", source_file="src/c.py", file_type="code")
|
|
G.add_node("py_d", label="ModuleD", source_file="src/d.py", file_type="code")
|
|
G.add_edge("py_a", "py_b", relation="calls", confidence="INFERRED",
|
|
weight=0.8, source_file="src/a.py")
|
|
G.add_edge("py_c", "py_d", relation="calls", confidence="EXTRACTED",
|
|
weight=1.0, source_file="src/c.py")
|
|
nc = {"py_a": 0, "py_b": 1, "py_c": 0, "py_d": 1}
|
|
score_inf, _ = _surprise_score(G, "py_a", "py_b", G.edges["py_a", "py_b"], nc,
|
|
"src/a.py", "src/b.py")
|
|
score_ext, _ = _surprise_score(G, "py_c", "py_d", G.edges["py_c", "py_d"], nc,
|
|
"src/c.py", "src/d.py")
|
|
assert score_inf > score_ext
|
|
|
|
|
|
def test_cross_language_extracted_calls_not_suppressed():
|
|
"""EXTRACTED cross-language edges are real structural facts — must not be penalised."""
|
|
G = _make_cross_lang_graph()
|
|
G.add_edge("py_auth", "ts_member", relation="calls", confidence="EXTRACTED",
|
|
weight=1.0, source_file="backend/auth.py")
|
|
nc = {"py_auth": 0, "ts_member": 1}
|
|
score, _ = _surprise_score(G, "py_auth", "ts_member",
|
|
G.edges["py_auth", "ts_member"], nc,
|
|
"backend/auth.py", "frontend/types.ts")
|
|
assert score >= 1
|
|
|
|
|
|
def test_surprising_connections_have_why_field():
|
|
G = make_graph()
|
|
communities = cluster(G)
|
|
for s in surprising_connections(G, communities):
|
|
assert "why" in s
|
|
assert isinstance(s["why"], str)
|
|
assert len(s["why"]) > 0
|
|
|
|
|
|
def test_file_category():
|
|
assert _file_category("model.py") == "code"
|
|
assert _file_category("flash.pdf") == "paper"
|
|
assert _file_category("diagram.png") == "image"
|
|
assert _file_category("notes.md") == "doc"
|
|
# Languages added in later releases — would misclassify as "doc" without detect.py import
|
|
assert _file_category("app.swift") == "code"
|
|
assert _file_category("plugin.lua") == "code"
|
|
assert _file_category("build.zig") == "code"
|
|
assert _file_category("deploy.ps1") == "code"
|
|
assert _file_category("server.ex") == "code"
|
|
assert _file_category("component.jsx") == "code"
|
|
assert _file_category("analysis.jl") == "code"
|
|
assert _file_category("view.m") == "code"
|
|
|
|
|
|
def test_is_concept_node_empty_source():
|
|
G = nx.Graph()
|
|
G.add_node("c1", source_file="")
|
|
assert _is_concept_node(G, "c1") is True
|
|
|
|
|
|
def test_is_concept_node_real_file():
|
|
G = nx.Graph()
|
|
G.add_node("n1", source_file="model.py")
|
|
assert _is_concept_node(G, "n1") is False
|
|
|
|
|
|
def test_surprising_connections_have_required_keys():
|
|
G = make_graph()
|
|
communities = cluster(G)
|
|
for s in surprising_connections(G, communities):
|
|
assert "source" in s
|
|
assert "target" in s
|
|
assert "source_files" in s
|
|
assert "confidence" in s
|
|
|
|
|
|
# --- graph_diff tests ---
|
|
|
|
def _make_simple_graph(nodes, edges):
|
|
"""Helper: build a small nx.Graph from node/edge specs."""
|
|
G = nx.Graph()
|
|
for node_id, label in nodes:
|
|
G.add_node(node_id, label=label, source_file="test.py")
|
|
for src, tgt, rel, conf in edges:
|
|
G.add_edge(src, tgt, relation=rel, confidence=conf)
|
|
return G
|
|
|
|
|
|
def test_graph_diff_new_nodes():
|
|
G_old = _make_simple_graph([("n1", "Alpha"), ("n2", "Beta")], [])
|
|
G_new = _make_simple_graph([("n1", "Alpha"), ("n2", "Beta"), ("n3", "Gamma")], [])
|
|
diff = graph_diff(G_old, G_new)
|
|
assert len(diff["new_nodes"]) == 1
|
|
assert diff["new_nodes"][0]["id"] == "n3"
|
|
assert diff["new_nodes"][0]["label"] == "Gamma"
|
|
assert diff["removed_nodes"] == []
|
|
assert "1 new node" in diff["summary"]
|
|
|
|
|
|
def test_graph_diff_removed_nodes():
|
|
G_old = _make_simple_graph([("n1", "Alpha"), ("n2", "Beta"), ("n3", "Gamma")], [])
|
|
G_new = _make_simple_graph([("n1", "Alpha"), ("n2", "Beta")], [])
|
|
diff = graph_diff(G_old, G_new)
|
|
assert diff["new_nodes"] == []
|
|
assert len(diff["removed_nodes"]) == 1
|
|
assert diff["removed_nodes"][0]["id"] == "n3"
|
|
assert "removed" in diff["summary"]
|
|
|
|
|
|
def test_graph_diff_new_edges():
|
|
nodes = [("n1", "Alpha"), ("n2", "Beta"), ("n3", "Gamma")]
|
|
G_old = _make_simple_graph(nodes, [("n1", "n2", "calls", "EXTRACTED")])
|
|
G_new = _make_simple_graph(
|
|
nodes,
|
|
[("n1", "n2", "calls", "EXTRACTED"), ("n2", "n3", "uses", "INFERRED")],
|
|
)
|
|
diff = graph_diff(G_old, G_new)
|
|
assert len(diff["new_edges"]) == 1
|
|
new_edge = diff["new_edges"][0]
|
|
assert new_edge["relation"] == "uses"
|
|
assert new_edge["confidence"] == "INFERRED"
|
|
assert diff["removed_edges"] == []
|
|
assert "new edge" in diff["summary"]
|
|
|
|
|
|
def test_graph_diff_empty_diff():
|
|
nodes = [("n1", "Alpha"), ("n2", "Beta")]
|
|
edges = [("n1", "n2", "calls", "EXTRACTED")]
|
|
G_old = _make_simple_graph(nodes, edges)
|
|
G_new = _make_simple_graph(nodes, edges)
|
|
diff = graph_diff(G_old, G_new)
|
|
assert diff["new_nodes"] == []
|
|
assert diff["removed_nodes"] == []
|
|
assert diff["new_edges"] == []
|
|
assert diff["removed_edges"] == []
|
|
assert diff["summary"] == "no changes"
|