129 lines
5.0 KiB
Python
Executable File
129 lines
5.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Test WebSocket - Simula comportamiento de app Flutter
|
|
Valida conexión, autenticación, mensajes y respuestas
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import httpx
|
|
from websockets import connect
|
|
|
|
async def test():
|
|
print("=" * 60)
|
|
print("TEST WEBSOCKET - Simulador App Flutter")
|
|
print("=" * 60)
|
|
|
|
# 1. Login
|
|
print("\n[1] Obteniendo token de autenticación...")
|
|
async with httpx.AsyncClient() as client:
|
|
r = await client.post('http://localhost:3030/auth/login',
|
|
json={'username': 'admin', 'password': 'admin'})
|
|
if r.status_code != 200:
|
|
print(f"ERROR: Login failed with status {r.status_code}")
|
|
return
|
|
token = r.json()['token']
|
|
print(f" Token obtenido: {token[:20]}...")
|
|
|
|
# 2. WebSocket
|
|
print("\n[2] Conectando a WebSocket...")
|
|
async with connect('ws://localhost:3030/ws/chat') as ws:
|
|
# Auth
|
|
await ws.send(json.dumps({'token': token}))
|
|
init = json.loads(await ws.recv())
|
|
print(f" Init response: {init['type']}")
|
|
|
|
# Connect to session
|
|
print("\n[3] Conectando a sesión 648211.captain_test...")
|
|
await ws.send(json.dumps({'type': 'connect_session', 'full_name': '648211.captain_test'}))
|
|
conn = json.loads(await ws.recv())
|
|
print(f" Session response: {conn}")
|
|
|
|
# Send message
|
|
print("\n[4] Enviando mensaje: 'di solo: OK'")
|
|
await ws.send(json.dumps({'type': 'message', 'content': 'di solo: OK'}))
|
|
|
|
# Receive responses - longer timeout for Claude response
|
|
print("\n[5] Esperando respuestas (timeout 30s)...")
|
|
outputs = []
|
|
output_count = 0
|
|
start_time = asyncio.get_event_loop().time()
|
|
max_wait = 30.0 # Max 30 seconds total
|
|
idle_timeout = 8.0 # 8 seconds without output = done
|
|
|
|
try:
|
|
while (asyncio.get_event_loop().time() - start_time) < max_wait:
|
|
try:
|
|
msg = await asyncio.wait_for(ws.recv(), timeout=idle_timeout)
|
|
data = json.loads(msg)
|
|
if data['type'] == 'output':
|
|
output_count += 1
|
|
content = data['content']
|
|
outputs.append(content)
|
|
preview = content[:100].replace('\n', '\\n')
|
|
print(f" Output #{output_count} ({len(content)} chars): {preview}...")
|
|
elif data['type'] == 'status':
|
|
print(f" Status: {data.get('status', data)}")
|
|
elif data['type'] == 'error':
|
|
print(f" ERROR: {data.get('message', data)}")
|
|
except asyncio.TimeoutError:
|
|
print(" (idle timeout - fin de respuestas)")
|
|
break
|
|
except Exception as e:
|
|
print(f" Error: {e}")
|
|
|
|
# Validate
|
|
full_output = ''.join(outputs)
|
|
|
|
print("\n" + "=" * 60)
|
|
print("VALIDACION DE RESULTADOS")
|
|
print("=" * 60)
|
|
|
|
# Validación 1: Contiene OK
|
|
contains_ok = 'OK' in full_output or 'ok' in full_output.lower()
|
|
print(f"\n[CHECK 1] Contiene 'OK': {'PASS' if contains_ok else 'FAIL'}")
|
|
|
|
# Validación 2: No duplicados
|
|
# Detectar si hay contenido repetido
|
|
has_duplicates = False
|
|
if len(outputs) > 1:
|
|
# Verificar si chunks consecutivos son idénticos
|
|
for i in range(len(outputs) - 1):
|
|
if outputs[i] == outputs[i+1] and len(outputs[i]) > 10:
|
|
has_duplicates = True
|
|
break
|
|
# Verificar si el contenido total tiene patrones repetidos
|
|
if len(full_output) > 100:
|
|
half = len(full_output) // 2
|
|
first_half = full_output[:half]
|
|
second_half = full_output[half:half*2]
|
|
if first_half == second_half:
|
|
has_duplicates = True
|
|
|
|
print(f"[CHECK 2] Sin duplicados: {'PASS' if not has_duplicates else 'FAIL'}")
|
|
if has_duplicates:
|
|
print(" WARNING: Se detectó contenido duplicado!")
|
|
|
|
# Validación 3: Longitud razonable
|
|
is_short = len(full_output) <= 500
|
|
print(f"[CHECK 3] Longitud <= 500 chars: {'PASS' if is_short else 'FAIL'}")
|
|
print(f" Longitud actual: {len(full_output)} caracteres")
|
|
|
|
# Resumen
|
|
all_passed = contains_ok and not has_duplicates and is_short
|
|
print("\n" + "=" * 60)
|
|
if all_passed:
|
|
print("RESULTADO FINAL: PASS - Todas las validaciones correctas")
|
|
else:
|
|
print("RESULTADO FINAL: FAIL - Hay validaciones fallidas")
|
|
print("=" * 60)
|
|
|
|
# Mostrar respuesta completa si es corta
|
|
if len(full_output) <= 500:
|
|
print(f"\nRespuesta completa:\n{full_output}")
|
|
else:
|
|
print(f"\nRespuesta (primeros 500 chars):\n{full_output[:500]}...")
|
|
print(f"\n... (truncado, total: {len(full_output)} chars)")
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(test())
|