#!/usr/bin/env python3 """ ADR-151 Context Graph Integration Tests (Phase 6: CP-43)
Integration tests for the context graph system including:
- ContextGraphBuilder with real database
- /cxq --graph integration
- MCP tools integration
- Token budget management
- Persistence and retrieval
Run: python3 scripts/context_graph/test_integration.py python3 scripts/context_graph/test_integration.py -v # verbose
Created: 2026-02-03 Author: Claude (Opus 4.5) Track: J (Memory Intelligence) Task: J.25.7.1 (CP-43) """
import json import os import sqlite3 import subprocess import sys import tempfile import unittest from pathlib import Path from typing import Dict, Any
Set up path for imports
_script_dir = Path(file).resolve().parent _coditect_root = _script_dir.parent.parent if str(_coditect_root) not in sys.path: sys.path.insert(0, str(_coditect_root))
class TestContextGraphBuilder(unittest.TestCase): """Integration tests for ContextGraphBuilder."""
@classmethod
def setUpClass(cls):
"""Set up test fixtures."""
from scripts.context_graph import ContextGraphBuilder
from scripts.core.paths import get_org_db_path, get_sessions_db_path
cls.org_db_path = get_org_db_path()
cls.sessions_db_path = get_sessions_db_path()
# Skip all tests if databases don't exist
if not cls.org_db_path.exists():
raise unittest.SkipTest(f"org.db not found at {cls.org_db_path}")
def test_build_with_semantic_strategy(self):
"""Test building context graph with semantic seed selection."""
from scripts.context_graph import ContextGraphBuilder
with ContextGraphBuilder() as builder:
graph = builder.build(
task_description="find decisions about database architecture",
seed_strategy="semantic",
token_budget=4000,
max_depth=3,
persist=False,
)
# Verify graph structure
self.assertIsNotNone(graph)
self.assertGreater(graph.node_count, 0)
self.assertLessEqual(graph.tokens_estimated, 4000)
# Verify stats
self.assertIn("seed_count", builder.last_build_stats)
self.assertIn("tokens_by_type", builder.last_build_stats)
def test_build_with_policy_first_strategy(self):
"""Test building context graph with policy_first strategy."""
from scripts.context_graph import ContextGraphBuilder
with ContextGraphBuilder() as builder:
graph = builder.build(
task_description="find ADR decisions",
seed_strategy="policy_first",
token_budget=2000,
persist=False,
)
self.assertIsNotNone(graph)
# Policy-first should prioritize policy/decision/ADR nodes
governance_nodes = [n for n in graph.nodes.values() if n.node_type in ("policy", "decision", "adr")]
self.assertGreater(len(governance_nodes), 0, "Should find policy/decision/ADR nodes")
def test_token_budget_enforcement(self):
"""Test that token budget is enforced via pruning."""
from scripts.context_graph import ContextGraphBuilder
with ContextGraphBuilder() as builder:
# Use a reasonable budget that can be achieved within min_nodes constraint
# min_nodes=5 means we'll keep at least 5 nodes regardless of budget
graph = builder.build(
task_description="find all components",
seed_strategy="semantic",
token_budget=2000, # Reasonable budget for min 5 nodes
max_nodes=100,
persist=False,
)
# Verify budget enforcement or hit min_nodes constraint
# Either tokens <= budget, or we hit the minimum node count (5)
min_nodes_constraint = 5
budget_met = graph.tokens_estimated <= 2000
min_nodes_hit = graph.node_count <= min_nodes_constraint
self.assertTrue(
budget_met or min_nodes_hit,
f"Expected tokens <= 2000 or node_count <= {min_nodes_constraint}, "
f"got tokens={graph.tokens_estimated}, nodes={graph.node_count}"
)
self.assertIn("budget_utilization_pct", builder.last_build_stats)
def test_token_breakdown_by_type(self):
"""Test token breakdown by node type reporting."""
from scripts.context_graph import ContextGraphBuilder
with ContextGraphBuilder() as builder:
graph = builder.build(
task_description="find error solutions",
token_budget=3000,
persist=False,
)
tokens_by_type = builder._get_tokens_by_type(graph)
# Verify structure
for node_type, stats in tokens_by_type.items():
self.assertIn("count", stats)
self.assertIn("tokens", stats)
self.assertIn("avg_tokens", stats)
def test_token_report_generation(self):
"""Test human-readable token report generation."""
from scripts.context_graph import ContextGraphBuilder
with ContextGraphBuilder() as builder:
graph = builder.build(
task_description="find security patterns",
token_budget=2000,
persist=False,
)
report = builder.get_token_report(graph)
self.assertIn("TOKEN BUDGET REPORT", report)
self.assertIn("Breakdown by Node Type", report)
self.assertIn("TOTAL", report)
def test_graph_persistence_and_load(self):
"""Test persisting and loading context graph."""
from scripts.context_graph import ContextGraphBuilder
with ContextGraphBuilder() as builder:
# Build and persist
graph = builder.build(
task_description="test persistence graph",
token_budget=1000,
persist=True,
)
graph_id = graph.id
# Load it back
loaded_graph = builder.load_graph(graph_id)
self.assertIsNotNone(loaded_graph)
self.assertEqual(loaded_graph.id, graph_id)
self.assertEqual(loaded_graph.task_description, graph.task_description)
self.assertEqual(loaded_graph.node_count, graph.node_count)
def test_serialization_formats(self):
"""Test different serialization formats."""
from scripts.context_graph import ContextGraphBuilder
with ContextGraphBuilder() as builder:
graph = builder.build(
task_description="test serialization",
token_budget=1000,
persist=False,
)
# Markdown format
md_output = builder.serialize_for_context(graph, format="markdown")
self.assertIn("# Context Graph", md_output)
# JSON format
json_output = builder.serialize_for_context(graph, format="json")
parsed = json.loads(json_output)
self.assertIn("nodes", parsed)
# Text format
text_output = builder.serialize_for_context(graph, format="text")
self.assertIn("Context Graph:", text_output)
def test_knowledge_graph_stats(self):
"""Test getting knowledge graph statistics."""
from scripts.context_graph import ContextGraphBuilder
with ContextGraphBuilder() as builder:
stats = builder.get_stats()
self.assertIn("total_nodes", stats)
self.assertIn("total_edges", stats)
self.assertIn("nodes_by_type", stats)
self.assertIn("edges_by_type", stats)
# Should have nodes in knowledge graph
self.assertGreater(stats["total_nodes"], 0)
class TestCxqGraphIntegration(unittest.TestCase): """Integration tests for /cxq --graph command."""
def test_cxq_graph_stats(self):
"""Test /cxq --graph-stats command."""
script_path = _coditect_root / "scripts" / "context-query.py"
result = subprocess.run(
[sys.executable, str(script_path), "--graph-stats", "--json"],
capture_output=True,
text=True,
cwd=str(_coditect_root),
)
self.assertEqual(result.returncode, 0, f"Failed: {result.stderr}")
# Parse JSON output
output = json.loads(result.stdout)
self.assertIn("total_nodes", output)
self.assertIn("edges_by_type", output)
def test_cxq_graph_build(self):
"""Test /cxq --graph command to build context graph."""
script_path = _coditect_root / "scripts" / "context-query.py"
result = subprocess.run(
[
sys.executable, str(script_path),
"--graph", "find security decisions",
"--graph-format", "text",
"--graph-budget", "1000",
],
capture_output=True,
text=True,
cwd=str(_coditect_root),
)
self.assertEqual(result.returncode, 0, f"Failed: {result.stderr}")
self.assertIn("Context Graph:", result.stdout)
self.assertIn("BUILD STATISTICS:", result.stdout)
def test_cxq_graph_json_output(self):
"""Test /cxq --graph with JSON output."""
script_path = _coditect_root / "scripts" / "context-query.py"
result = subprocess.run(
[
sys.executable, str(script_path),
"--graph", "find error handling",
"--json",
],
capture_output=True,
text=True,
cwd=str(_coditect_root),
)
self.assertEqual(result.returncode, 0, f"Failed: {result.stderr}")
# Should be valid JSON
output = json.loads(result.stdout)
self.assertIn("nodes", output)
class TestMcpContextGraphServer(unittest.TestCase): """Integration tests for MCP context graph server."""
def test_server_stats_command(self):
"""Test MCP server stats CLI command."""
server_path = _coditect_root / "tools" / "mcp-context-graph" / "server.py"
result = subprocess.run(
[sys.executable, str(server_path), "stats"],
capture_output=True,
text=True,
cwd=str(_coditect_root),
)
self.assertEqual(result.returncode, 0, f"Failed: {result.stderr}")
output = json.loads(result.stdout)
self.assertIn("total_nodes", output)
self.assertIn("summary", output)
def test_server_build_command(self):
"""Test MCP server build CLI command."""
server_path = _coditect_root / "tools" / "mcp-context-graph" / "server.py"
result = subprocess.run(
[
sys.executable, str(server_path),
"build", "find database decisions",
"--budget", "1000",
"--format", "text",
"--no-persist",
],
capture_output=True,
text=True,
cwd=str(_coditect_root),
)
self.assertEqual(result.returncode, 0, f"Failed: {result.stderr}")
self.assertIn("Context Graph:", result.stdout)
def test_server_list_command(self):
"""Test MCP server list CLI command."""
server_path = _coditect_root / "tools" / "mcp-context-graph" / "server.py"
result = subprocess.run(
[sys.executable, str(server_path), "list", "--limit", "5"],
capture_output=True,
text=True,
cwd=str(_coditect_root),
)
self.assertEqual(result.returncode, 0, f"Failed: {result.stderr}")
self.assertIn("Context Graphs", result.stdout)
class TestPruningStrategies(unittest.TestCase): """Integration tests for pruning strategies."""
def test_prune_by_token_budget(self):
"""Test token budget pruning."""
from scripts.context_graph import ContextGraphBuilder
from scripts.context_graph.pruning import prune_by_token_budget
with ContextGraphBuilder() as builder:
# Build a large graph
graph = builder.build(
task_description="find all components",
token_budget=10000, # Large budget initially
max_nodes=100,
persist=False,
)
if graph.tokens_estimated <= 2000:
self.skipTest("Graph too small for pruning test")
original_tokens = graph.tokens_estimated
original_nodes = graph.node_count
# Prune to smaller budget (reasonable for min_nodes=5)
pruned = prune_by_token_budget(graph, token_budget=2000)
# Verify either budget met or hit min_nodes constraint
min_nodes_constraint = 5
budget_met = pruned.tokens_estimated <= 2000
min_nodes_hit = pruned.node_count <= min_nodes_constraint
self.assertTrue(
budget_met or min_nodes_hit,
f"Expected tokens <= 2000 or node_count <= {min_nodes_constraint}, "
f"got tokens={pruned.tokens_estimated}, nodes={pruned.node_count}"
)
# Verify pruning occurred (node count reduced or stayed at min)
self.assertLessEqual(pruned.node_count, original_nodes)
def test_prune_by_relevance(self):
"""Test relevance threshold pruning."""
from scripts.context_graph import ContextGraphBuilder
from scripts.context_graph.pruning import prune_by_relevance_threshold
with ContextGraphBuilder() as builder:
graph = builder.build(
task_description="find decisions",
token_budget=5000,
persist=False,
)
original_count = graph.node_count
# Prune with high threshold
pruned = prune_by_relevance_threshold(graph, threshold=0.5)
# Should remove low-relevance nodes
self.assertLessEqual(pruned.node_count, original_count)
class TestEdgeCases(unittest.TestCase): """Tests for edge cases and error handling."""
def test_empty_task_description(self):
"""Test handling of empty task description."""
from scripts.context_graph import ContextGraphBuilder
with ContextGraphBuilder() as builder:
graph = builder.build(
task_description="",
token_budget=1000,
persist=False,
)
# Should still return a graph (possibly empty)
self.assertIsNotNone(graph)
def test_very_small_budget(self):
"""Test handling of very small token budget."""
from scripts.context_graph import ContextGraphBuilder
with ContextGraphBuilder() as builder:
graph = builder.build(
task_description="find decisions",
token_budget=100, # Very small - may hit min_nodes constraint
persist=False,
)
# With very small budget, we expect either:
# 1. Budget is met (tokens <= 100), OR
# 2. min_nodes constraint is hit (we keep at least 5 nodes)
# The pruning algorithm won't prune below min_nodes=5
min_nodes_constraint = 5
budget_met = graph.tokens_estimated <= 100
min_nodes_hit = graph.node_count <= min_nodes_constraint
self.assertTrue(
budget_met or min_nodes_hit,
f"Expected tokens <= 100 or node_count <= {min_nodes_constraint}, "
f"got tokens={graph.tokens_estimated}, nodes={graph.node_count}"
)
# Verify we got a valid graph
self.assertIsNotNone(graph)
def test_nonexistent_graph_id(self):
"""Test loading a non-existent graph."""
from scripts.context_graph import ContextGraphBuilder
with ContextGraphBuilder() as builder:
graph = builder.load_graph("nonexistent:graph:id")
self.assertIsNone(graph)
def run_tests(verbosity: int = 2) -> bool: """ Run all integration tests.
Args:
verbosity: Test output verbosity (0-2)
Returns:
True if all tests passed
"""
loader = unittest.TestLoader()
suite = unittest.TestSuite()
# Add all test classes
suite.addTests(loader.loadTestsFromTestCase(TestContextGraphBuilder))
suite.addTests(loader.loadTestsFromTestCase(TestCxqGraphIntegration))
suite.addTests(loader.loadTestsFromTestCase(TestMcpContextGraphServer))
suite.addTests(loader.loadTestsFromTestCase(TestPruningStrategies))
suite.addTests(loader.loadTestsFromTestCase(TestEdgeCases))
runner = unittest.TextTestRunner(verbosity=verbosity)
result = runner.run(suite)
return result.wasSuccessful()
if name == "main": import argparse
parser = argparse.ArgumentParser(
description="ADR-151 Context Graph Integration Tests"
)
parser.add_argument(
"-v", "--verbose",
action="store_true",
help="Verbose test output"
)
args = parser.parse_args()
verbosity = 2 if args.verbose else 1
success = run_tests(verbosity)
sys.exit(0 if success else 1)