""" Tests pour le WebSocket et le service WebSocketManager. Couvre: - Connexion/déconnexion - Broadcast de messages - Gestion des connexions multiples - Nettoyage des connexions mortes """ import pytest from unittest.mock import AsyncMock, MagicMock, patch from fastapi import WebSocket from fastapi.testclient import TestClient pytestmark = pytest.mark.unit class TestWebSocketManager: """Tests pour WebSocketManager.""" def test_initial_state(self): """État initial: aucune connexion.""" from app.services.websocket_service import WebSocketManager manager = WebSocketManager() assert manager.active_connections == [] assert manager.connection_count == 0 @pytest.mark.asyncio async def test_connect_adds_connection(self): """Connect ajoute une connexion.""" from app.services.websocket_service import WebSocketManager manager = WebSocketManager() mock_ws = AsyncMock(spec=WebSocket) await manager.connect(mock_ws) assert manager.connection_count == 1 assert mock_ws in manager.active_connections mock_ws.accept.assert_called_once() @pytest.mark.asyncio async def test_connect_multiple_clients(self): """Plusieurs clients peuvent se connecter.""" from app.services.websocket_service import WebSocketManager manager = WebSocketManager() ws1 = AsyncMock(spec=WebSocket) ws2 = AsyncMock(spec=WebSocket) ws3 = AsyncMock(spec=WebSocket) await manager.connect(ws1) await manager.connect(ws2) await manager.connect(ws3) assert manager.connection_count == 3 def test_disconnect_removes_connection(self): """Disconnect supprime une connexion.""" from app.services.websocket_service import WebSocketManager manager = WebSocketManager() mock_ws = MagicMock(spec=WebSocket) manager.active_connections.append(mock_ws) manager.disconnect(mock_ws) assert mock_ws not in manager.active_connections assert manager.connection_count == 0 def test_disconnect_nonexistent_is_safe(self): """Disconnect d'une connexion inexistante ne plante pas.""" from app.services.websocket_service import WebSocketManager manager = WebSocketManager() mock_ws = MagicMock(spec=WebSocket) # Should not raise manager.disconnect(mock_ws) assert manager.connection_count == 0 @pytest.mark.asyncio async def test_broadcast_sends_to_all(self): """Broadcast envoie à tous les clients.""" from app.services.websocket_service import WebSocketManager manager = WebSocketManager() ws1 = AsyncMock(spec=WebSocket) ws2 = AsyncMock(spec=WebSocket) manager.active_connections = [ws1, ws2] message = {"type": "test", "data": "hello"} await manager.broadcast(message) ws1.send_json.assert_called_once_with(message) ws2.send_json.assert_called_once_with(message) @pytest.mark.asyncio async def test_broadcast_removes_dead_connections(self): """Broadcast nettoie les connexions mortes.""" from app.services.websocket_service import WebSocketManager manager = WebSocketManager() alive_ws = AsyncMock(spec=WebSocket) dead_ws = AsyncMock(spec=WebSocket) dead_ws.send_json.side_effect = Exception("Connection closed") manager.active_connections = [alive_ws, dead_ws] await manager.broadcast({"type": "test"}) # Dead connection should be removed assert dead_ws not in manager.active_connections assert alive_ws in manager.active_connections @pytest.mark.asyncio async def test_send_to_client_success(self): """Envoi à un client spécifique.""" from app.services.websocket_service import WebSocketManager manager = WebSocketManager() mock_ws = AsyncMock(spec=WebSocket) manager.active_connections.append(mock_ws) message = {"type": "private", "data": "secret"} await manager.send_to_client(mock_ws, message) mock_ws.send_json.assert_called_once_with(message) @pytest.mark.asyncio async def test_send_to_client_failure_disconnects(self): """Échec d'envoi déconnecte le client.""" from app.services.websocket_service import WebSocketManager manager = WebSocketManager() mock_ws = AsyncMock(spec=WebSocket) mock_ws.send_json.side_effect = Exception("Failed") manager.active_connections.append(mock_ws) await manager.send_to_client(mock_ws, {"type": "test"}) assert mock_ws not in manager.active_connections class TestWebSocketEndpoint: """Tests pour l'endpoint WebSocket /ws.""" def test_websocket_connection_integration(self, app): """Test d'intégration basique WebSocket.""" from fastapi.testclient import TestClient client = TestClient(app) with client.websocket_connect("/ws") as websocket: # Connection should be accepted # We can't easily test the full flow without more setup pass @pytest.mark.asyncio async def test_websocket_broadcast_event_types(self): """Vérifie les types d'événements broadcastés.""" from app.services.websocket_service import WebSocketManager manager = WebSocketManager() mock_ws = AsyncMock(spec=WebSocket) manager.active_connections = [mock_ws] # Task started event await manager.broadcast({ "type": "task_started", "data": {"task_id": "123", "playbook": "test.yml"} }) call_args = mock_ws.send_json.call_args[0][0] assert call_args["type"] == "task_started" assert "task_id" in call_args["data"] @pytest.mark.asyncio async def test_websocket_message_format(self): """Vérifie le format des messages.""" from app.services.websocket_service import WebSocketManager manager = WebSocketManager() mock_ws = AsyncMock(spec=WebSocket) manager.active_connections = [mock_ws] # Host updated event await manager.broadcast({ "type": "host_updated", "data": { "id": "host-1", "name": "server.local", "status": "online" } }) call_args = mock_ws.send_json.call_args[0][0] assert "type" in call_args assert "data" in call_args assert isinstance(call_args["data"], dict) class TestWebSocketEventTypes: """Tests pour les différents types d'événements WebSocket.""" @pytest.fixture def ws_manager(self): """WebSocketManager avec un client mocké.""" from app.services.websocket_service import WebSocketManager manager = WebSocketManager() mock_ws = AsyncMock(spec=WebSocket) manager.active_connections = [mock_ws] return manager, mock_ws @pytest.mark.asyncio async def test_schedule_started_event(self, ws_manager): """Événement schedule_started.""" manager, mock_ws = ws_manager await manager.broadcast({ "type": "schedule_started", "data": { "schedule_id": "sched-1", "schedule_name": "Daily Backup", "run_id": "run-123" } }) msg = mock_ws.send_json.call_args[0][0] assert msg["type"] == "schedule_started" assert msg["data"]["schedule_id"] == "sched-1" @pytest.mark.asyncio async def test_schedule_completed_event(self, ws_manager): """Événement schedule_completed.""" manager, mock_ws = ws_manager await manager.broadcast({ "type": "schedule_completed", "data": { "schedule_id": "sched-1", "status": "success", "duration": 15.5 } }) msg = mock_ws.send_json.call_args[0][0] assert msg["type"] == "schedule_completed" assert msg["data"]["status"] == "success" @pytest.mark.asyncio async def test_hosts_synced_event(self, ws_manager): """Événement hosts_synced.""" manager, mock_ws = ws_manager await manager.broadcast({ "type": "hosts_synced", "data": { "created": 5, "updated": 10, "total": 15 } }) msg = mock_ws.send_json.call_args[0][0] assert msg["type"] == "hosts_synced" assert msg["data"]["total"] == 15