chat/tests/test_notifications.py

#!/usr/bin/env python3
"""
🔔 ТЕСТ СИСТЕМЫ УВЕДОМЛЕНИЙ
Проверяем работу уведомлений всем пользователям в радиусе при экстренных событиях
"""
import requests
import json
import asyncio
import websockets
import threading
import time
from datetime import datetime

# Конфигурация
BASE_URL = "http://localhost:8002"
WS_URL = "ws://localhost:8002/api/v1/emergency/ws"

# Тестовые пользователи (нужно использовать реальные токены)
TEST_USERS = [
    {
        "name": "User1 (shadow85)",
        "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIyIiwiZW1haWwiOiJzaGFkb3c4NUBsaXN0LnJ1IiwiZXhwIjoxNzYwNzgxNzk5fQ.cAG66Xqpxs_-NNkL6Sz82HuFV_-bNv3dEhYAntgbVRg",
        "user_id": 2
    },
    {
        "name": "User2 (Raisa)", 
        "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIzIiwiZW1haWwiOiJSYWlzYUBtYWlsLnJ1IiwiZXhwIjoxNzYwNzgxOTM3fQ.8gZeMsOmnqOMfiz8azGVJ_SxweaaLH6UIRImi9aCK4U",
        "user_id": 3
    }
]

# Координаты для тестирования (близко друг к другу)
TEST_COORDINATES = {
    "emergency_location": {"latitude": 35.1815, "longitude": 126.8108},
    "nearby_user1": {"latitude": 35.1820, "longitude": 126.8110},  # ~500m away
    "nearby_user2": {"latitude": 35.1825, "longitude": 126.8115}   # ~1km away
}

class WebSocketListener:
    def __init__(self, user_name, token, user_id):
        self.user_name = user_name
        self.token = token
        self.user_id = user_id
        self.notifications_received = []
        self.connected = False
        
    async def listen(self):
        """Подключение к WebSocket и прослушивание уведомлений"""
        uri = f"{WS_URL}/current_user_id?token={self.token}"
        
        try:
            print(f"🔌 {self.user_name}: Подключение к WebSocket...")
            async with websockets.connect(uri) as websocket:
                self.connected = True
                print(f"✅ {self.user_name}: WebSocket подключен")
                
                # Слушаем уведомления в течение 30 секунд
                try:
                    await asyncio.wait_for(self._listen_messages(websocket), timeout=30.0)
                except asyncio.TimeoutError:
                    print(f"⏰ {self.user_name}: Таймаут WebSocket соединения")
                    
        except Exception as e:
            print(f"❌ {self.user_name}: Ошибка WebSocket: {e}")
        finally:
            self.connected = False
            print(f"🔌 {self.user_name}: WebSocket отключен")
            
    async def _listen_messages(self, websocket):
        """Прослушивание входящих сообщений"""
        async for message in websocket:
            try:
                data = json.loads(message)
                if data.get("type") == "emergency_alert":
                    self.notifications_received.append({
                        "timestamp": datetime.now(),
                        "data": data
                    })
                    distance = data.get("distance_km", "unknown")
                    print(f"🔔 {self.user_name}: Получено экстренное уведомление! Расстояние: {distance}км")
                    print(f"   📍 Alert ID: {data.get('alert_id')}")
                    print(f"   🚨 Type: {data.get('alert_type')}")
                    print(f"   💬 Message: {data.get('message')}")
                else:
                    print(f"📨 {self.user_name}: Получено сообщение: {data}")
            except Exception as e:
                print(f"❌ {self.user_name}: Ошибка парсинга сообщения: {e}")


def test_notification_system():
    """Основная функция тестирования системы уведомлений"""
    print("🔔 ТЕСТИРОВАНИЕ СИСТЕМЫ УВЕДОМЛЕНИЙ")
    print("=" * 60)
    
    # Шаг 1: Создать WebSocket подключения для тестовых пользователей
    listeners = []
    for user in TEST_USERS:
        listener = WebSocketListener(user["name"], user["token"], user["user_id"])
        listeners.append(listener)
    
    # Шаг 2: Запустить WebSocket соединения в отдельных потоках
    async def run_all_listeners():
        tasks = [listener.listen() for listener in listeners]
        await asyncio.gather(*tasks, return_exceptions=True)
    
    # Запуск WebSocket в отдельном потоке
    ws_thread = threading.Thread(target=lambda: asyncio.run(run_all_listeners()))
    ws_thread.daemon = True
    ws_thread.start()
    
    # Подождем подключения
    print("⏳ Ожидание подключения WebSocket...")
    time.sleep(3)
    
    # Проверим статус подключений
    connected_users = [l for l in listeners if l.connected]
    print(f"📊 Подключено пользователей: {len(connected_users)}/{len(listeners)}")
    
    # Шаг 3: Создать экстренное событие
    print(f"\n🚨 Создание экстренного события...")
    
    emergency_data = {
        "latitude": TEST_COORDINATES["emergency_location"]["latitude"],
        "longitude": TEST_COORDINATES["emergency_location"]["longitude"], 
        "alert_type": "general",
        "message": "Тестирование системы уведомлений - это учебное событие",
        "address": "Тестовая локация"
    }
    
    try:
        # Используем токен первого пользователя для создания события
        headers = {"Authorization": f"Bearer {TEST_USERS[0]['token']}"}
        
        response = requests.post(
            f"{BASE_URL}/api/v1/emergency/events",
            json=emergency_data,
            headers=headers,
            timeout=10
        )
        
        if response.status_code == 200:
            alert_data = response.json()
            alert_id = alert_data.get("id")
            print(f"✅ Экстренное событие создано! ID: {alert_id}")
            print(f"📍 Координаты: {emergency_data['latitude']}, {emergency_data['longitude']}")
        else:
            print(f"❌ Не удалось создать событие: {response.status_code}")
            print(f"📄 Ответ: {response.text}")
            return
            
    except Exception as e:
        print(f"❌ Ошибка создания события: {e}")
        return
    
    # Шаг 4: Ждем уведомления
    print(f"\n⏳ Ожидание уведомлений... (15 секунд)")
    time.sleep(15)
    
    # Шаг 5: Анализ результатов
    print(f"\n📊 РЕЗУЛЬТАТЫ ТЕСТИРОВАНИЯ")
    print("=" * 60)
    
    total_notifications = 0
    for listener in listeners:
        count = len(listener.notifications_received)
        total_notifications += count
        status = "✅ Получил" if count > 0 else "❌ Не получил"
        print(f"{status} {listener.user_name}: {count} уведомлений")
        
        # Показать детали уведомлений
        for i, notification in enumerate(listener.notifications_received, 1):
            data = notification["data"]
            timestamp = notification["timestamp"].strftime("%H:%M:%S")
            distance = data.get("distance_km", "unknown")
            print(f"   {i}. [{timestamp}] Alert ID {data.get('alert_id')} ({distance}км)")
    
    # Итоговый отчет
    print(f"\n🎯 ИТОГИ:")
    print(f"📊 Всего уведомлений получено: {total_notifications}")
    print(f"👥 Подключенных пользователей: {len(connected_users)}")
    print(f"🎯 Ожидаемо уведомлений: {len(connected_users)}")
    
    if total_notifications >= len(connected_users):
        print(f"✅ УСПЕХ: Система уведомлений работает правильно!")
    else:
        print(f"⚠️ ПРЕДУПРЕЖДЕНИЕ: Не все пользователи получили уведомления")
    
    print(f"\n💡 Примечание: Уведомления отправляются пользователям в радиусе 5км от события")
    print(f"📱 Также отправляются push-уведомления через Notification Service")


if __name__ == "__main__":
    test_notification_system()