test_api.py

#!/usr/bin/env python3
"""
Test script for the HuggingFace Segment-Based Video Highlights API
"""

import requests
import time
import json
from pathlib import Path

# API configuration
API_BASE = "http://localhost:7860"  # Change to your deployed URL
TEST_VIDEO = "../test_video/test.mp4"  # Adjust path as needed

def test_api():
    """Test the complete API workflow"""
    print("🧪 Testing HuggingFace Segment-Based Video Highlights API")
    
    # Check if test video exists
    if not Path(TEST_VIDEO).exists():
        print(f"❌ Test video not found: {TEST_VIDEO}")
        return
    
    try:
        # 1. Test health endpoint
        print("\n1️⃣ Testing health endpoint...")
        response = requests.get(f"{API_BASE}/health")
        print(f"Health check: {response.status_code} - {response.json()}")
        
        # 2. Upload video
        print("\n2️⃣ Uploading video...")
        with open(TEST_VIDEO, 'rb') as video_file:
            files = {'video': video_file}
            data = {
                'segment_length': 5.0,
                'model_name': 'HuggingFaceTB/SmolVLM2-256M-Video-Instruct',
                'with_effects': True
            }
            response = requests.post(f"{API_BASE}/upload-video", files=files, data=data)
        
        if response.status_code != 200:
            print(f"❌ Upload failed: {response.status_code} - {response.text}")
            return
        
        job_data = response.json()
        job_id = job_data['job_id']
        print(f"✅ Video uploaded successfully! Job ID: {job_id}")
        
        # 3. Monitor job status
        print("\n3️⃣ Monitoring job progress...")
        while True:
            response = requests.get(f"{API_BASE}/job-status/{job_id}")
            if response.status_code != 200:
                print(f"❌ Status check failed: {response.status_code}")
                break
            
            status_data = response.json()
            print(f"Status: {status_data['status']} - {status_data['message']} ({status_data['progress']}%)")
            
            if status_data['status'] == 'completed':
                print(f"✅ Processing completed!")
                print(f"📹 Highlights URL: {status_data['highlights_url']}")
                print(f"📊 Analysis URL: {status_data['analysis_url']}")
                print(f"🎬 Segments: {status_data['selected_segments']}/{status_data['total_segments']}")
                print(f"📈 Compression: {status_data['compression_ratio']:.1%}")
                break
            elif status_data['status'] == 'failed':
                print(f"❌ Processing failed: {status_data['message']}")
                break
            
            time.sleep(5)  # Wait 5 seconds before checking again
        
        # 4. Download results (optional)
        if status_data['status'] == 'completed':
            print("\n4️⃣ Download URLs available:")
            print(f"Highlights: {API_BASE}{status_data['highlights_url']}")
            print(f"Analysis: {API_BASE}{status_data['analysis_url']}")
    
    except requests.exceptions.ConnectionError:
        print(f"❌ Cannot connect to API at {API_BASE}")
        print("Make sure the API server is running with: uvicorn app:app --host 0.0.0.0 --port 7860")
    except Exception as e:
        print(f"❌ Test failed: {str(e)}")

if __name__ == "__main__":
    test_api()