154d466ff1
Add GameNamespaceHandler with full event handling for real-time gameplay: - handle_join: Join/rejoin games with visibility-filtered state - handle_action: Execute actions and broadcast state to participants - handle_resign: Process resignation and end game - handle_disconnect: Notify opponent of disconnection - Broadcast helpers for state, game over, and opponent status Includes 28 unit tests covering all handler methods. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
934 lines
29 KiB
Python
934 lines
29 KiB
Python
"""Tests for GameNamespaceHandler.
|
|
|
|
This module tests the WebSocket event handlers for the /game namespace.
|
|
All tests use dependency injection to mock GameService and ConnectionManager,
|
|
following the project's no-monkey-patching testing pattern.
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from app.core.enums import GameEndReason, TurnPhase
|
|
from app.core.models.game_state import GameState, PlayerState
|
|
from app.core.visibility import VisibleGameState
|
|
from app.schemas.ws_messages import ConnectionStatus, WSErrorCode
|
|
from app.services.connection_manager import ConnectionInfo
|
|
from app.services.game_service import (
|
|
ForcedActionRequiredError,
|
|
GameActionResult,
|
|
GameAlreadyEndedError,
|
|
GameEndResult,
|
|
GameJoinResult,
|
|
GameNotFoundError,
|
|
InvalidActionError,
|
|
NotPlayerTurnError,
|
|
PendingForcedAction,
|
|
)
|
|
from app.socketio.game_namespace import GameNamespaceHandler
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_game_service() -> AsyncMock:
|
|
"""Create a mock GameService.
|
|
|
|
The GameService orchestrates game operations - join, execute action,
|
|
resign, end game, and state retrieval.
|
|
"""
|
|
service = AsyncMock()
|
|
service.join_game = AsyncMock()
|
|
service.execute_action = AsyncMock()
|
|
service.resign_game = AsyncMock()
|
|
service.end_game = AsyncMock()
|
|
service.get_game_state = AsyncMock()
|
|
return service
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_connection_manager() -> AsyncMock:
|
|
"""Create a mock ConnectionManager.
|
|
|
|
The ConnectionManager tracks WebSocket connections and their
|
|
association with games.
|
|
"""
|
|
cm = AsyncMock()
|
|
cm.join_game = AsyncMock(return_value=True)
|
|
cm.leave_game = AsyncMock()
|
|
cm.get_connection = AsyncMock(return_value=None)
|
|
cm.get_game_user_sids = AsyncMock(return_value={})
|
|
cm.get_opponent_sid = AsyncMock(return_value=None)
|
|
return cm
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_sio() -> AsyncMock:
|
|
"""Create a mock Socket.IO AsyncServer.
|
|
|
|
The server is used to emit events and manage rooms.
|
|
"""
|
|
sio = AsyncMock()
|
|
sio.emit = AsyncMock()
|
|
sio.enter_room = AsyncMock()
|
|
sio.leave_room = AsyncMock()
|
|
return sio
|
|
|
|
|
|
@pytest.fixture
|
|
def handler(
|
|
mock_game_service: AsyncMock,
|
|
mock_connection_manager: AsyncMock,
|
|
) -> GameNamespaceHandler:
|
|
"""Create a GameNamespaceHandler with injected mock dependencies."""
|
|
return GameNamespaceHandler(
|
|
game_svc=mock_game_service,
|
|
conn_manager=mock_connection_manager,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_visible_state() -> VisibleGameState:
|
|
"""Create a sample visible game state for testing.
|
|
|
|
This represents what a player would see after joining a game.
|
|
"""
|
|
from app.core.visibility import VisiblePlayerState, VisibleZone
|
|
|
|
return VisibleGameState(
|
|
game_id="game-123",
|
|
viewer_id="player-1",
|
|
players={
|
|
"player-1": VisiblePlayerState(
|
|
player_id="player-1",
|
|
is_current_player=True,
|
|
deck_count=40,
|
|
hand=VisibleZone(count=7, cards=[], zone_type="hand"),
|
|
),
|
|
"player-2": VisiblePlayerState(
|
|
player_id="player-2",
|
|
is_current_player=False,
|
|
deck_count=40,
|
|
hand=VisibleZone(count=7, cards=[], zone_type="hand"),
|
|
),
|
|
},
|
|
current_player_id="player-1",
|
|
turn_number=1,
|
|
phase=TurnPhase.MAIN,
|
|
is_my_turn=True,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_game_state() -> GameState:
|
|
"""Create a sample game state for testing broadcasts."""
|
|
player1 = PlayerState(player_id="player-1")
|
|
player2 = PlayerState(player_id="player-2")
|
|
|
|
return GameState(
|
|
game_id="game-123",
|
|
players={"player-1": player1, "player-2": player2},
|
|
current_player_id="player-1",
|
|
turn_number=1,
|
|
phase=TurnPhase.MAIN,
|
|
)
|
|
|
|
|
|
class TestHandleJoin:
|
|
"""Tests for the handle_join event handler."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_join_success(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_connection_manager: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
sample_visible_state: VisibleGameState,
|
|
) -> None:
|
|
"""Test successful game join returns game state.
|
|
|
|
When a player joins a game they're a participant in, they should
|
|
receive the filtered game state and be added to the game room.
|
|
"""
|
|
mock_game_service.join_game.return_value = GameJoinResult(
|
|
success=True,
|
|
game_id="game-123",
|
|
player_id="player-1",
|
|
visible_state=sample_visible_state,
|
|
is_your_turn=True,
|
|
game_over=False,
|
|
)
|
|
|
|
result = await handler.handle_join(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"game_id": "game-123", "message_id": "msg-1"},
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["game_id"] == "game-123"
|
|
assert result["is_your_turn"] is True
|
|
assert "state" in result
|
|
|
|
# Should have joined the connection manager
|
|
mock_connection_manager.join_game.assert_called_once_with("sid-123", "game-123")
|
|
|
|
# Should have entered the Socket.IO room
|
|
mock_sio.enter_room.assert_called_once_with("sid-123", "game:game-123", namespace="/game")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_join_missing_game_id(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test that missing game_id returns error.
|
|
|
|
The game_id is required to join a game.
|
|
"""
|
|
result = await handler.handle_join(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"message_id": "msg-1"}, # No game_id
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"]["code"] == WSErrorCode.INVALID_MESSAGE.value
|
|
assert "game_id" in result["error"]["message"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_join_game_not_found(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test joining non-existent game returns error.
|
|
|
|
When GameService indicates the game wasn't found, we should
|
|
return a game_not_found error.
|
|
"""
|
|
mock_game_service.join_game.return_value = GameJoinResult(
|
|
success=False,
|
|
game_id="nonexistent",
|
|
player_id="player-1",
|
|
message="Game not found",
|
|
)
|
|
|
|
result = await handler.handle_join(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"game_id": "nonexistent"},
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"]["code"] == WSErrorCode.GAME_NOT_FOUND.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_join_not_participant(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test joining as non-participant returns error.
|
|
|
|
Only players in the game can join it.
|
|
"""
|
|
mock_game_service.join_game.return_value = GameJoinResult(
|
|
success=False,
|
|
game_id="game-123",
|
|
player_id="stranger",
|
|
message="You are not a participant in this game",
|
|
)
|
|
|
|
result = await handler.handle_join(
|
|
mock_sio,
|
|
"sid-123",
|
|
"stranger",
|
|
{"game_id": "game-123"},
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"]["code"] == WSErrorCode.NOT_IN_GAME.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_join_includes_pending_forced_action(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_connection_manager: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
sample_visible_state: VisibleGameState,
|
|
) -> None:
|
|
"""Test that pending forced action is included in join result.
|
|
|
|
When rejoining a game with a pending forced action, the client
|
|
needs to know what action is required.
|
|
"""
|
|
mock_game_service.join_game.return_value = GameJoinResult(
|
|
success=True,
|
|
game_id="game-123",
|
|
player_id="player-1",
|
|
visible_state=sample_visible_state,
|
|
is_your_turn=True,
|
|
pending_forced_action=PendingForcedAction(
|
|
player_id="player-1",
|
|
action_type="select_active",
|
|
reason="Your active Pokemon was knocked out.",
|
|
params={"available_bench_ids": ["bench-1"]},
|
|
),
|
|
)
|
|
|
|
result = await handler.handle_join(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"game_id": "game-123"},
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert "pending_forced_action" in result
|
|
assert result["pending_forced_action"]["action_type"] == "select_active"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_join_notifies_opponent(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_connection_manager: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
sample_visible_state: VisibleGameState,
|
|
) -> None:
|
|
"""Test that joining notifies the opponent.
|
|
|
|
When a player joins/rejoins, the opponent should be notified
|
|
of their connection status.
|
|
"""
|
|
mock_game_service.join_game.return_value = GameJoinResult(
|
|
success=True,
|
|
game_id="game-123",
|
|
player_id="player-1",
|
|
visible_state=sample_visible_state,
|
|
is_your_turn=True,
|
|
)
|
|
mock_connection_manager.get_opponent_sid.return_value = "opponent-sid"
|
|
|
|
await handler.handle_join(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"game_id": "game-123"},
|
|
)
|
|
|
|
# Should have tried to notify opponent
|
|
mock_connection_manager.get_opponent_sid.assert_called()
|
|
|
|
# Should have emitted opponent status to opponent
|
|
emit_calls = [
|
|
call for call in mock_sio.emit.call_args_list if call[0][0] == "game:opponent_status"
|
|
]
|
|
assert len(emit_calls) == 1
|
|
|
|
|
|
class TestHandleAction:
|
|
"""Tests for the handle_action event handler."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_success(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_connection_manager: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
sample_game_state: GameState,
|
|
) -> None:
|
|
"""Test successful action execution.
|
|
|
|
A valid action should be executed and the state should be
|
|
broadcast to all participants.
|
|
"""
|
|
mock_game_service.execute_action.return_value = GameActionResult(
|
|
success=True,
|
|
game_id="game-123",
|
|
action_type="pass",
|
|
message="Turn ended",
|
|
turn_changed=True,
|
|
current_player_id="player-2",
|
|
)
|
|
mock_game_service.get_game_state.return_value = sample_game_state
|
|
mock_connection_manager.get_game_user_sids.return_value = {
|
|
"player-1": "sid-1",
|
|
"player-2": "sid-2",
|
|
}
|
|
|
|
result = await handler.handle_action(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{
|
|
"game_id": "game-123",
|
|
"action": {"type": "pass"},
|
|
"message_id": "msg-1",
|
|
},
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["action_type"] == "pass"
|
|
assert result["turn_changed"] is True
|
|
|
|
# Should have broadcast state to participants
|
|
assert mock_sio.emit.called
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_missing_game_id(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test that missing game_id returns error."""
|
|
result = await handler.handle_action(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"action": {"type": "pass"}},
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"]["code"] == WSErrorCode.INVALID_MESSAGE.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_missing_action(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test that missing action returns error."""
|
|
result = await handler.handle_action(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"game_id": "game-123"},
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"]["code"] == WSErrorCode.INVALID_MESSAGE.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_invalid_action_format(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test that invalid action format returns error.
|
|
|
|
The action data must match a valid Action type.
|
|
"""
|
|
result = await handler.handle_action(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"game_id": "game-123", "action": {"type": "invalid_action_type"}},
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"]["code"] == WSErrorCode.INVALID_ACTION.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_game_not_found(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test action on non-existent game returns error."""
|
|
mock_game_service.execute_action.side_effect = GameNotFoundError("game-123")
|
|
|
|
result = await handler.handle_action(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"game_id": "game-123", "action": {"type": "pass"}},
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"]["code"] == WSErrorCode.GAME_NOT_FOUND.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_not_your_turn(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test action when not player's turn returns error."""
|
|
mock_game_service.execute_action.side_effect = NotPlayerTurnError(
|
|
"game-123", "player-2", "player-1"
|
|
)
|
|
|
|
result = await handler.handle_action(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-2",
|
|
{"game_id": "game-123", "action": {"type": "pass"}},
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"]["code"] == WSErrorCode.NOT_YOUR_TURN.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_forced_action_required(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test action when forced action required returns error."""
|
|
mock_game_service.execute_action.side_effect = ForcedActionRequiredError(
|
|
"game-123", "player-1", "select_active", "pass"
|
|
)
|
|
|
|
result = await handler.handle_action(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"game_id": "game-123", "action": {"type": "pass"}},
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"]["code"] == WSErrorCode.ACTION_NOT_ALLOWED.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_invalid_action(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test invalid action (rejected by engine) returns error."""
|
|
mock_game_service.execute_action.side_effect = InvalidActionError(
|
|
"game-123", "player-1", "Not enough energy"
|
|
)
|
|
|
|
result = await handler.handle_action(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"game_id": "game-123", "action": {"type": "attack", "attack_index": 0}},
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"]["code"] == WSErrorCode.INVALID_ACTION.value
|
|
assert "Not enough energy" in result["error"]["message"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_game_over(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_connection_manager: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
sample_game_state: GameState,
|
|
) -> None:
|
|
"""Test action that ends game broadcasts game over.
|
|
|
|
When an action results in game over, the result should include
|
|
game_over=True and the game should be ended.
|
|
"""
|
|
mock_game_service.execute_action.return_value = GameActionResult(
|
|
success=True,
|
|
game_id="game-123",
|
|
action_type="attack",
|
|
game_over=True,
|
|
winner_id="player-1",
|
|
end_reason=GameEndReason.PRIZES_TAKEN,
|
|
)
|
|
mock_game_service.get_game_state.return_value = sample_game_state
|
|
mock_game_service.end_game.return_value = GameEndResult(
|
|
success=True,
|
|
game_id="game-123",
|
|
winner_id="player-1",
|
|
)
|
|
mock_connection_manager.get_game_user_sids.return_value = {}
|
|
|
|
result = await handler.handle_action(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"game_id": "game-123", "action": {"type": "attack", "attack_index": 0}},
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["game_over"] is True
|
|
assert result["winner_id"] == "player-1"
|
|
|
|
# Should have called end_game
|
|
mock_game_service.end_game.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_action_includes_pending_forced_action(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_connection_manager: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
sample_game_state: GameState,
|
|
) -> None:
|
|
"""Test that action result includes pending forced action.
|
|
|
|
When an action creates a forced action (e.g., knockout triggers
|
|
select_active), the result should include it.
|
|
"""
|
|
mock_game_service.execute_action.return_value = GameActionResult(
|
|
success=True,
|
|
game_id="game-123",
|
|
action_type="attack",
|
|
pending_forced_action=PendingForcedAction(
|
|
player_id="player-2",
|
|
action_type="select_active",
|
|
reason="Your active was knocked out.",
|
|
),
|
|
)
|
|
mock_game_service.get_game_state.return_value = sample_game_state
|
|
mock_connection_manager.get_game_user_sids.return_value = {}
|
|
|
|
result = await handler.handle_action(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"game_id": "game-123", "action": {"type": "attack", "attack_index": 0}},
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert "pending_forced_action" in result
|
|
assert result["pending_forced_action"]["action_type"] == "select_active"
|
|
|
|
|
|
class TestHandleResign:
|
|
"""Tests for the handle_resign event handler."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resign_success(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_connection_manager: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
sample_game_state: GameState,
|
|
) -> None:
|
|
"""Test successful resignation.
|
|
|
|
Resignation should end the game with the opponent as winner.
|
|
"""
|
|
mock_game_service.resign_game.return_value = GameActionResult(
|
|
success=True,
|
|
game_id="game-123",
|
|
action_type="resign",
|
|
game_over=True,
|
|
winner_id="player-2",
|
|
end_reason=GameEndReason.RESIGNATION,
|
|
)
|
|
mock_game_service.get_game_state.return_value = sample_game_state
|
|
mock_game_service.end_game.return_value = GameEndResult(
|
|
success=True,
|
|
game_id="game-123",
|
|
winner_id="player-2",
|
|
)
|
|
mock_connection_manager.get_game_user_sids.return_value = {}
|
|
|
|
result = await handler.handle_resign(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"game_id": "game-123"},
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["game_over"] is True
|
|
assert result["winner_id"] == "player-2"
|
|
|
|
# Should have called end_game
|
|
mock_game_service.end_game.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resign_missing_game_id(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test resignation with missing game_id returns error."""
|
|
result = await handler.handle_resign(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{},
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"]["code"] == WSErrorCode.INVALID_MESSAGE.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resign_game_not_found(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test resignation from non-existent game returns error."""
|
|
mock_game_service.resign_game.side_effect = GameNotFoundError("game-123")
|
|
|
|
result = await handler.handle_resign(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"game_id": "game-123"},
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"]["code"] == WSErrorCode.GAME_NOT_FOUND.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resign_game_already_ended(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test resignation from ended game returns error."""
|
|
mock_game_service.resign_game.side_effect = GameAlreadyEndedError("game-123")
|
|
|
|
result = await handler.handle_resign(
|
|
mock_sio,
|
|
"sid-123",
|
|
"player-1",
|
|
{"game_id": "game-123"},
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"]["code"] == WSErrorCode.GAME_ENDED.value
|
|
|
|
|
|
class TestHandleDisconnect:
|
|
"""Tests for the handle_disconnect event handler."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disconnect_notifies_opponent(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_connection_manager: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test that disconnect notifies opponent.
|
|
|
|
When a player disconnects from a game, the opponent should
|
|
be notified of the disconnection.
|
|
"""
|
|
from datetime import UTC, datetime
|
|
|
|
mock_connection_manager.get_connection.return_value = ConnectionInfo(
|
|
sid="sid-123",
|
|
user_id="player-1",
|
|
game_id="game-123",
|
|
connected_at=datetime.now(UTC),
|
|
last_seen=datetime.now(UTC),
|
|
)
|
|
mock_connection_manager.get_opponent_sid.return_value = "opponent-sid"
|
|
|
|
await handler.handle_disconnect(mock_sio, "sid-123", "player-1")
|
|
|
|
# Should have looked up connection to find game
|
|
mock_connection_manager.get_connection.assert_called_once_with("sid-123")
|
|
|
|
# Should have emitted opponent status to opponent
|
|
emit_calls = [
|
|
call for call in mock_sio.emit.call_args_list if call[0][0] == "game:opponent_status"
|
|
]
|
|
assert len(emit_calls) == 1
|
|
assert emit_calls[0][1]["to"] == "opponent-sid"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disconnect_no_game(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_connection_manager: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test disconnect when not in a game does nothing.
|
|
|
|
If the player wasn't in a game, there's no opponent to notify.
|
|
"""
|
|
mock_connection_manager.get_connection.return_value = None
|
|
|
|
await handler.handle_disconnect(mock_sio, "sid-123", "player-1")
|
|
|
|
# Should not emit anything
|
|
mock_sio.emit.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disconnect_no_game_id_in_connection(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_connection_manager: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test disconnect when connection exists but not in game.
|
|
|
|
The connection may exist but game_id may be None if they
|
|
haven't joined a game yet.
|
|
"""
|
|
from datetime import UTC, datetime
|
|
|
|
mock_connection_manager.get_connection.return_value = ConnectionInfo(
|
|
sid="sid-123",
|
|
user_id="player-1",
|
|
game_id=None, # Not in a game
|
|
connected_at=datetime.now(UTC),
|
|
last_seen=datetime.now(UTC),
|
|
)
|
|
|
|
await handler.handle_disconnect(mock_sio, "sid-123", "player-1")
|
|
|
|
# Should not emit anything
|
|
mock_sio.emit.assert_not_called()
|
|
|
|
|
|
class TestBroadcastGameState:
|
|
"""Tests for the _broadcast_game_state helper method."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_sends_to_each_player(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_connection_manager: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
sample_game_state: GameState,
|
|
) -> None:
|
|
"""Test that broadcast sends filtered state to each player.
|
|
|
|
Each player should receive their own visibility-filtered
|
|
view of the game state.
|
|
"""
|
|
mock_game_service.get_game_state.return_value = sample_game_state
|
|
mock_connection_manager.get_game_user_sids.return_value = {
|
|
"player-1": "sid-1",
|
|
"player-2": "sid-2",
|
|
}
|
|
|
|
await handler._broadcast_game_state(mock_sio, "game-123")
|
|
|
|
# Should have emitted to both players
|
|
assert mock_sio.emit.call_count == 2
|
|
|
|
# Verify each emission was to a different player
|
|
emit_calls = mock_sio.emit.call_args_list
|
|
to_sids = {call[1]["to"] for call in emit_calls}
|
|
assert to_sids == {"sid-1", "sid-2"}
|
|
|
|
# All emissions should be game:state events
|
|
for call in emit_calls:
|
|
assert call[0][0] == "game:state"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_handles_game_not_found(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_game_service: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test that broadcast handles missing game gracefully.
|
|
|
|
If the game can't be found (e.g., already archived), the
|
|
broadcast should log a warning but not raise an error.
|
|
"""
|
|
mock_game_service.get_game_state.side_effect = GameNotFoundError("game-123")
|
|
|
|
# Should not raise
|
|
await handler._broadcast_game_state(mock_sio, "game-123")
|
|
|
|
# Should not have emitted anything
|
|
mock_sio.emit.assert_not_called()
|
|
|
|
|
|
class TestNotifyOpponentStatus:
|
|
"""Tests for the _notify_opponent_status helper method."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notify_sends_to_opponent(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_connection_manager: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test that notify sends status to opponent.
|
|
|
|
The opponent's connection should receive a status update
|
|
about the player's connection state.
|
|
"""
|
|
mock_connection_manager.get_opponent_sid.return_value = "opponent-sid"
|
|
|
|
await handler._notify_opponent_status(
|
|
mock_sio,
|
|
"sender-sid",
|
|
"game-123",
|
|
"player-1",
|
|
ConnectionStatus.CONNECTED,
|
|
)
|
|
|
|
mock_sio.emit.assert_called_once()
|
|
call = mock_sio.emit.call_args
|
|
assert call[0][0] == "game:opponent_status"
|
|
assert call[1]["to"] == "opponent-sid"
|
|
message = call[0][1]
|
|
assert message["opponent_id"] == "player-1"
|
|
assert message["status"] == ConnectionStatus.CONNECTED.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notify_no_opponent_connected(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
mock_connection_manager: AsyncMock,
|
|
mock_sio: AsyncMock,
|
|
) -> None:
|
|
"""Test that notify does nothing when opponent not connected.
|
|
|
|
If the opponent isn't connected, we can't notify them.
|
|
"""
|
|
mock_connection_manager.get_opponent_sid.return_value = None
|
|
|
|
await handler._notify_opponent_status(
|
|
mock_sio,
|
|
"sender-sid",
|
|
"game-123",
|
|
"player-1",
|
|
ConnectionStatus.CONNECTED,
|
|
)
|
|
|
|
mock_sio.emit.assert_not_called()
|
|
|
|
|
|
class TestErrorResponse:
|
|
"""Tests for the _error_response helper method."""
|
|
|
|
def test_error_response_format(
|
|
self,
|
|
handler: GameNamespaceHandler,
|
|
) -> None:
|
|
"""Test that error response has correct format.
|
|
|
|
Error responses should include success=False, error code,
|
|
message, and the original request's message_id.
|
|
"""
|
|
result = handler._error_response(
|
|
WSErrorCode.GAME_NOT_FOUND,
|
|
"Game not found",
|
|
"msg-123",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["error"]["code"] == WSErrorCode.GAME_NOT_FOUND.value
|
|
assert result["error"]["message"] == "Game not found"
|
|
assert result["request_message_id"] == "msg-123"
|