test_preflight_simulation.py•6.3 kB
#!/usr/bin/env python3
"""Simulate FastMCP Cloud pre-flight check to validate startup fixes."""
import asyncio
import os
import sys
import time
from fastmcp import Client
from catabus_mcp.server import server
async def test_preflight_simulation():
"""Simulate the exact conditions of FastMCP Cloud pre-flight validation."""
print("🚀 FastMCP Cloud Pre-Flight Simulation")
print("=" * 50)
# Simulate cloud environment
print("\n1. Setting up cloud environment simulation...")
os.environ['FASTMCP_CLOUD'] = '1'
print(" ✅ FASTMCP_CLOUD environment variable set")
# Test 1: Ultra-fast server startup validation
print("\n2. Testing server startup speed...")
try:
start_time = time.time()
# This is what FastMCP Cloud does - immediate server validation
tools = await server.get_tools()
startup_time = time.time() - start_time
print(f" ✅ Server startup: {startup_time:.3f}s")
print(f" Tools available: {len(tools)}")
if startup_time > 2.0:
print(" ⚠️ SLOW - may fail pre-flight checks")
return False
else:
print(" ✅ FAST - should pass pre-flight checks")
except Exception as e:
print(f" ❌ Server startup failed: {e}")
return False
# Test 2: Health check under cloud conditions
print("\n3. Testing health check in cloud mode...")
try:
client = Client(server)
async with client:
start_time = time.time()
# This simulates the exact pre-flight health check
result = await asyncio.wait_for(client.call_tool("health_check", {}), timeout=3)
health_time = time.time() - start_time
print(f" ✅ Health check: {health_time:.3f}s")
# Extract and validate result
import json
if hasattr(result, 'content') and isinstance(result.content, list) and len(result.content) > 0:
text_content = result.content[0]
if hasattr(text_content, 'text'):
health = json.loads(text_content.text)
print(f" Status: {health['status']}")
print(f" Environment: {health['environment']}")
print(f" Mode: {health['startup_mode']}")
if health['environment'] != 'cloud':
print(" ⚠️ Environment detection not working")
return False
if health['startup_mode'] != 'optimized':
print(" ⚠️ Not using optimized startup mode")
return False
print(" ✅ Cloud environment properly detected")
if health_time > 1.0:
print(" ⚠️ SLOW health check - may timeout in pre-flight")
return False
else:
print(" ✅ FAST health check - perfect for pre-flight")
except asyncio.TimeoutError:
print(" ❌ Health check timed out - CRITICAL FAILURE")
return False
except Exception as e:
print(f" ❌ Health check failed: {e}")
return False
# Test 3: Verify no blocking operations during startup
print("\n4. Testing multiple rapid health checks...")
try:
client = Client(server)
async with client:
times = []
# Simulate rapid pre-flight checks
for i in range(5):
start_time = time.time()
result = await asyncio.wait_for(client.call_tool("health_check", {}), timeout=2)
check_time = time.time() - start_time
times.append(check_time)
print(f" Check {i+1}: {check_time:.3f}s")
avg_time = sum(times) / len(times)
max_time = max(times)
print(f" Average: {avg_time:.3f}s, Max: {max_time:.3f}s")
if max_time > 1.0:
print(" ⚠️ Inconsistent performance - may fail under load")
return False
else:
print(" ✅ Consistent fast performance")
except Exception as e:
print(f" ❌ Rapid health check test failed: {e}")
return False
# Test 4: Test with network disabled (simulating network issues)
print("\n5. Testing resilience to network failures...")
try:
client = Client(server)
async with client:
# Health check should still work even if network is down
result = await asyncio.wait_for(client.call_tool("health_check", {}), timeout=2)
print(" ✅ Health check works without network dependencies")
# Try to initialize data - should fail gracefully
start_time = time.time()
result = await asyncio.wait_for(client.call_tool("initialize_data", {}), timeout=20)
init_time = time.time() - start_time
print(f" ✅ Data initialization completes in {init_time:.3f}s (with fallbacks)")
if init_time > 25:
print(" ⚠️ Data initialization too slow")
return False
except Exception as e:
print(f" ⚠️ Network failure test: {e} (may be expected)")
# Clean up environment
del os.environ['FASTMCP_CLOUD']
# Final validation
print("\n" + "=" * 50)
print("🎉 PRE-FLIGHT SIMULATION: PASSED")
print("✅ Server startup: <2s")
print("✅ Health check: <1s consistently")
print("✅ Cloud environment detection working")
print("✅ No blocking operations in critical path")
print("✅ Graceful fallbacks for network failures")
print("\n📋 FastMCP Cloud Deployment Status: READY")
print("The server should now pass pre-flight validation successfully!")
print("=" * 50)
return True
if __name__ == "__main__":
success = asyncio.run(test_preflight_simulation())
sys.exit(0 if success else 1)