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