#!/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)