chat/tests/test_websocket_direct.py

#!/usr/bin/env python3
"""
Простое тестирование WebSocket подключений без авторизации
"""

import asyncio
import json
import websockets
from datetime import datetime


BASE_URL = "192.168.219.108"
EMERGENCY_PORT = "8002"


async def test_websocket_direct():
    """Прямое тестирование WebSocket подключения"""
    print("🔌 Тестирование WebSocket подключения напрямую...")
    
    # Используем тестовый JWT токен из наших предыдущих тестов
    test_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxLCJlbWFpbCI6InNoYWRvdzg1QGxpc3QucnUiLCJleHAiOjE3NjEzMTczMzl9.W6_k8VbYA73kKL7sUGFJKwl7Oez3ErGjjR5F29O-NZw"
    
    ws_url = f"ws://{BASE_URL}:{EMERGENCY_PORT}/api/v1/emergency/ws/current_user_id?token={test_token}"
    
    try:
        print(f"🌐 Подключение к: {ws_url}")
        
        async with websockets.connect(ws_url) as websocket:
            print("✅ WebSocket подключен!")
            
            # Ждем приветственное сообщение
            try:
                message = await asyncio.wait_for(websocket.recv(), timeout=5.0)
                print(f"📨 Получено сообщение: {message}")
                
                # Парсим JSON
                try:
                    data = json.loads(message)
                    print(f"📋 Данные сообщения:")
                    for key, value in data.items():
                        print(f"   - {key}: {value}")
                except json.JSONDecodeError:
                    print("⚠️ Сообщение не в формате JSON")
                
            except asyncio.TimeoutError:
                print("⏰ Тайм-аут ожидания сообщения")
            
            # Держим соединение открытым
            print("🔄 Держим соединение открытым 10 секунд...")
            
            # Слушаем дополнительные сообщения
            try:
                while True:
                    message = await asyncio.wait_for(websocket.recv(), timeout=10.0)
                    print(f"📨 Дополнительное сообщение: {message}")
                    
            except asyncio.TimeoutError:
                print("✅ Соединение стабильно в течение 10 секунд")
            except websockets.exceptions.ConnectionClosed:
                print("❌ Соединение закрыто сервером")
            
    except websockets.exceptions.InvalidStatusCode as e:
        print(f"❌ Ошибка статус-кода: {e}")
        if e.status_code == 403:
            print("🔒 Проблема с авторизацией - токен может быть недействительным")
        elif e.status_code == 404:
            print("🔍 Endpoint не найден")
    except Exception as e:
        print(f"❌ Ошибка подключения: {e}")


async def test_multiple_connections():
    """Тест множественных подключений"""
    print("\n" + "="*60)
    print("🚀 ТЕСТИРОВАНИЕ МНОЖЕСТВЕННЫХ WEBSOCKET ПОДКЛЮЧЕНИЙ")
    print("="*60)
    
    # Список тестовых токенов (если у нас есть разные пользователи)
    tokens = [
        "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxLCJlbWFpbCI6InNoYWRvdzg1QGxpc3QucnUiLCJleHAiOjE3NjEzMTczMzl9.W6_k8VbYA73kKL7sUGFJKwl7Oez3ErGjjR5F29O-NZw"
    ]
    
    connections = []
    
    # Создаем несколько подключений
    for i, token in enumerate(tokens):
        try:
            ws_url = f"ws://{BASE_URL}:{EMERGENCY_PORT}/api/v1/emergency/ws/current_user_id?token={token}"
            websocket = await websockets.connect(ws_url)
            connections.append((i+1, websocket))
            print(f"✅ Подключение {i+1} успешно установлено")
            
            # Ждем приветственное сообщение
            try:
                message = await asyncio.wait_for(websocket.recv(), timeout=2.0)
                print(f"   📨 Сообщение: {message}")
            except asyncio.TimeoutError:
                print("   ⏰ Нет приветственного сообщения")
                
        except Exception as e:
            print(f"❌ Ошибка подключения {i+1}: {e}")
    
    print(f"\n📊 Установлено подключений: {len(connections)}")
    
    if connections:
        print("⏱️ Держим подключения открытыми 5 секунд...")
        await asyncio.sleep(5)
        
        # Закрываем подключения
        for conn_id, websocket in connections:
            try:
                await websocket.close()
                print(f"🔚 Подключение {conn_id} закрыто")
            except Exception as e:
                print(f"❌ Ошибка закрытия подключения {conn_id}: {e}")


def check_websocket_manager_directly():
    """Проверить WebSocketManager напрямую"""
    print("\n" + "="*60)
    print("🔍 ПРОВЕРКА WEBSOCKETMANAGER ЧЕРЕЗ HTTP")
    print("="*60)
    
    import requests
    
    # Пробуем получить статистику через простой HTTP-запрос к health endpoint
    try:
        health_response = requests.get(f"http://{BASE_URL}:{EMERGENCY_PORT}/health")
        if health_response.status_code == 200:
            print("✅ Emergency Service работает")
            print(f"   Ответ: {health_response.json()}")
        else:
            print(f"❌ Emergency Service недоступен: {health_response.status_code}")
            
    except Exception as e:
        print(f"❌ Ошибка проверки health: {e}")


async def main():
    """Главная функция"""
    print("🚀 WebSocket Direct Test v1.0")
    print(f"🌐 Сервер: {BASE_URL}:{EMERGENCY_PORT}")
    print("=" * 60)
    
    # 1. Проверяем сервер
    check_websocket_manager_directly()
    
    # 2. Тестируем одно WebSocket подключение
    await test_websocket_direct()
    
    # 3. Тестируем множественные подключения
    await test_multiple_connections()
    
    print("\n" + "="*60)
    print("✅ ТЕСТИРОВАНИЕ ЗАВЕРШЕНО")
    print("="*60)


if __name__ == "__main__":
    asyncio.run(main())