test_mcp_integration.py

#!/usr/bin/env python3
"""
Test script to verify MCP integration with STT service
"""

import asyncio
import json
import sys
import os

# Add current directory to path for imports
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

def test_imports():
    """Test that all required imports work"""
    print("🧪 Testing imports...")
    
    try:
        import gradio as gr
        print("✅ Gradio import successful")
    except ImportError as e:
        print(f"❌ Gradio import failed: {e}")
        return False
    
    try:
        import mcp
        from mcp.server import Server
        from mcp.types import Tool, TextContent
        print("✅ MCP imports successful")
        return True
    except ImportError as e:
        print(f"❌ MCP imports failed: {e}")
        print("💡 Install with: pip install mcp>=1.0.0")
        return False

def test_mcp_server_creation():
    """Test MCP server creation"""
    print("\n🧪 Testing MCP server creation...")
    
    try:
        from app import mcp_server, MCP_AVAILABLE
        
        if not MCP_AVAILABLE:
            print("❌ MCP not available in app")
            return False
        
        if mcp_server is None:
            print("❌ MCP server not initialized")
            return False
        
        print("✅ MCP server created successfully")
        return True
        
    except Exception as e:
        print(f"❌ MCP server creation failed: {e}")
        return False

async def test_mcp_tools():
    """Test MCP tool listing"""
    print("\n🧪 Testing MCP tool definitions...")
    
    try:
        from app import mcp_server, MCP_AVAILABLE
        
        if not MCP_AVAILABLE:
            print("❌ MCP not available")
            return False
        
        # Get the list_tools handler
        tools = await mcp_server.list_tools()()
        
        print(f"✅ Found {len(tools)} MCP tools:")
        for tool in tools:
            print(f"  - {tool.name}: {tool.description}")
        
        # Verify expected tools exist
        tool_names = [tool.name for tool in tools]
        expected_tools = ["stt_transcribe", "stt_batch_transcribe", "stt_get_info"]
        
        for expected in expected_tools:
            if expected in tool_names:
                print(f"✅ Tool '{expected}' found")
            else:
                print(f"❌ Tool '{expected}' missing")
                return False
        
        return True
        
    except Exception as e:
        print(f"❌ MCP tools test failed: {e}")
        return False

async def test_system_info_tool():
    """Test the system info MCP tool"""
    print("\n🧪 Testing stt_get_info tool...")
    
    try:
        from app import mcp_server, MCP_AVAILABLE
        
        if not MCP_AVAILABLE:
            print("❌ MCP not available")
            return False
        
        # Call the get_info tool
        result = await mcp_server.call_tool()("stt_get_info", {})
        
        if not result or len(result) == 0:
            print("❌ No result from stt_get_info")
            return False
        
        info_text = result[0].text
        info_data = json.loads(info_text)
        
        print("✅ System info tool response:")
        print(f"  - MCP Status: {info_data.get('mcp_status', 'Unknown')}")
        print(f"  - Available Tools: {info_data.get('available_tools', [])}")
        print(f"  - Supported Languages: {len(info_data.get('supported_languages', []))} languages")
        print(f"  - Supported Models: {info_data.get('supported_models', [])}")
        
        return True
        
    except Exception as e:
        print(f"❌ System info tool test failed: {e}")
        return False

def test_gradio_functions():
    """Test that original Gradio functions still work"""
    print("\n🧪 Testing original Gradio functions...")
    
    try:
        from app import get_system_info, LANGUAGES, MODEL_SIZES
        
        # Test system info function
        info = get_system_info()
        if "MCP Server" in info:
            print("✅ get_system_info() includes MCP status")
        else:
            print("❌ get_system_info() missing MCP status")
            return False
        
        # Test language and model definitions
        print(f"✅ Languages available: {len(LANGUAGES)}")
        print(f"✅ Model sizes available: {len(MODEL_SIZES)}")
        
        return True
        
    except Exception as e:
        print(f"❌ Gradio functions test failed: {e}")
        return False

async def main():
    """Run all tests"""
    print("🚀 STT Service MCP Integration Tests")
    print("=" * 50)
    
    tests = [
        ("Import Test", test_imports),
        ("MCP Server Creation", test_mcp_server_creation),
        ("MCP Tools", test_mcp_tools),
        ("System Info Tool", test_system_info_tool),
        ("Gradio Functions", test_gradio_functions)
    ]
    
    results = []
    
    for test_name, test_func in tests:
        try:
            if asyncio.iscoroutinefunction(test_func):
                result = await test_func()
            else:
                result = test_func()
            results.append((test_name, result))
        except Exception as e:
            print(f"❌ {test_name} crashed: {e}")
            results.append((test_name, False))
    
    # Summary
    print("\n" + "=" * 50)
    print("📊 Test Results Summary")
    print("=" * 50)
    
    passed = 0
    total = len(results)
    
    for test_name, result in results:
        status = "✅ PASS" if result else "❌ FAIL"
        print(f"{status}: {test_name}")
        if result:
            passed += 1
    
    print(f"\n🎯 Results: {passed}/{total} tests passed")
    
    if passed == total:
        print("🎉 All tests passed! MCP integration is working correctly.")
        print("\n📝 Next steps:")
        print("1. Deploy updated service to Hugging Face")
        print("2. Test with actual MCP client")
        print("3. Integrate with ChatCal voice pipeline")
    else:
        print("❌ Some tests failed. Please check the implementation.")
        return 1
    
    return 0

if __name__ == "__main__":
    exit_code = asyncio.run(main())
    sys.exit(exit_code)