""" API tests module """ import os import sys from pathlib import Path from typing import Dict, Optional, List from tests.utils.logger import Logger from tests.utils.http_utils import HTTPUtils from tests.utils.api_analyzer import APIAnalyzer class APITests: """Test all API endpoints""" def __init__(self, logger: Logger, base_url: str = "http://localhost:8000", project_dir: str = None, version: str = "unknown"): self.logger = logger self.version = version self.http_utils = HTTPUtils(logger, base_url) self.project_dir = Path(project_dir) if project_dir else Path(__file__).parent.parent.parent self.api_analyzer = APIAnalyzer(str(self.project_dir), version=version) self.test_user = { "username": "testadmin", "email": "testadmin@example.com", "password": "TestAdmin123!" } self.access_token: Optional[str] = None self.user_id: Optional[str] = None def run_all_tests(self, auto_fix: bool = True) -> bool: """Run all API tests""" self.logger.subheader(f"API Tests (v{self.version})") # Create test admin user if not self._create_test_user(auto_fix): return False # Test authentication flow if not self._test_authentication(): return False # Test health endpoint if not self._test_health_endpoint(): return False # Test stream endpoints if not self._test_stream_endpoints(): return False # Test comment endpoints if not self._test_comment_endpoints(): return False # Test tag endpoints if not self._test_tag_endpoints(): return False # Test settings endpoints if not self._test_settings_endpoints(): return False # Test watch history endpoints if not self._test_watch_history_endpoints(): return False # Test user endpoints if not self._test_user_endpoints(): return False # Test admin endpoints if not self._test_admin_endpoints(): return False # Test backup endpoints if not self._test_backup_endpoints(): return False # Test chat endpoints if not self._test_chat_endpoints(): return False # Test twitch auth endpoints if not self._test_twitch_auth_endpoints(): return False # Test manual import with chat.json if not self._test_manual_import(): return False # Generate endpoint coverage report self._generate_endpoint_coverage_report() self.logger.success("API", "All API tests passed") return True def _create_test_user(self, auto_fix: bool) -> bool: """Create test admin user if not exists""" self.logger.info("API", "Creating test admin user...") # Try to register the user success, status_code, response = self.http_utils.post( "/api/auth/register", json_data=self.test_user ) if success and status_code == 201: self.logger.success("API", "Test admin user created successfully") # Login after successful registration to get token if not self._login_user(): return False # Set user as admin if not already admin return self._set_user_as_admin() elif success and status_code == 400: # User might already exist, try to login self.logger.info("API", "User might already exist, attempting login...") if not self._login_user(): return False # Set user as admin if not already admin return self._set_user_as_admin() else: self.logger.error("API", f"Failed to create test user: {response}") if auto_fix: self.logger.info("API", "Attempting to login with existing user...") if not self._login_user(): return False # Set user as admin if not already admin return self._set_user_as_admin() return False def _login_user(self) -> bool: """Login as test user""" success, status_code, response = self.http_utils.post( "/api/auth/login", json_data={ "username": self.test_user["username"], "password": self.test_user["password"] } ) if success and status_code == 200: self.access_token = response.get("access_token") self.user_id = response.get("user", {}).get("id") self.http_utils.set_auth_token(self.access_token) # Store cookies from response for refresh test if self.http_utils.last_response: cookies = self.http_utils.last_response.cookies for cookie_name, cookie_value in cookies.items(): self.http_utils.session.cookies.set(cookie_name, cookie_value) self.logger.success("API", "Login successful") return True else: self.logger.error("API", f"Login failed: {response}") return False def _set_user_as_admin(self) -> bool: """Set test user as admin directly in database""" try: import subprocess import uuid self.logger.info("API", "Setting test user as admin...") # Get user_id from /me endpoint success, status_code, response = self.http_utils.get("/api/auth/me") if not success or status_code != 200: self.logger.error("API", f"Failed to get user info: {response}") return False user_id = response.get("id") if not user_id: self.logger.error("API", "User ID not found in response") return False # Validate UUID format before using in SQL to prevent injection try: uuid.UUID(str(user_id)) except ValueError: self.logger.error("API", f"Invalid user ID format: {user_id}") return False # Update user role in database directly cmd = [ "docker", "compose", "exec", "postgres", "psql", "-U", "streama", "-d", "streama2026", "-c", f"UPDATE users SET role_id = (SELECT id FROM roles WHERE name = 'admin') WHERE id = '{user_id}'::uuid;" ] result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) if result.returncode == 0: self.logger.success("API", "Test user set as admin successfully") return True else: self.logger.error("API", f"Failed to set user as admin: {result.stderr}") return False except Exception as e: self.logger.error("API", f"Failed to set user as admin: {e}") return False def _test_authentication(self) -> bool: """Test authentication flow""" self.logger.info("API", "Testing authentication flow...") # Test /me endpoint success, status_code, response = self.http_utils.get("/api/auth/me") if success and status_code == 200: self.logger.success("API", "GET /api/auth/me: 200") else: self.logger.error("API", f"GET /api/auth/me failed: {status_code}") return False # Test token refresh endpoint success, status_code, response = self.http_utils.post("/api/auth/refresh") if success and status_code == 200: self.logger.success("API", "POST /api/auth/refresh: 200") else: self.logger.warning("API", f"POST /api/auth/refresh failed: {status_code} (may require cookie)") return True def _test_health_endpoint(self) -> bool: """Test health endpoint""" self.logger.info("API", "Testing health endpoint...") success, status_code, response = self.http_utils.get("/api/health") if success and status_code == 200: self.logger.success("API", "GET /api/health: 200") return True else: self.logger.error("API", f"GET /api/health failed: {status_code}") return False def _test_stream_endpoints(self) -> bool: """Test stream endpoints""" self.logger.info("API", "Testing stream endpoints...") endpoints = [ ("GET /api/streams", "/api/streams"), ("GET /api/streams/latest", "/api/streams/latest"), ("GET /api/streams/popular", "/api/streams/popular"), ("GET /api/streams/random", "/api/streams/random"), ("GET /api/streams/admin/all", "/api/streams/admin/all"), ] all_ok = True for name, endpoint in endpoints: success, status_code, response = self.http_utils.get(endpoint) if success and status_code == 200: self.logger.success("API", f"{name}: 200") elif success and status_code == 401: self.logger.warning("API", f"{name}: 401 (authentication required)") elif success and status_code == 403: self.logger.warning("API", f"{name}: 403 (admin required)") else: self.logger.error("API", f"{name} failed: {status_code}") all_ok = False # Test search endpoint success, status_code, response = self.http_utils.get("/api/streams/search", params={"q": "test"}) if success and status_code == 200: self.logger.success("API", "GET /api/streams/search: 200") else: self.logger.warning("API", "GET /api/streams/search: no results (expected)") return all_ok def _test_comment_endpoints(self) -> bool: """Test comment endpoints""" self.logger.info("API", "Testing comment endpoints...") # Test comments endpoint (requires stream_id) # Comments are under /api/streams/{stream_id}/comments # Test with invalid stream_id to check if endpoint exists success, status_code, response = self.http_utils.get("/api/streams/00000000-0000-0000-0000-000000000000/comments") if success and status_code == 401: self.logger.info("API", "GET /api/streams/{stream_id}/comments: 401 (authentication required)") elif success and status_code == 404: self.logger.info("API", "GET /api/streams/{stream_id}/comments: 404 (invalid stream_id - endpoint exists)") elif success and status_code == 403: self.logger.info("API", "GET /api/streams/{stream_id}/comments: 403 (insufficient permissions)") elif success: self.logger.warning("API", f"GET /api/streams/{{stream_id}}/comments: {status_code}") else: self.logger.error("API", "GET /api/streams/{stream_id}/comments: request failed") return True # Not critical def _test_tag_endpoints(self) -> bool: """Test tag endpoints""" self.logger.info("API", "Testing tag endpoints...") # List tags (requires authentication) success, status_code, response = self.http_utils.get("/api/tags") if success and status_code == 200: self.logger.success("API", "GET /api/tags: 200") return True elif success and status_code == 401: self.logger.warning("API", "GET /api/tags: 401 (authentication required - test user may not be logged in)") return True # Not critical - endpoint exists but requires auth else: self.logger.error("API", f"GET /api/tags failed: {status_code}") return False def _test_settings_endpoints(self) -> bool: """Test settings endpoints""" self.logger.info("API", "Testing settings endpoints...") # Test public settings endpoint success, status_code, response = self.http_utils.get("/api/settings") if success and status_code == 200: self.logger.success("API", "GET /api/settings: 200") return True else: self.logger.warning("API", f"GET /api/settings failed: {status_code} (may be optional)") return True # Not critical def _test_watch_history_endpoints(self) -> bool: """Test watch history endpoints""" self.logger.info("API", "Testing watch history endpoints...") # Test watch history list endpoint success, status_code, response = self.http_utils.get("/api/users/me/watch-history") if success and status_code == 200: self.logger.success("API", "GET /api/users/me/watch-history: 200") return True else: self.logger.warning("API", f"GET /api/users/me/watch-history failed: {status_code} (may be optional)") return True # Not critical def _test_user_endpoints(self) -> bool: """Test user endpoints""" self.logger.info("API", "Testing user endpoints...") # Test /api/users/me (already tested in auth flow) # Test /api/users/me/watch-history (already tested separately) # Test profile image endpoint success, status_code, response = self.http_utils.post("/api/users/me/profile-image") if success and status_code == 401: self.logger.info("API", "POST /api/users/me/profile-image: 401 (authentication required)") elif success and status_code == 422: self.logger.info("API", "POST /api/users/me/profile-image: 422 (no image data sent - expected)") elif success: self.logger.warning("API", f"POST /api/users/me/profile-image: {status_code}") else: self.logger.error("API", f"POST /api/users/me/profile-image failed: {status_code}") # Test admin users endpoint success, status_code, response = self.http_utils.get("/api/admin/users") if success and status_code == 200: self.logger.success("API", "GET /api/admin/users: 200") elif success and status_code == 401: self.logger.warning("API", "GET /api/admin/users: 401 (not admin)") elif success and status_code == 403: self.logger.warning("API", "GET /api/admin/users: 403 (insufficient permissions)") else: self.logger.warning("API", f"GET /api/admin/users failed: {status_code}") return True # Not critical def _test_admin_endpoints(self) -> bool: """Test admin endpoints""" self.logger.info("API", "Testing admin endpoints...") # Test admin streams endpoint success, status_code, response = self.http_utils.get("/api/streams/admin/all") if success and status_code == 200: self.logger.success("API", "GET /api/streams/admin/all: 200") return True elif success and status_code == 403: self.logger.warning("API", "GET /api/streams/admin/all: 403 (not admin)") return True # Not critical for basic tests else: self.logger.error("API", f"GET /api/streams/admin/all failed: {status_code}") return False def _test_backup_endpoints(self) -> bool: """Test backup endpoints""" self.logger.info("API", "Testing backup endpoints...") # List backups (admin only) - under /api/admin/backup success, status_code, response = self.http_utils.get("/api/admin/backup") if success and status_code == 200: self.logger.success("API", "GET /api/admin/backup: 200") return True elif success and status_code == 403: self.logger.warning("API", "GET /api/admin/backup: 403 (not admin)") return True # Not critical for basic tests else: self.logger.warning("API", f"GET /api/admin/backup failed: {status_code}") return True # Not critical def _test_chat_endpoints(self) -> bool: """Test chat endpoints""" self.logger.info("API", "Testing chat endpoints...") # Test chat list endpoint success, status_code, response = self.http_utils.get("/api/chat") if success and status_code == 200: self.logger.success("API", "GET /api/chat: 200") elif success and status_code == 401: self.logger.info("API", "GET /api/chat: 401 (authentication required)") elif success and status_code == 404: self.logger.info("API", "GET /api/chat: 404 (chat feature not implemented)") else: self.logger.warning("API", f"GET /api/chat: {status_code}") # Test chat proxy image endpoint success, status_code, response = self.http_utils.get("/api/chat/proxy-image") if success and status_code in [400, 401]: self.logger.info("API", "GET /api/chat/proxy-image: endpoint exists (requires parameters)") elif success and status_code == 404: self.logger.info("API", "GET /api/chat/proxy-image: 404 (chat feature not implemented)") elif success: self.logger.warning("API", f"GET /api/chat/proxy-image: {status_code}") else: self.logger.info("API", "GET /api/chat/proxy-image: request failed") return True # Not critical def _test_twitch_auth_endpoints(self) -> bool: """Test twitch auth endpoints""" self.logger.info("API", "Testing twitch auth endpoints...") # Test twitch login endpoint success, status_code, response = self.http_utils.get("/api/twitch-auth/login") if success and status_code in [302, 307]: self.logger.success("API", "GET /api/twitch-auth/login: redirect (expected)") elif success and status_code == 404: self.logger.info("API", "GET /api/twitch-auth/login: 404 (twitch auth not implemented)") elif success: self.logger.warning("API", f"GET /api/twitch-auth/login: {status_code}") else: self.logger.info("API", "GET /api/twitch-auth/login: request failed") # Test twitch unlink endpoint success, status_code, response = self.http_utils.post("/api/twitch-auth/unlink") if success and status_code == 401: self.logger.info("API", "POST /api/twitch-auth/unlink: 401 (authentication required)") elif success and status_code == 204: self.logger.success("API", "POST /api/twitch-auth/unlink: 204 (no content)") elif success and status_code == 404: self.logger.info("API", "POST /api/twitch-auth/unlink: 404 (twitch auth not implemented)") else: self.logger.info("API", f"POST /api/twitch-auth/unlink: {status_code}") return True # Not critical def cleanup(self): """Cleanup test user if needed""" if self.user_id: self.logger.info("API", "Cleaning up test user via PostgreSQL...") try: import subprocess from pathlib import Path project_root = Path(__file__).parent.parent.parent # Delete test user directly via PostgreSQL cmd = [ "docker", "compose", "exec", "-T", "postgres", "psql", "-U", "streama", "-d", "streama2026", "-c", f"DELETE FROM users WHERE username = '{self.test_user['username']}';" ] result = subprocess.run( cmd, cwd=str(project_root), capture_output=True, text=True, timeout=10 ) if result.returncode == 0: self.logger.success("API", "Test user deleted successfully via PostgreSQL") else: self.logger.warning("API", f"Failed to delete test user: {result.stderr}") except Exception as e: self.logger.warning("API", f"Failed to cleanup test user via PostgreSQL: {e}") def _generate_endpoint_coverage_report(self) -> None: """Generate endpoint coverage report""" self.logger.info("API", "Generating endpoint coverage report...") try: coverage_report = self.api_analyzer.generate_endpoint_coverage_report() self.logger.info("API", f"Total backend endpoints: {coverage_report['total_backend_endpoints']}") self.logger.info("API", f"Total frontend calls: {coverage_report['total_frontend_calls']}") self.logger.info("API", f"Covered endpoints: {coverage_report['covered_count']} ({coverage_report['covered_percentage']}%)") self.logger.info("API", f"Unused endpoints: {coverage_report['unused_count']} ({coverage_report['unused_percentage']}%)") if coverage_report.get('intentionally_unused_count', 0) > 0: self.logger.info("API", f"Intentionally unused (internal): {coverage_report['intentionally_unused_count']} ({coverage_report.get('intentionally_unused_percentage', 0)}%)") if coverage_report['new_endpoints']: self.logger.info("API", f"New endpoints detected: {coverage_report['new_endpoints_count']}") for endpoint in coverage_report['new_endpoints']: self.logger.info("API", f" - {endpoint['method']} {endpoint['path']} (new)") if coverage_report['unused_endpoints']: self.logger.info("API", f"Unused endpoints (may be intentional):") for endpoint in coverage_report['unused_endpoints'][:5]: # Show first 5 self.logger.info("API", f" - {endpoint['method']} {endpoint['path']}") if len(coverage_report['unused_endpoints']) > 5: self.logger.info("API", f" ... and {len(coverage_report['unused_endpoints']) - 5} more") if coverage_report.get('intentionally_unused_endpoints'): self.logger.info("API", f"Intentionally unused (internal features):") for endpoint in coverage_report['intentionally_unused_endpoints'][:5]: self.logger.info("API", f" - {endpoint['method']} {endpoint['path']}") if len(coverage_report['intentionally_unused_endpoints']) > 5: self.logger.info("API", f" ... and {len(coverage_report['intentionally_unused_endpoints']) - 5} more") if coverage_report['missing_endpoints']: self.logger.warning("API", f"Missing backend endpoints: {coverage_report['missing_count']}") for endpoint in coverage_report['missing_endpoints']: self.logger.warning("API", f" - {endpoint['method']} {endpoint['path']} called from {endpoint['file']}:{endpoint['line']}") except Exception as e: self.logger.error("API", f"Failed to generate endpoint coverage report: {e}") def _test_manual_import(self) -> bool: """Test manual stream import with chat.json""" self.logger.info("API", "Testing manual import with chat.json...") try: from pathlib import Path project_root = Path(__file__).parent.parent.parent chat_json_path = project_root / "tests" / "test_data" / "chat.json" if not chat_json_path.exists(): self.logger.warning("API", "chat.json not found - skipping manual import test") return True # Not critical if file doesn't exist # Read chat.json file with open(chat_json_path, 'r', encoding='utf-8') as f: chat_json_content = f.read() # Create manual stream with chat.json files = { 'chat_json_file': ('chat.json', chat_json_content, 'application/json') } data = { 'title': 'Test Stream from chat.json', 'description': 'Test import with Factor_KS stream', 'is_public': False } success, status_code, response = self.http_utils.post_with_files( "/api/streams/manual", data=data, files=files ) if success and status_code in [200, 201]: stream_id = response.get("id") stream_status = response.get("status") stream_hls_path = response.get("hls_path") self.logger.success("API", f"Manual import successful: {stream_id}") self.logger.info("API", f" - Status: {stream_status}") self.logger.info("API", f" - HLS Path: {stream_hls_path}") # Check if stream is ready or pending if stream_status == 'ready': self.logger.success("API", " ✓ Stream is ready") elif stream_status == 'pending': if stream_hls_path: self.logger.warning("API", " ⚠ Stream is pending but has HLS path - should be ready") else: self.logger.info("API", " ⚠ Stream is pending (no HLS path)") else: self.logger.warning("API", f" ⚠ Stream status: {stream_status}") # Get stream details to verify success, status_code, response = self.http_utils.get(f"/api/streams/{stream_id}") if success and status_code == 200: stream_details = response self.logger.info("API", "Stream details verified:") self.logger.info("API", f" - Title: {stream_details.get('title')}") self.logger.info("API", f" - Status: {stream_details.get('status')}") self.logger.info("API", f" - HLS Path: {stream_details.get('hls_path')}") self.logger.info("API", f" - Thumbnail URL: {stream_details.get('thumbnail_url')}") self.logger.info("API", f" - Duration: {stream_details.get('duration_seconds')}s") self.logger.info("API", f" - Streamer: {stream_details.get('streamer')}") self.logger.info("API", f" - Twitch VOD ID: {stream_details.get('twitch_vod_id')}") # Check if thumbnail URL is set if stream_details.get('thumbnail_url'): self.logger.success("API", " ✓ Twitch thumbnail URL set") else: self.logger.warning("API", " ⚠ Thumbnail URL not set") # Delete the test stream after successful import self.logger.info("API", f"Cleaning up test stream {stream_id}...") success, status_code, response = self.http_utils.delete(f"/api/streams/{stream_id}") if success and status_code in [200, 204]: self.logger.success("API", f"Test stream {stream_id} deleted successfully") else: self.logger.warning("API", f"Failed to delete test stream {stream_id}: {status_code}") return True else: self.logger.error("API", f"Manual import failed: {status_code}") self.logger.error("API", f"Response: {response}") return False except Exception as e: self.logger.error("API", f"Manual import test crashed: {e}") return False