test_mcp_server.py

#!/usr/bin/env python3
"""
Test script for the MCP Sentiment Analysis Server

This script tests the MCP server functionality both locally and remotely.
"""

import requests
import json
import time
import sys
from typing import Dict, Any


def test_web_interface(base_url: str = "http://localhost:7860") -> bool:
    """Test if the web interface is accessible."""
    try:
        response = requests.get(base_url, timeout=10)
        if response.status_code == 200:
            print(f"✅ Web interface accessible at {base_url}")
            return True
        else:
            print(f"❌ Web interface returned status {response.status_code}")
            return False
    except requests.exceptions.RequestException as e:
        print(f"❌ Failed to connect to web interface: {e}")
        return False


def test_mcp_endpoint(base_url: str = "http://localhost:7860") -> bool:
    """Test if the MCP endpoint is accessible."""
    mcp_url = f"{base_url}/gradio_api/mcp/sse"
    try:
        # Test basic connectivity
        response = requests.get(mcp_url, timeout=10)
        print(f"✅ MCP endpoint accessible at {mcp_url}")
        return True
    except requests.exceptions.RequestException as e:
        print(f"❌ Failed to connect to MCP endpoint: {e}")
        return False


def test_sentiment_functions() -> bool:
    """Test the sentiment analysis functions directly."""
    try:
        # Import the functions from app.py
        import sys
        import os
        sys.path.append(os.path.dirname(__file__))
        
        from app import analyze_sentiment, get_sentiment_score, classify_emotion, batch_analyze
        
        # Test analyze_sentiment
        result = analyze_sentiment("I love this product!")
        result_data = json.loads(result)
        assert "sentiment" in result_data
        assert result_data["sentiment"] == "Positive"
        print("✅ analyze_sentiment function works")
        
        # Test get_sentiment_score
        result = get_sentiment_score("This is terrible!")
        result_data = json.loads(result)
        assert "sentiment_score" in result_data
        assert result_data["sentiment_score"] < 0
        print("✅ get_sentiment_score function works")
        
        # Test classify_emotion
        result = classify_emotion("I'm so happy today!")
        result_data = json.loads(result)
        assert "emotion" in result_data
        print("✅ classify_emotion function works")
        
        # Test batch_analyze
        result = batch_analyze("I love this!\nThis is bad.\nNeutral statement.")
        result_data = json.loads(result)
        assert "summary" in result_data
        assert result_data["summary"]["total_texts"] == 3
        print("✅ batch_analyze function works")
        
        return True
        
    except Exception as e:
        print(f"❌ Function tests failed: {e}")
        return False


def test_gradio_api(base_url: str = "http://localhost:7860") -> bool:
    """Test the Gradio API endpoints."""
    try:
        # Test the analyze_sentiment endpoint
        api_url = f"{base_url}/api/predict"
        
        # This is a basic test - actual Gradio API testing would require
        # more specific endpoint knowledge
        print("ℹ️  Gradio API testing requires running server")
        return True
        
    except Exception as e:
        print(f"❌ Gradio API test failed: {e}")
        return False


def wait_for_server(base_url: str = "http://localhost:7860", max_wait: int = 60) -> bool:
    """Wait for the server to become available."""
    print(f"Waiting for server at {base_url}...")
    
    for i in range(max_wait):
        try:
            response = requests.get(base_url, timeout=5)
            if response.status_code == 200:
                print(f"✅ Server is ready after {i+1} seconds")
                return True
        except requests.exceptions.RequestException:
            pass
        
        time.sleep(1)
        if (i + 1) % 10 == 0:
            print(f"Still waiting... ({i+1}/{max_wait} seconds)")
    
    print(f"❌ Server did not become available within {max_wait} seconds")
    return False


def main():
    """Run all tests."""
    print("🧪 Testing MCP Sentiment Analysis Server")
    print("=" * 50)
    
    # Test functions directly (doesn't require server)
    print("\n📋 Testing sentiment analysis functions...")
    functions_ok = test_sentiment_functions()
    
    # Check if we should test the server
    test_server = "--server" in sys.argv or len(sys.argv) == 1
    base_url = "http://localhost:7860"
    
    # Allow custom URL
    for arg in sys.argv:
        if arg.startswith("--url="):
            base_url = arg.split("=", 1)[1]
    
    if test_server:
        print(f"\n🌐 Testing server at {base_url}...")
        
        # Wait for server if needed
        if "--wait" in sys.argv:
            server_ready = wait_for_server(base_url)
            if not server_ready:
                print("❌ Server tests skipped - server not available")
                return 1 if not functions_ok else 0
        
        # Test server endpoints
        web_ok = test_web_interface(base_url)
        mcp_ok = test_mcp_endpoint(base_url)
        api_ok = test_gradio_api(base_url)
        
        server_ok = web_ok and mcp_ok and api_ok
    else:
        print("\n⏭️  Server tests skipped (use --server to enable)")
        server_ok = True
    
    # Summary
    print("\n📊 Test Summary")
    print("=" * 50)
    print(f"Functions: {'✅ PASS' if functions_ok else '❌ FAIL'}")
    if test_server:
        print(f"Server:    {'✅ PASS' if server_ok else '❌ FAIL'}")
    
    overall_success = functions_ok and (server_ok if test_server else True)
    print(f"Overall:   {'✅ PASS' if overall_success else '❌ FAIL'}")
    
    if overall_success:
        print("\n🎉 All tests passed!")
        return 0
    else:
        print("\n💥 Some tests failed!")
        return 1


if __name__ == "__main__":
    if "--help" in sys.argv or "-h" in sys.argv:
        print("Usage: python test_mcp_server.py [options]")
        print("\nOptions:")
        print("  --server          Test server endpoints (default: enabled)")
        print("  --wait            Wait for server to become available")
        print("  --url=URL         Use custom server URL (default: http://localhost:7860)")
        print("  --help, -h        Show this help message")
        print("\nExamples:")
        print("  python test_mcp_server.py")
        print("  python test_mcp_server.py --wait")
        print("  python test_mcp_server.py --url=https://myspace.hf.space")
        sys.exit(0)
    
    sys.exit(main())