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>
446 lines
12 KiB
Python
446 lines
12 KiB
Python
"""
|
|
Shared fixtures for WebSocket handler tests.
|
|
|
|
Provides reusable mock objects and test utilities for all WebSocket test files.
|
|
Follows the pattern established in test_manual_outcome_handlers.py.
|
|
|
|
Author: Claude
|
|
Date: 2025-01-27
|
|
"""
|
|
|
|
import asyncio
|
|
import pytest
|
|
from contextlib import asynccontextmanager
|
|
from uuid import uuid4
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pendulum
|
|
|
|
from app.models.game_models import (
|
|
GameState,
|
|
LineupPlayerState,
|
|
TeamLineupState,
|
|
ManualOutcomeSubmission,
|
|
)
|
|
from app.config.result_charts import PlayOutcome
|
|
from app.core.roll_types import AbRoll, RollType
|
|
from app.core.play_resolver import PlayResult
|
|
from app.core.substitution_manager import SubstitutionResult
|
|
|
|
|
|
# ============================================================================
|
|
# CONNECTION MANAGER FIXTURES
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_manager():
|
|
"""
|
|
Mock ConnectionManager with AsyncMock methods for event emission.
|
|
|
|
Tracks calls to emit_to_user and broadcast_to_game for assertion.
|
|
"""
|
|
manager = MagicMock()
|
|
manager.emit_to_user = AsyncMock()
|
|
manager.broadcast_to_game = AsyncMock()
|
|
manager.join_game = AsyncMock()
|
|
manager.leave_game = AsyncMock()
|
|
manager.connect = AsyncMock()
|
|
manager.disconnect = AsyncMock()
|
|
manager.update_activity = AsyncMock() # Session activity tracking
|
|
# Tracking dictionaries
|
|
manager.user_sessions = {}
|
|
manager.game_rooms = {}
|
|
return manager
|
|
|
|
|
|
# ============================================================================
|
|
# STATE MANAGER FIXTURES
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_state_manager():
|
|
"""
|
|
Mock state_manager singleton with game_lock context manager.
|
|
|
|
Provides:
|
|
- get_state() returning mock game state
|
|
- update_state() tracking calls
|
|
- game_lock() as async context manager
|
|
- get_lineup() returning mock lineup
|
|
- recover_game() for state recovery
|
|
"""
|
|
state_mgr = MagicMock()
|
|
state_mgr.get_state = MagicMock(return_value=None)
|
|
state_mgr.update_state = MagicMock()
|
|
state_mgr.get_lineup = MagicMock(return_value=None)
|
|
state_mgr.set_lineup = MagicMock()
|
|
state_mgr.recover_game = AsyncMock(return_value=None)
|
|
|
|
# Create a proper async context manager for game_lock
|
|
@asynccontextmanager
|
|
async def mock_game_lock(game_id, timeout=30.0):
|
|
yield
|
|
|
|
state_mgr.game_lock = mock_game_lock
|
|
return state_mgr
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_state_manager_with_timeout():
|
|
"""
|
|
Mock state_manager that raises TimeoutError on game_lock.
|
|
|
|
Use this to test timeout handling in handlers.
|
|
"""
|
|
state_mgr = MagicMock()
|
|
state_mgr.get_state = MagicMock(return_value=None)
|
|
state_mgr.update_state = MagicMock()
|
|
|
|
@asynccontextmanager
|
|
async def timeout_lock(game_id, timeout=30.0):
|
|
raise asyncio.TimeoutError(f"Lock timeout for game {game_id}")
|
|
yield # Never reached, but needed for syntax
|
|
|
|
state_mgr.game_lock = timeout_lock
|
|
return state_mgr
|
|
|
|
|
|
# ============================================================================
|
|
# GAME ENGINE FIXTURES
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_game_engine():
|
|
"""
|
|
Mock game_engine singleton.
|
|
|
|
Provides:
|
|
- resolve_manual_play() for outcome resolution
|
|
- submit_defensive_decision() for defense strategy
|
|
- submit_offensive_decision() for offense strategy
|
|
"""
|
|
engine = MagicMock()
|
|
engine.resolve_manual_play = AsyncMock(return_value=None)
|
|
engine.submit_defensive_decision = AsyncMock(return_value=None)
|
|
engine.submit_offensive_decision = AsyncMock(return_value=None)
|
|
engine.start_game = AsyncMock(return_value=None)
|
|
return engine
|
|
|
|
|
|
# ============================================================================
|
|
# DICE SYSTEM FIXTURES
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_dice_system(mock_ab_roll):
|
|
"""
|
|
Mock dice_system singleton.
|
|
|
|
Returns mock_ab_roll from roll_ab().
|
|
"""
|
|
dice = MagicMock()
|
|
dice.roll_ab = MagicMock(return_value=mock_ab_roll)
|
|
return dice
|
|
|
|
|
|
# ============================================================================
|
|
# GAME STATE FIXTURES
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_game_id():
|
|
"""Generate a unique game ID for each test."""
|
|
return uuid4()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_game_state(sample_game_id):
|
|
"""
|
|
Create a mock active game state with current batter set.
|
|
|
|
Returns a GameState configured for:
|
|
- SBA league
|
|
- Active status
|
|
- Inning 1, top half
|
|
- 0 outs, 0-0 score
|
|
- Current batter assigned
|
|
"""
|
|
return GameState(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
current_batter=LineupPlayerState(
|
|
lineup_id=1,
|
|
card_id=100,
|
|
position="CF",
|
|
batting_order=1,
|
|
),
|
|
status="active",
|
|
inning=1,
|
|
half="top",
|
|
outs=0,
|
|
home_score=0,
|
|
away_score=0,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_game_state_with_runners(mock_game_state):
|
|
"""
|
|
Create game state with runners on base.
|
|
|
|
Runner on first and third (corners situation).
|
|
"""
|
|
mock_game_state.runner_on_first = LineupPlayerState(
|
|
lineup_id=2, card_id=200, position="LF", batting_order=2
|
|
)
|
|
mock_game_state.runner_on_third = LineupPlayerState(
|
|
lineup_id=3, card_id=300, position="RF", batting_order=3
|
|
)
|
|
return mock_game_state
|
|
|
|
|
|
# ============================================================================
|
|
# ROLL FIXTURES
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_ab_roll(sample_game_id):
|
|
"""
|
|
Create a mock AB roll for dice operations.
|
|
|
|
Standard roll values for predictable test behavior.
|
|
"""
|
|
return AbRoll(
|
|
roll_id="test_roll_123",
|
|
roll_type=RollType.AB,
|
|
league_id="sba",
|
|
timestamp=pendulum.now("UTC"),
|
|
game_id=sample_game_id,
|
|
d6_one=3,
|
|
d6_two_a=4,
|
|
d6_two_b=3,
|
|
chaos_d20=10,
|
|
resolution_d20=12,
|
|
d6_two_total=7,
|
|
check_wild_pitch=False,
|
|
check_passed_ball=False,
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# PLAY RESULT FIXTURES
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_play_result():
|
|
"""
|
|
Create a mock play result for outcome resolution.
|
|
|
|
Standard groundball out result.
|
|
"""
|
|
return PlayResult(
|
|
outcome=PlayOutcome.GROUNDBALL_C,
|
|
outs_recorded=1,
|
|
runs_scored=0,
|
|
batter_result=None,
|
|
runners_advanced=[],
|
|
description="Groundball to shortstop",
|
|
ab_roll=None,
|
|
is_hit=False,
|
|
is_out=True,
|
|
is_walk=False,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_play_result_hit():
|
|
"""Create a mock play result for a hit."""
|
|
return PlayResult(
|
|
outcome=PlayOutcome.SINGLE,
|
|
outs_recorded=0,
|
|
runs_scored=0,
|
|
batter_result="1B",
|
|
runners_advanced=[("B", "1B")],
|
|
description="Single to left field",
|
|
ab_roll=None,
|
|
is_hit=True,
|
|
is_out=False,
|
|
is_walk=False,
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# SUBSTITUTION FIXTURES
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_substitution_result_success():
|
|
"""Create a successful substitution result."""
|
|
return SubstitutionResult(
|
|
success=True,
|
|
player_out_lineup_id=1,
|
|
player_in_card_id=999,
|
|
new_lineup_id=100,
|
|
new_position="CF",
|
|
new_batting_order=1,
|
|
error_message=None,
|
|
error_code=None,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_substitution_result_failure():
|
|
"""Create a failed substitution result."""
|
|
return SubstitutionResult(
|
|
success=False,
|
|
player_out_lineup_id=1,
|
|
player_in_card_id=999,
|
|
new_lineup_id=None,
|
|
new_position=None,
|
|
new_batting_order=None,
|
|
error_message="Player not found in roster",
|
|
error_code="PLAYER_NOT_IN_ROSTER",
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# LINEUP FIXTURES
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_lineup_state():
|
|
"""
|
|
Create a mock team lineup state.
|
|
|
|
9 players in standard batting order with positions.
|
|
"""
|
|
players = []
|
|
positions = ["C", "1B", "2B", "SS", "3B", "LF", "CF", "RF", "DH"]
|
|
for i in range(9):
|
|
players.append(
|
|
LineupPlayerState(
|
|
lineup_id=i + 1,
|
|
card_id=(i + 1) * 100,
|
|
position=positions[i],
|
|
batting_order=i + 1,
|
|
)
|
|
)
|
|
return TeamLineupState(team_id=1, players=players)
|
|
|
|
|
|
# ============================================================================
|
|
# SOCKET.IO SERVER FIXTURES
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def sio_server():
|
|
"""
|
|
Create a bare Socket.io AsyncServer for handler registration.
|
|
|
|
Use this when you need to control patching manually.
|
|
"""
|
|
from socketio import AsyncServer
|
|
|
|
return AsyncServer()
|
|
|
|
|
|
@pytest.fixture
|
|
def sio_with_mocks(mock_manager, mock_state_manager, mock_game_engine, mock_dice_system):
|
|
"""
|
|
Create Socket.io server with handlers registered and all dependencies mocked.
|
|
|
|
This fixture patches all singletons BEFORE registering handlers to ensure
|
|
the handlers capture the mocked versions in their closures.
|
|
|
|
Returns a tuple of (sio, patches_dict) where patches_dict contains the
|
|
mock objects for additional configuration in tests.
|
|
"""
|
|
from socketio import AsyncServer
|
|
|
|
sio = AsyncServer()
|
|
|
|
with patch("app.websocket.handlers.state_manager", mock_state_manager), \
|
|
patch("app.websocket.handlers.game_engine", mock_game_engine), \
|
|
patch("app.websocket.handlers.dice_system", mock_dice_system):
|
|
from app.websocket.handlers import register_handlers
|
|
|
|
register_handlers(sio, mock_manager)
|
|
yield sio, {
|
|
"manager": mock_manager,
|
|
"state_manager": mock_state_manager,
|
|
"game_engine": mock_game_engine,
|
|
"dice_system": mock_dice_system,
|
|
}
|
|
|
|
|
|
# ============================================================================
|
|
# UTILITY FUNCTIONS
|
|
# ============================================================================
|
|
|
|
|
|
def get_handler(sio, handler_name: str):
|
|
"""
|
|
Get a specific event handler from the Socket.io server.
|
|
|
|
Args:
|
|
sio: AsyncServer instance with registered handlers
|
|
handler_name: Name of the event handler (e.g., 'roll_dice')
|
|
|
|
Returns:
|
|
The async handler function
|
|
"""
|
|
return sio.handlers["/"][handler_name]
|
|
|
|
|
|
def assert_error_emitted(mock_manager, expected_message_substring: str):
|
|
"""
|
|
Assert that an error was emitted to the user.
|
|
|
|
Args:
|
|
mock_manager: The mock ConnectionManager
|
|
expected_message_substring: Text that should appear in error message
|
|
"""
|
|
mock_manager.emit_to_user.assert_called()
|
|
call_args = mock_manager.emit_to_user.call_args
|
|
event_name = call_args[0][1]
|
|
data = call_args[0][2]
|
|
|
|
assert event_name in ("error", "substitution_error", "outcome_rejected"), \
|
|
f"Expected error event, got {event_name}"
|
|
assert expected_message_substring in data.get("message", ""), \
|
|
f"Expected '{expected_message_substring}' in message, got: {data}"
|
|
|
|
|
|
def assert_broadcast_sent(mock_manager, event_name: str, game_id: str = None):
|
|
"""
|
|
Assert that a broadcast was sent to the game room.
|
|
|
|
Args:
|
|
mock_manager: The mock ConnectionManager
|
|
event_name: Expected event name (e.g., 'dice_rolled')
|
|
game_id: Optional game ID to verify destination
|
|
"""
|
|
mock_manager.broadcast_to_game.assert_called()
|
|
call_args = mock_manager.broadcast_to_game.call_args
|
|
actual_event = call_args[0][1]
|
|
|
|
assert actual_event == event_name, \
|
|
f"Expected broadcast event '{event_name}', got '{actual_event}'"
|
|
|
|
if game_id:
|
|
actual_game_id = call_args[0][0]
|
|
assert actual_game_id == game_id, \
|
|
f"Expected game_id '{game_id}', got '{actual_game_id}'"
|