SYNC_API/test_sync_status.py

#!/usr/bin/env python3
"""
Script de prueba para validar la nueva lógica de estados de sincronización
"""

import os
import sys
from datetime import datetime, timedelta
from pathlib import Path

# Agregar el directorio padre al path para importar los módulos
sys.path.insert(0, str(Path(__file__).parent / 'sync_api'))

from routers.nodes import determine_sync_status
from models.sync_models import EstadoSincronizacion


def create_test_files():
    """Crear archivos de prueba para validar la lógica"""
    
    # Crear directorios de prueba si no existen
    os.makedirs("D:\\sftp", exist_ok=True)
    os.makedirs("D:\\BackupSFTP", exist_ok=True)
    
    # Crear archivo de prueba en D:\sftp
    with open("D:\\sftp\\NODO_TEST1.zip", "w") as f:
        f.write("archivo de prueba")
    
    # Crear archivo de prueba en D:\BackupSFTP
    with open("D:\\BackupSFTP\\NODO_TEST2.zip", "w") as f:
        f.write("archivo de prueba")
    
    print("Archivos de prueba creados:")
    print("- D:\\sftp\\NODO_TEST1.zip")
    print("- D:\\BackupSFTP\\NODO_TEST2.zip")


def test_sync_status_logic():
    """Probar la lógica de determinación de estados"""
    
    print("\n=== PRUEBAS DE LÓGICA DE ESTADOS ===\n")
    
    # Prueba 1: Nodo con archivo en D:\sftp (debe ser "Recibido")
    print("1. Prueba: Archivo en D:\\sftp")
    status = determine_sync_status("NODO_TEST1")
    print(f"   Estado: {status}")
    print(f"   Esperado: {EstadoSincronizacion.RECIBIDO.value}")
    print(f"   ✓ Correcto" if status == EstadoSincronizacion.RECIBIDO.value else "   ✗ Incorrecto")
    
    # Prueba 2: Nodo con archivo en D:\BackupSFTP (debe ser "Procesando")
    print("\n2. Prueba: Archivo en D:\\BackupSFTP")
    status = determine_sync_status("NODO_TEST2")
    print(f"   Estado: {status}")
    print(f"   Esperado: {EstadoSincronizacion.PROCESANDO.value}")
    print(f"   ✓ Correcto" if status == EstadoSincronizacion.PROCESANDO.value else "   ✗ Incorrecto")
    
    # Prueba 3: Nodo sin archivo, con fecha reciente (debe ser "Actualizada")
    print("\n3. Prueba: Sin archivo, fecha reciente")
    recent_date = datetime.now() - timedelta(hours=1)
    status = determine_sync_status("NODO_TEST3", recent_date)
    print(f"   Estado: {status}")
    print(f"   Esperado: {EstadoSincronizacion.ACTUALIZADA.value}")
    print(f"   ✓ Correcto" if status == EstadoSincronizacion.ACTUALIZADA.value else "   ✗ Incorrecto")
    
    # Prueba 4: Nodo sin archivo, fecha antigua (debe ser "Error")
    print("\n4. Prueba: Sin archivo, fecha antigua")
    old_date = datetime.now() - timedelta(days=10)
    status = determine_sync_status("NODO_TEST4", old_date)
    print(f"   Estado: {status}")
    print(f"   Esperado: {EstadoSincronizacion.ERROR.value}")
    print(f"   ✓ Correcto" if status == EstadoSincronizacion.ERROR.value else "   ✗ Incorrecto")
    
    # Prueba 5: Nodo sin archivo, sin fecha (debe ser "Error")
    print("\n5. Prueba: Sin archivo, sin fecha")
    status = determine_sync_status("NODO_TEST5", None)
    print(f"   Estado: {status}")
    print(f"   Esperado: {EstadoSincronizacion.ERROR.value}")
    print(f"   ✓ Correcto" if status == EstadoSincronizacion.ERROR.value else "   ✗ Incorrecto")
    
    # Prueba 6: Archivo en BackupSFTP con fecha más reciente que el archivo (debe ser "Actualizada")
    print("\n6. Prueba: Archivo en BackupSFTP, fecha más reciente")
    future_date = datetime.now() + timedelta(hours=1)  # Fecha futura para simular más reciente
    status = determine_sync_status("NODO_TEST2", future_date)
    print(f"   Estado: {status}")
    print(f"   Esperado: {EstadoSincronizacion.ACTUALIZADA.value}")
    print(f"   ✓ Correcto" if status == EstadoSincronizacion.ACTUALIZADA.value else "   ✗ Incorrecto")


def cleanup_test_files():
    """Limpiar archivos de prueba"""
    test_files = [
        "D:\\sftp\\NODO_TEST1.zip",
        "D:\\BackupSFTP\\NODO_TEST2.zip"
    ]
    
    for file_path in test_files:
        try:
            if os.path.exists(file_path):
                os.remove(file_path)
                print(f"Archivo eliminado: {file_path}")
        except Exception as e:
            print(f"Error eliminando {file_path}: {e}")


if __name__ == "__main__":
    print("=== SCRIPT DE PRUEBA PARA ESTADOS DE SINCRONIZACIÓN ===")
    
    try:
        # Solo crear archivos de prueba si los directorios existen
        if os.path.exists("D:\\") and os.access("D:\\", os.W_OK):
            create_test_files()
            test_sync_status_logic()
            
            # Preguntar si limpiar archivos de prueba
            response = input("\n¿Deseas eliminar los archivos de prueba? (s/n): ")
            if response.lower() in ['s', 'si', 'yes', 'y']:
                cleanup_test_files()
        else:
            print("No se puede acceder al disco D: o no hay permisos de escritura")
            print("Probando solo la lógica sin archivos reales...")
            test_sync_status_logic()
            
    except Exception as e:
        print(f"Error durante las pruebas: {e}")
        import traceback
        traceback.print_exc()