"""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()