run_full_tests.py•5.82 kB
#!/usr/bin/env python3
"""
Full Test Suite Runner
Runs all tests including:
1. MCP Server startup tests
2. Elasticsearch entity search tests for all 11 entity types
"""
import sys
import os
import subprocess
from pathlib import Path
def print_header(title):
    """Print a formatted header."""
    print("\n" + "="*70)
    print(f"  {title}")
    print("="*70 + "\n")
def run_command(cmd, description):
    """Run a command and return success status."""
    print(f"Running: {description}")
    print(f"Command: {' '.join(cmd)}\n")
    
    try:
        result = subprocess.run(cmd, check=False)
        return result.returncode == 0
    except Exception as e:
        print(f"❌ Error: {e}")
        return False
def check_elasticsearch():
    """Check if Elasticsearch is running."""
    print_header("Checking Elasticsearch Connection")
    
    try:
        from elasticsearch import Elasticsearch
        
        es_host = os.getenv("ES_HOST", "localhost")
        es_port = int(os.getenv("ES_PORT", "9200"))
        
        es = Elasticsearch([f"http://{es_host}:{es_port}"])
        
        if es.ping():
            info = es.info()
            print(f"✓ Elasticsearch is running")
            print(f"  Version: {info['version']['number']}")
            print(f"  Cluster: {info['cluster_name']}")
            print(f"  Host: {es_host}:{es_port}")
            return True
        else:
            print(f"✗ Elasticsearch is not responding at {es_host}:{es_port}")
            return False
            
    except Exception as e:
        print(f"✗ Cannot connect to Elasticsearch: {e}")
        print(f"  Make sure Elasticsearch is running on {es_host}:{es_port}")
        return False
def check_dependencies():
    """Check if required dependencies are installed."""
    print_header("Checking Dependencies")
    
    required = {
        'pytest': 'pytest',
        'elasticsearch': 'elasticsearch',
        'fastmcp': 'fastmcp',
    }
    
    missing = []
    
    for package, import_name in required.items():
        try:
            __import__(import_name)
            print(f"✓ {package} installed")
        except ImportError:
            print(f"✗ {package} not installed")
            missing.append(package)
    
    if missing:
        print(f"\n❌ Missing packages: {', '.join(missing)}")
        print(f"Install with: pip install {' '.join(missing)}")
        return False
    
    return True
def run_mcp_server_tests():
    """Run MCP server startup tests."""
    print_header("MCP Server Startup Tests")
    
    cmd = [
        sys.executable, "-m", "pytest",
        "tests/test_mcp_server.py",
        "-v",
        "--tb=short",
        "-s"
    ]
    
    return run_command(cmd, "MCP Server Tests")
def run_elasticsearch_tests():
    """Run Elasticsearch entity search tests."""
    print_header("Elasticsearch Entity Search Tests")
    
    cmd = [
        sys.executable, "-m", "pytest",
        "tests/test_elasticsearch_entities.py",
        "-v",
        "--tb=short",
        "-s"
    ]
    
    return run_command(cmd, "Elasticsearch Entity Tests")
def run_all_tests():
    """Run all tests together."""
    print_header("Running All Tests Together")
    
    cmd = [
        sys.executable, "-m", "pytest",
        "tests/test_mcp_server.py",
        "tests/test_elasticsearch_entities.py",
        "-v",
        "--tb=short",
        "-s"
    ]
    
    return run_command(cmd, "All Tests")
def main():
    """Main test runner."""
    print_header("Full Test Suite Runner")
    
    # Change to project root
    project_root = Path(__file__).parent
    os.chdir(project_root)
    
    print(f"Project root: {project_root}")
    print(f"Python: {sys.version}")
    
    # Check dependencies
    if not check_dependencies():
        return 1
    
    # Check Elasticsearch
    es_running = check_elasticsearch()
    
    # Determine which tests to run
    if len(sys.argv) > 1:
        test_type = sys.argv[1].lower()
        
        if test_type == "mcp":
            print_header("Running MCP Server Tests Only")
            success = run_mcp_server_tests()
            
        elif test_type == "es" or test_type == "elasticsearch":
            if not es_running:
                print("\n❌ Elasticsearch is not running. Cannot run Elasticsearch tests.")
                return 1
            print_header("Running Elasticsearch Tests Only")
            success = run_elasticsearch_tests()
            
        elif test_type == "all":
            if not es_running:
                print("\n⚠ Warning: Elasticsearch is not running. Some tests will fail.")
            success = run_all_tests()
            
        else:
            print(f"\n❌ Unknown test type: {test_type}")
            print("Usage: python run_full_tests.py [mcp|es|all]")
            return 1
    else:
        # Run all tests by default
        if not es_running:
            print("\n⚠ Warning: Elasticsearch is not running. Some tests will fail.")
            print("To skip Elasticsearch tests, run: python run_full_tests.py mcp")
        
        # Run MCP tests first
        mcp_success = run_mcp_server_tests()
        
        # Run Elasticsearch tests if ES is running
        if es_running:
            es_success = run_elasticsearch_tests()
        else:
            print("\n⚠ Skipping Elasticsearch tests (service not running)")
            es_success = True  # Don't fail overall if ES not running
        
        success = mcp_success and es_success
    
    # Print final summary
    print_header("Test Summary")
    
    if success:
        print("✅ All tests passed!")
        return 0
    else:
        print("❌ Some tests failed. Check output above for details.")
        return 1
if __name__ == "__main__":
    sys.exit(main())