test_runway_urls.py

#!/usr/bin/env python3
"""
Test script to verify different image URLs work with Runway video generation.
This helps identify which URL formats work best.
"""

import json
import time
from typing import List, Dict

# Test URLs - mix of working and potentially problematic ones
TEST_URLS = [
    {
        "name": "Unsplash Simple",
        "url": "https://images.unsplash.com/photo-1506905925346-21bda4d32df4?w=800&q=80",
        "expected": "should_work"
    },
    {
        "name": "Unsplash with Width",
        "url": "https://images.unsplash.com/photo-1441974231531-c6227db76b6e?w=1024",
        "expected": "should_work"
    },
    {
        "name": "Unsplash No Parameters",
        "url": "https://images.unsplash.com/photo-1439066615861-d1af74d74000",
        "expected": "might_work"
    },
    {
        "name": "Unsplash Complex Parameters",
        "url": "https://images.unsplash.com/photo-1506905925346-21bda4d32df4?ixlib=rb-4.0.3&auto=format&fit=crop&w=2000&q=80",
        "expected": "might_fail"
    }
]


def test_url_with_api(url: str, name: str) -> Dict:
    """Test a single URL with the API."""
    print(f"\n🧪 Testing: {name}")
    print(f"   URL: {url}")

    try:
        import requests

        payload = {
            "image_url": url,
            "prompt_text": f"Test video for {name}",
            "ratio": "1280:720",
            "duration": 5
        }

        response = requests.post(
            "http://localhost:8000/generate-video",
            json=payload,
            timeout=120  # 2 minute timeout for quick test
        )

        if response.status_code == 200:
            result = response.json()
            status = result.get("status", "unknown")

            if status == "success":
                print(f"   ✅ Success! Video URL: {result.get('video_url', 'N/A')}")
                return {"name": name, "url": url, "result": "success", "video_url": result.get('video_url')}
            else:
                error = result.get("error", "Unknown error")
                print(f"   ❌ Failed: {error}")
                return {"name": name, "url": url, "result": "failed", "error": error}
        else:
            print(f"   ❌ HTTP {response.status_code}: {response.text}")
            return {"name": name, "url": url, "result": "http_error", "error": response.text}

    except requests.exceptions.Timeout:
        print(f"   ⏰ Timeout - video might still be processing")
        return {"name": name, "url": url, "result": "timeout"}
    except Exception as e:
        print(f"   ❌ Error: {e}")
        return {"name": name, "url": url, "result": "error", "error": str(e)}


def test_url_optimization():
    """Test the URL optimization function directly."""
    print("\n🔧 Testing URL optimization function...")

    try:
        from src.file_processors.runway_processor import RunwayProcessor
        processor = RunwayProcessor()

        test_cases = [
            "https://images.unsplash.com/photo-1506905925346-21bda4d32df4",
            "https://images.unsplash.com/photo-1506905925346-21bda4d32df4?ixlib=rb-4.0.3&auto=format",
            "https://images.unsplash.com/photo-1506905925346-21bda4d32df4?w=500",
            "https://example.com/image.jpg"
        ]

        for url in test_cases:
            optimized = processor._optimize_image_url(url)
            print(f"   Original:  {url}")
            print(f"   Optimized: {optimized}")
            print(f"   Changed:   {'Yes' if url != optimized else 'No'}")
            print()

    except Exception as e:
        print(f"   ❌ Error testing optimization: {e}")


def main():
    """Run all URL tests."""
    print("🎬 Runway URL Compatibility Test Suite")
    print("=" * 50)

    # Test URL optimization function
    test_url_optimization()

    # Test actual video generation with different URLs
    print("\n🌐 Testing actual video generation...")
    results = []

    for test_case in TEST_URLS:
        result = test_url_with_api(test_case["url"], test_case["name"])
        results.append(result)

        # Small delay between tests to avoid rate limiting
        time.sleep(2)

    # Summary
    print("\n📊 Test Results Summary:")
    print("-" * 30)

    success_count = 0
    for result in results:
        status_emoji = {
            "success": "✅",
            "failed": "❌",
            "timeout": "⏰",
            "http_error": "🚫",
            "error": "💥"
        }.get(result["result"], "❓")

        print(f"{status_emoji} {result['name']}: {result['result']}")
        if result["result"] == "success":
            success_count += 1

    print(f"\n🎯 Success Rate: {success_count}/{len(results)} ({success_count/len(results)*100:.1f}%)")

    # Save detailed results
    timestamp = time.strftime("%Y%m%d_%H%M%S")
    output_file = f"runway_url_test_results_{timestamp}.json"

    with open(output_file, 'w') as f:
        json.dump({
            "timestamp": timestamp,
            "test_count": len(results),
            "success_count": success_count,
            "results": results
        }, f, indent=2)

    print(f"📁 Detailed results saved to: {output_file}")

    # Recommendations
    print("\n💡 Recommendations:")
    if success_count == len(results):
        print("   🎉 All URLs worked! Your Runway integration is solid.")
    elif success_count > 0:
        print("   ⚠️  Some URLs failed. Use simpler URLs with fewer parameters.")
        print("   ✅ Working URLs can be used as templates for future requests.")
    else:
        print("   🚨 All URLs failed. Check your Runway API key and internet connection.")


if __name__ == "__main__":
    main()