""" Integration test for video generation end-to-end workflow Requirements: - PEXELS_API_KEY environment variable - HF_TTS environment variable - Sufficient disk space for video generation Run with: python tests/integration_test.py """ import asyncio import os import sys import time from pathlib import Path # Add parent directory to path sys.path.insert(0, str(Path(__file__).parent.parent)) from config import Config from video_creator.libraries.tts_client import TTSClient from video_creator.libraries.whisper_client import WhisperClient from video_creator.libraries.pexels_client import PexelsClient from video_creator.music_manager import MusicManager from video_creator.short_creator import ShortCreator from models.schemas import SceneInput, RenderConfig, VoiceEnum, OrientationEnum async def main(): print("=== Short Video Maker Integration Test ===\n") # Load configuration config = Config() if not config.pexels_api_key: print("ERROR: PEXELS_API_KEY not set!") return if not config.hf_tts: print("ERROR: HF_TTS not set!") return print(f"✓ Configuration loaded") print(f" - Data directory: {config.base_data_dir}") print(f" - Whisper model: {config.whisper_model}") # Initialize components print("\n Initializing components...") tts_client = TTSClient(config.hf_tts) print("✓ TTS client initialized") whisper_client = WhisperClient(config.whisper_model, config.whisper_model_dir) print("✓ Whisper client initialized") pexels_client = PexelsClient(config.pexels_api_key) print("✓ Pexels client initialized") music_manager = MusicManager(config.music_dir_path) try: music_manager.ensure_music_files_exist() print(f"✓ Music manager initialized ({len(music_manager.get_music_list())} files)") except Exception as e: print(f"⚠ Music manager warning: {e}") # Create short creator short_creator = ShortCreator( config=config, tts_client=tts_client, whisper_client=whisper_client, pexels_client=pexels_client, music_manager=music_manager ) print("✓ Short creator initialized") # Test video creation print("\n🎬 Creating test video...") scenes = [ SceneInput( text="Welcome to the test video!", searchTerms=["nature", "forest"] ), SceneInput( text="This video was automatically generated using Python and FastAPI.", searchTerms=["technology", "computer"] ) ] render_config = RenderConfig( voice=VoiceEnum.af_heart, orientation=OrientationEnum.portrait, paddingBack=1000 ) video_id = short_creator.add_to_queue(scenes, render_config) print(f"✓ Video added to queue: {video_id}") # Wait for processing print("\n⏳ Waiting for video to process...") max_wait = 300 # 5 minutes start_time = time.time() while time.time() - start_time < max_wait: status = short_creator.get_status(video_id) print(f" Status: {status.value}") if status.value == "ready": print("\n✅ Video created successfully!") break elif status.value == "failed": print("\n❌ Video creation failed!") return await asyncio.sleep(5) else: print("\n⏰ Timeout waiting for video") return # Verify video file video_path = short_creator.get_video_path(video_id) if video_path.exists(): size_mb = video_path.stat().st_size / (1024 * 1024) print(f"\n📹 Video file: {video_path}") print(f" Size: {size_mb:.2f} MB") print(f"\n✅ INTEGRATION TEST PASSED!") else: print("\n❌ Video file not found!") if __name__ == "__main__": asyncio.run(main())