Delete test_sync_status.py
This commit is contained in:
@@ -1,126 +0,0 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user