""" Unit tests for WebSocket messaging (ws/server/websocket_messaging.py). Tests the WebSocket messaging utilities: - BatchedUpdate: Message batching and accumulation - ComplexHandler: JSON serialization for complex objects - serverClass: Server state tracking - Message sending functions: sendToUser, sendEventMessage, sendUserInfo, sendDict Run with: pytest tests/unit/test_websocket_messaging.py -v """ import pytest import json from datetime import datetime from unittest.mock import Mock, AsyncMock, MagicMock, patch import sys from pathlib import Path from types import SimpleNamespace # Add ws directory to path ws_dir = Path(__file__).parent.parent.parent / 'ws' sys.path.insert(0, str(ws_dir)) from server.websocket_messaging import ( BatchedUpdate, serverClass, ComplexHandler, sendToUser, sendEventMessage, sendUserInfo, sendDict, ) # ============================================================================ # FIXTURES # ============================================================================ @pytest.fixture def mock_websocket(): """Create a mock WebSocket connection""" ws = AsyncMock() ws.userID = "test_user_123" ws.send = AsyncMock() return ws @pytest.fixture def mock_player(): """Create a mock player object""" # Use a simple class instead of Mock to avoid recursion issues with serialization class MockPlayer: def __init__(self): self.userID = "test_user_123" self.c = SimpleNamespace( firstname="John", lastname="Doe", money=1000, energy=100 ) return MockPlayer() @pytest.fixture def mock_event(): """Create a mock event object""" event = Mock() event.__dict__ = { 'id': 'event_123', 'type': 'messageEvent', 'title': 'Test Event', 'message': 'Test message' } return event # ============================================================================ # BATCHED UPDATE TESTS (5 tests) # ============================================================================ def test_batched_update_collects_messages(): """Test that messages are added to the batch correctly""" # Arrange batch = BatchedUpdate() # Act batch.add('energy', 95) batch.add('money', 1050) batch.add('hourOfDay', 14) # Assert result = batch.to_dict() assert result is not None assert result['type'] == 'batch_update' assert result['updates']['energy'] == 95 assert result['updates']['money'] == 1050 assert result['updates']['hourOfDay'] == 14 assert len(result['updates']) == 3 def test_batched_update_is_empty(): """Test that is_empty() correctly identifies empty batches""" # Arrange batch = BatchedUpdate() # Assert - initially empty assert batch.is_empty() is True # Act - add item batch.add('test', 'value') # Assert - no longer empty assert batch.is_empty() is False def test_batched_update_to_dict_returns_none_when_empty(): """Test that to_dict() returns None for empty batches""" # Arrange batch = BatchedUpdate() # Act result = batch.to_dict() # Assert assert result is None def test_batched_update_to_json(): """Test that to_json() correctly serializes batch to JSON string""" # Arrange batch = BatchedUpdate() batch.add('energy', 80) batch.add('money', 500) # Act json_str = batch.to_json() # Assert assert json_str is not None data = json.loads(json_str) assert data['type'] == 'batch_update' assert data['updates']['energy'] == 80 assert data['updates']['money'] == 500 def test_batched_update_preserves_order(): """Test that message order is preserved when adding to batch""" # Arrange batch = BatchedUpdate() keys = ['first', 'second', 'third', 'fourth'] # Act for i, key in enumerate(keys): batch.add(key, i) # Assert result = batch.to_dict() updates = result['updates'] # Dict keys maintain insertion order in Python 3.7+ assert list(updates.keys()) == keys assert updates['first'] == 0 assert updates['second'] == 1 assert updates['third'] == 2 assert updates['fourth'] == 3 # ============================================================================ # SERIALIZATION TESTS (5 tests) # ============================================================================ def test_complex_handler_serializes_set(): """Test that ComplexHandler converts set → list""" # Arrange test_set = {1, 2, 3, 4, 5} # Act result = ComplexHandler(test_set) # Assert assert isinstance(result, list) assert set(result) == test_set def test_complex_handler_serializes_namespace(): """Test that ComplexHandler converts SimpleNamespace → dict""" # Arrange ns = SimpleNamespace(name="John", age=30, city="NYC") # Act result = ComplexHandler(ns) # Assert assert isinstance(result, dict) assert result['name'] == "John" assert result['age'] == 30 assert result['city'] == "NYC" def test_complex_handler_serializes_object_with_jsonable(): """Test that ComplexHandler uses jsonable() method if available""" # Arrange class CustomObject: def jsonable(self): return {'custom': 'data', 'value': 42} obj = CustomObject() # Act result = ComplexHandler(obj) # Assert assert result == {'custom': 'data', 'value': 42} def test_complex_handler_serializes_nested_objects(): """Test that ComplexHandler handles nested objects with __dict__""" # Arrange class InnerObject: def __init__(self): self.inner_value = "nested" class OuterObject: def __init__(self): self.outer_value = "outer" self.inner = InnerObject() obj = OuterObject() # Act result = ComplexHandler(obj) # Assert assert isinstance(result, dict) assert result['outer_value'] == "outer" assert 'inner' in result def test_serialize_unknown_type(): """Test that unknown types raise TypeError gracefully""" # Arrange class NoSerialize: # No jsonable(), no __dict__ (using __slots__) __slots__ = ['value'] def __init__(self): self.value = 42 obj = NoSerialize() # Act & Assert with pytest.raises(TypeError) as exc_info: ComplexHandler(obj) assert "is not JSON serializable" in str(exc_info.value) # ============================================================================ # MESSAGE TYPES TESTS (6 tests) # ============================================================================ @pytest.mark.asyncio async def test_send_player_object_message(mock_websocket, mock_player): """Test sendUserInfo sends player object in correct format""" # Arrange with patch('server.websocket_registry.USERS', {mock_websocket.userID: mock_websocket}): # Act await sendUserInfo(mock_player, mock_websocket) # Assert mock_websocket.send.assert_called_once() sent_data = mock_websocket.send.call_args[0][0] parsed = json.loads(sent_data) assert 'userID' in parsed assert parsed['userID'] == "test_user_123" @pytest.mark.asyncio async def test_send_message_event(mock_websocket, mock_event): """Test sendEventMessage sends event in correct format""" # Arrange with patch('server.websocket_registry.USERS', {mock_websocket.userID: mock_websocket}): # Act await sendEventMessage(mock_websocket, mock_event) # Assert mock_websocket.send.assert_called_once() sent_data = mock_websocket.send.call_args[0][0] parsed = json.loads(sent_data) assert parsed['id'] == 'event_123' assert parsed['type'] == 'messageEvent' assert parsed['title'] == 'Test Event' @pytest.mark.asyncio async def test_send_dict_message(mock_websocket): """Test sendDict sends dictionary in correct format""" # Arrange test_dict = { 'type': 'update', 'energy': 85, 'money': 1200 } with patch('server.websocket_registry.USERS', {mock_websocket.userID: mock_websocket}): # Act await sendDict(mock_websocket, test_dict) # Assert mock_websocket.send.assert_called_once() sent_data = mock_websocket.send.call_args[0][0] parsed = json.loads(sent_data) assert parsed['type'] == 'update' assert parsed['energy'] == 85 assert parsed['money'] == 1200 @pytest.mark.asyncio async def test_send_to_user_with_valid_connection(mock_websocket): """Test sendToUser sends message when user connection exists""" # Arrange message = json.dumps({'type': 'test', 'value': 123}) with patch('server.websocket_registry.USERS', {mock_websocket.userID: mock_websocket}): # Act await sendToUser(mock_websocket, message) # Assert mock_websocket.send.assert_called_once_with(message) @pytest.mark.asyncio async def test_send_to_user_with_no_connection(mock_websocket, capsys): """Test sendToUser handles missing user connection gracefully""" # Arrange message = json.dumps({'type': 'test'}) with patch('server.websocket_registry.USERS', {}): # Empty users dict # Act await sendToUser(mock_websocket, message) # Assert - should not raise exception captured = capsys.readouterr() assert "No active websocket found" in captured.out @pytest.mark.asyncio async def test_send_to_user_handles_send_error(mock_websocket, capsys): """Test sendToUser handles send errors gracefully""" # Arrange message = json.dumps({'type': 'test'}) mock_websocket.send.side_effect = Exception("Connection lost") with patch('server.websocket_registry.USERS', {mock_websocket.userID: mock_websocket}): # Act await sendToUser(mock_websocket, message) # Assert - should not raise exception captured = capsys.readouterr() assert "Error sending to" in captured.out # ============================================================================ # SERVER CLASS TESTS (3 tests) # ============================================================================ def test_server_class_initialization(): """Test that serverClass initializes with correct default values""" # Arrange & Act server = serverClass() # Assert assert hasattr(server, 'ticks') assert server.ticks == 0 def test_server_class_tick_increment(): """Test that server ticks can be incremented""" # Arrange server = serverClass() # Act server.ticks += 1 server.ticks += 1 # Assert assert server.ticks == 2 def test_server_class_tick_reset(): """Test that server ticks can be reset""" # Arrange server = serverClass() server.ticks = 100 # Act server.ticks = 0 # Assert assert server.ticks == 0 # ============================================================================ # EDGE CASES & INTEGRATION TESTS (3 tests) # ============================================================================ def test_complex_handler_with_datetime(): """Test ComplexHandler with datetime objects (if they have __dict__)""" # Arrange class CustomDateTime: def __init__(self): self.year = 2025 self.month = 11 self.day = 14 dt = CustomDateTime() # Act result = ComplexHandler(dt) # Assert assert isinstance(result, dict) assert result['year'] == 2025 assert result['month'] == 11 assert result['day'] == 14 def test_batched_update_with_complex_objects(): """Test that BatchedUpdate can handle complex objects via ComplexHandler""" # Arrange batch = BatchedUpdate() test_set = {1, 2, 3} # Act batch.add('events', test_set) json_str = batch.to_json() # Assert assert json_str is not None data = json.loads(json_str) assert isinstance(data['updates']['events'], list) assert set(data['updates']['events']) == test_set @pytest.mark.asyncio async def test_send_event_with_none_websocket(mock_event): """Test sendEventMessage handles None websocket gracefully""" # Arrange & Act await sendEventMessage(None, mock_event) # Assert - should not raise exception (no assertion needed, just shouldn't crash)