test: add end-to-end pipeline integration test

Covers detect → extract → build → cluster → analyze → report → export
using existing fixtures. AST-only (no LLM calls), catches regressions
in how modules connect, not just individual module behaviour.

tests/test_pipeline.py

@@ -0,0 +1,158 @@
+"""
+End-to-end pipeline test: detect → extract → build → cluster → analyze → report → export.
+Uses the existing test fixtures (code + markdown). No LLM calls - AST extraction only.
+Catches regressions in how modules connect, not just individual module behaviour.
+"""
+import json
+import tempfile
+from pathlib import Path
+
+import pytest
+
+from graphify.detect import detect
+from graphify.extract import collect_files, extract
+from graphify.build import build_from_json
+from graphify.cluster import cluster, score_all
+from graphify.analyze import god_nodes, surprising_connections, suggest_questions
+from graphify.report import generate
+from graphify.export import to_json, to_html, to_obsidian
+
+FIXTURES = Path(__file__).parent / "fixtures"
+
+
+def run_pipeline(tmp_path: Path) -> dict:
+    """Run the full pipeline on the fixtures directory. Returns a dict of outputs."""
+    # Step 1: detect
+    detection = detect(FIXTURES)
+    assert detection["total_files"] > 0
+    # fixtures corpus is intentionally small (< 5k words), so needs_graph may be False
+    assert "files" in detection
+
+    # Step 2: extract (AST only - no LLM)
+    code_files = [Path(f) for f in detection["files"].get("code", [])]
+    assert len(code_files) > 0
+    extraction = extract(code_files)
+    assert len(extraction["nodes"]) > 0
+    assert len(extraction["edges"]) > 0
+
+    # Step 3: build
+    G = build_from_json(extraction)
+    assert G.number_of_nodes() > 0
+    assert G.number_of_edges() > 0
+
+    # Step 4: cluster
+    communities = cluster(G)
+    assert len(communities) > 0
+    cohesion = score_all(G, communities)
+    assert len(cohesion) == len(communities)
+    for score in cohesion.values():
+        assert 0.0 <= score <= 1.0
+
+    # Step 5: analyze
+    gods = god_nodes(G)
+    assert len(gods) > 0
+    assert all("id" in g and "edges" in g for g in gods)
+
+    surprises = surprising_connections(G, communities)
+    assert isinstance(surprises, list)
+
+    labels = {cid: f"Group {cid}" for cid in communities}
+    questions = suggest_questions(G, communities, labels)
+    assert isinstance(questions, list)
+
+    # Step 6: report
+    tokens = {"input": 0, "output": 0}
+    report = generate(G, communities, cohesion, labels, gods, surprises, detection, tokens, str(FIXTURES), suggested_questions=questions)
+    assert "God Nodes" in report
+    assert "Communities" in report
+    assert len(report) > 100
+
+    # Step 7: export - JSON
+    json_path = tmp_path / "graph.json"
+    to_json(G, communities, str(json_path))
+    assert json_path.exists()
+    data = json.loads(json_path.read_text())
+    assert "nodes" in data and "links" in data
+    assert all("community" in n for n in data["nodes"])
+
+    # Step 8: export - HTML
+    html_path = tmp_path / "graph.html"
+    to_html(G, communities, str(html_path), community_labels=labels)
+    assert html_path.exists()
+    html = html_path.read_text()
+    assert "vis-network" in html
+    assert "RAW_NODES" in html
+
+    # Step 9: export - Obsidian vault
+    vault_path = tmp_path / "obsidian"
+    n_notes = to_obsidian(G, communities, str(vault_path), community_labels=labels, cohesion=cohesion)
+    assert n_notes > 0
+    assert (vault_path / ".obsidian" / "graph.json").exists()
+    md_files = list(vault_path.glob("*.md"))
+    assert len(md_files) > 0
+
+    return {
+        "detection": detection,
+        "extraction": extraction,
+        "graph": G,
+        "communities": communities,
+        "cohesion": cohesion,
+        "gods": gods,
+        "surprises": surprises,
+        "questions": questions,
+        "report": report,
+    }
+
+
+def test_pipeline_runs_end_to_end(tmp_path):
+    result = run_pipeline(tmp_path)
+    assert result["graph"].number_of_nodes() > 0
+
+
+def test_pipeline_graph_has_edges(tmp_path):
+    result = run_pipeline(tmp_path)
+    assert result["graph"].number_of_edges() > 0
+
+
+def test_pipeline_all_nodes_have_community(tmp_path):
+    result = run_pipeline(tmp_path)
+    G = result["graph"]
+    communities = result["communities"]
+    all_community_nodes = {n for nodes in communities.values() for n in nodes}
+    for node in G.nodes():
+        assert node in all_community_nodes, f"Node {node!r} has no community"
+
+
+def test_pipeline_report_mentions_top_god_node(tmp_path):
+    result = run_pipeline(tmp_path)
+    top_god = result["gods"][0]["label"]
+    assert top_god in result["report"]
+
+
+def test_pipeline_detection_finds_code_and_docs(tmp_path):
+    result = run_pipeline(tmp_path)
+    assert len(result["detection"]["files"].get("code", [])) > 0
+    assert len(result["detection"]["files"].get("document", [])) > 0
+
+
+def test_pipeline_incremental_update(tmp_path):
+    """Second run on unchanged corpus should produce identical node/edge counts."""
+    result1 = run_pipeline(tmp_path)
+    result2 = run_pipeline(tmp_path)
+    assert result1["graph"].number_of_nodes() == result2["graph"].number_of_nodes()
+    assert result1["graph"].number_of_edges() == result2["graph"].number_of_edges()
+
+
+def test_pipeline_extraction_confidence_labels(tmp_path):
+    result = run_pipeline(tmp_path)
+    extraction = result["extraction"]
+    valid = {"EXTRACTED", "INFERRED", "AMBIGUOUS"}
+    for edge in extraction["edges"]:
+        assert edge["confidence"] in valid, f"Invalid confidence: {edge['confidence']}"
+
+
+def test_pipeline_no_self_loops(tmp_path):
+    result = run_pipeline(tmp_path)
+    G = result["graph"]
+    for u, v in G.edges():
+        assert u != v, f"Self-loop found on node {u!r}"