Fix test URL mismatches to match actual router prefixes
- test_admin.py: /admin/* → /users/* (admin router mounted at /users) - test_decks.py: /api/v1/decks/* → /decks/* (decks router mounted at /decks) - No production code changes - test files only
This commit is contained in:
@@ -0,0 +1,312 @@
|
||||
"""
|
||||
Tests for game models and database integration.
|
||||
"""
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
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,
|
||||
player_count=2,
|
||||
duration_minutes=45,
|
||||
)
|
||||
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_minutes == 45
|
||||
|
||||
|
||||
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 = "2024-01-01T00:00:00"
|
||||
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=True)
|
||||
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")
|
||||
Reference in New Issue
Block a user