163 lines
4.6 KiB
Python
163 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to verify COBY system fixes.
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
import os
|
|
|
|
# Add the current directory to Python path
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
try:
|
|
from api.rest_api import create_app
|
|
from caching.redis_manager import redis_manager
|
|
from utils.logging import get_logger, setup_logging
|
|
print("✓ All imports successful")
|
|
except ImportError as e:
|
|
print(f"✗ Import error: {e}")
|
|
sys.exit(1)
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
async def test_health_endpoints():
|
|
"""Test health endpoints"""
|
|
print("\n--- Testing Health Endpoints ---")
|
|
|
|
try:
|
|
# Test Redis manager
|
|
await redis_manager.initialize()
|
|
ping_result = await redis_manager.ping()
|
|
print(f"✓ Redis ping: {ping_result}")
|
|
|
|
# Test app creation
|
|
app = create_app()
|
|
print("✓ FastAPI app created successfully")
|
|
|
|
# Test health endpoint logic
|
|
from api.response_formatter import ResponseFormatter
|
|
formatter = ResponseFormatter()
|
|
|
|
health_data = {
|
|
'status': 'healthy' if ping_result else 'degraded',
|
|
'redis': 'connected' if ping_result else 'disconnected',
|
|
'version': '1.0.0'
|
|
}
|
|
|
|
response = formatter.status_response(health_data)
|
|
print(f"✓ Health response format: {type(response)}")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"✗ Health endpoint test failed: {e}")
|
|
return False
|
|
|
|
async def test_static_files():
|
|
"""Test static file serving"""
|
|
print("\n--- Testing Static Files ---")
|
|
|
|
try:
|
|
static_path = os.path.join(os.path.dirname(__file__), "web", "static")
|
|
index_path = os.path.join(static_path, "index.html")
|
|
|
|
if os.path.exists(static_path):
|
|
print(f"✓ Static directory exists: {static_path}")
|
|
else:
|
|
print(f"✗ Static directory missing: {static_path}")
|
|
return False
|
|
|
|
if os.path.exists(index_path):
|
|
print(f"✓ Index.html exists: {index_path}")
|
|
|
|
# Test reading the file
|
|
with open(index_path, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
if "COBY" in content:
|
|
print("✓ Index.html contains COBY content")
|
|
else:
|
|
print("✗ Index.html missing COBY content")
|
|
return False
|
|
else:
|
|
print(f"✗ Index.html missing: {index_path}")
|
|
return False
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"✗ Static files test failed: {e}")
|
|
return False
|
|
|
|
async def test_websocket_config():
|
|
"""Test WebSocket configuration"""
|
|
print("\n--- Testing WebSocket Configuration ---")
|
|
|
|
try:
|
|
from api.simple_websocket_server import WebSocketServer
|
|
from simple_config import config
|
|
|
|
ws_server = WebSocketServer(
|
|
host=config.api.host,
|
|
port=config.api.websocket_port
|
|
)
|
|
print(f"✓ WebSocket server configured: {config.api.host}:{config.api.websocket_port}")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"✗ WebSocket config test failed: {e}")
|
|
return False
|
|
|
|
async def main():
|
|
"""Run all tests"""
|
|
print("COBY System Fix Verification")
|
|
print("=" * 40)
|
|
|
|
# Setup logging
|
|
setup_logging(level='INFO')
|
|
|
|
tests = [
|
|
test_health_endpoints,
|
|
test_static_files,
|
|
test_websocket_config
|
|
]
|
|
|
|
results = []
|
|
for test in tests:
|
|
try:
|
|
result = await test()
|
|
results.append(result)
|
|
except Exception as e:
|
|
print(f"✗ Test {test.__name__} failed with exception: {e}")
|
|
results.append(False)
|
|
|
|
print("\n" + "=" * 40)
|
|
print("Test Results Summary:")
|
|
|
|
passed = sum(results)
|
|
total = len(results)
|
|
|
|
for i, result in enumerate(results):
|
|
status = "✓ PASS" if result else "✗ FAIL"
|
|
print(f" Test {i+1}: {status}")
|
|
|
|
print(f"\nOverall: {passed}/{total} tests passed")
|
|
|
|
if passed == total:
|
|
print("🎉 All tests passed! COBY system should work correctly.")
|
|
return 0
|
|
else:
|
|
print("❌ Some tests failed. Please check the issues above.")
|
|
return 1
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
exit_code = asyncio.run(main())
|
|
sys.exit(exit_code)
|
|
except KeyboardInterrupt:
|
|
print("\nTest interrupted by user")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"Test failed with error: {e}")
|
|
sys.exit(1) |