"""End-to-end test script for the content creation platform.""" import requests import time import json from typing import Optional class ContentCreatorClient: """Client for testing the content creation API.""" def __init__(self, base_url: str = "http://localhost:8000"): """Initialize client. Args: base_url: Base URL of the API gateway """ self.base_url = base_url self.token: Optional[str] = None def register(self, email: str, username: str, password: str) -> dict: """Register a new user. Args: email: User email username: Username password: Password Returns: User data """ response = requests.post( f"{self.base_url}/auth/register", json={ "email": email, "username": username, "password": password } ) response.raise_for_status() return response.json() def login(self, username: str, password: str) -> str: """Login and get access token. Args: username: Username or email password: Password Returns: Access token """ response = requests.post( f"{self.base_url}/auth/login", params={"username": username, "password": password} ) response.raise_for_status() data = response.json() self.token = data["access_token"] return self.token def create_project( self, name: str, prompt: str, audio_duration: int = 30, audio_style: str = "cinematic", video_resolution: str = "1080p" ) -> dict: """Create a content generation project. Args: name: Project name prompt: Text prompt audio_duration: Audio duration in seconds audio_style: Music style video_resolution: Video resolution Returns: Project data """ if not self.token: raise ValueError("Must login first") response = requests.post( f"{self.base_url}/projects", headers={"Authorization": f"Bearer {self.token}"}, json={ "name": name, "prompt": prompt, "audio_settings": { "duration": audio_duration, "style": audio_style }, "video_settings": { "resolution": video_resolution, "provider": "mock" } } ) response.raise_for_status() return response.json() def get_project(self, project_id: str) -> dict: """Get project status. Args: project_id: Project ID Returns: Project data """ if not self.token: raise ValueError("Must login first") response = requests.get( f"{self.base_url}/projects/{project_id}", headers={"Authorization": f"Bearer {self.token}"} ) response.raise_for_status() return response.json() def list_projects(self) -> list: """List all projects. Returns: List of projects """ if not self.token: raise ValueError("Must login first") response = requests.get( f"{self.base_url}/projects", headers={"Authorization": f"Bearer {self.token}"} ) response.raise_for_status() return response.json() def wait_for_project(self, project_id: str, timeout: int = 120) -> dict: """Wait for project to complete. Args: project_id: Project ID timeout: Maximum wait time in seconds Returns: Completed project data """ start_time = time.time() while time.time() - start_time < timeout: project = self.get_project(project_id) if project["status"] == "completed": return project elif project["status"] == "failed": raise Exception(f"Project failed: {project.get('error_message')}") print(f"Status: {project['status']}... waiting") time.sleep(3) raise TimeoutError(f"Project did not complete within {timeout}s") def run_test(): """Run end-to-end test.""" print("\n" + "="*60) print("🧪 AI Content Creation Platform - End-to-End Test") print("="*60) client = ContentCreatorClient() # Test 1: Health check print("\n1️⃣ Testing health endpoint...") response = requests.get("http://localhost:8000/health") print(f"✅ Health check: {response.json()['status']}") # Test 2: User registration print("\n2️⃣ Registering user...") try: user = client.register( email=f"test{int(time.time())}@example.com", username=f"testuser{int(time.time())}", password="testpass123" ) print(f"✅ User registered: {user['username']}") except Exception as e: print(f"ℹ️ Registration skipped (user may exist): {e}") # Test 3: Login print("\n3️⃣ Logging in...") username = f"testuser{int(time.time())}" if 'user' in locals() else "admin" password = "testpass123" if 'user' in locals() else "admin" try: token = client.login(username, password) print(f"✅ Logged in successfully") print(f" Token: {token[:20]}...") except: # Try default admin token = client.login("admin", "admin") print(f"✅ Logged in as admin") # Test 4: Create project print("\n4️⃣ Creating content project...") project = client.create_project( name="Test Epic Music Video", prompt="Epic cinematic orchestral music with dramatic mountain landscapes at sunset", audio_duration=15, # Short duration for testing audio_style="cinematic", video_resolution="720p" ) print(f"✅ Project created: {project['id']}") print(f" Name: {project['name']}") print(f" Status: {project['status']}") # Test 5: Monitor project print("\n5️⃣ Monitoring project progress...") print(" (This will take a moment as content is generated...)") try: completed_project = client.wait_for_project(project['id'], timeout=120) print(f"\n✅ Project completed!") print(f" Status: {completed_project['status']}") if completed_project.get('audio_file'): print(f" Audio: {completed_project['audio_file']}") if completed_project.get('video_file'): print(f" Video: {completed_project['video_file']}") if completed_project.get('final_file'): print(f" Final: {completed_project['final_file']}") except TimeoutError: print("⏱️ Project still processing (check status later)") current_status = client.get_project(project['id']) print(f" Current status: {current_status['status']}") # Test 6: List projects print("\n6️⃣ Listing all projects...") projects = client.list_projects() print(f"✅ Found {len(projects)} projects") for i, p in enumerate(projects[:3], 1): print(f" {i}. {p['name']} - {p['status']}") # Summary print("\n" + "="*60) print("🎉 Test Complete!") print("="*60) print("\nNext steps:") print("1. Check project details at http://localhost:8000/docs") print("2. View generated files in ./media directory") print("3. Try creating more projects with different prompts") print("\n") if __name__ == "__main__": try: run_test() except requests.exceptions.ConnectionError: print("\n❌ Error: Could not connect to API") print("Make sure the API gateway is running:") print(" python gateway/main.py") except Exception as e: print(f"\n❌ Test failed: {e}") import traceback traceback.print_exc()