test_websocket.py

#!/usr/bin/env python3
"""
WebSocket Test Scripti
"""

import websockets
import asyncio
import json

async def test_websocket():
    """WebSocket bağlantısını test et"""
    print("🔍 WebSocket bağlantısı test ediliyor...")
    
    try:
        uri = "ws://localhost:8000/ws/temperature"
        async with websockets.connect(uri) as websocket:
            print("✅ WebSocket bağlantısı başarılı!")
            print("📡 Mesajlar dinleniyor...")
            
            # 10 saniye boyunca mesajları dinle
            start_time = asyncio.get_event_loop().time()
            message_count = 0
            
            while asyncio.get_event_loop().time() - start_time < 10:
                try:
                    message = await asyncio.wait_for(websocket.recv(), timeout=1)
                    data = json.loads(message)
                    
                    if data['type'] == 'temperature_update':
                        temp_value = data['data']['temperature']['value']
                        timestamp = data['timestamp']
                        print(f"📊 Sıcaklık: {temp_value}°C | {timestamp}")
                        message_count += 1
                        
                except asyncio.TimeoutError:
                    continue
                except Exception as e:
                    print(f"❌ Mesaj alma hatası: {e}")
                    break
            
            print(f"✅ {message_count} WebSocket mesajı alındı!")
            return message_count > 0
            
    except Exception as e:
        print(f"❌ WebSocket bağlantı hatası: {e}")
        return False

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