5ed19c8f0f
- Introduced a new test suite for the PlaylistService covering various functionalities including creation, retrieval, updating, and deletion of playlists. - Added tests for handling sounds within playlists, ensuring correct behavior when adding/removing sounds and managing current playlists. - Refactored socket service tests for improved readability by adjusting function signatures. - Cleaned up unnecessary whitespace in sound normalizer and sound scanner tests for consistency. - Enhanced audio utility tests to ensure accurate hash and size calculations, including edge cases for nonexistent files. - Removed redundant blank lines in cookie utility tests for cleaner code.
281 lines
9.2 KiB
Python
281 lines
9.2 KiB
Python
"""Tests for socket service."""
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
import socketio
|
|
|
|
from app.services.socket import SocketManager
|
|
|
|
|
|
class TestSocketManager:
|
|
"""Test socket manager service."""
|
|
|
|
@pytest.fixture
|
|
def socket_manager(self):
|
|
"""Create a fresh socket manager for testing."""
|
|
return SocketManager()
|
|
|
|
@pytest.fixture
|
|
def mock_sio(self, socket_manager):
|
|
"""Mock the socket.io server."""
|
|
socket_manager.sio = AsyncMock(spec=socketio.AsyncServer)
|
|
return socket_manager.sio
|
|
|
|
def test_init_creates_socket_server(self):
|
|
"""Test that socket manager initializes with proper configuration."""
|
|
manager = SocketManager()
|
|
|
|
assert manager.sio is not None
|
|
assert isinstance(manager.user_rooms, dict)
|
|
assert isinstance(manager.socket_users, dict)
|
|
assert len(manager.user_rooms) == 0
|
|
assert len(manager.socket_users) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_to_user_success(self, socket_manager, mock_sio):
|
|
"""Test sending message to connected user."""
|
|
user_id = "123"
|
|
room_id = "user_123"
|
|
socket_manager.user_rooms[user_id] = room_id
|
|
|
|
event = "test_event"
|
|
data = {"message": "hello"}
|
|
|
|
result = await socket_manager.send_to_user(user_id, event, data)
|
|
|
|
assert result is True
|
|
mock_sio.emit.assert_called_once_with(event, data, room=room_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_to_user_not_connected(self, socket_manager, mock_sio):
|
|
"""Test sending message to user who is not connected."""
|
|
user_id = "999"
|
|
event = "test_event"
|
|
data = {"message": "hello"}
|
|
|
|
result = await socket_manager.send_to_user(user_id, event, data)
|
|
|
|
assert result is False
|
|
mock_sio.emit.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_to_all(self, socket_manager, mock_sio):
|
|
"""Test broadcasting message to all users."""
|
|
event = "broadcast_event"
|
|
data = {"message": "announcement"}
|
|
|
|
await socket_manager.broadcast_to_all(event, data)
|
|
|
|
mock_sio.emit.assert_called_once_with(event, data)
|
|
|
|
def test_get_connected_users(self, socket_manager):
|
|
"""Test getting list of connected users."""
|
|
# Add some users
|
|
socket_manager.user_rooms["1"] = "user_1"
|
|
socket_manager.user_rooms["2"] = "user_2"
|
|
socket_manager.user_rooms["3"] = "user_3"
|
|
|
|
connected_users = socket_manager.get_connected_users()
|
|
|
|
assert len(connected_users) == 3
|
|
assert "1" in connected_users
|
|
assert "2" in connected_users
|
|
assert "3" in connected_users
|
|
|
|
def test_get_room_info(self, socket_manager):
|
|
"""Test getting room information."""
|
|
# Add some users
|
|
socket_manager.user_rooms["1"] = "user_1"
|
|
socket_manager.user_rooms["2"] = "user_2"
|
|
|
|
room_info = socket_manager.get_room_info()
|
|
|
|
assert room_info["total_users"] == 2
|
|
assert room_info["connected_users"] == ["1", "2"]
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("app.services.socket.extract_access_token_from_cookies")
|
|
@patch("app.services.socket.JWTUtils.decode_access_token")
|
|
async def test_connect_handler_success(
|
|
self, mock_decode, mock_extract_token, socket_manager, mock_sio
|
|
):
|
|
"""Test successful connection with valid token."""
|
|
# Setup mocks
|
|
mock_extract_token.return_value = "valid_token"
|
|
mock_decode.return_value = {"sub": "123"}
|
|
|
|
# Mock environment
|
|
environ = {"HTTP_COOKIE": "access_token=valid_token"}
|
|
|
|
# Access the connect handler directly
|
|
handlers = {}
|
|
original_event = socket_manager.sio.event
|
|
|
|
def mock_event(func):
|
|
handlers[func.__name__] = func
|
|
return func
|
|
|
|
socket_manager.sio.event = mock_event
|
|
socket_manager._setup_handlers()
|
|
|
|
# Call the connect handler
|
|
await handlers["connect"]("test_sid", environ)
|
|
|
|
# Verify token extraction and validation
|
|
mock_extract_token.assert_called_once_with("access_token=valid_token")
|
|
mock_decode.assert_called_once_with("valid_token")
|
|
|
|
# Verify user tracking
|
|
assert socket_manager.socket_users["test_sid"] == "123"
|
|
assert socket_manager.user_rooms["123"] == "user_123"
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("app.services.socket.extract_access_token_from_cookies")
|
|
async def test_connect_handler_no_token(
|
|
self, mock_extract_token, socket_manager, mock_sio
|
|
):
|
|
"""Test connection with no access token."""
|
|
# Setup mocks
|
|
mock_extract_token.return_value = None
|
|
|
|
# Mock environment
|
|
environ = {"HTTP_COOKIE": ""}
|
|
|
|
# Access the connect handler directly
|
|
handlers = {}
|
|
original_event = socket_manager.sio.event
|
|
|
|
def mock_event(func):
|
|
handlers[func.__name__] = func
|
|
return func
|
|
|
|
socket_manager.sio.event = mock_event
|
|
socket_manager._setup_handlers()
|
|
|
|
# Call the connect handler
|
|
await handlers["connect"]("test_sid", environ)
|
|
|
|
# Verify disconnection
|
|
mock_sio.disconnect.assert_called_once_with("test_sid")
|
|
|
|
# Verify no user tracking
|
|
assert "test_sid" not in socket_manager.socket_users
|
|
assert len(socket_manager.user_rooms) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("app.services.socket.extract_access_token_from_cookies")
|
|
@patch("app.services.socket.JWTUtils.decode_access_token")
|
|
async def test_connect_handler_invalid_token(
|
|
self, mock_decode, mock_extract_token, socket_manager, mock_sio
|
|
):
|
|
"""Test connection with invalid token."""
|
|
# Setup mocks
|
|
mock_extract_token.return_value = "invalid_token"
|
|
mock_decode.side_effect = Exception("Invalid token")
|
|
|
|
# Mock environment
|
|
environ = {"HTTP_COOKIE": "access_token=invalid_token"}
|
|
|
|
# Access the connect handler directly
|
|
handlers = {}
|
|
original_event = socket_manager.sio.event
|
|
|
|
def mock_event(func):
|
|
handlers[func.__name__] = func
|
|
return func
|
|
|
|
socket_manager.sio.event = mock_event
|
|
socket_manager._setup_handlers()
|
|
|
|
# Call the connect handler
|
|
await handlers["connect"]("test_sid", environ)
|
|
|
|
# Verify disconnection
|
|
mock_sio.disconnect.assert_called_once_with("test_sid")
|
|
|
|
# Verify no user tracking
|
|
assert "test_sid" not in socket_manager.socket_users
|
|
assert len(socket_manager.user_rooms) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("app.services.socket.extract_access_token_from_cookies")
|
|
@patch("app.services.socket.JWTUtils.decode_access_token")
|
|
async def test_connect_handler_missing_user_id(
|
|
self, mock_decode, mock_extract_token, socket_manager, mock_sio
|
|
):
|
|
"""Test connection with token missing user ID."""
|
|
# Setup mocks
|
|
mock_extract_token.return_value = "token_without_user_id"
|
|
mock_decode.return_value = {"other_field": "value"} # Missing 'sub'
|
|
|
|
# Mock environment
|
|
environ = {"HTTP_COOKIE": "access_token=token_without_user_id"}
|
|
|
|
# Access the connect handler directly
|
|
handlers = {}
|
|
original_event = socket_manager.sio.event
|
|
|
|
def mock_event(func):
|
|
handlers[func.__name__] = func
|
|
return func
|
|
|
|
socket_manager.sio.event = mock_event
|
|
socket_manager._setup_handlers()
|
|
|
|
# Call the connect handler
|
|
await handlers["connect"]("test_sid", environ)
|
|
|
|
# Verify disconnection
|
|
mock_sio.disconnect.assert_called_once_with("test_sid")
|
|
|
|
# Verify no user tracking
|
|
assert "test_sid" not in socket_manager.socket_users
|
|
assert len(socket_manager.user_rooms) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disconnect_handler(self, socket_manager, mock_sio):
|
|
"""Test disconnect handler."""
|
|
# Setup initial state
|
|
socket_manager.socket_users["test_sid"] = "123"
|
|
socket_manager.user_rooms["123"] = "user_123"
|
|
|
|
# Access the disconnect handler directly
|
|
handlers = {}
|
|
original_event = socket_manager.sio.event
|
|
|
|
def mock_event(func):
|
|
handlers[func.__name__] = func
|
|
return func
|
|
|
|
socket_manager.sio.event = mock_event
|
|
socket_manager._setup_handlers()
|
|
|
|
# Call the disconnect handler
|
|
await handlers["disconnect"]("test_sid")
|
|
|
|
# Verify cleanup
|
|
assert "test_sid" not in socket_manager.socket_users
|
|
assert "123" not in socket_manager.user_rooms
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disconnect_handler_unknown_socket(self, socket_manager, mock_sio):
|
|
"""Test disconnect handler with unknown socket."""
|
|
# Access the disconnect handler directly
|
|
handlers = {}
|
|
original_event = socket_manager.sio.event
|
|
|
|
def mock_event(func):
|
|
handlers[func.__name__] = func
|
|
return func
|
|
|
|
socket_manager.sio.event = mock_event
|
|
socket_manager._setup_handlers()
|
|
|
|
# Call the disconnect handler with unknown socket
|
|
await handlers["disconnect"]("unknown_sid")
|
|
|
|
# Should not raise any errors and state should remain clean
|
|
assert len(socket_manager.socket_users) == 0
|
|
assert len(socket_manager.user_rooms) == 0
|