#!/usr/bin/env python3
"""
Comprehensive API endpoint tester for all MCP services
Tests all endpoints across HTTP bridge (8000) and Smart Search API (8002)
"""
import requests
import json
import time
import sys
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from enum import Enum
class TestStatus(Enum):
PASS = "✅ PASS"
FAIL = "❌ FAIL"
SKIP = "⏭️ SKIP"
WARNING = "⚠️ WARNING"
@dataclass
class TestResult:
endpoint: str
method: str
status: TestStatus
response_code: Optional[int]
response_time: float
message: str
response_data: Optional[Dict] = None
class APITester:
def __init__(self):
self.results: List[TestResult] = []
self.base_urls = {
'http_bridge': 'http://localhost:8000',
'smart_search': 'http://localhost:8002'
}
def test_endpoint(
self,
service: str,
endpoint: str,
method: str = 'GET',
data: Optional[Dict] = None,
expected_status: int = 200,
timeout: int = 10
) -> TestResult:
"""Test a single endpoint"""
url = f"{self.base_urls[service]}{endpoint}"
start_time = time.time()
try:
if method == 'GET':
response = requests.get(url, timeout=timeout)
elif method == 'POST':
response = requests.post(url, json=data, timeout=timeout)
else:
response = requests.request(method, url, json=data, timeout=timeout)
response_time = time.time() - start_time
response_data = None
try:
response_data = response.json()
except:
response_data = {"text": response.text}
if response.status_code == expected_status:
status = TestStatus.PASS
message = f"Response received in {response_time:.2f}s"
elif response.status_code in [404, 405]:
status = TestStatus.SKIP
message = f"Endpoint not available (HTTP {response.status_code})"
else:
status = TestStatus.WARNING
message = f"Expected {expected_status}, got {response.status_code}"
return TestResult(
endpoint=endpoint,
method=method,
status=status,
response_code=response.status_code,
response_time=response_time,
message=message,
response_data=response_data
)
except requests.Timeout:
return TestResult(
endpoint=endpoint,
method=method,
status=TestStatus.FAIL,
response_code=None,
response_time=time.time() - start_time,
message="Request timeout"
)
except Exception as e:
return TestResult(
endpoint=endpoint,
method=method,
status=TestStatus.FAIL,
response_code=None,
response_time=time.time() - start_time,
message=f"Error: {str(e)}"
)
def test_all_endpoints(self):
"""Test all known endpoints across all services"""
print("🚀 Starting Comprehensive API Endpoint Testing")
print("=" * 60)
# Test data for POST endpoints
test_question = {
"question": "What tables are available in the database?",
"database": "db3",
"include_sql": True,
"include_semantic": True,
"include_schema": True
}
test_chat = {
"message": "Hello, what can you help me with?",
"conversation_id": "test-123"
}
# Define all endpoints to test
endpoints = [
# HTTP Bridge (Port 8000) - Core endpoints
('http_bridge', '/health', 'GET'),
('http_bridge', '/api/health', 'GET'),
('http_bridge', '/api/info', 'GET'),
# HTTP Bridge - Conversation endpoints
('http_bridge', '/api/conversation/chat', 'POST', test_chat),
('http_bridge', '/api/conversation/chat/stream', 'POST', test_chat),
('http_bridge', '/api/conversation/list', 'GET'),
('http_bridge', '/api/conversation/clear', 'POST'),
# HTTP Bridge - MCP Database endpoints
('http_bridge', '/api/database/query', 'POST', {"query": "SELECT 1"}),
('http_bridge', '/api/database/schema', 'GET'),
('http_bridge', '/api/database/tables', 'GET'),
# Smart Search API (Port 8002) - Core endpoints
('smart_search', '/health', 'GET'),
('smart_search', '/api/info', 'GET'),
('smart_search', '/api/databases', 'GET'),
('smart_search', '/api/models', 'GET'),
# Smart Search API - Main functionality
('smart_search', '/api/smart-search', 'POST', test_question),
]
# Test service availability first
print("🔍 Testing Service Availability")
print("-" * 40)
for service, base_url in self.base_urls.items():
try:
response = requests.get(f"{base_url}/health", timeout=5)
if response.status_code == 200:
print(f"✅ {service.replace('_', ' ').title()}: {base_url} - ONLINE")
else:
print(f"⚠️ {service.replace('_', ' ').title()}: {base_url} - HTTP {response.status_code}")
except Exception as e:
print(f"❌ {service.replace('_', ' ').title()}: {base_url} - OFFLINE ({str(e)})")
print()
print("🧪 Testing Individual Endpoints")
print("-" * 40)
# Test all endpoints
for endpoint_data in endpoints:
service, endpoint, method = endpoint_data[:3]
data = endpoint_data[3] if len(endpoint_data) > 3 else None
result = self.test_endpoint(service, endpoint, method, data)
self.results.append(result)
# Print real-time results
service_name = service.replace('_', ' ').title()
print(f"{result.status.value} {service_name}: {method} {endpoint}")
if result.response_time > 0:
print(f" Time: {result.response_time:.2f}s | {result.message}")
else:
print(f" {result.message}")
# Show interesting response data
if result.status == TestStatus.PASS and result.response_data:
if endpoint == '/api/models' and 'models' in result.response_data:
models = result.response_data['models']
print(f" Found {len(models)} models: {models[:3]}{'...' if len(models) > 3 else ''}")
elif endpoint == '/api/databases' and 'databases' in result.response_data:
dbs = result.response_data['databases']
print(f" Available databases: {dbs}")
elif endpoint == '/api/smart-search' and 'strategy' in result.response_data:
strategy = result.response_data['strategy']
time_taken = result.response_data.get('processing_time', 0)
print(f" Strategy: {strategy} | Processing: {time_taken:.2f}s")
print()
def print_summary(self):
"""Print test summary and statistics"""
print("📊 Test Summary")
print("=" * 60)
# Count results by status
status_counts = {}
for result in self.results:
status_counts[result.status] = status_counts.get(result.status, 0) + 1
total_tests = len(self.results)
passed = status_counts.get(TestStatus.PASS, 0)
failed = status_counts.get(TestStatus.FAIL, 0)
warnings = status_counts.get(TestStatus.WARNING, 0)
skipped = status_counts.get(TestStatus.SKIP, 0)
print(f"Total Endpoints Tested: {total_tests}")
print(f"✅ Passed: {passed}")
print(f"❌ Failed: {failed}")
print(f"⚠️ Warnings: {warnings}")
print(f"⏭️ Skipped: {skipped}")
print()
# Calculate success rate
success_rate = (passed / total_tests * 100) if total_tests > 0 else 0
print(f"Success Rate: {success_rate:.1f}%")
print()
# Show failed tests
if failed > 0:
print("❌ Failed Tests:")
for result in self.results:
if result.status == TestStatus.FAIL:
print(f" {result.method} {result.endpoint}: {result.message}")
print()
# Show warnings
if warnings > 0:
print("⚠️ Warnings:")
for result in self.results:
if result.status == TestStatus.WARNING:
print(f" {result.method} {result.endpoint}: {result.message}")
print()
# Performance summary
response_times = [r.response_time for r in self.results if r.response_time > 0]
if response_times:
avg_time = sum(response_times) / len(response_times)
max_time = max(response_times)
print(f"🕐 Performance:")
print(f" Average Response Time: {avg_time:.2f}s")
print(f" Slowest Response: {max_time:.2f}s")
print()
# Overall status
if failed == 0 and warnings == 0:
print("🎉 All critical endpoints are working perfectly!")
elif failed == 0:
print("✅ All critical endpoints are working (some warnings)")
else:
print("🚨 Some endpoints have issues that need attention")
def test_smart_search_scenarios(self):
"""Test smart search with different question types"""
print("🧠 Testing Smart Search Scenarios")
print("-" * 40)
test_scenarios = [
{
"name": "Schema Question",
"question": "What tables are available in the database?",
"expected_strategy": "schema"
},
{
"name": "Count Query",
"question": "How many records are in each table?",
"expected_strategy": "sql"
},
{
"name": "General Database Question",
"question": "Tell me about the database structure",
"expected_strategy": "hybrid"
}
]
for scenario in test_scenarios:
print(f"Testing: {scenario['name']}")
data = {
"question": scenario["question"],
"database": "db3",
"include_sql": True,
"include_semantic": True,
"include_schema": True
}
result = self.test_endpoint(
'smart_search',
'/api/smart-search',
'POST',
data
)
if result.status == TestStatus.PASS:
strategy = result.response_data.get('strategy', 'unknown')
processing_time = result.response_data.get('processing_time', 0)
answer_length = len(result.response_data.get('answer_markdown', ''))
print(f" ✅ Strategy: {strategy} | Time: {processing_time:.2f}s | Answer: {answer_length} chars")
else:
print(f" ❌ {result.message}")
print()
def main():
"""Main testing function"""
tester = APITester()
try:
# Run basic endpoint tests
tester.test_all_endpoints()
# Run smart search scenario tests
tester.test_smart_search_scenarios()
# Print summary
tester.print_summary()
except KeyboardInterrupt:
print("\n🛑 Testing interrupted by user")
except Exception as e:
print(f"\n💥 Unexpected error during testing: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
print("🔧 MCP API Endpoint Comprehensive Tester")
print("Testing all endpoints on ports 8000 and 8002")
print()
main()
