29ab0b3d84
- Implement GameStateManager with Redis-primary, Postgres-backup pattern - Cache operations: save_to_cache, load_from_cache, delete_from_cache - DB operations: persist_to_db, load_from_db, delete_from_db - High-level: load_state (cache-first), delete_game, recover_active_games - Query helpers: get_active_game_count, get_player_active_games - Add 22 tests for GameStateManager (87% coverage) - Add 6 __repr__ tests for all DB models (100% model coverage)
951 lines
32 KiB
Python
951 lines
32 KiB
Python
"""Tests for SQLAlchemy database models.
|
|
|
|
This module contains comprehensive tests for all database models:
|
|
- User: OAuth accounts, premium subscriptions
|
|
- Collection: Card ownership tracking
|
|
- Deck: Deck configurations with JSONB
|
|
- CampaignProgress: Single-player campaign state
|
|
- ActiveGame: In-progress games
|
|
- GameHistory: Completed game records
|
|
|
|
Each test is isolated via transaction rollback - no cleanup needed.
|
|
"""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.models import (
|
|
CardSource,
|
|
EndReason,
|
|
GameType,
|
|
User,
|
|
)
|
|
from tests.factories import (
|
|
ActiveGameFactory,
|
|
CampaignProgressFactory,
|
|
CollectionFactory,
|
|
DeckFactory,
|
|
GameHistoryFactory,
|
|
UserFactory,
|
|
)
|
|
|
|
# =============================================================================
|
|
# User Model Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestUserModel:
|
|
"""Tests for the User model."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_user(self, db_session: AsyncSession):
|
|
"""Test creating a basic user with required fields.
|
|
|
|
Verifies that a user can be created with OAuth credentials
|
|
and that the ID is generated automatically.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
|
|
assert user.id is not None
|
|
assert user.email is not None
|
|
assert user.display_name is not None
|
|
assert user.oauth_provider == "google"
|
|
assert user.oauth_id is not None
|
|
assert user.is_premium is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_has_timestamps(self, db_session: AsyncSession):
|
|
"""Test that users get created_at and updated_at timestamps.
|
|
|
|
These are inherited from Base and should be auto-populated.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
|
|
assert user.created_at is not None
|
|
assert user.updated_at is not None
|
|
assert isinstance(user.created_at, datetime)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_email_unique(self, db_session: AsyncSession):
|
|
"""Test that duplicate emails are rejected.
|
|
|
|
Email is a unique constraint on the User model.
|
|
"""
|
|
email = "duplicate@example.com"
|
|
await UserFactory.create(db_session, email=email)
|
|
|
|
# Attempt to create another user with same email
|
|
with pytest.raises(IntegrityError):
|
|
await UserFactory.create(db_session, email=email)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_oauth_unique(self, db_session: AsyncSession):
|
|
"""Test that duplicate OAuth provider+id combinations are rejected.
|
|
|
|
The (oauth_provider, oauth_id) pair must be unique.
|
|
"""
|
|
provider = "discord"
|
|
oauth_id = "discord_123456"
|
|
|
|
await UserFactory.create(db_session, oauth_provider=provider, oauth_id=oauth_id)
|
|
|
|
# Attempt to create another user with same OAuth credentials
|
|
with pytest.raises(IntegrityError):
|
|
await UserFactory.create(db_session, oauth_provider=provider, oauth_id=oauth_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_premium_user(self, db_session: AsyncSession):
|
|
"""Test creating a premium user with subscription expiration.
|
|
|
|
Verifies that premium fields are stored correctly.
|
|
"""
|
|
premium_until = datetime.now(UTC) + timedelta(days=30)
|
|
user = await UserFactory.create(
|
|
db_session,
|
|
is_premium=True,
|
|
premium_until=premium_until,
|
|
)
|
|
|
|
assert user.is_premium is True
|
|
assert user.premium_until is not None
|
|
assert user.premium_until > datetime.now(UTC)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_has_active_premium_property(self, db_session: AsyncSession):
|
|
"""Test the has_active_premium property logic.
|
|
|
|
Premium is active only if is_premium=True AND premium_until is in future.
|
|
"""
|
|
# Active premium
|
|
active = await UserFactory.create_premium(db_session, days_remaining=30)
|
|
assert active.has_active_premium is True
|
|
|
|
# Expired premium
|
|
expired = await UserFactory.create(
|
|
db_session,
|
|
is_premium=True,
|
|
premium_until=datetime.now(UTC) - timedelta(days=1),
|
|
)
|
|
assert expired.has_active_premium is False
|
|
|
|
# Non-premium
|
|
free = await UserFactory.create(db_session, is_premium=False)
|
|
assert free.has_active_premium is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_max_decks_property(self, db_session: AsyncSession):
|
|
"""Test that max_decks returns correct limits based on premium status.
|
|
|
|
Free users: 5 decks
|
|
Premium users: 999 decks (effectively unlimited)
|
|
"""
|
|
free_user = await UserFactory.create(db_session)
|
|
assert free_user.max_decks == 5
|
|
|
|
premium_user = await UserFactory.create_premium(db_session)
|
|
assert premium_user.max_decks == 999
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_user(self, db_session: AsyncSession):
|
|
"""Test updating user fields.
|
|
|
|
Verifies that changes persist after flush.
|
|
"""
|
|
user = await UserFactory.create(db_session, display_name="Original")
|
|
|
|
user.display_name = "Updated Name"
|
|
user.last_login = datetime.now(UTC)
|
|
await db_session.flush()
|
|
await db_session.refresh(user)
|
|
|
|
assert user.display_name == "Updated Name"
|
|
assert user.last_login is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_user_by_id(self, db_session: AsyncSession):
|
|
"""Test reading a user by their ID.
|
|
|
|
Verifies that we can query users back from the database.
|
|
"""
|
|
created = await UserFactory.create(db_session)
|
|
|
|
# Read back
|
|
result = await db_session.execute(select(User).where(User.id == created.id))
|
|
fetched = result.scalar_one()
|
|
|
|
assert fetched.id == created.id
|
|
assert fetched.email == created.email
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_user_by_oauth(self, db_session: AsyncSession):
|
|
"""Test finding a user by OAuth credentials.
|
|
|
|
This is the primary lookup method during authentication.
|
|
"""
|
|
user = await UserFactory.create(
|
|
db_session, oauth_provider="discord", oauth_id="discord_unique_123"
|
|
)
|
|
|
|
result = await db_session.execute(
|
|
select(User).where(
|
|
User.oauth_provider == "discord",
|
|
User.oauth_id == "discord_unique_123",
|
|
)
|
|
)
|
|
fetched = result.scalar_one()
|
|
|
|
assert fetched.id == user.id
|
|
|
|
|
|
# =============================================================================
|
|
# Collection Model Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestCollectionModel:
|
|
"""Tests for the Collection model."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_collection_entry(self, db_session: AsyncSession):
|
|
"""Test creating a collection entry for a user.
|
|
|
|
Verifies basic card ownership tracking.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
entry = await CollectionFactory.create(
|
|
db_session,
|
|
user_id=user.id,
|
|
card_definition_id="pikachu_base_001",
|
|
quantity=3,
|
|
)
|
|
|
|
assert entry.id is not None
|
|
assert entry.user_id == user.id
|
|
assert entry.card_definition_id == "pikachu_base_001"
|
|
assert entry.quantity == 3
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collection_card_source_enum(self, db_session: AsyncSession):
|
|
"""Test that all CardSource enum values work correctly.
|
|
|
|
Each source type should be valid and stored properly.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
|
|
for source in CardSource:
|
|
entry = await CollectionFactory.create(
|
|
db_session,
|
|
user_id=user.id,
|
|
source=source,
|
|
)
|
|
assert entry.source == source
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collection_unique_constraint(self, db_session: AsyncSession):
|
|
"""Test that (user_id, card_definition_id) is unique.
|
|
|
|
Users can't have duplicate entries for the same card.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
card_id = "duplicate_card"
|
|
|
|
await CollectionFactory.create(db_session, user_id=user.id, card_definition_id=card_id)
|
|
|
|
with pytest.raises(IntegrityError):
|
|
await CollectionFactory.create(db_session, user_id=user.id, card_definition_id=card_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collection_same_card_different_users(self, db_session: AsyncSession):
|
|
"""Test that different users can own the same card.
|
|
|
|
The unique constraint is per-user, not global.
|
|
"""
|
|
user1 = await UserFactory.create(db_session)
|
|
user2 = await UserFactory.create(db_session)
|
|
card_id = "shared_card"
|
|
|
|
entry1 = await CollectionFactory.create(
|
|
db_session, user_id=user1.id, card_definition_id=card_id
|
|
)
|
|
entry2 = await CollectionFactory.create(
|
|
db_session, user_id=user2.id, card_definition_id=card_id
|
|
)
|
|
|
|
assert entry1.id != entry2.id
|
|
assert entry1.user_id != entry2.user_id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collection_timestamps(self, db_session: AsyncSession):
|
|
"""Test that obtained_at tracks when card was added.
|
|
|
|
obtained_at is separate from created_at for historical tracking.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
before = datetime.now(UTC)
|
|
|
|
entry = await CollectionFactory.create(db_session, user_id=user.id)
|
|
|
|
assert entry.obtained_at is not None
|
|
assert entry.obtained_at >= before
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_collection_quantity(self, db_session: AsyncSession):
|
|
"""Test updating card quantity in collection.
|
|
|
|
Quantity should be updateable without constraint issues.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
entry = await CollectionFactory.create(db_session, user_id=user.id, quantity=1)
|
|
|
|
entry.quantity = 4
|
|
await db_session.flush()
|
|
await db_session.refresh(entry)
|
|
|
|
assert entry.quantity == 4
|
|
|
|
|
|
# =============================================================================
|
|
# Deck Model Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestDeckModel:
|
|
"""Tests for the Deck model."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_empty_deck(self, db_session: AsyncSession):
|
|
"""Test creating an empty deck.
|
|
|
|
Decks start empty and are populated later.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
deck = await DeckFactory.create(db_session, user_id=user.id)
|
|
|
|
assert deck.id is not None
|
|
assert deck.user_id == user.id
|
|
assert deck.cards == {}
|
|
assert deck.energy_cards == {}
|
|
assert deck.is_valid is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deck_jsonb_cards(self, db_session: AsyncSession):
|
|
"""Test storing card data in JSONB format.
|
|
|
|
Cards are stored as {card_id: quantity} mapping.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
cards = {
|
|
"pikachu_base_001": 4,
|
|
"raichu_base_001": 2,
|
|
"lightning_energy_001": 10,
|
|
}
|
|
|
|
deck = await DeckFactory.create(db_session, user_id=user.id, cards=cards)
|
|
|
|
assert deck.cards == cards
|
|
assert deck.cards["pikachu_base_001"] == 4
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deck_jsonb_energy(self, db_session: AsyncSession):
|
|
"""Test storing energy card data separately.
|
|
|
|
Energy is tracked by type for easier UI display.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
energy = {"lightning": 8, "fire": 4, "colorless": 4}
|
|
|
|
deck = await DeckFactory.create(db_session, user_id=user.id, energy_cards=energy)
|
|
|
|
assert deck.energy_cards == energy
|
|
assert deck.energy_cards["lightning"] == 8
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deck_total_cards_property(self, db_session: AsyncSession):
|
|
"""Test the total_cards computed property.
|
|
|
|
Should sum all card quantities (excluding energy).
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
deck = await DeckFactory.create(
|
|
db_session,
|
|
user_id=user.id,
|
|
cards={"card_a": 4, "card_b": 3, "card_c": 2},
|
|
energy_cards={"fire": 10},
|
|
)
|
|
|
|
assert deck.total_cards == 9 # 4 + 3 + 2
|
|
assert deck.total_energy == 10
|
|
assert deck.deck_size == 19 # 9 + 10
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deck_validation_errors_jsonb(self, db_session: AsyncSession):
|
|
"""Test storing validation errors in JSONB.
|
|
|
|
validation_errors can be null or a list of error messages.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
errors = ["Deck too small (30/40)", "Missing basic Pokemon"]
|
|
|
|
deck = await DeckFactory.create(
|
|
db_session,
|
|
user_id=user.id,
|
|
is_valid=False,
|
|
validation_errors=errors,
|
|
)
|
|
|
|
assert deck.validation_errors == errors
|
|
assert len(deck.validation_errors) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_starter_deck(self, db_session: AsyncSession):
|
|
"""Test creating a starter deck.
|
|
|
|
Starter decks have special flags for type identification.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
deck = await DeckFactory.create_starter_deck(db_session, user, starter_type="water")
|
|
|
|
assert deck.is_starter is True
|
|
assert deck.starter_type == "water"
|
|
assert "Water" in deck.name
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_multiple_decks(self, db_session: AsyncSession):
|
|
"""Test that users can have multiple decks.
|
|
|
|
No unique constraint on (user_id, name) - names can be duplicated.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
|
|
deck1 = await DeckFactory.create(db_session, user_id=user.id, name="Deck A")
|
|
deck2 = await DeckFactory.create(db_session, user_id=user.id, name="Deck B")
|
|
deck3 = await DeckFactory.create(db_session, user_id=user.id, name="Deck A")
|
|
|
|
assert deck1.id != deck2.id != deck3.id
|
|
# Same name is allowed
|
|
assert deck1.name == deck3.name
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_deck_cards(self, db_session: AsyncSession):
|
|
"""Test updating deck card composition.
|
|
|
|
JSONB fields should be fully replaceable.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
deck = await DeckFactory.create(db_session, user_id=user.id, cards={"old_card": 4})
|
|
|
|
new_cards = {"new_card_a": 4, "new_card_b": 4}
|
|
deck.cards = new_cards
|
|
deck.is_valid = True
|
|
await db_session.flush()
|
|
await db_session.refresh(deck)
|
|
|
|
assert deck.cards == new_cards
|
|
assert "old_card" not in deck.cards
|
|
|
|
|
|
# =============================================================================
|
|
# CampaignProgress Model Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestCampaignProgressModel:
|
|
"""Tests for the CampaignProgress model."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_campaign_progress(self, db_session: AsyncSession):
|
|
"""Test creating initial campaign progress.
|
|
|
|
New campaigns start at grass_club with zero progress.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
progress = await CampaignProgressFactory.create(db_session, user_id=user.id)
|
|
|
|
assert progress.id is not None
|
|
assert progress.user_id == user.id
|
|
assert progress.current_club == "grass_club"
|
|
assert progress.medals == []
|
|
assert progress.mantibucks == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_campaign_one_per_user(self, db_session: AsyncSession):
|
|
"""Test that each user can only have one campaign progress.
|
|
|
|
user_id has a unique constraint (one-to-one relationship).
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
await CampaignProgressFactory.create(db_session, user_id=user.id)
|
|
|
|
with pytest.raises(IntegrityError):
|
|
await CampaignProgressFactory.create(db_session, user_id=user.id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_campaign_medals_jsonb(self, db_session: AsyncSession):
|
|
"""Test storing medals as JSONB array.
|
|
|
|
Medals are a list of medal IDs earned.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
medals = ["grass_medal", "fire_medal", "water_medal"]
|
|
|
|
progress = await CampaignProgressFactory.create(db_session, user_id=user.id, medals=medals)
|
|
|
|
assert progress.medals == medals
|
|
assert progress.medal_count == 3
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_campaign_defeated_npcs_jsonb(self, db_session: AsyncSession):
|
|
"""Test storing defeated NPCs as JSONB array.
|
|
|
|
Tracks which NPCs have been beaten (for reward eligibility).
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
npcs = ["trainer_1", "trainer_2", "leader_grass"]
|
|
|
|
progress = await CampaignProgressFactory.create(
|
|
db_session, user_id=user.id, defeated_npcs=npcs
|
|
)
|
|
|
|
assert progress.defeated_npcs == npcs
|
|
assert progress.has_defeated("trainer_1") is True
|
|
assert progress.has_defeated("trainer_99") is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_campaign_has_medal_helper(self, db_session: AsyncSession):
|
|
"""Test the has_medal helper method.
|
|
|
|
Convenient way to check medal ownership.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
progress = await CampaignProgressFactory.create(
|
|
db_session, user_id=user.id, medals=["fire_medal"]
|
|
)
|
|
|
|
assert progress.has_medal("fire_medal") is True
|
|
assert progress.has_medal("water_medal") is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_campaign_win_rate_property(self, db_session: AsyncSession):
|
|
"""Test the win_rate computed property.
|
|
|
|
Should calculate wins / total games.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
|
|
# 75% win rate
|
|
progress = await CampaignProgressFactory.create(
|
|
db_session, user_id=user.id, total_wins=15, total_losses=5
|
|
)
|
|
assert progress.win_rate == 0.75
|
|
|
|
# No games played
|
|
new_progress = await CampaignProgressFactory.create(
|
|
db_session,
|
|
user_id=(await UserFactory.create(db_session)).id,
|
|
total_wins=0,
|
|
total_losses=0,
|
|
)
|
|
assert new_progress.win_rate == 0.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_campaign_can_face_grand_masters(self, db_session: AsyncSession):
|
|
"""Test the can_face_grand_masters property.
|
|
|
|
Requires 8 medals to unlock Grand Masters.
|
|
"""
|
|
user1 = await UserFactory.create(db_session)
|
|
user2 = await UserFactory.create(db_session)
|
|
|
|
# Not enough medals
|
|
progress1 = await CampaignProgressFactory.create(
|
|
db_session, user_id=user1.id, medals=["m1", "m2", "m3"]
|
|
)
|
|
assert progress1.can_face_grand_masters is False
|
|
|
|
# Enough medals
|
|
all_medals = [f"medal_{i}" for i in range(8)]
|
|
progress2 = await CampaignProgressFactory.create(
|
|
db_session, user_id=user2.id, medals=all_medals
|
|
)
|
|
assert progress2.can_face_grand_masters is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_campaign_progress(self, db_session: AsyncSession):
|
|
"""Test updating campaign progress after a match.
|
|
|
|
Simulates winning a match and earning rewards.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
progress = await CampaignProgressFactory.create(db_session, user_id=user.id)
|
|
|
|
# Win a match
|
|
progress.total_wins += 1
|
|
progress.booster_packs += 2
|
|
progress.mantibucks += 100
|
|
progress.defeated_npcs = progress.defeated_npcs + ["new_npc"]
|
|
await db_session.flush()
|
|
await db_session.refresh(progress)
|
|
|
|
assert progress.total_wins == 1
|
|
assert progress.booster_packs == 2
|
|
assert progress.mantibucks == 100
|
|
assert "new_npc" in progress.defeated_npcs
|
|
|
|
|
|
# =============================================================================
|
|
# ActiveGame Model Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestActiveGameModel:
|
|
"""Tests for the ActiveGame model."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_campaign_game(self, db_session: AsyncSession):
|
|
"""Test creating an active campaign game.
|
|
|
|
Campaign games have one player vs an NPC.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
game = await ActiveGameFactory.create_campaign_game(db_session, user)
|
|
|
|
assert game.id is not None
|
|
assert game.game_type == GameType.CAMPAIGN
|
|
assert game.player1_id == user.id
|
|
assert game.player2_id is None
|
|
assert game.npc_id is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_pvp_game(self, db_session: AsyncSession):
|
|
"""Test creating an active PvP game.
|
|
|
|
PvP games have two players and no NPC.
|
|
"""
|
|
user1 = await UserFactory.create(db_session)
|
|
user2 = await UserFactory.create(db_session)
|
|
|
|
game = await ActiveGameFactory.create_pvp_game(db_session, user1, user2)
|
|
|
|
assert game.player1_id == user1.id
|
|
assert game.player2_id == user2.id
|
|
assert game.npc_id is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_game_type_enum(self, db_session: AsyncSession):
|
|
"""Test that all GameType enum values work.
|
|
|
|
CAMPAIGN, FREEPLAY, RANKED, PRACTICE should all be valid.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
|
|
for game_type in GameType:
|
|
game = await ActiveGameFactory.create(
|
|
db_session, player1_id=user.id, game_type=game_type
|
|
)
|
|
assert game.game_type == game_type
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_game_state_jsonb(self, db_session: AsyncSession):
|
|
"""Test storing game state in JSONB.
|
|
|
|
The full GameState serializes to JSONB for persistence.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
state = {
|
|
"turn": 5,
|
|
"phase": "attack",
|
|
"players": {
|
|
"player1": {"active": "pikachu_inst_1", "bench": []},
|
|
"player2": {"active": "charizard_inst_1", "bench": []},
|
|
},
|
|
}
|
|
|
|
game = await ActiveGameFactory.create(db_session, player1_id=user.id, game_state=state)
|
|
|
|
assert game.game_state == state
|
|
assert game.game_state["turn"] == 5
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_game_rules_config_jsonb(self, db_session: AsyncSession):
|
|
"""Test storing rules configuration in JSONB.
|
|
|
|
Different game modes may have different rules.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
rules = {
|
|
"prize_count": 6,
|
|
"deck_size": 60,
|
|
"max_hand_size": 7,
|
|
"first_turn_attack": False,
|
|
}
|
|
|
|
game = await ActiveGameFactory.create(db_session, player1_id=user.id, rules_config=rules)
|
|
|
|
assert game.rules_config == rules
|
|
assert game.rules_config["prize_count"] == 6
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_game_timestamps(self, db_session: AsyncSession):
|
|
"""Test that game timing fields work correctly.
|
|
|
|
started_at, last_action_at track game timing.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
game = await ActiveGameFactory.create(db_session, player1_id=user.id)
|
|
|
|
assert game.started_at is not None
|
|
assert game.last_action_at is not None
|
|
|
|
# Update last action
|
|
new_time = datetime.now(UTC)
|
|
game.last_action_at = new_time
|
|
game.turn_number += 1
|
|
await db_session.flush()
|
|
|
|
assert game.last_action_at == new_time
|
|
assert game.turn_number == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_game_turn_deadline(self, db_session: AsyncSession):
|
|
"""Test optional turn deadline for timed games.
|
|
|
|
turn_deadline is set for games with time controls.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
deadline = datetime.now(UTC) + timedelta(minutes=2)
|
|
|
|
game = await ActiveGameFactory.create(
|
|
db_session, player1_id=user.id, turn_deadline=deadline
|
|
)
|
|
|
|
assert game.turn_deadline is not None
|
|
assert game.turn_deadline == deadline
|
|
|
|
|
|
# =============================================================================
|
|
# GameHistory Model Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestGameHistoryModel:
|
|
"""Tests for the GameHistory model."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_player_win(self, db_session: AsyncSession):
|
|
"""Test recording a player victory.
|
|
|
|
Player wins are tracked with winner_id pointing to the player.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
history = await GameHistoryFactory.create_player_win(db_session, user)
|
|
|
|
assert history.id is not None
|
|
assert history.winner_id == user.id
|
|
assert history.winner_is_npc is False
|
|
assert history.end_reason == EndReason.PRIZES_TAKEN
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_player_loss(self, db_session: AsyncSession):
|
|
"""Test recording a player loss to NPC.
|
|
|
|
NPC wins have winner_is_npc=True and winner_id=None.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
history = await GameHistoryFactory.create_player_loss(db_session, user)
|
|
|
|
assert history.winner_id is None
|
|
assert history.winner_is_npc is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_end_reason_enum(self, db_session: AsyncSession):
|
|
"""Test that all EndReason enum values work.
|
|
|
|
All ways a game can end should be valid.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
|
|
for reason in EndReason:
|
|
history = await GameHistoryFactory.create(
|
|
db_session, player1_id=user.id, end_reason=reason
|
|
)
|
|
assert history.end_reason == reason
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_is_draw_property(self, db_session: AsyncSession):
|
|
"""Test the is_draw computed property.
|
|
|
|
Draws have end_reason=DRAW and no winner.
|
|
"""
|
|
user1 = await UserFactory.create(db_session)
|
|
user2 = await UserFactory.create(db_session)
|
|
|
|
draw_game = await GameHistoryFactory.create_pvp_game(db_session, user1, user2, winner=None)
|
|
assert draw_game.is_draw is True
|
|
assert draw_game.winner_id is None
|
|
|
|
won_game = await GameHistoryFactory.create_pvp_game(db_session, user1, user2, winner=user1)
|
|
assert won_game.is_draw is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_game_statistics(self, db_session: AsyncSession):
|
|
"""Test that game statistics are stored correctly.
|
|
|
|
turn_count and duration_seconds track game length.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
history = await GameHistoryFactory.create(
|
|
db_session,
|
|
player1_id=user.id,
|
|
turn_count=25,
|
|
duration_seconds=600,
|
|
)
|
|
|
|
assert history.turn_count == 25
|
|
assert history.duration_seconds == 600
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_replay_data_jsonb(self, db_session: AsyncSession):
|
|
"""Test storing replay data in JSONB.
|
|
|
|
Replay data captures all actions for playback.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
replay = {
|
|
"version": "1.0",
|
|
"actions": [
|
|
{"type": "play_card", "card": "pikachu", "turn": 1},
|
|
{"type": "attack", "attack": "thunder_shock", "turn": 1},
|
|
],
|
|
"seed": 12345,
|
|
}
|
|
|
|
history = await GameHistoryFactory.create(
|
|
db_session, player1_id=user.id, replay_data=replay
|
|
)
|
|
|
|
assert history.replay_data == replay
|
|
assert len(history.replay_data["actions"]) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pvp_game_history(self, db_session: AsyncSession):
|
|
"""Test recording a PvP match.
|
|
|
|
PvP games have both player1_id and player2_id set.
|
|
"""
|
|
user1 = await UserFactory.create(db_session)
|
|
user2 = await UserFactory.create(db_session)
|
|
|
|
history = await GameHistoryFactory.create_pvp_game(
|
|
db_session,
|
|
user1,
|
|
user2,
|
|
winner=user2,
|
|
game_type=GameType.RANKED,
|
|
)
|
|
|
|
assert history.player1_id == user1.id
|
|
assert history.player2_id == user2.id
|
|
assert history.winner_id == user2.id
|
|
assert history.game_type == GameType.RANKED
|
|
assert history.npc_id is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_played_at_timestamp(self, db_session: AsyncSession):
|
|
"""Test that played_at records completion time.
|
|
|
|
played_at should be set when the game ends.
|
|
"""
|
|
user = await UserFactory.create(db_session)
|
|
before = datetime.now(UTC)
|
|
|
|
history = await GameHistoryFactory.create(db_session, player1_id=user.id)
|
|
|
|
assert history.played_at is not None
|
|
assert history.played_at >= before
|
|
|
|
|
|
# =============================================================================
|
|
# Model __repr__ Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestModelRepr:
|
|
"""Tests for model __repr__ methods.
|
|
|
|
These ensure debugging output is useful and doesn't break.
|
|
"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_repr(self, db_session: AsyncSession):
|
|
"""Test User __repr__ includes id and email."""
|
|
user = await UserFactory.create(db_session, email="repr_test@example.com")
|
|
repr_str = repr(user)
|
|
|
|
assert "User" in repr_str
|
|
assert str(user.id) in repr_str
|
|
assert "repr_test@example.com" in repr_str
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collection_repr(self, db_session: AsyncSession):
|
|
"""Test Collection __repr__ includes relevant info."""
|
|
user = await UserFactory.create(db_session)
|
|
entry = await CollectionFactory.create(
|
|
db_session, user_id=user.id, card_definition_id="test_card_repr"
|
|
)
|
|
repr_str = repr(entry)
|
|
|
|
assert "Collection" in repr_str
|
|
# Collection repr uses user_id and card, not the collection id
|
|
assert str(user.id) in repr_str
|
|
assert "test_card_repr" in repr_str
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deck_repr(self, db_session: AsyncSession):
|
|
"""Test Deck __repr__ includes id and name."""
|
|
user = await UserFactory.create(db_session)
|
|
deck = await DeckFactory.create_for_user(db_session, user, name="Repr Test Deck")
|
|
repr_str = repr(deck)
|
|
|
|
assert "Deck" in repr_str
|
|
assert str(deck.id) in repr_str
|
|
assert "Repr Test Deck" in repr_str
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_campaign_progress_repr(self, db_session: AsyncSession):
|
|
"""Test CampaignProgress __repr__ includes user_id and club."""
|
|
user = await UserFactory.create(db_session)
|
|
progress = await CampaignProgressFactory.create_for_user(
|
|
db_session, user, current_club="fire_club"
|
|
)
|
|
repr_str = repr(progress)
|
|
|
|
assert "CampaignProgress" in repr_str
|
|
# CampaignProgress repr uses user_id, not the progress id
|
|
assert str(user.id) in repr_str
|
|
assert "fire_club" in repr_str
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_active_game_repr(self, db_session: AsyncSession):
|
|
"""Test ActiveGame __repr__ includes id, type, and turn."""
|
|
user = await UserFactory.create(db_session)
|
|
game = await ActiveGameFactory.create_campaign_game(db_session, user)
|
|
repr_str = repr(game)
|
|
|
|
assert "ActiveGame" in repr_str
|
|
assert str(game.id) in repr_str
|
|
assert "campaign" in repr_str.lower() or "CAMPAIGN" in repr_str
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_game_history_repr(self, db_session: AsyncSession):
|
|
"""Test GameHistory __repr__ includes id, type, turns, and end reason."""
|
|
user = await UserFactory.create(db_session)
|
|
history = await GameHistoryFactory.create_player_win(db_session, user)
|
|
repr_str = repr(history)
|
|
|
|
assert "GameHistory" in repr_str
|
|
assert str(history.id) in repr_str
|