From 797cee8695ae6a4ab02e971035c9fbf7d1414f8e Mon Sep 17 00:00:00 2001 From: Geutebruck API Developer Date: Tue, 9 Dec 2025 14:39:25 +0100 Subject: [PATCH] Fix: Add missing Optional import in crossswitch router --- src/api/routers/crossswitch.py | 3 +- test_api.py | 286 +++++++++++++++++++++++++++++++++ 2 files changed, 287 insertions(+), 2 deletions(-) create mode 100644 test_api.py diff --git a/src/api/routers/crossswitch.py b/src/api/routers/crossswitch.py index dd0803c..dba8817 100644 --- a/src/api/routers/crossswitch.py +++ b/src/api/routers/crossswitch.py @@ -1,6 +1,7 @@ """ Cross-switch router for camera-to-monitor routing operations """ +from typing import Optional from fastapi import APIRouter, Depends, status, HTTPException, Query, Request from sqlalchemy.ext.asyncio import AsyncSession import structlog @@ -275,8 +276,6 @@ async def get_routing_history( - Investigate when a camera was last displayed on a monitor - Track operator actions """ - from typing import Optional # Import for type hints - crossswitch_service = CrossSwitchService(db) logger.info("get_routing_history_request", diff --git a/test_api.py b/test_api.py new file mode 100644 index 0000000..07d356c --- /dev/null +++ b/test_api.py @@ -0,0 +1,286 @@ +#!/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)