Change PIN to 1451

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

apps/captain-mobile-v2/backend/test_final.py

@@ -0,0 +1,118 @@
+#!/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())