287 lines
9.1 KiB
Python
287 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Simple API Test Script
|
||
|
||
Tests the Geutebruck API without requiring GeViServer.
|
||
Tests authentication, health checks, and basic endpoints.
|
||
"""
|
||
import requests
|
||
import json
|
||
import sys
|
||
from time import sleep
|
||
|
||
# Configuration
|
||
API_URL = "http://localhost:8000"
|
||
USERNAME = "admin"
|
||
PASSWORD = "admin123"
|
||
|
||
# Colors for output
|
||
class Colors:
|
||
GREEN = '\033[92m'
|
||
RED = '\033[91m'
|
||
YELLOW = '\033[93m'
|
||
BLUE = '\033[94m'
|
||
END = '\033[0m'
|
||
BOLD = '\033[1m'
|
||
|
||
def print_header(text):
|
||
print(f"\n{Colors.BOLD}{Colors.BLUE}{'='*60}{Colors.END}")
|
||
print(f"{Colors.BOLD}{Colors.BLUE}{text:^60}{Colors.END}")
|
||
print(f"{Colors.BOLD}{Colors.BLUE}{'='*60}{Colors.END}\n")
|
||
|
||
def print_success(text):
|
||
print(f"{Colors.GREEN}✓ {text}{Colors.END}")
|
||
|
||
def print_error(text):
|
||
print(f"{Colors.RED}✗ {text}{Colors.END}")
|
||
|
||
def print_info(text):
|
||
print(f"{Colors.YELLOW}ℹ {text}{Colors.END}")
|
||
|
||
def test_health():
|
||
"""Test health endpoint"""
|
||
print_header("Testing Health Endpoint")
|
||
|
||
try:
|
||
response = requests.get(f"{API_URL}/health", timeout=5)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
print_success(f"API is running (v{data.get('version', 'unknown')})")
|
||
print_info(f"Environment: {data.get('environment', 'unknown')}")
|
||
|
||
# Check components
|
||
components = data.get('components', {})
|
||
for name, status in components.items():
|
||
if status.get('status') == 'healthy':
|
||
print_success(f"{name}: {status['status']}")
|
||
else:
|
||
print_error(f"{name}: {status.get('status', 'unknown')} - {status.get('error', '')}")
|
||
|
||
return True
|
||
else:
|
||
print_error(f"Health check failed: {response.status_code}")
|
||
return False
|
||
|
||
except requests.exceptions.ConnectionError:
|
||
print_error("Cannot connect to API. Is it running?")
|
||
print_info(f"Make sure to start the API: python src/api/main.py")
|
||
return False
|
||
except Exception as e:
|
||
print_error(f"Error: {e}")
|
||
return False
|
||
|
||
def test_authentication():
|
||
"""Test authentication"""
|
||
print_header("Testing Authentication")
|
||
|
||
try:
|
||
# Login
|
||
print_info(f"Logging in as '{USERNAME}'...")
|
||
response = requests.post(
|
||
f"{API_URL}/api/v1/auth/login",
|
||
json={"username": USERNAME, "password": PASSWORD},
|
||
timeout=5
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
token = data.get('access_token')
|
||
user = data.get('user', {})
|
||
|
||
print_success(f"Login successful!")
|
||
print_info(f"User: {user.get('username')} (Role: {user.get('role')})")
|
||
print_info(f"Token expires in: {data.get('expires_in')} seconds")
|
||
|
||
return token
|
||
else:
|
||
print_error(f"Login failed: {response.status_code}")
|
||
print_error(f"Response: {response.text}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
print_error(f"Authentication error: {e}")
|
||
return None
|
||
|
||
def test_cameras(token):
|
||
"""Test camera endpoints"""
|
||
print_header("Testing Camera Endpoints")
|
||
|
||
if not token:
|
||
print_error("No token available, skipping camera tests")
|
||
return
|
||
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
try:
|
||
# List cameras
|
||
print_info("Fetching camera list...")
|
||
response = requests.get(f"{API_URL}/api/v1/cameras", headers=headers, timeout=5)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
print_success(f"Found {data.get('total', 0)} cameras")
|
||
|
||
cameras = data.get('cameras', [])
|
||
if cameras:
|
||
for cam in cameras[:3]: # Show first 3
|
||
print_info(f" Camera {cam['id']}: {cam['name']} ({cam['status']})")
|
||
if len(cameras) > 3:
|
||
print_info(f" ... and {len(cameras) - 3} more")
|
||
else:
|
||
print_info(" No cameras found (expected if SDK Bridge not connected)")
|
||
else:
|
||
print_error(f"Camera list failed: {response.status_code}")
|
||
|
||
except Exception as e:
|
||
print_error(f"Camera test error: {e}")
|
||
|
||
def test_monitors(token):
|
||
"""Test monitor endpoints"""
|
||
print_header("Testing Monitor Endpoints")
|
||
|
||
if not token:
|
||
print_error("No token available, skipping monitor tests")
|
||
return
|
||
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
try:
|
||
# List monitors
|
||
print_info("Fetching monitor list...")
|
||
response = requests.get(f"{API_URL}/api/v1/monitors", headers=headers, timeout=5)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
print_success(f"Found {data.get('total', 0)} monitors")
|
||
|
||
monitors = data.get('monitors', [])
|
||
if monitors:
|
||
for mon in monitors[:3]: # Show first 3
|
||
cam_id = mon.get('current_camera_id')
|
||
status = f"showing camera {cam_id}" if cam_id else "idle"
|
||
print_info(f" Monitor {mon['id']}: {mon['name']} ({status})")
|
||
if len(monitors) > 3:
|
||
print_info(f" ... and {len(monitors) - 3} more")
|
||
else:
|
||
print_info(" No monitors found (expected if SDK Bridge not connected)")
|
||
else:
|
||
print_error(f"Monitor list failed: {response.status_code}")
|
||
|
||
except Exception as e:
|
||
print_error(f"Monitor test error: {e}")
|
||
|
||
def test_crossswitch(token):
|
||
"""Test cross-switch endpoints"""
|
||
print_header("Testing Cross-Switch Endpoints")
|
||
|
||
if not token:
|
||
print_error("No token available, skipping cross-switch tests")
|
||
return
|
||
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
try:
|
||
# Get routing state
|
||
print_info("Fetching routing state...")
|
||
response = requests.get(f"{API_URL}/api/v1/crossswitch/routing", headers=headers, timeout=5)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
print_success(f"Found {data.get('total', 0)} active routes")
|
||
|
||
routes = data.get('routes', [])
|
||
if routes:
|
||
for route in routes[:3]:
|
||
print_info(f" Camera {route['camera_id']} → Monitor {route['monitor_id']}")
|
||
else:
|
||
print_info(" No active routes")
|
||
else:
|
||
print_error(f"Routing state failed: {response.status_code}")
|
||
|
||
# Note about execution
|
||
print_info("\nNote: Cross-switch execution requires:")
|
||
print_info(" 1. Operator or Administrator role")
|
||
print_info(" 2. SDK Bridge connected to GeViServer")
|
||
print_info(" 3. Valid camera and monitor IDs")
|
||
|
||
except Exception as e:
|
||
print_error(f"Cross-switch test error: {e}")
|
||
|
||
def test_metrics():
|
||
"""Test metrics endpoint"""
|
||
print_header("Testing Metrics Endpoint")
|
||
|
||
try:
|
||
response = requests.get(f"{API_URL}/metrics", timeout=5)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
print_success("Metrics endpoint working")
|
||
|
||
routes = data.get('routes', {})
|
||
print_info(f"Total routes: {routes.get('total', 0)}")
|
||
print_info(f" Auth: {routes.get('auth', 0)}")
|
||
print_info(f" Cameras: {routes.get('cameras', 0)}")
|
||
print_info(f" Monitors: {routes.get('monitors', 0)}")
|
||
print_info(f" Cross-switch: {routes.get('crossswitch', 0)}")
|
||
|
||
features = data.get('features', {})
|
||
enabled = [k for k, v in features.items() if v]
|
||
print_success(f"Features enabled: {len(enabled)}")
|
||
for feature in enabled:
|
||
print_info(f" ✓ {feature}")
|
||
|
||
else:
|
||
print_error(f"Metrics failed: {response.status_code}")
|
||
|
||
except Exception as e:
|
||
print_error(f"Metrics test error: {e}")
|
||
|
||
def main():
|
||
"""Run all tests"""
|
||
print_header("Geutebruck API Test Suite")
|
||
print_info(f"Testing API at: {API_URL}")
|
||
print_info(f"Username: {USERNAME}\n")
|
||
|
||
# Test 1: Health Check
|
||
if not test_health():
|
||
print_error("\nAPI is not running or not reachable!")
|
||
print_info("Start the API with: python src/api/main.py")
|
||
sys.exit(1)
|
||
|
||
# Test 2: Metrics
|
||
test_metrics()
|
||
|
||
# Test 3: Authentication
|
||
token = test_authentication()
|
||
|
||
if token:
|
||
# Test 4: Cameras
|
||
test_cameras(token)
|
||
|
||
# Test 5: Monitors
|
||
test_monitors(token)
|
||
|
||
# Test 6: Cross-switching
|
||
test_crossswitch(token)
|
||
|
||
# Summary
|
||
print_header("Test Summary")
|
||
print_success("Basic API functionality verified!")
|
||
print_info("\nNext steps:")
|
||
print_info(" 1. Start SDK Bridge to connect to GeViServer")
|
||
print_info(" 2. Verify cameras and monitors appear")
|
||
print_info(" 3. Test cross-switch execution")
|
||
print_info("\nDocumentation:")
|
||
print_info(" API Docs: http://localhost:8000/docs")
|
||
print_info(" Usage Guide: docs/usage-guide.md")
|
||
print_info(" Deployment: docs/deployment.md")
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except KeyboardInterrupt:
|
||
print_error("\n\nTest interrupted by user")
|
||
sys.exit(1)
|