test_runtime_quick.py•2.41 kB
#!/usr/bin/env python3
"""Quick runtime test: start server, call ibkr_connect, check for loop errors"""
import asyncio
import time
import subprocess
import sys
async def test_connect():
import httpx
# Wait for server startup
await asyncio.sleep(3)
base_url = "http://localhost:8000/api/v1"
async with httpx.AsyncClient(timeout=10.0) as client:
# Try to call ibkr_connect tool
print("Testing ibkr_connect tool...")
try:
# For SSE-based MCP, we need to establish SSE connection first
# For now, just test that the endpoint doesn't crash
response = await client.get(f"{base_url}/sse")
print(f"SSE endpoint status: {response.status_code}")
# Read a bit of the SSE stream
if response.status_code == 200:
print("✓ Server started successfully")
print("✓ SSE endpoint is accessible")
return True
else:
print(f"✗ Unexpected status: {response.status_code}")
return False
except Exception as e:
print(f"✗ Error: {e}")
return False
if __name__ == "__main__":
# Start server in background
print("Starting server...")
server_proc = subprocess.Popen(
["uv", "run", "python", "main.py"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
cwd="/home/zhaoli/dev/git/ibkr-tws-mcp-server"
)
try:
# Run test
result = asyncio.run(test_connect())
# Check server stderr for loop errors
time.sleep(1)
server_proc.terminate()
server_proc.wait(timeout=2)
stdout, stderr = server_proc.communicate(timeout=1)
stderr_text = stderr.decode('utf-8', errors='ignore')
if "different loop" in stderr_text:
print("\n✗ FAILED: Loop error still present in logs")
print("Error excerpt:")
for line in stderr_text.split('\n'):
if 'loop' in line.lower():
print(f" {line}")
sys.exit(1)
else:
print("\n✓ SUCCESS: No loop errors detected")
sys.exit(0 if result else 1)
finally:
try:
server_proc.kill()
server_proc.wait(timeout=1)
except:
pass