test_hf_spaces.py

#!/usr/bin/env python3
"""
Script de prueba para verificar la integración con espacios de Hugging Face
"""

import sys
import os
import time
from typing import Dict, Any

# Agregar el directorio actual al path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

try:
    from hf_spaces_config import get_hf_spaces_manager, test_space_connection, list_available_spaces
    print("✅ Módulo hf_spaces_config importado correctamente")
except ImportError as e:
    print(f"❌ Error importando hf_spaces_config: {e}")
    sys.exit(1)

def test_basic_functionality():
    """Probar funcionalidad básica del gestor de espacios"""
    print("\n🧪 Probando funcionalidad básica...")
    
    try:
        manager = get_hf_spaces_manager()
        print("✅ Gestor de espacios creado correctamente")
        
        # Verificar configuración
        spaces_config = manager.spaces_config
        print(f"📊 Espacios configurados: {len(spaces_config)}")
        
        for space_name, config in spaces_config.items():
            print(f"  • {space_name}: {config['type']} - {'✅' if config['enabled'] else '❌'}")
        
        return True
        
    except Exception as e:
        print(f"❌ Error en funcionalidad básica: {e}")
        return False

def test_space_status():
    """Probar verificación de estado de espacios"""
    print("\n🌐 Probando verificación de estado...")
    
    try:
        # Obtener estado de todos los espacios
        status = list_available_spaces()
        
        print("📊 Estado de espacios:")
        for space_name, space_status in status.items():
            status_icon = "🟢" if space_status["status"] == "online" else "🔴"
            print(f"  {status_icon} {space_name}: {space_status['status']} - {space_status['message']}")
        
        # Contar espacios por estado
        online_count = sum(1 for s in status.values() if s["status"] == "online")
        disabled_count = sum(1 for s in status.values() if s["status"] == "disabled")
        error_count = sum(1 for s in status.values() if s["status"] in ["error", "offline", "timeout"])
        
        print(f"\n📈 Resumen:")
        print(f"  🟢 En línea: {online_count}")
        print(f"  🟡 Deshabilitados: {disabled_count}")
        print(f"  🔴 Con errores: {error_count}")
        
        return True
        
    except Exception as e:
        print(f"❌ Error verificando estado: {e}")
        return False

def test_individual_space_connection():
    """Probar conexión individual con cada espacio"""
    print("\n🔗 Probando conexiones individuales...")
    
    try:
        manager = get_hf_spaces_manager()
        enabled_spaces = manager.get_enabled_spaces()
        
        print(f"🔍 Probando {len(enabled_spaces)} espacios habilitados...")
        
        for space_name in enabled_spaces.keys():
            print(f"\n  🔍 Probando {space_name}...")
            
            try:
                status = test_space_connection(space_name)
                status_icon = "✅" if status["status"] == "online" else "❌"
                print(f"    {status_icon} {status['status']}: {status['message']}")
                
                # Pequeña pausa para no sobrecargar
                time.sleep(1)
                
            except Exception as e:
                print(f"    ❌ Error probando {space_name}: {e}")
        
        return True
        
    except Exception as e:
        print(f"❌ Error en pruebas individuales: {e}")
        return False

def test_api_payload_preparation():
    """Probar preparación de payloads para diferentes tipos de espacios"""
    print("\n📦 Probando preparación de payloads...")
    
    try:
        manager = get_hf_spaces_manager()
        
        test_prompt = "Un gato bailando en la luna"
        test_params = {
            "num_frames": 16,
            "num_inference_steps": 20,
            "guidance_scale": 7.5
        }
        
        for space_name, config in manager.spaces_config.items():
            if config["enabled"]:
                print(f"\n  📦 {space_name} ({config['type']}):")
                
                try:
                    payload = manager._prepare_payload(space_name, test_prompt, **test_params)
                    print(f"    ✅ Payload preparado: {payload}")
                except Exception as e:
                    print(f"    ❌ Error preparando payload: {e}")
        
        return True
        
    except Exception as e:
        print(f"❌ Error en preparación de payloads: {e}")
        return False

def test_video_generation_simulation():
    """Simular generación de video (sin hacer llamadas reales)"""
    print("\n🎬 Simulando generación de video...")
    
    try:
        manager = get_hf_spaces_manager()
        enabled_spaces = manager.get_enabled_spaces()
        
        test_prompt = "Un gato espacial explorando la galaxia"
        test_params = {
            "num_frames": 16,
            "num_inference_steps": 20
        }
        
        print(f"🎯 Simulando generación con prompt: '{test_prompt}'")
        
        for space_name in enabled_spaces.keys():
            config = enabled_spaces[space_name]
            
            if config["type"] in ["text-to-video", "animation", "stable-video-diffusion"]:
                print(f"\n  🎬 {space_name}:")
                print(f"    📝 Tipo: {config['type']}")
                print(f"    ⏱️ Timeout: {config['timeout']}s")
                print(f"    🔄 Reintentos: {config['retries']}")
                
                # Simular payload
                try:
                    payload = manager._prepare_payload(space_name, test_prompt, **test_params)
                    print(f"    📦 Payload simulado: {payload}")
                    print(f"    ✅ Simulación exitosa")
                except Exception as e:
                    print(f"    ❌ Error en simulación: {e}")
        
        return True
        
    except Exception as e:
        print(f"❌ Error en simulación: {e}")
        return False

def run_all_tests():
    """Ejecutar todas las pruebas"""
    print("🚀 Iniciando pruebas de integración con espacios de HF...")
    print("=" * 60)
    
    tests = [
        ("Funcionalidad básica", test_basic_functionality),
        ("Estado de espacios", test_space_status),
        ("Conexiones individuales", test_individual_space_connection),
        ("Preparación de payloads", test_api_payload_preparation),
        ("Simulación de generación", test_video_generation_simulation)
    ]
    
    results = []
    
    for test_name, test_func in tests:
        print(f"\n{'='*20} {test_name} {'='*20}")
        try:
            success = test_func()
            results.append((test_name, success))
        except Exception as e:
            print(f"❌ Error ejecutando {test_name}: {e}")
            results.append((test_name, False))
    
    # Resumen final
    print("\n" + "=" * 60)
    print("📊 RESUMEN DE PRUEBAS")
    print("=" * 60)
    
    passed = 0
    total = len(results)
    
    for test_name, success in results:
        status = "✅ PASÓ" if success else "❌ FALLÓ"
        print(f"  {status} {test_name}")
        if success:
            passed += 1
    
    print(f"\n🎯 Resultado: {passed}/{total} pruebas pasaron")
    
    if passed == total:
        print("🎉 ¡Todas las pruebas pasaron! La integración está lista.")
        return True
    else:
        print("⚠️ Algunas pruebas fallaron. Revisa los errores arriba.")
        return False

def main():
    """Función principal"""
    try:
        success = run_all_tests()
        
        if success:
            print("\n🚀 La integración con espacios de HF está lista para usar.")
            print("💡 Puedes usar app_enhanced.py para generar videos con los espacios.")
        else:
            print("\n⚠️ Hay problemas que necesitan ser resueltos antes de usar la integración.")
        
        return 0 if success else 1
        
    except KeyboardInterrupt:
        print("\n⏹️ Pruebas interrumpidas por el usuario.")
        return 1
    except Exception as e:
        print(f"\n❌ Error inesperado: {e}")
        return 1

if __name__ == "__main__":
    exit_code = main()
    sys.exit(exit_code)