f6e8ab5f67
Create 16 integration tests across 7 test classes covering: - JWT authentication (valid/invalid/expired tokens) - Game join flow with Redis connection tracking - Action execution and state broadcasting - Turn timeout Redis operations (start/get/cancel/extend) - Disconnection cleanup - Spectator filtered state - Reconnection tracking Uses testcontainers for real Redis/Postgres integration. Completes Phase 4 (Game Service + WebSocket) with all 18 tasks finished. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
334 lines
10 KiB
Python
334 lines
10 KiB
Python
"""Fixtures for Socket.IO integration tests.
|
|
|
|
This module provides fixtures for end-to-end testing of the WebSocket game flow
|
|
using real testcontainer infrastructure (Redis + Postgres) and the python-socketio
|
|
AsyncClient for simulating real connections.
|
|
|
|
Key Features:
|
|
- Real Socket.IO client connections to test ASGI app
|
|
- Testcontainer-backed Redis/Postgres for state persistence
|
|
- JWT token generation for authentication testing
|
|
- Game creation helpers for common test scenarios
|
|
|
|
Usage:
|
|
@pytest.mark.asyncio
|
|
async def test_game_join(sio_client, game_with_players):
|
|
game_id, p1_token, p2_token = game_with_players
|
|
|
|
async with sio_client(p1_token) as client:
|
|
result = await client.emit("game:join", {"game_id": game_id})
|
|
assert result["success"] is True
|
|
"""
|
|
|
|
from collections.abc import AsyncGenerator, Callable
|
|
from contextlib import asynccontextmanager
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
import socketio
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import RulesConfig
|
|
from app.core.enums import CardType, EnergyType, PokemonStage
|
|
from app.core.models.card import Attack, CardDefinition
|
|
from app.db.models.user import User
|
|
from app.services.jwt_service import create_access_token as jwt_create_access_token
|
|
|
|
|
|
@pytest.fixture
|
|
def create_access_token() -> Callable[[str], str]:
|
|
"""Factory fixture to create valid JWT access tokens.
|
|
|
|
Returns a function that takes a user_id string and returns
|
|
a valid JWT access token for testing authentication.
|
|
"""
|
|
|
|
def _create_token(user_id: str) -> str:
|
|
return jwt_create_access_token(UUID(user_id))
|
|
|
|
return _create_token
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_user(db_session: AsyncSession) -> User:
|
|
"""Create a test user in the database.
|
|
|
|
Returns a User object that can be used for authentication
|
|
and game creation tests.
|
|
"""
|
|
user = User(
|
|
id=uuid4(),
|
|
email="test@example.com",
|
|
display_name="Test Player",
|
|
oauth_provider="google",
|
|
oauth_id=f"google-{uuid4()}",
|
|
created_at=datetime.now(UTC),
|
|
)
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
return user
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_user_pair(db_session: AsyncSession) -> tuple[User, User]:
|
|
"""Create a pair of test users for two-player game tests.
|
|
|
|
Returns (player1, player2) User objects.
|
|
"""
|
|
player1 = User(
|
|
id=uuid4(),
|
|
email="player1@example.com",
|
|
display_name="Player One",
|
|
oauth_provider="google",
|
|
oauth_id=f"google-{uuid4()}",
|
|
created_at=datetime.now(UTC),
|
|
)
|
|
player2 = User(
|
|
id=uuid4(),
|
|
email="player2@example.com",
|
|
display_name="Player Two",
|
|
oauth_provider="google",
|
|
oauth_id=f"google-{uuid4()}",
|
|
created_at=datetime.now(UTC),
|
|
)
|
|
db_session.add_all([player1, player2])
|
|
await db_session.commit()
|
|
await db_session.refresh(player1)
|
|
await db_session.refresh(player2)
|
|
return player1, player2
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_card_definitions() -> dict[str, CardDefinition]:
|
|
"""Create sample card definitions for testing.
|
|
|
|
Provides a minimal set of cards needed to create valid decks:
|
|
- Basic Pokemon (Pikachu)
|
|
- Energy card (Lightning)
|
|
"""
|
|
pikachu = CardDefinition(
|
|
id="pikachu-base-001",
|
|
name="Pikachu",
|
|
card_type=CardType.POKEMON,
|
|
stage=PokemonStage.BASIC,
|
|
hp=60,
|
|
pokemon_type=EnergyType.LIGHTNING,
|
|
attacks=[
|
|
Attack(name="Thunder Shock", damage=20, cost=[EnergyType.LIGHTNING]),
|
|
Attack(
|
|
name="Thunderbolt", damage=50, cost=[EnergyType.LIGHTNING, EnergyType.COLORLESS]
|
|
),
|
|
],
|
|
retreat_cost=1,
|
|
)
|
|
|
|
lightning_energy = CardDefinition(
|
|
id="energy-basic-lightning",
|
|
name="Lightning Energy",
|
|
card_type=CardType.ENERGY,
|
|
energy_type=EnergyType.LIGHTNING,
|
|
)
|
|
|
|
return {
|
|
"pikachu-base-001": pikachu,
|
|
"energy-basic-lightning": lightning_energy,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def rules_config() -> RulesConfig:
|
|
"""Create a default RulesConfig for testing.
|
|
|
|
Uses standard rules with turn timer disabled for simpler testing.
|
|
"""
|
|
config = RulesConfig()
|
|
config.win_conditions.turn_timer_enabled = False
|
|
return config
|
|
|
|
|
|
@pytest.fixture
|
|
def sio_client_factory() -> Callable[[str | None], socketio.AsyncClient]:
|
|
"""Factory to create Socket.IO async clients for testing.
|
|
|
|
Returns a factory function that creates pre-configured AsyncClient
|
|
instances. Use with async context manager for proper cleanup.
|
|
|
|
Usage:
|
|
client = sio_client_factory(token)
|
|
await client.connect("http://testserver", ...)
|
|
# ... test ...
|
|
await client.disconnect()
|
|
"""
|
|
|
|
def _create_client(token: str | None = None) -> socketio.AsyncClient:
|
|
client = socketio.AsyncClient(
|
|
logger=False,
|
|
engineio_logger=False,
|
|
)
|
|
# Store token for connection
|
|
client._test_token = token # type: ignore[attr-defined]
|
|
return client
|
|
|
|
return _create_client
|
|
|
|
|
|
class SocketIOTestClient:
|
|
"""Helper class for Socket.IO integration testing.
|
|
|
|
Wraps the AsyncClient with convenience methods for testing
|
|
game events and handling authentication.
|
|
|
|
Attributes:
|
|
client: The underlying socketio.AsyncClient.
|
|
token: JWT token for authentication.
|
|
received_events: List of (event_name, data) tuples received.
|
|
"""
|
|
|
|
def __init__(self, token: str | None = None) -> None:
|
|
"""Initialize the test client.
|
|
|
|
Args:
|
|
token: JWT access token for authentication. If None, connects
|
|
without authentication (for testing auth rejection).
|
|
"""
|
|
self.client = socketio.AsyncClient(logger=False, engineio_logger=False)
|
|
self.token = token
|
|
self.received_events: list[tuple[str, Any]] = []
|
|
self._setup_handlers()
|
|
|
|
def _setup_handlers(self) -> None:
|
|
"""Set up event handlers to capture received events."""
|
|
|
|
@self.client.on("*", namespace="/game")
|
|
async def catch_all(event: str, data: Any) -> None:
|
|
self.received_events.append((event, data))
|
|
|
|
@self.client.on("game:state", namespace="/game")
|
|
async def on_state(data: Any) -> None:
|
|
self.received_events.append(("game:state", data))
|
|
|
|
@self.client.on("game:reconnected", namespace="/game")
|
|
async def on_reconnected(data: Any) -> None:
|
|
self.received_events.append(("game:reconnected", data))
|
|
|
|
@self.client.on("game:opponent_status", namespace="/game")
|
|
async def on_opponent_status(data: Any) -> None:
|
|
self.received_events.append(("game:opponent_status", data))
|
|
|
|
@self.client.on("game:game_over", namespace="/game")
|
|
async def on_game_over(data: Any) -> None:
|
|
self.received_events.append(("game:game_over", data))
|
|
|
|
@self.client.on("auth_error", namespace="/game")
|
|
async def on_auth_error(data: Any) -> None:
|
|
self.received_events.append(("auth_error", data))
|
|
|
|
async def connect(self, url: str = "http://testserver") -> bool:
|
|
"""Connect to the Socket.IO server.
|
|
|
|
Args:
|
|
url: Server URL to connect to.
|
|
|
|
Returns:
|
|
True if connection succeeded, False otherwise.
|
|
"""
|
|
auth = {"token": self.token} if self.token else None
|
|
try:
|
|
await self.client.connect(
|
|
url,
|
|
namespaces=["/game"],
|
|
auth=auth,
|
|
wait=True,
|
|
wait_timeout=5.0,
|
|
)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
async def disconnect(self) -> None:
|
|
"""Disconnect from the server."""
|
|
if self.client.connected:
|
|
await self.client.disconnect()
|
|
|
|
async def emit_and_wait(
|
|
self,
|
|
event: str,
|
|
data: dict[str, Any],
|
|
timeout: float = 5.0,
|
|
) -> dict[str, Any]:
|
|
"""Emit an event and wait for the callback response.
|
|
|
|
Args:
|
|
event: Event name to emit.
|
|
data: Event data to send.
|
|
timeout: Seconds to wait for response.
|
|
|
|
Returns:
|
|
The callback response data.
|
|
"""
|
|
return await self.client.call(
|
|
event,
|
|
data,
|
|
namespace="/game",
|
|
timeout=timeout,
|
|
)
|
|
|
|
def get_events(self, event_name: str) -> list[Any]:
|
|
"""Get all received events of a specific type.
|
|
|
|
Args:
|
|
event_name: Name of event to filter for.
|
|
|
|
Returns:
|
|
List of event data for matching events.
|
|
"""
|
|
return [data for name, data in self.received_events if name == event_name]
|
|
|
|
def clear_events(self) -> None:
|
|
"""Clear the received events list."""
|
|
self.received_events.clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def create_sio_test_client() -> Callable[[str | None], SocketIOTestClient]:
|
|
"""Factory fixture to create SocketIOTestClient instances.
|
|
|
|
Usage:
|
|
client = create_sio_test_client(token)
|
|
await client.connect()
|
|
result = await client.emit_and_wait("game:join", {"game_id": "..."})
|
|
await client.disconnect()
|
|
"""
|
|
|
|
def _create(token: str | None = None) -> SocketIOTestClient:
|
|
return SocketIOTestClient(token)
|
|
|
|
return _create
|
|
|
|
|
|
@asynccontextmanager
|
|
async def connected_client(
|
|
token: str | None,
|
|
url: str = "http://testserver",
|
|
) -> AsyncGenerator[SocketIOTestClient, None]:
|
|
"""Context manager for a connected Socket.IO test client.
|
|
|
|
Handles connection and disconnection automatically.
|
|
|
|
Usage:
|
|
async with connected_client(token) as client:
|
|
result = await client.emit_and_wait("game:join", {...})
|
|
"""
|
|
client = SocketIOTestClient(token)
|
|
try:
|
|
connected = await client.connect(url)
|
|
if not connected:
|
|
raise ConnectionError(f"Failed to connect to {url}")
|
|
yield client
|
|
finally:
|
|
await client.disconnect()
|