server/tests/test_unified_manager.py

"""
Test Unified Data Source Manager

Demonstrates how the system handles multiple data sources across languages.
"""

import asyncio
import sys
import os

sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))


async def test_unified_manager():
    """Test the unified data source manager"""
    from app.services.unified_data_manager import UnifiedDataSourceManager
    
    print("=" * 70)
    print("🧪 TESTING UNIFIED DATA SOURCE MANAGER")
    print("=" * 70)
    
    # Initialize without database (will use defaults)
    manager = UnifiedDataSourceManager(db_session=None)
    
    # Test scenarios
    test_cases = [
        {
            "query": "mujhe bukhar hai",
            "website_id": 123,
            "industry": "healthcare",
            "description": "Roman Urdu - Healthcare symptom query"
        },
        {
            "query": "What is diabetes?",
            "website_id": 123,
            "industry": "healthcare",
            "description": "English - Medical information query"
        },
        {
            "query": "داخلے کے لیے کیا ضرورت ہے؟",
            "website_id": 456,
            "industry": "education",
            "description": "Urdu - Academic admission query"
        },
        {
            "query": "GPA calculate kaise karun?",
            "website_id": 456,
            "industry": "education",
            "description": "Roman Urdu - Academic process query"
        }
    ]
    
    for idx, test in enumerate(test_cases, 1):
        print(f"\n{'='*70}")
        print(f"TEST {idx}: {test['description']}")
        print(f"{'='*70}")
        print(f"Query: \"{test['query']}\"")
        print(f"Industry: {test['industry']}")
        
        # Process query
        result = await manager.query(
            user_query=test['query'],
            website_id=test['website_id'],
            industry=test['industry'],
            session_id=f"test_session_{idx}"
        )
        
        # Display results
        print(f"\n📊 RESULTS:")
        print(f"  Language Detected: {result['language_detected']}")
        print(f"  Intent: {result['intent']}")
        print(f"  Data Source Used: {result['source']}")
        print(f"  Confidence: {result['confidence']:.0%}")
        print(f"  Sources Checked: {', '.join(result['data_sources_checked'])}")
        print(f"\n💬 ANSWER:")
        print(f"  {result['answer'][:150]}...")
        
        # Show metadata
        if result['metadata']:
            print(f"\n📋 Metadata:")
            for key, value in result['metadata'].items():
                print(f"  {key}: {value}")
    
    print(f"\n{'='*70}")
    print("✅ ALL TESTS COMPLETED")
    print(f"{'='*70}")
    
    # Summary
    print(f"\n📊 SYSTEM CAPABILITIES DEMONSTRATED:")
    print(f"  ✅ Multi-language support (English, Urdu, Roman Urdu)")
    print(f"  ✅ Multi-industry support (Healthcare, Education)")
    print(f"  ✅ Intent-based routing")
    print(f"  ✅ Priority-based data source selection")
    print(f"  ✅ Automatic fallback handling")
    print(f"  ✅ Graceful handling of empty data sources")


if __name__ == "__main__":
    print("\n🚀 Starting Unified Data Source Manager Tests\n")
    asyncio.run(test_unified_manager())