python-sdk/pkg/test_integration.py

#!/usr/bin/env python3
"""Integration test to verify all Hanzo packages work together."""

import sys
import asyncio
from pathlib import Path

# Add all packages to path
packages = [
    ("hanzo-mcp", ""),
    ("hanzo-agents", ""),
    ("hanzo-network", "src"),
    ("hanzo-memory", "src")
]
for pkg, subdir in packages:
    path = Path(__file__).parent / pkg
    if subdir:
        path = path / subdir
    sys.path.insert(0, str(path))

async def test_integration():
    """Test that all packages can be imported and used together."""
    print("🧪 Running Hanzo Python SDK Integration Tests")
    print("=" * 60)
    
    # Test 1: Import all packages
    print("\n1. Testing package imports...")
    try:
        import hanzo_mcp
        print(f"✅ hanzo-mcp imported successfully (v{hanzo_mcp.__version__})")
        
        import hanzo_agents
        print(f"✅ hanzo-agents imported successfully (v{hanzo_agents.__version__})")
        
        import hanzo_network
        print(f"✅ hanzo-network imported successfully (v{hanzo_network.__version__})")
        
        try:
            import hanzo_memory
            print(f"✅ hanzo-memory imported successfully (v{hanzo_memory.__version__})")
        except ImportError:
            print("⚠️  hanzo-memory has additional dependencies (install with pip install hanzo-memory[all])")
    except ImportError as e:
        print(f"❌ Import failed: {e}")
        return False
    
    # Test 2: Test MCP tools
    print("\n2. Testing MCP tools...")
    try:
        from hanzo_mcp.server import HanzoMCPServer
        from hanzo_mcp.tools import TOOL_REGISTRY
        
        print(f"✅ MCP has {len(TOOL_REGISTRY)} tools available")
        
        # Test a simple tool
        from hanzo_mcp.tools.filesystem import fs_tools
        read_tool = next(t for t in fs_tools.TOOLS if t.name == "read")
        if read_tool:
            print("✅ MCP read tool is available")
        else:
            print("⚠️  MCP read tool not found")
    except Exception as e:
        print(f"⚠️  MCP tool test skipped: {e}")
    
    # Test 3: Test agent creation
    print("\n3. Testing agent creation...")
    try:
        from hanzo_agents import Agent, State, Network
        from dataclasses import dataclass
        
        @dataclass
        class TestState(State):
            test: str = "initial"
            done: bool = False
        
        class TestAgent(Agent[TestState]):
            name = "test_agent"
            description = "Test agent"
            
            async def run(self, state, history, network):
                state.test = "modified"
                state.done = True
                return {"content": "Test completed"}
        
        # Create network
        def test_router(network, call_count, last_result, agents):
            if network.state.done:
                return None
            return TestAgent
        
        network = Network(
            state=TestState(),
            agents=[TestAgent],
            router=test_router
        )
        
        print("✅ Agent and Network created successfully")
    except Exception as e:
        print(f"❌ Agent test failed: {e}")
        return False
    
    # Test 4: Test network components
    print("\n4. Testing network components...")
    try:
        from hanzo_network import create_agent, create_network, NetworkState
        
        agent = create_agent(
            name="network_test",
            description="Test agent for network",
            instructions="Test agent for network"
        )
        
        print("✅ Network agent created successfully")
    except Exception as e:
        print(f"❌ Network test failed: {e}")
        return False
    
    # Test 5: Test memory models
    print("\n5. Testing memory models...")
    try:
        from hanzo_memory.models import Memory, KnowledgeBase, Fact, ChatMessage
        from datetime import datetime
        
        # Create test models
        memory = Memory(
            id="test_mem",
            content="Test memory",
            user_id="test_user",
            created_at=datetime.now()
        )
        
        kb = KnowledgeBase(
            id="test_kb",
            name="Test KB",
            created_at=datetime.now()
        )
        
        fact = Fact(
            id="test_fact",
            kb_id=kb.id,
            content="Test fact",
            created_at=datetime.now()
        )
        
        print("✅ Memory models created successfully")
    except Exception as e:
        print(f"⚠️  Memory test skipped (requires hanzo-memory[all]): {e}")
    
    # Test 6: Test inter-package compatibility
    print("\n6. Testing inter-package compatibility...")
    try:
        # Test MCP with agents
        from hanzo_mcp.tools.agent import AgentTool
        agent_tool = AgentTool()
        
        # Test network with memory
        from hanzo_network import LocalComputeOrchestrator
        if hasattr(hanzo_network, 'LocalComputeOrchestrator'):
            print("✅ Local compute features available")
        
        print("✅ Inter-package compatibility verified")
    except Exception as e:
        print(f"⚠️  Some advanced features not available: {e}")
    
    print("\n" + "=" * 60)
    print("✅ All integration tests passed!")
    print("\nPackage Summary:")
    print(f"  • hanzo-mcp: v{hanzo_mcp.__version__} - MCP tools and server")
    print(f"  • hanzo-agents: v{hanzo_agents.__version__ if 'hanzo_agents' in locals() else '0.1.0'} - Agent orchestration")
    print(f"  • hanzo-network: v{hanzo_network.__version__} - Network management")
    print(f"  • hanzo-memory: v{hanzo_memory.__version__} - Memory service")
    
    return True

if __name__ == "__main__":
    success = asyncio.run(test_integration())
    sys.exit(0 if success else 1)