119 lines
4.2 KiB
Python
119 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test final de Captain Mobile - Valida flujo completo
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import httpx
|
|
from websockets import connect
|
|
|
|
SESSION = "655551.qa_test"
|
|
|
|
async def test():
|
|
print("=" * 60)
|
|
print("TEST FINAL - Captain Mobile")
|
|
print(f"Sesión: {SESSION}")
|
|
print("=" * 60)
|
|
|
|
# 1. Login
|
|
print("\n[1] Obteniendo token...")
|
|
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 {r.status_code}")
|
|
return
|
|
token = r.json()['token']
|
|
print(f" Token OK: {token[:20]}...")
|
|
|
|
# 2. WebSocket
|
|
print("\n[2] Conectando WebSocket...")
|
|
async with connect('ws://localhost:3030/ws/chat') as ws:
|
|
await ws.send(json.dumps({'token': token}))
|
|
init = json.loads(await ws.recv())
|
|
print(f" Init: {init['type']}")
|
|
|
|
# 3. Connect to session
|
|
print(f"\n[3] Conectando a sesión {SESSION}...")
|
|
await ws.send(json.dumps({'type': 'connect_session', 'full_name': SESSION}))
|
|
conn = json.loads(await ws.recv())
|
|
print(f" Resultado: {conn['type']}")
|
|
|
|
if conn['type'] != 'session_connected':
|
|
print(f" ERROR: No se pudo conectar a la sesión")
|
|
return
|
|
|
|
# 4. Esperar un poco para que se establezca la conexión
|
|
print("\n[4] Esperando 2s para estabilizar conexión...")
|
|
await asyncio.sleep(2)
|
|
|
|
# 5. Enviar mensaje simple
|
|
mensaje = "responde EXACTAMENTE: TEST_OK_12345"
|
|
print(f"\n[5] Enviando mensaje: '{mensaje}'")
|
|
await ws.send(json.dumps({'type': 'message', 'content': mensaje}))
|
|
|
|
# 6. Recibir respuestas
|
|
print("\n[6] Esperando respuestas (max 45s)...")
|
|
outputs = []
|
|
start = asyncio.get_event_loop().time()
|
|
|
|
try:
|
|
while (asyncio.get_event_loop().time() - start) < 45:
|
|
try:
|
|
msg = await asyncio.wait_for(ws.recv(), timeout=10.0)
|
|
data = json.loads(msg)
|
|
if data['type'] == 'output':
|
|
content = data['content']
|
|
outputs.append(content)
|
|
# Mostrar preview
|
|
preview = content[:80].replace('\n', '\\n')
|
|
print(f" Output #{len(outputs)} ({len(content)} chars): {preview}...")
|
|
|
|
# Buscar la respuesta esperada
|
|
if 'TEST_OK_12345' in content:
|
|
print(f" *** RESPUESTA ENCONTRADA! ***")
|
|
break
|
|
except asyncio.TimeoutError:
|
|
print(" (timeout 10s sin output)")
|
|
if outputs:
|
|
break
|
|
except Exception as e:
|
|
print(f" Error: {e}")
|
|
|
|
# 7. Validación
|
|
full_output = ''.join(outputs)
|
|
|
|
print("\n" + "=" * 60)
|
|
print("RESULTADOS")
|
|
print("=" * 60)
|
|
|
|
print(f"\nOutputs recibidos: {len(outputs)}")
|
|
print(f"Total caracteres: {len(full_output)}")
|
|
|
|
# Check 1: Respuesta encontrada
|
|
has_response = 'TEST_OK_12345' in full_output
|
|
print(f"\n[CHECK 1] Respuesta 'TEST_OK_12345': {'PASS' if has_response else 'FAIL'}")
|
|
|
|
# Check 2: Sin duplicados excesivos
|
|
no_duplicates = len(full_output) < 2000 # Respuesta simple debería ser corta
|
|
print(f"[CHECK 2] Sin duplicados excesivos (<2000 chars): {'PASS' if no_duplicates else 'FAIL'}")
|
|
|
|
# Check 3: UI filtrada
|
|
has_ui = 'bypass permissions' in full_output or '───' in full_output
|
|
no_ui = not has_ui
|
|
print(f"[CHECK 3] UI de Claude filtrada: {'PASS' if no_ui else 'FAIL'}")
|
|
|
|
# Resultado final
|
|
all_pass = has_response and no_duplicates and no_ui
|
|
print("\n" + "=" * 60)
|
|
print(f"RESULTADO FINAL: {'PASS' if all_pass else 'FAIL'}")
|
|
print("=" * 60)
|
|
|
|
if len(full_output) < 500:
|
|
print(f"\nRespuesta completa:\n{full_output}")
|
|
else:
|
|
print(f"\nRespuesta (primeros 500 chars):\n{full_output[:500]}...")
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(test())
|