4253b71db9
WebSocket Infrastructure: - Connection manager: Improved connection/disconnection handling - Handlers: Enhanced event handlers for game operations Test Coverage (148 new tests): - test_connection_handlers.py: Connection lifecycle tests - test_connection_manager.py: Manager operations tests - test_handler_locking.py: Concurrency/locking tests - test_query_handlers.py: Game query handler tests - test_rate_limiting.py: Rate limit enforcement tests - test_substitution_handlers.py: Player substitution tests - test_manual_outcome_handlers.py: Manual outcome workflow tests - conftest.py: Shared WebSocket test fixtures 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
818 lines
26 KiB
Python
818 lines
26 KiB
Python
"""
|
|
Tests for ConnectionManager class.
|
|
|
|
Verifies session management, room membership, event broadcasting,
|
|
and session expiration functionality that forms the foundation of
|
|
real-time WebSocket communication.
|
|
|
|
Author: Claude
|
|
Date: 2025-01-27
|
|
Updated: 2025-01-27 - Added session expiration tests
|
|
"""
|
|
|
|
import asyncio
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pendulum
|
|
|
|
from app.websocket.connection_manager import ConnectionManager, SessionInfo
|
|
|
|
|
|
# ============================================================================
|
|
# CONNECTION TESTS
|
|
# ============================================================================
|
|
|
|
|
|
class TestConnect:
|
|
"""Tests for connection registration."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_registers_session(self):
|
|
"""
|
|
Verify connect() stores user_id in user_sessions dict.
|
|
|
|
The session ID (sid) should map to the user_id for later lookups.
|
|
"""
|
|
sio = MagicMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
|
|
assert "sid_123" in manager.user_sessions
|
|
assert manager.user_sessions["sid_123"] == "user_456"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_duplicate_sid_replaces(self):
|
|
"""
|
|
Verify connecting with same SID replaces existing user_id.
|
|
|
|
This handles reconnection scenarios where the same socket reconnects
|
|
with a different user (edge case).
|
|
"""
|
|
sio = MagicMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_A")
|
|
await manager.connect("sid_123", "user_B")
|
|
|
|
assert manager.user_sessions["sid_123"] == "user_B"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connect_multiple_users(self):
|
|
"""Verify multiple users can connect simultaneously."""
|
|
sio = MagicMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_1", "user_1")
|
|
await manager.connect("sid_2", "user_2")
|
|
await manager.connect("sid_3", "user_3")
|
|
|
|
assert len(manager.user_sessions) == 3
|
|
assert manager.user_sessions["sid_1"] == "user_1"
|
|
assert manager.user_sessions["sid_2"] == "user_2"
|
|
assert manager.user_sessions["sid_3"] == "user_3"
|
|
|
|
|
|
# ============================================================================
|
|
# DISCONNECT TESTS
|
|
# ============================================================================
|
|
|
|
|
|
class TestDisconnect:
|
|
"""Tests for disconnection handling."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disconnect_removes_session(self):
|
|
"""
|
|
Verify disconnect() removes user from user_sessions dict.
|
|
|
|
After disconnection, the SID should no longer be tracked.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
await manager.disconnect("sid_123")
|
|
|
|
assert "sid_123" not in manager.user_sessions
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disconnect_removes_from_all_rooms(self):
|
|
"""
|
|
Verify disconnect() removes user from all game rooms they joined.
|
|
|
|
User should be cleaned up from every game room to prevent stale
|
|
participants.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.enter_room = AsyncMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
# Connect and join multiple games
|
|
await manager.connect("sid_123", "user_456")
|
|
await manager.join_game("sid_123", "game_A", "home")
|
|
await manager.join_game("sid_123", "game_B", "away")
|
|
|
|
# Verify in both rooms
|
|
assert "sid_123" in manager.game_rooms["game_A"]
|
|
assert "sid_123" in manager.game_rooms["game_B"]
|
|
|
|
# Disconnect
|
|
await manager.disconnect("sid_123")
|
|
|
|
# Should be removed from all rooms
|
|
assert "sid_123" not in manager.game_rooms["game_A"]
|
|
assert "sid_123" not in manager.game_rooms["game_B"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disconnect_nonexistent_noop(self):
|
|
"""
|
|
Verify disconnecting unknown SID is a safe no-op.
|
|
|
|
Should not raise exception when SID was never connected.
|
|
"""
|
|
sio = MagicMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
# Should not raise
|
|
await manager.disconnect("unknown_sid")
|
|
|
|
assert "unknown_sid" not in manager.user_sessions
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disconnect_broadcasts_to_rooms(self):
|
|
"""
|
|
Verify disconnect() broadcasts user_disconnected to game rooms.
|
|
|
|
Other participants in the game should be notified when a user leaves.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.enter_room = AsyncMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
await manager.join_game("sid_123", "game_A", "home")
|
|
|
|
# Clear previous emit calls
|
|
sio.emit.reset_mock()
|
|
|
|
await manager.disconnect("sid_123")
|
|
|
|
# Should broadcast user_disconnected
|
|
sio.emit.assert_called()
|
|
call_args = sio.emit.call_args_list[-1]
|
|
assert call_args[0][0] == "user_disconnected"
|
|
assert call_args[0][1]["user_id"] == "user_456"
|
|
|
|
|
|
# ============================================================================
|
|
# ROOM MANAGEMENT TESTS
|
|
# ============================================================================
|
|
|
|
|
|
class TestRoomManagement:
|
|
"""Tests for game room join/leave functionality."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_join_game_creates_room(self):
|
|
"""
|
|
Verify join_game() creates room entry if game doesn't exist.
|
|
|
|
First user joining a game should create the room tracking.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.enter_room = AsyncMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
await manager.join_game("sid_123", "game_A", "home")
|
|
|
|
assert "game_A" in manager.game_rooms
|
|
assert "sid_123" in manager.game_rooms["game_A"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_join_game_adds_to_existing(self):
|
|
"""
|
|
Verify join_game() adds to existing room.
|
|
|
|
Multiple users can join the same game room.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.enter_room = AsyncMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_1", "user_1")
|
|
await manager.connect("sid_2", "user_2")
|
|
|
|
await manager.join_game("sid_1", "game_A", "home")
|
|
await manager.join_game("sid_2", "game_A", "away")
|
|
|
|
assert len(manager.game_rooms["game_A"]) == 2
|
|
assert "sid_1" in manager.game_rooms["game_A"]
|
|
assert "sid_2" in manager.game_rooms["game_A"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_join_game_enters_socketio_room(self):
|
|
"""
|
|
Verify join_game() calls Socket.io enter_room.
|
|
|
|
Must register with Socket.io's room system for broadcasting to work.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.enter_room = AsyncMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
await manager.join_game("sid_123", "game_A", "home")
|
|
|
|
sio.enter_room.assert_called_once_with("sid_123", "game_A")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_join_game_broadcasts_user_connected(self):
|
|
"""
|
|
Verify join_game() broadcasts user_connected to the room.
|
|
|
|
Other participants should be notified when a new user joins.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.enter_room = AsyncMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
await manager.join_game("sid_123", "game_A", "home")
|
|
|
|
# Should broadcast user_connected
|
|
sio.emit.assert_called()
|
|
call_args = sio.emit.call_args
|
|
assert call_args[0][0] == "user_connected"
|
|
assert call_args[0][1]["user_id"] == "user_456"
|
|
assert call_args[0][1]["role"] == "home"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leave_game_removes_from_room(self):
|
|
"""
|
|
Verify leave_game() removes user from specific game room.
|
|
|
|
User should only be removed from the specified game.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.enter_room = AsyncMock()
|
|
sio.leave_room = AsyncMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
await manager.join_game("sid_123", "game_A", "home")
|
|
await manager.join_game("sid_123", "game_B", "away")
|
|
|
|
await manager.leave_game("sid_123", "game_A")
|
|
|
|
assert "sid_123" not in manager.game_rooms["game_A"]
|
|
assert "sid_123" in manager.game_rooms["game_B"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_leave_game_leaves_socketio_room(self):
|
|
"""
|
|
Verify leave_game() calls Socket.io leave_room.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.enter_room = AsyncMock()
|
|
sio.leave_room = AsyncMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
await manager.join_game("sid_123", "game_A", "home")
|
|
await manager.leave_game("sid_123", "game_A")
|
|
|
|
sio.leave_room.assert_called_once_with("sid_123", "game_A")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_game_participants(self):
|
|
"""
|
|
Verify get_game_participants() returns all SIDs in room.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.enter_room = AsyncMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_1", "user_1")
|
|
await manager.connect("sid_2", "user_2")
|
|
await manager.join_game("sid_1", "game_A", "home")
|
|
await manager.join_game("sid_2", "game_A", "away")
|
|
|
|
participants = manager.get_game_participants("game_A")
|
|
|
|
assert len(participants) == 2
|
|
assert "sid_1" in participants
|
|
assert "sid_2" in participants
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_game_participants_empty_room(self):
|
|
"""
|
|
Verify get_game_participants() returns empty set for unknown game.
|
|
"""
|
|
sio = MagicMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
participants = manager.get_game_participants("nonexistent_game")
|
|
|
|
assert participants == set()
|
|
|
|
|
|
# ============================================================================
|
|
# BROADCASTING TESTS
|
|
# ============================================================================
|
|
|
|
|
|
class TestBroadcasting:
|
|
"""Tests for event emission functionality."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_to_game_emits_to_room(self):
|
|
"""
|
|
Verify broadcast_to_game() emits to the correct Socket.io room.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.broadcast_to_game("game_A", "test_event", {"key": "value"})
|
|
|
|
sio.emit.assert_called_once_with(
|
|
"test_event", {"key": "value"}, room="game_A"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_to_empty_room_noop(self):
|
|
"""
|
|
Verify broadcasting to empty room doesn't raise exception.
|
|
|
|
Socket.io handles empty rooms gracefully - we just emit.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
# Should not raise
|
|
await manager.broadcast_to_game("empty_game", "event", {})
|
|
|
|
sio.emit.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_emit_to_user_targets_sid(self):
|
|
"""
|
|
Verify emit_to_user() sends to specific session ID.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.emit_to_user("sid_123", "private_event", {"secret": "data"})
|
|
|
|
sio.emit.assert_called_once_with(
|
|
"private_event", {"secret": "data"}, room="sid_123"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_emit_to_user_vs_broadcast_isolation(self):
|
|
"""
|
|
Verify emit_to_user and broadcast_to_game are independent.
|
|
|
|
Private message should not go to game room, and vice versa.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.emit = AsyncMock()
|
|
sio.enter_room = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
await manager.join_game("sid_123", "game_A", "home")
|
|
|
|
# Clear previous emits
|
|
sio.emit.reset_mock()
|
|
|
|
# Send private message
|
|
await manager.emit_to_user("sid_123", "private", {"data": 1})
|
|
|
|
# Verify it went to sid, not game room
|
|
call_args = sio.emit.call_args
|
|
assert call_args[1]["room"] == "sid_123" # Not "game_A"
|
|
|
|
|
|
# ============================================================================
|
|
# SESSION INFO TESTS
|
|
# ============================================================================
|
|
|
|
|
|
class TestSessionInfo:
|
|
"""Tests for SessionInfo dataclass."""
|
|
|
|
def test_session_info_creates_with_defaults(self):
|
|
"""
|
|
Verify SessionInfo creates with required fields and defaults.
|
|
|
|
games should default to empty set, ip_address to None.
|
|
"""
|
|
now = pendulum.now("UTC")
|
|
info = SessionInfo(
|
|
user_id="user_123",
|
|
connected_at=now,
|
|
last_activity=now,
|
|
)
|
|
|
|
assert info.user_id == "user_123"
|
|
assert info.connected_at == now
|
|
assert info.last_activity == now
|
|
assert info.games == set()
|
|
assert info.ip_address is None
|
|
|
|
def test_session_info_with_ip_address(self):
|
|
"""Verify SessionInfo stores IP address when provided."""
|
|
now = pendulum.now("UTC")
|
|
info = SessionInfo(
|
|
user_id="user_123",
|
|
connected_at=now,
|
|
last_activity=now,
|
|
ip_address="192.168.1.100",
|
|
)
|
|
|
|
assert info.ip_address == "192.168.1.100"
|
|
|
|
def test_inactive_seconds_fresh_session(self):
|
|
"""
|
|
Verify inactive_seconds returns near-zero for fresh session.
|
|
|
|
A session just created should have very low inactivity.
|
|
"""
|
|
now = pendulum.now("UTC")
|
|
info = SessionInfo(
|
|
user_id="user_123",
|
|
connected_at=now,
|
|
last_activity=now,
|
|
)
|
|
|
|
# Should be very close to 0 (within 1 second)
|
|
assert info.inactive_seconds() < 1.0
|
|
|
|
def test_inactive_seconds_old_session(self):
|
|
"""
|
|
Verify inactive_seconds returns correct duration for old session.
|
|
|
|
A session inactive for 5 minutes should report ~300 seconds.
|
|
"""
|
|
now = pendulum.now("UTC")
|
|
five_min_ago = now.subtract(minutes=5)
|
|
info = SessionInfo(
|
|
user_id="user_123",
|
|
connected_at=five_min_ago,
|
|
last_activity=five_min_ago,
|
|
)
|
|
|
|
# Should be approximately 300 seconds (allow some tolerance)
|
|
inactive = info.inactive_seconds()
|
|
assert 299.0 < inactive < 302.0
|
|
|
|
|
|
# ============================================================================
|
|
# SESSION ACTIVITY TESTS
|
|
# ============================================================================
|
|
|
|
|
|
class TestSessionActivity:
|
|
"""Tests for session activity tracking."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_activity_refreshes_timestamp(self):
|
|
"""
|
|
Verify update_activity() refreshes last_activity timestamp.
|
|
|
|
After calling update_activity, the session should show recent activity.
|
|
"""
|
|
sio = MagicMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
|
|
# Get original timestamp
|
|
session = manager.get_session("sid_123")
|
|
original_activity = session.last_activity
|
|
|
|
# Small delay to ensure time difference
|
|
await asyncio.sleep(0.01)
|
|
|
|
# Update activity
|
|
await manager.update_activity("sid_123")
|
|
|
|
# Timestamp should be updated
|
|
updated_session = manager.get_session("sid_123")
|
|
assert updated_session.last_activity > original_activity
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_activity_unknown_sid_noop(self):
|
|
"""
|
|
Verify update_activity() for unknown SID is safe no-op.
|
|
|
|
Should not raise exception for non-existent session.
|
|
"""
|
|
sio = MagicMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
# Should not raise
|
|
await manager.update_activity("unknown_sid")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session_returns_info(self):
|
|
"""
|
|
Verify get_session() returns SessionInfo for valid SID.
|
|
"""
|
|
sio = MagicMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456", ip_address="10.0.0.1")
|
|
|
|
session = manager.get_session("sid_123")
|
|
|
|
assert session is not None
|
|
assert session.user_id == "user_456"
|
|
assert session.ip_address == "10.0.0.1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session_unknown_returns_none(self):
|
|
"""
|
|
Verify get_session() returns None for unknown SID.
|
|
"""
|
|
sio = MagicMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
session = manager.get_session("unknown_sid")
|
|
|
|
assert session is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_id_returns_user(self):
|
|
"""
|
|
Verify get_user_id() returns user_id for valid SID.
|
|
"""
|
|
sio = MagicMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
|
|
user_id = manager.get_user_id("sid_123")
|
|
|
|
assert user_id == "user_456"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_id_unknown_returns_none(self):
|
|
"""
|
|
Verify get_user_id() returns None for unknown SID.
|
|
"""
|
|
sio = MagicMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
user_id = manager.get_user_id("unknown_sid")
|
|
|
|
assert user_id is None
|
|
|
|
|
|
# ============================================================================
|
|
# SESSION EXPIRATION TESTS
|
|
# ============================================================================
|
|
|
|
|
|
class TestSessionExpiration:
|
|
"""Tests for session expiration functionality."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_expire_inactive_sessions_removes_old(self):
|
|
"""
|
|
Verify expire_inactive_sessions() removes sessions beyond timeout.
|
|
|
|
Sessions inactive longer than timeout should be disconnected.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.emit = AsyncMock()
|
|
sio.disconnect = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
|
|
# Manually set session to be old
|
|
manager._sessions["sid_123"].last_activity = pendulum.now("UTC").subtract(minutes=10)
|
|
|
|
# Expire with 5 minute timeout
|
|
expired = await manager.expire_inactive_sessions(timeout_seconds=300)
|
|
|
|
assert "sid_123" in expired
|
|
assert "sid_123" not in manager._sessions
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_expire_inactive_sessions_keeps_active(self):
|
|
"""
|
|
Verify expire_inactive_sessions() keeps recently active sessions.
|
|
|
|
Sessions with recent activity should not be expired.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
await manager.update_activity("sid_123") # Fresh activity
|
|
|
|
# Expire with 5 minute timeout
|
|
expired = await manager.expire_inactive_sessions(timeout_seconds=300)
|
|
|
|
assert "sid_123" not in expired
|
|
assert "sid_123" in manager._sessions
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_expire_inactive_calls_socketio_disconnect(self):
|
|
"""
|
|
Verify expire_inactive_sessions() calls Socket.io disconnect.
|
|
|
|
Expired sessions should be forcefully disconnected from Socket.io.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.emit = AsyncMock()
|
|
sio.disconnect = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
manager._sessions["sid_123"].last_activity = pendulum.now("UTC").subtract(minutes=10)
|
|
|
|
await manager.expire_inactive_sessions(timeout_seconds=300)
|
|
|
|
sio.disconnect.assert_called_once_with("sid_123")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_expire_inactive_handles_disconnect_error(self):
|
|
"""
|
|
Verify expire_inactive_sessions() handles Socket.io disconnect errors.
|
|
|
|
Should not raise if Socket.io disconnect fails (connection may already be gone).
|
|
"""
|
|
sio = MagicMock()
|
|
sio.emit = AsyncMock()
|
|
sio.disconnect = AsyncMock(side_effect=Exception("Already disconnected"))
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_123", "user_456")
|
|
manager._sessions["sid_123"].last_activity = pendulum.now("UTC").subtract(minutes=10)
|
|
|
|
# Should not raise
|
|
expired = await manager.expire_inactive_sessions(timeout_seconds=300)
|
|
|
|
assert "sid_123" in expired
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_expire_returns_list_of_expired_sids(self):
|
|
"""
|
|
Verify expire_inactive_sessions() returns list of expired session IDs.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.emit = AsyncMock()
|
|
sio.disconnect = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
# Connect multiple sessions
|
|
await manager.connect("sid_1", "user_1")
|
|
await manager.connect("sid_2", "user_2")
|
|
await manager.connect("sid_3", "user_3")
|
|
|
|
# Make only some sessions old
|
|
manager._sessions["sid_1"].last_activity = pendulum.now("UTC").subtract(minutes=10)
|
|
manager._sessions["sid_3"].last_activity = pendulum.now("UTC").subtract(minutes=10)
|
|
# sid_2 stays fresh
|
|
|
|
expired = await manager.expire_inactive_sessions(timeout_seconds=300)
|
|
|
|
assert len(expired) == 2
|
|
assert "sid_1" in expired
|
|
assert "sid_3" in expired
|
|
assert "sid_2" not in expired
|
|
|
|
|
|
# ============================================================================
|
|
# CONNECTION STATISTICS TESTS
|
|
# ============================================================================
|
|
|
|
|
|
class TestConnectionStatistics:
|
|
"""Tests for connection statistics functionality."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_stats_empty(self):
|
|
"""
|
|
Verify get_stats() returns zero counts when no connections.
|
|
"""
|
|
sio = MagicMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
stats = manager.get_stats()
|
|
|
|
assert stats["total_sessions"] == 0
|
|
assert stats["unique_users"] == 0
|
|
assert stats["active_game_rooms"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_stats_with_sessions(self):
|
|
"""
|
|
Verify get_stats() counts active sessions correctly.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.enter_room = AsyncMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_1", "user_1")
|
|
await manager.connect("sid_2", "user_2")
|
|
await manager.connect("sid_3", "user_1") # Same user, different session
|
|
|
|
stats = manager.get_stats()
|
|
|
|
assert stats["total_sessions"] == 3
|
|
assert stats["unique_users"] == 2 # Only 2 unique users
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_stats_with_games(self):
|
|
"""
|
|
Verify get_stats() counts active game rooms.
|
|
"""
|
|
sio = MagicMock()
|
|
sio.enter_room = AsyncMock()
|
|
sio.emit = AsyncMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_1", "user_1")
|
|
await manager.connect("sid_2", "user_2")
|
|
await manager.join_game("sid_1", "game_A", "home")
|
|
await manager.join_game("sid_2", "game_A", "away")
|
|
await manager.join_game("sid_1", "game_B", "home")
|
|
|
|
stats = manager.get_stats()
|
|
|
|
assert stats["active_game_rooms"] == 2
|
|
assert "game_A" in stats["sessions_per_game"]
|
|
assert stats["sessions_per_game"]["game_A"] == 2
|
|
assert stats["sessions_per_game"]["game_B"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_stats_inactivity_distribution(self):
|
|
"""
|
|
Verify get_stats() categorizes sessions by inactivity.
|
|
"""
|
|
sio = MagicMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
await manager.connect("sid_1", "user_1")
|
|
await manager.connect("sid_2", "user_2")
|
|
await manager.connect("sid_3", "user_3")
|
|
|
|
# Make sessions have different inactivity levels
|
|
now = pendulum.now("UTC")
|
|
manager._sessions["sid_1"].last_activity = now # Active (<1m)
|
|
manager._sessions["sid_2"].last_activity = now.subtract(minutes=3) # 1-5m
|
|
manager._sessions["sid_3"].last_activity = now.subtract(minutes=20) # >15m
|
|
|
|
stats = manager.get_stats()
|
|
|
|
assert stats["inactivity_distribution"]["<1m"] == 1
|
|
assert stats["inactivity_distribution"]["1-5m"] == 1
|
|
assert stats["inactivity_distribution"][">15m"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_stats_session_ages(self):
|
|
"""
|
|
Verify get_stats() calculates session age statistics.
|
|
"""
|
|
sio = MagicMock()
|
|
manager = ConnectionManager(sio)
|
|
|
|
now = pendulum.now("UTC")
|
|
|
|
await manager.connect("sid_1", "user_1")
|
|
# Override connected_at to simulate older connection
|
|
manager._sessions["sid_1"].connected_at = now.subtract(minutes=30)
|
|
|
|
await manager.connect("sid_2", "user_2")
|
|
# Fresh connection
|
|
|
|
stats = manager.get_stats()
|
|
|
|
# Oldest should be around 30 minutes = 1800 seconds
|
|
assert stats["oldest_session_seconds"] > 1700
|
|
|
|
# Average should be around 15 minutes = 900 seconds
|
|
assert 800 < stats["avg_session_seconds"] < 1000
|