test_hierarchical_coordination.py

import sys
import asyncio
from pathlib import Path

# Add project root to path
project_root = Path(__file__).parent
sys.path.append(str(project_root))

from core.coordinator import coordinator
from core.session import session_manager

async def test_hierarchical_coordination():
    """Test the hierarchical coordination system"""
    print("=== Hierarchical Coordination System Test ===")
    print()
    
    # Test user query
    user_query = "What are the key principles of effective time management?"
    user_id = "test_user"
    
    print(f"User Query: {user_query}")
    print()
    
    # Test coordination status
    print("1. Testing Coordination Status:")
    try:
        coord_status = coordinator.get_coordination_status()
        print(f"   Tavily Available: {coord_status.get('tavily_available', False)}")
        print(f"   Weather Available: {coord_status.get('weather_available', False)}")
        print(f"   Web Search Enabled: {coord_status.get('web_search_enabled', False)}")
        print("   ✅ Coordination Status Check Passed")
    except Exception as e:
        print(f"   ❌ Coordination Status Check Failed: {e}")
    
    print()
    
    # Test hierarchical conversation coordination
    print("2. Testing Hierarchical Conversation Coordination:")
    try:
        print("   Starting hierarchical coordination...")
        response_count = 0
        
        async for response_chunk in coordinator.coordinate_hierarchical_conversation(user_id, user_query):
            response_count += 1
            print(f"   Chunk {response_count}: {response_chunk['type']} - {response_chunk['content'][:50]}...")
            
            # Limit output for readability
            if response_count >= 5:
                print("   ... (truncated for brevity)")
                break
        
        print("   ✅ Hierarchical Coordination Test Passed")
    except Exception as e:
        print(f"   ❌ Hierarchical Coordination Test Failed: {e}")
    
    print()
    
    # Test hierarchical session tracking
    print("3. Testing Hierarchical Session Tracking:")
    try:
        # Update with test coordination data
        test_data = {
            'hf_engaged': True,
            'ollama_responded': True,
            'success': True
        }
        update_result = session_manager.update_hierarchical_coordination(user_id, test_data)
        print(f"   Update Result: {'✅ Success' if update_result else '❌ Failed'}")
        
        # Get hierarchical stats
        stats = session_manager.get_hierarchical_stats(user_id)
        print(f"   Total Conversations: {stats.get('total_conversations', 0)}")
        print(f"   HF Engagements: {stats.get('hf_engagements', 0)}")
        print(f"   Ollama Responses: {stats.get('ollama_responses', 0)}")
        print("   ✅ Hierarchical Session Tracking Passed")
    except Exception as e:
        print(f"   ❌ Hierarchical Session Tracking Failed: {e}")
    
    print()
    print("🎉 Hierarchical Coordination System Test Completed!")

if __name__ == "__main__":
    asyncio.run(test_hierarchical_coordination())