Copy #!/usr/bin/env python3
# testing/health-check-suite.py
import requests
import time
import sys
from typing import Dict, List, Any
class ProductionHealthChecker:
def __init__(self, base_url: str = "https://api.jaegis.ai"):
self.base_url = base_url
self.health_checks = [
self.check_api_health,
self.check_authentication,
self.check_processing_pipeline,
self.check_database_connectivity,
self.check_cache_connectivity,
self.check_external_dependencies,
self.check_performance_metrics
]
def run_all_checks(self) -> Dict[str, Any]:
"""Run all health checks"""
results = {
'overall_status': 'HEALTHY',
'timestamp': time.time(),
'checks': {}
}
for check in self.health_checks:
check_name = check.__name__
try:
result = check()
results['checks'][check_name] = result
if not result.get('healthy', False):
results['overall_status'] = 'UNHEALTHY'
except Exception as e:
results['checks'][check_name] = {
'healthy': False,
'error': str(e)
}
results['overall_status'] = 'UNHEALTHY'
return results
def check_api_health(self) -> Dict[str, Any]:
"""Check API health endpoint"""
response = requests.get(f"{self.base_url}/health", timeout=10)
return {
'healthy': response.status_code == 200,
'response_time_ms': response.elapsed.total_seconds() * 1000,
'status_code': response.status_code
}
def check_authentication(self) -> Dict[str, Any]:
"""Check authentication system"""
# Test invalid auth
response = requests.get(f"{self.base_url}/status", timeout=10)
return {
'healthy': response.status_code == 401, # Should require auth
'response_time_ms': response.elapsed.total_seconds() * 1000,
'status_code': response.status_code
}
def check_processing_pipeline(self) -> Dict[str, Any]:
"""Check processing pipeline with test request"""
test_payload = {
'input_text': 'Test processing pipeline health check',
'mode': 'standard'
}
# This would need valid API key in production
headers = {'Authorization': 'Bearer test-api-key'}
response = requests.post(
f"{self.base_url}/process",
json=test_payload,
headers=headers,
timeout=30
)
return {
'healthy': response.status_code in [200, 401], # 401 if no valid key
'response_time_ms': response.elapsed.total_seconds() * 1000,
'status_code': response.status_code
}
def check_database_connectivity(self) -> Dict[str, Any]:
"""Check database connectivity through API"""
# This would be implemented as a dedicated health check endpoint
response = requests.get(f"{self.base_url}/health/database", timeout=10)
return {
'healthy': response.status_code == 200,
'response_time_ms': response.elapsed.total_seconds() * 1000
}
def check_cache_connectivity(self) -> Dict[str, Any]:
"""Check Redis cache connectivity"""
response = requests.get(f"{self.base_url}/health/cache", timeout=10)
return {
'healthy': response.status_code == 200,
'response_time_ms': response.elapsed.total_seconds() * 1000
}
def check_external_dependencies(self) -> Dict[str, Any]:
"""Check external API dependencies"""
dependencies = {
'openrouter': 'https://openrouter.ai',
'github': 'https://api.github.com'
}
results = {}
all_healthy = True
for name, url in dependencies.items():
try:
response = requests.get(url, timeout=10)
results[name] = {
'healthy': response.status_code < 500,
'response_time_ms': response.elapsed.total_seconds() * 1000
}
if not results[name]['healthy']:
all_healthy = False
except Exception as e:
results[name] = {'healthy': False, 'error': str(e)}
all_healthy = False
return {
'healthy': all_healthy,
'dependencies': results
}
def check_performance_metrics(self) -> Dict[str, Any]:
"""Check performance metrics"""
response = requests.get(f"{self.base_url}/metrics", timeout=10)
return {
'healthy': response.status_code in [200, 401], # May require auth
'response_time_ms': response.elapsed.total_seconds() * 1000
}
if __name__ == "__main__":
checker = ProductionHealthChecker()
results = checker.run_all_checks()
print(f"Overall Status: {results['overall_status']}")
for check_name, result in results['checks'].items():
status = "✅ PASS" if result.get('healthy', False) else "❌ FAIL"
print(f"{status} {check_name}")
if 'response_time_ms' in result:
print(f" Response Time: {result['response_time_ms']:.1f}ms")
if 'error' in result:
print(f" Error: {result['error']}")
# Exit with error code if unhealthy
sys.exit(0 if results['overall_status'] == 'HEALTHY' else 1)