Spaces:
Sleeping
Sleeping
File size: 4,856 Bytes
63ed3a7 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 |
#!/usr/bin/env python3
"""
Test script for the integrated Inference Server setup
This script verifies that both the FastAPI API and Gradio UI work correctly.
"""
import asyncio
import sys
import time
from pathlib import Path
import httpx
# Add src to path
src_path = Path(__file__).parent / "src"
sys.path.insert(0, str(src_path))
async def test_api_endpoints():
"""Test that the FastAPI endpoints work correctly."""
base_url = "http://localhost:7860"
async with httpx.AsyncClient(timeout=10.0) as client:
try:
# Test main health endpoint
print("🔍 Testing main health endpoint...")
response = await client.get(f"{base_url}/api/health")
assert response.status_code == 200
data = response.json()
print(f"✅ Health check passed: {data}")
# Test API docs availability
print("📖 Testing API docs availability...")
response = await client.get(f"{base_url}/api/docs")
assert response.status_code == 200
print("✅ API docs available")
# Test OpenAPI schema
print("📋 Testing OpenAPI schema...")
response = await client.get(f"{base_url}/api/openapi.json")
assert response.status_code == 200
schema = response.json()
print(f"✅ OpenAPI schema available: {schema['info']['title']}")
# Test sessions endpoint
print("📝 Testing sessions endpoint...")
response = await client.get(f"{base_url}/api/sessions")
assert response.status_code == 200
sessions = response.json()
print(f"✅ Sessions endpoint works: {len(sessions)} sessions")
# Test Gradio UI availability
print("🎨 Testing Gradio UI availability...")
response = await client.get(f"{base_url}/")
assert response.status_code == 200
print("✅ Gradio UI available")
print("\n🎉 All tests passed! The integrated setup is working correctly.")
return True
except Exception as e:
print(f"❌ Test failed: {e}")
return False
async def test_session_creation():
"""Test creating a session through the API."""
base_url = "http://localhost:7860"
async with httpx.AsyncClient(timeout=30.0) as client:
try:
# Create a test session
print("🔧 Testing session creation...")
session_data = {
"session_id": "test-session",
"policy_path": "./checkpoints/act_so101_beyond",
"camera_names": ["front"],
"arena_server_url": "http://localhost:8000",
}
response = await client.post(f"{base_url}/api/sessions", json=session_data)
if response.status_code == 200:
print("✅ Session creation successful")
return True
print(
f"⚠️ Session creation failed (expected if no model exists): {response.status_code}"
)
return True # This is expected if no model exists
except Exception as e:
print(f"⚠️ Session creation test failed (expected if no model exists): {e}")
return True # This is expected if no model exists
def print_integration_info():
"""Print information about the integrated setup."""
print("=" * 60)
print("🤖 Integrated Inference Server Test Results")
print("=" * 60)
print()
print("📍 Access Points:")
print(" 🎨 Gradio UI: http://localhost:7860/")
print(" 📖 API Docs: http://localhost:7860/api/docs")
print(" 🔄 Health Check: http://localhost:7860/api/health")
print(" 📋 OpenAPI Schema: http://localhost:7860/api/openapi.json")
print(" 📝 Sessions API: http://localhost:7860/api/sessions")
print()
print("🔧 Features Available:")
print(" ✅ Direct session management through UI")
print(" ✅ Full REST API for programmatic access")
print(" ✅ Interactive API documentation")
print(" ✅ Single port deployment")
print(" ✅ CORS enabled for frontend integration")
print()
if __name__ == "__main__":
print("🧪 Testing Integrated Inference Server Setup")
print("Make sure the server is running with: python launch_simple.py")
print()
# Wait a moment for server to be ready
time.sleep(2)
# Run tests
try:
success = asyncio.run(test_api_endpoints())
if success:
asyncio.run(test_session_creation())
print_integration_info()
except KeyboardInterrupt:
print("\n🛑 Tests interrupted by user")
except Exception as e:
print(f"❌ Test runner error: {e}")
sys.exit(1)
|