#!/usr/bin/env python3
"""
Integration Test Runner - Agent Orchestration Platform
Simple test runner for integration testing without complex dependencies.
Focuses on critical functionality validation and system integration.
Author: Adder_1 | Created: 2025-06-26
"""
import asyncio
import sys
import traceback
from pathlib import Path
from typing import Dict, Any, List
import tempfile
import json
# Add project root to path for imports
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
class IntegrationTestRunner:
"""Simple integration test runner for system validation."""
def __init__(self):
self.test_results: Dict[str, Any] = {
"total_tests": 0,
"passed": 0,
"failed": 0,
"errors": 0,
"test_details": []
}
async def run_all_tests(self):
"""Run all integration tests."""
print("🚀 Starting Integration Test Suite")
print("=" * 50)
# Test categories
test_categories = [
("Import Tests", self.test_imports),
("Type System Tests", self.test_type_system),
("Basic Functionality Tests", self.test_basic_functionality),
("MCP Tools Structure Tests", self.test_mcp_tools_structure),
("Configuration Tests", self.test_configuration)
]
for category_name, test_func in test_categories:
print(f"\n📋 Running {category_name}")
print("-" * 30)
await self.run_test_category(category_name, test_func)
# Print summary
self.print_summary()
return self.test_results["failed"] == 0 and self.test_results["errors"] == 0
async def run_test_category(self, category_name: str, test_func):
"""Run a category of tests."""
try:
await test_func()
except Exception as e:
self.record_error(category_name, str(e), traceback.format_exc())
def record_test(self, test_name: str, passed: bool, message: str = "", error: str = ""):
"""Record test result."""
self.test_results["total_tests"] += 1
if error:
self.test_results["errors"] += 1
status = "ERROR"
icon = "❌"
elif passed:
self.test_results["passed"] += 1
status = "PASS"
icon = "✅"
else:
self.test_results["failed"] += 1
status = "FAIL"
icon = "❌"
print(f" {icon} {test_name}: {status}")
if message:
print(f" {message}")
if error:
print(f" Error: {error}")
self.test_results["test_details"].append({
"name": test_name,
"status": status,
"message": message,
"error": error
})
def record_error(self, test_name: str, message: str, traceback_str: str):
"""Record test error."""
self.record_test(test_name, False, message, traceback_str)
async def test_imports(self):
"""Test that all critical modules can be imported."""
# Test type imports
try:
from src.models.ids import AgentId, SessionId
from src.models.agent import AgentState, AgentStatus
from src.models.session import SessionState
from src.models.security import SecurityLevel, SecurityContext
self.record_test("Core Types Import", True, "All core types imported successfully")
except Exception as e:
self.record_test("Core Types Import", False, error=str(e))
# Test utils imports
try:
from src.utils.work_preservation import WorkPreservationHandler
from src.utils.cascade_deletion import CascadeDeletionOrchestrator
self.record_test("Utils Import", True, "Utility modules imported successfully")
except Exception as e:
self.record_test("Utils Import", False, error=str(e))
# Test validators import
try:
from src.validators.session_deletion import SessionDeletionValidator
self.record_test("Validators Import", True, "Validator modules imported successfully")
except Exception as e:
self.record_test("Validators Import", False, error=str(e))
async def test_type_system(self):
"""Test type system functionality."""
try:
from src.models.ids import AgentId, SessionId
# Test ID creation
agent_id = AgentId("test_agent_123")
session_id = SessionId("test_session_456")
self.record_test("ID Creation", True, f"Created AgentId: {agent_id}, SessionId: {session_id}")
except Exception as e:
self.record_test("ID Creation", False, error=str(e))
try:
from src.models.security import SecurityLevel, SecurityContext
# Test security types
sec_level = SecurityLevel.HIGH
sec_context = SecurityContext()
self.record_test("Security Types", True, f"Created SecurityLevel: {sec_level}")
except Exception as e:
self.record_test("Security Types", False, error=str(e))
async def test_basic_functionality(self):
"""Test basic functionality without external dependencies."""
# Test work preservation handler initialization
try:
from src.utils.work_preservation import WorkPreservationHandler
with tempfile.TemporaryDirectory() as temp_dir:
handler = WorkPreservationHandler(preservation_root=Path(temp_dir))
# Check that preservation directory is created
if handler.preservation_root.exists():
self.record_test("Work Preservation Handler", True, "Handler initialized and directory created")
else:
self.record_test("Work Preservation Handler", False, "Preservation directory not created")
except Exception as e:
self.record_test("Work Preservation Handler", False, error=str(e))
# Test cascade deletion orchestrator
try:
from src.utils.cascade_deletion import CascadeDeletionOrchestrator
orchestrator = CascadeDeletionOrchestrator()
self.record_test("Cascade Deletion Orchestrator", True, "Orchestrator initialized successfully")
except Exception as e:
self.record_test("Cascade Deletion Orchestrator", False, error=str(e))
async def test_mcp_tools_structure(self):
"""Test MCP tools structure without full initialization."""
# Test that MCP tools file exists and has expected structure
try:
mcp_tools_path = Path(__file__).parent.parent / "src" / "interfaces" / "mcp_tools.py"
if not mcp_tools_path.exists():
self.record_test("MCP Tools File", False, "mcp_tools.py file not found")
return
# Read file and check for key methods
content = mcp_tools_path.read_text()
expected_methods = [
"create_agent",
"delete_agent",
"create_session",
"get_session_status",
"delete_session",
"send_message_to_agent"
]
missing_methods = []
for method in expected_methods:
if f"async def {method}(" not in content:
missing_methods.append(method)
if missing_methods:
self.record_test("MCP Tools Methods", False, f"Missing methods: {missing_methods}")
else:
self.record_test("MCP Tools Methods", True, f"All {len(expected_methods)} MCP tools methods found")
except Exception as e:
self.record_test("MCP Tools Structure", False, error=str(e))
async def test_configuration(self):
"""Test configuration and setup."""
# Test pyproject.toml exists
try:
pyproject_path = Path(__file__).parent.parent / "pyproject.toml"
if pyproject_path.exists():
self.record_test("Configuration File", True, "pyproject.toml found")
# Check for key dependencies
content = pyproject_path.read_text()
key_deps = ["fastmcp", "hypothesis", "pytest"]
found_deps = []
for dep in key_deps:
if dep in content:
found_deps.append(dep)
if len(found_deps) == len(key_deps):
self.record_test("Dependencies", True, f"Found all key dependencies: {found_deps}")
else:
missing = set(key_deps) - set(found_deps)
self.record_test("Dependencies", False, f"Missing dependencies: {missing}")
else:
self.record_test("Configuration File", False, "pyproject.toml not found")
except Exception as e:
self.record_test("Configuration", False, error=str(e))
def print_summary(self):
"""Print test summary."""
results = self.test_results
print("\n" + "=" * 50)
print("🏁 Integration Test Summary")
print("=" * 50)
print(f"📊 Total Tests: {results['total_tests']}")
print(f"✅ Passed: {results['passed']}")
print(f"❌ Failed: {results['failed']}")
print(f"💥 Errors: {results['errors']}")
success_rate = (results['passed'] / results['total_tests'] * 100) if results['total_tests'] > 0 else 0
print(f"📈 Success Rate: {success_rate:.1f}%")
if results['failed'] > 0 or results['errors'] > 0:
print("\n🔍 Failed/Error Tests:")
for test in results['test_details']:
if test['status'] in ['FAIL', 'ERROR']:
print(f" - {test['name']}: {test['status']}")
if test['message']:
print(f" Message: {test['message']}")
# Overall status
if results['failed'] == 0 and results['errors'] == 0:
print("\n🎉 All tests passed!")
return True
else:
print(f"\n⚠️ {results['failed'] + results['errors']} tests failed")
return False
async def main():
"""Main test runner entry point."""
runner = IntegrationTestRunner()
success = await runner.run_all_tests()
# Save results to file
results_path = Path(__file__).parent / "integration_test_results.json"
with open(results_path, 'w') as f:
json.dump(runner.test_results, f, indent=2)
print(f"\n📁 Results saved to: {results_path}")
# Exit with appropriate code
sys.exit(0 if success else 1)
if __name__ == "__main__":
asyncio.run(main())
