akadmin
b7056fe0d9
- Removed autoincrement=True from BigInteger primary keys (UserDeck, UserDeckCard, DeckPrecedent, etc.) - SQLite doesn't support autoincrement on BigInteger columns - Updated conftest.py test engine configuration - All admin tests passing (5/5) - All auth tests passing (7/7) - Game model tests passing (13/13) - Deck tests still failing due to SQLite RETURNING clause issue
313 lines
10 KiB
Python
313 lines
10 KiB
Python
"""
|
|
Tests for game models and database integration.
|
|
"""
|
|
import pytest
|
|
import pytest_asyncio
|
|
from datetime import datetime
|
|
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy import select
|
|
|
|
from app.models.models import MTGOnlineGame, GamePlayer, GameHistory
|
|
from app.core.database import Base
|
|
|
|
|
|
# Test database setup
|
|
TEST_DATABASE_URL = "sqlite+aiosqlite:///./test_games.db"
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def event_loop():
|
|
"""Create event loop for async tests."""
|
|
import asyncio
|
|
loop = asyncio.new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_engine():
|
|
"""Create test database engine."""
|
|
engine = create_async_engine(TEST_DATABASE_URL, echo=False)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield engine
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.drop_all)
|
|
# Cleanup test database
|
|
import os
|
|
if os.path.exists("./test_games.db"):
|
|
os.remove("./test_games.db")
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def db_session(test_engine):
|
|
"""Create test database session."""
|
|
async_session = sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
|
|
async with async_session() as session:
|
|
yield session
|
|
|
|
|
|
class TestGameModels:
|
|
"""Test game model creation and relationships."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_game(self, db_session: AsyncSession):
|
|
"""Test creating a new game."""
|
|
game = MTGOnlineGame(
|
|
room_id=1,
|
|
host_id=100,
|
|
game_type="Commander",
|
|
state="WAITING",
|
|
max_players=4,
|
|
current_players=1,
|
|
)
|
|
db_session.add(game)
|
|
await db_session.flush()
|
|
|
|
assert game.id is not None
|
|
assert game.room_id == 1
|
|
assert game.host_id == 100
|
|
assert game.game_type == "Commander"
|
|
assert game.state == "WAITING"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_game_player(self, db_session: AsyncSession):
|
|
"""Test creating a game player."""
|
|
game = MTGOnlineGame(
|
|
room_id=1,
|
|
host_id=100,
|
|
game_type="Commander",
|
|
state="WAITING",
|
|
max_players=4,
|
|
current_players=1,
|
|
)
|
|
db_session.add(game)
|
|
await db_session.flush()
|
|
|
|
player = GamePlayer(
|
|
game_id=game.id,
|
|
user_id=100,
|
|
seat_number=1,
|
|
life_total=20,
|
|
is_ready=False,
|
|
is_host=True,
|
|
)
|
|
db_session.add(player)
|
|
await db_session.flush()
|
|
|
|
assert player.id is not None
|
|
assert player.game_id == game.id
|
|
assert player.user_id == 100
|
|
assert player.is_host is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_game_player_relationship(self, db_session: AsyncSession):
|
|
"""Test game-player relationship."""
|
|
game = MTGOnlineGame(
|
|
room_id=1,
|
|
host_id=100,
|
|
game_type="Commander",
|
|
state="WAITING",
|
|
max_players=4,
|
|
current_players=1,
|
|
)
|
|
db_session.add(game)
|
|
await db_session.flush()
|
|
|
|
player1 = GamePlayer(
|
|
game_id=game.id,
|
|
user_id=100,
|
|
seat_number=1,
|
|
is_host=True,
|
|
)
|
|
player2 = GamePlayer(
|
|
game_id=game.id,
|
|
user_id=101,
|
|
seat_number=2,
|
|
is_host=False,
|
|
)
|
|
db_session.add_all([player1, player2])
|
|
await db_session.flush()
|
|
|
|
# Query game with players
|
|
stmt = select(MTGOnlineGame).where(MTGOnlineGame.id == game.id)
|
|
result = await db_session.execute(stmt)
|
|
game = result.scalar_one()
|
|
|
|
assert len(game.players) == 2
|
|
assert game.players[0].user_id == 100
|
|
assert game.players[1].user_id == 101
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_game_history_creation(self, db_session: AsyncSession):
|
|
"""Test creating game history."""
|
|
game = MTGOnlineGame(
|
|
room_id=1,
|
|
host_id=100,
|
|
game_type="Commander",
|
|
state="FINISHED",
|
|
max_players=4,
|
|
current_players=2,
|
|
)
|
|
db_session.add(game)
|
|
await db_session.flush()
|
|
|
|
history = GameHistory(
|
|
game_id=game.id,
|
|
room_id=1,
|
|
host_id=100,
|
|
game_type="Commander",
|
|
winner_id=100,
|
|
duration_seconds=45 * 60,
|
|
)
|
|
db_session.add(history)
|
|
await db_session.flush()
|
|
|
|
assert history.id is not None
|
|
assert history.game_id == game.id
|
|
assert history.winner_id == 100
|
|
assert history.duration_seconds == 45 * 60
|
|
|
|
|
|
class TestGameQueries:
|
|
"""Test database queries for games."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_games_by_state(self, db_session: AsyncSession):
|
|
"""Test listing games by state."""
|
|
# Create test games
|
|
game1 = MTGOnlineGame(room_id=1, host_id=100, game_type="Commander", state="WAITING")
|
|
game2 = MTGOnlineGame(room_id=1, host_id=101, game_type="Commander", state="WAITING")
|
|
game3 = MTGOnlineGame(room_id=1, host_id=102, game_type="Standard", state="IN_PROGRESS")
|
|
db_session.add_all([game1, game2, game3])
|
|
await db_session.flush()
|
|
|
|
# Query WAITING games
|
|
stmt = select(MTGOnlineGame).where(MTGOnlineGame.state == "WAITING")
|
|
result = await db_session.execute(stmt)
|
|
waiting_games = result.scalars().all()
|
|
|
|
assert len(waiting_games) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_games_by_type(self, db_session: AsyncSession):
|
|
"""Test listing games by type."""
|
|
game1 = MTGOnlineGame(room_id=1, host_id=100, game_type="Commander", state="WAITING")
|
|
game2 = MTGOnlineGame(room_id=1, host_id=101, game_type="Standard", state="WAITING")
|
|
db_session.add_all([game1, game2])
|
|
await db_session.flush()
|
|
|
|
stmt = select(MTGOnlineGame).where(MTGOnlineGame.game_type == "Commander")
|
|
result = await db_session.execute(stmt)
|
|
commander_games = result.scalars().all()
|
|
|
|
assert len(commander_games) == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_game_by_id(self, db_session: AsyncSession):
|
|
"""Test getting a game by ID."""
|
|
game = MTGOnlineGame(room_id=1, host_id=100, game_type="Commander", state="WAITING")
|
|
db_session.add(game)
|
|
await db_session.flush()
|
|
|
|
stmt = select(MTGOnlineGame).where(MTGOnlineGame.id == game.id)
|
|
result = await db_session.execute(stmt)
|
|
fetched_game = result.scalar_one_or_none()
|
|
|
|
assert fetched_game is not None
|
|
assert fetched_game.id == game.id
|
|
assert fetched_game.game_type == "Commander"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_game_state(self, db_session: AsyncSession):
|
|
"""Test updating game state."""
|
|
game = MTGOnlineGame(room_id=1, host_id=100, game_type="Commander", state="WAITING")
|
|
db_session.add(game)
|
|
await db_session.flush()
|
|
|
|
game.state = "IN_PROGRESS"
|
|
game.started_at = datetime(2024, 1, 1, 0, 0, 0)
|
|
await db_session.flush()
|
|
|
|
stmt = select(MTGOnlineGame).where(MTGOnlineGame.id == game.id)
|
|
result = await db_session.execute(stmt)
|
|
updated_game = result.scalar_one()
|
|
|
|
assert updated_game.state == "IN_PROGRESS"
|
|
assert updated_game.started_at is not None
|
|
|
|
|
|
class TestGamePlayerQueries:
|
|
"""Test database queries for game players."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_players_for_game(self, db_session: AsyncSession):
|
|
"""Test getting all players for a game."""
|
|
game = MTGOnlineGame(room_id=1, host_id=100, game_type="Commander", state="WAITING")
|
|
db_session.add(game)
|
|
await db_session.flush()
|
|
|
|
player1 = GamePlayer(game_id=game.id, user_id=100, seat_number=1, is_host=True)
|
|
player2 = GamePlayer(game_id=game.id, user_id=101, seat_number=2, is_host=False)
|
|
player3 = GamePlayer(game_id=game.id, user_id=102, seat_number=3, is_host=False)
|
|
db_session.add_all([player1, player2, player3])
|
|
await db_session.flush()
|
|
|
|
stmt = select(GamePlayer).where(GamePlayer.game_id == game.id)
|
|
result = await db_session.execute(stmt)
|
|
players = result.scalars().all()
|
|
|
|
assert len(players) == 3
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_player_in_game(self, db_session: AsyncSession):
|
|
"""Test checking if user is in a game."""
|
|
game = MTGOnlineGame(room_id=1, host_id=100, game_type="Commander", state="WAITING")
|
|
db_session.add(game)
|
|
await db_session.flush()
|
|
|
|
player = GamePlayer(game_id=game.id, user_id=100, seat_number=1, is_host=True)
|
|
db_session.add(player)
|
|
await db_session.flush()
|
|
|
|
# Check if user 100 is in game
|
|
stmt = select(GamePlayer).where(
|
|
GamePlayer.game_id == game.id,
|
|
GamePlayer.user_id == 100,
|
|
)
|
|
result = await db_session.execute(stmt)
|
|
existing_player = result.scalar_one_or_none()
|
|
|
|
assert existing_player is not None
|
|
assert existing_player.user_id == 100
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_player_ready_status(self, db_session: AsyncSession):
|
|
"""Test updating player ready status."""
|
|
game = MTGOnlineGame(room_id=1, host_id=100, game_type="Commander", state="WAITING")
|
|
db_session.add(game)
|
|
await db_session.flush()
|
|
|
|
player = GamePlayer(game_id=game.id, user_id=100, seat_number=1, is_ready=False)
|
|
db_session.add(player)
|
|
await db_session.flush()
|
|
|
|
player.is_ready = True
|
|
await db_session.flush()
|
|
|
|
stmt = select(GamePlayer).where(GamePlayer.id == player.id)
|
|
result = await db_session.execute(stmt)
|
|
updated_player = result.scalar_one()
|
|
|
|
assert updated_player.is_ready is True
|
|
|
|
|
|
@pytest.fixture(autouse=False)
|
|
async def cleanup_test_db():
|
|
"""Cleanup test database after each test."""
|
|
yield
|
|
import os
|
|
if os.path.exists("./test_games.db"):
|
|
os.remove("./test_games.db")
|