Bruno Charest
ecefbc8611
Some checks failed
Tests / Backend Tests (Python) (3.10) (push) Has been cancelled
Tests / Backend Tests (Python) (3.11) (push) Has been cancelled
Tests / Backend Tests (Python) (3.12) (push) Has been cancelled
Tests / Frontend Tests (JS) (push) Has been cancelled
Tests / Integration Tests (push) Has been cancelled
Tests / All Tests Passed (push) Has been cancelled
280 lines
8.9 KiB
Python
280 lines
8.9 KiB
Python
"""
|
|
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
|