mantimon-tcg/backend/tests/services/test_deck_service.py
Cal Corum 3ec670753b Fix security and validation issues from code review
Critical fixes:
- Add admin API key authentication for admin endpoints
- Add race condition protection via unique partial index for starter decks
- Make starter deck selection atomic with combined method

Moderate fixes:
- Fix DI pattern violation in validate_deck_endpoint
- Add card ID format validation (regex pattern)
- Add card quantity validation (1-99 range)
- Fix exception chaining with from None (B904)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-28 14:16:07 -06:00

804 lines
25 KiB
Python

"""Integration tests for DeckService.
Tests deck management operations with real PostgreSQL database.
Uses dev containers (docker compose up -d) for database access.
The backend is stateless - deck rules come from DeckConfig parameter.
These tests verify that the service correctly accepts and applies config
from the caller.
These tests verify:
- Creating decks with validation
- Updating decks with re-validation
- Deleting decks
- Deck slot limits (free vs premium)
- Ownership validation modes
- Starter deck creation
"""
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import DeckConfig
from app.db.models.collection import CardSource
from app.repositories.postgres.collection import PostgresCollectionRepository
from app.repositories.postgres.deck import PostgresDeckRepository
from app.services.card_service import CardService
from app.services.collection_service import CollectionService
from app.services.deck_service import (
DeckLimitExceededError,
DeckNotFoundError,
DeckService,
)
from tests.factories import DeckFactory, UserFactory
# =============================================================================
# Fixtures
# =============================================================================
@pytest_asyncio.fixture
async def card_service() -> CardService:
"""Create a CardService with real card data loaded."""
service = CardService()
await service.load_all()
return service
@pytest.fixture
def default_config() -> DeckConfig:
"""Standard deck config for testing."""
return DeckConfig()
@pytest.fixture
def deck_service(db_session: AsyncSession, card_service: CardService) -> DeckService:
"""Create a DeckService with PostgreSQL repositories."""
deck_repo = PostgresDeckRepository(db_session)
collection_repo = PostgresCollectionRepository(db_session)
return DeckService(deck_repo, card_service, collection_repo)
@pytest.fixture
def collection_service(db_session: AsyncSession, card_service: CardService) -> CollectionService:
"""Create a CollectionService for granting cards."""
repo = PostgresCollectionRepository(db_session)
return CollectionService(repo, card_service)
def make_valid_deck_cards() -> tuple[dict[str, int], dict[str, int]]:
"""Create card and energy dicts that form a valid 40+20 deck.
Uses real card IDs from the a1 set for proper validation testing.
"""
# 40 cards total - using real card IDs from a1 set
cards = {
# Basic Pokemon
"a1-001-bulbasaur": 4,
"a1-033-charmander": 4,
"a1-053-squirtle": 4,
"a1-094-pikachu": 4,
# Stage 1 Pokemon
"a1-002-ivysaur": 4,
"a1-034-charmeleon": 4,
"a1-054-wartortle": 4,
# Trainers
"a1-211-potion": 4,
"a1-213-professor-oaks-research": 4,
"a1-214-red": 4,
}
# 20 energy
energy_cards = {"grass": 10, "colorless": 10}
return cards, energy_cards
# =============================================================================
# Create Deck Tests
# =============================================================================
class TestCreateDeck:
"""Tests for deck creation."""
@pytest.mark.asyncio
async def test_create_deck_basic(
self,
db_session: AsyncSession,
deck_service: DeckService,
default_config: DeckConfig,
):
"""
Test that create_deck creates a deck with provided data.
The deck should be persisted with all provided attributes.
"""
user = await UserFactory.create(db_session)
cards, energy = make_valid_deck_cards()
deck = await deck_service.create_deck(
user_id=user.id,
name="My Test Deck",
cards=cards,
energy_cards=energy,
deck_config=default_config,
max_decks=5,
validate_ownership=False, # Skip ownership for this test
description="A test deck",
)
assert deck.name == "My Test Deck"
assert deck.cards == cards
assert deck.energy_cards == energy
assert deck.description == "A test deck"
assert deck.user_id == user.id
assert deck.is_starter is False
@pytest.mark.asyncio
async def test_create_deck_stores_validation_errors(
self,
db_session: AsyncSession,
deck_service: DeckService,
default_config: DeckConfig,
):
"""
Test that invalid decks are saved with validation errors.
Invalid decks CAN be saved (with errors) to support work-in-progress
deck building. The validation errors are stored for display.
"""
user = await UserFactory.create(db_session)
# Invalid deck: wrong counts
cards = {"a1-001-bulbasaur": 10} # Only 10 cards, not 40
energy = {"grass": 5} # Only 5 energy, not 20
deck = await deck_service.create_deck(
user_id=user.id,
name="Invalid Deck",
cards=cards,
energy_cards=energy,
deck_config=default_config,
max_decks=5,
validate_ownership=False,
)
assert deck.is_valid is False
assert deck.validation_errors is not None
assert len(deck.validation_errors) > 0
@pytest.mark.asyncio
async def test_create_deck_respects_custom_config(
self,
db_session: AsyncSession,
deck_service: DeckService,
):
"""
Test that create_deck uses the provided DeckConfig.
The backend is stateless - rules come from the request.
Custom config should be applied during validation.
"""
user = await UserFactory.create(db_session)
# Custom config with smaller deck sizes
custom_config = DeckConfig(
min_size=20,
max_size=20,
energy_deck_size=10,
max_copies_per_card=10,
)
# Deck that's valid for custom config but invalid for default
cards = {"a1-001-bulbasaur": 10, "a1-002-ivysaur": 10} # 20 cards
energy = {"grass": 10} # 10 energy
deck = await deck_service.create_deck(
user_id=user.id,
name="Custom Config Deck",
cards=cards,
energy_cards=energy,
deck_config=custom_config,
max_decks=5,
validate_ownership=False,
)
# Should have errors only for card IDs not existing, not for counts
# (since counts match custom config)
if deck.validation_errors:
# The only error should be about invalid card IDs, not counts
count_errors = [
e
for e in deck.validation_errors
if "cards" in e.lower() and "must have" in e.lower()
]
assert (
len(count_errors) == 0
), f"Should have no count errors with custom config: {deck.validation_errors}"
@pytest.mark.asyncio
async def test_create_deck_enforces_deck_limit(
self,
db_session: AsyncSession,
deck_service: DeckService,
default_config: DeckConfig,
):
"""
Test that create_deck raises error when at deck limit.
Free users are limited to 5 decks by default.
"""
user = await UserFactory.create(db_session)
cards, energy = make_valid_deck_cards()
# Create 5 decks (the limit)
for i in range(5):
await DeckFactory.create_for_user(db_session, user, name=f"Deck {i}")
# Try to create a 6th deck
with pytest.raises(DeckLimitExceededError, match="Deck limit reached"):
await deck_service.create_deck(
user_id=user.id,
name="One Too Many",
cards=cards,
energy_cards=energy,
deck_config=default_config,
max_decks=5,
validate_ownership=False,
)
@pytest.mark.asyncio
async def test_create_deck_premium_unlimited(
self,
db_session: AsyncSession,
deck_service: DeckService,
default_config: DeckConfig,
):
"""
Test that premium users can have more decks.
Premium users have max_decks=999 (effectively unlimited).
"""
user = await UserFactory.create_premium(db_session)
cards, energy = make_valid_deck_cards()
# Create many decks
for i in range(10):
await deck_service.create_deck(
user_id=user.id,
name=f"Deck {i}",
cards=cards,
energy_cards=energy,
deck_config=default_config,
max_decks=999, # Premium limit
validate_ownership=False,
)
# Should have 10 decks
decks = await deck_service.get_user_decks(user.id)
assert len(decks) == 10
# =============================================================================
# Update Deck Tests
# =============================================================================
class TestUpdateDeck:
"""Tests for deck updates."""
@pytest.mark.asyncio
async def test_update_deck_name_only(
self,
db_session: AsyncSession,
deck_service: DeckService,
default_config: DeckConfig,
):
"""
Test that updating only the name doesn't re-validate.
Name changes should preserve existing validation state.
"""
user = await UserFactory.create(db_session)
deck = await DeckFactory.create_for_user(
db_session, user, is_valid=True, validation_errors=None
)
updated = await deck_service.update_deck(
user_id=user.id,
deck_id=deck.id,
deck_config=default_config,
name="New Name",
)
assert updated.name == "New Name"
assert updated.is_valid is True # Preserved
@pytest.mark.asyncio
async def test_update_deck_cards_revalidates(
self,
db_session: AsyncSession,
deck_service: DeckService,
default_config: DeckConfig,
):
"""
Test that updating cards triggers re-validation.
Card changes should always re-validate the deck.
"""
user = await UserFactory.create(db_session)
deck = await DeckFactory.create_for_user(
db_session, user, is_valid=True, validation_errors=None
)
# Update with invalid cards
updated = await deck_service.update_deck(
user_id=user.id,
deck_id=deck.id,
deck_config=default_config,
cards={"a1-001-bulbasaur": 5}, # Invalid: only 5 cards
validate_ownership=False,
)
assert updated.is_valid is False
assert updated.validation_errors is not None
@pytest.mark.asyncio
async def test_update_deck_not_found(
self,
db_session: AsyncSession,
deck_service: DeckService,
default_config: DeckConfig,
):
"""
Test that updating non-existent deck raises error.
Should raise DeckNotFoundError for invalid deck_id.
"""
user = await UserFactory.create(db_session)
from uuid import uuid4
with pytest.raises(DeckNotFoundError):
await deck_service.update_deck(
user_id=user.id,
deck_id=uuid4(), # Non-existent
deck_config=default_config,
name="New Name",
)
@pytest.mark.asyncio
async def test_update_deck_wrong_user(
self,
db_session: AsyncSession,
deck_service: DeckService,
default_config: DeckConfig,
):
"""
Test that users can only update their own decks.
Should raise DeckNotFoundError when trying to update another user's deck.
"""
user1 = await UserFactory.create(db_session)
user2 = await UserFactory.create(db_session)
deck = await DeckFactory.create_for_user(db_session, user1)
with pytest.raises(DeckNotFoundError):
await deck_service.update_deck(
user_id=user2.id, # Wrong user
deck_id=deck.id,
deck_config=default_config,
name="Stolen Deck",
)
# =============================================================================
# Delete Deck Tests
# =============================================================================
class TestDeleteDeck:
"""Tests for deck deletion."""
@pytest.mark.asyncio
async def test_delete_deck_removes_deck(
self, db_session: AsyncSession, deck_service: DeckService
):
"""
Test that delete_deck removes the deck.
Deck should no longer be retrievable after deletion.
"""
user = await UserFactory.create(db_session)
deck = await DeckFactory.create_for_user(db_session, user)
result = await deck_service.delete_deck(user.id, deck.id)
assert result is True
with pytest.raises(DeckNotFoundError):
await deck_service.get_deck(user.id, deck.id)
@pytest.mark.asyncio
async def test_delete_deck_not_found(self, db_session: AsyncSession, deck_service: DeckService):
"""
Test that deleting non-existent deck raises error.
Should raise DeckNotFoundError for invalid deck_id.
"""
user = await UserFactory.create(db_session)
from uuid import uuid4
with pytest.raises(DeckNotFoundError):
await deck_service.delete_deck(user.id, uuid4())
@pytest.mark.asyncio
async def test_delete_deck_wrong_user(
self, db_session: AsyncSession, deck_service: DeckService
):
"""
Test that users can only delete their own decks.
Should raise DeckNotFoundError when trying to delete another user's deck.
"""
user1 = await UserFactory.create(db_session)
user2 = await UserFactory.create(db_session)
deck = await DeckFactory.create_for_user(db_session, user1)
with pytest.raises(DeckNotFoundError):
await deck_service.delete_deck(user2.id, deck.id)
# =============================================================================
# Get Deck Tests
# =============================================================================
class TestGetDeck:
"""Tests for retrieving decks."""
@pytest.mark.asyncio
async def test_get_deck_returns_owned_deck(
self, db_session: AsyncSession, deck_service: DeckService
):
"""
Test that get_deck returns a deck owned by user.
Should return complete deck details.
"""
user = await UserFactory.create(db_session)
deck = await DeckFactory.create_for_user(db_session, user, name="My Deck")
result = await deck_service.get_deck(user.id, deck.id)
assert result.id == deck.id
assert result.name == "My Deck"
assert result.user_id == user.id
@pytest.mark.asyncio
async def test_get_deck_not_found(self, db_session: AsyncSession, deck_service: DeckService):
"""
Test that get_deck raises error for non-existent deck.
Should raise DeckNotFoundError for invalid deck_id.
"""
user = await UserFactory.create(db_session)
from uuid import uuid4
with pytest.raises(DeckNotFoundError):
await deck_service.get_deck(user.id, uuid4())
@pytest.mark.asyncio
async def test_get_deck_wrong_user(self, db_session: AsyncSession, deck_service: DeckService):
"""
Test that users can only get their own decks.
Should raise DeckNotFoundError for another user's deck.
"""
user1 = await UserFactory.create(db_session)
user2 = await UserFactory.create(db_session)
deck = await DeckFactory.create_for_user(db_session, user1)
with pytest.raises(DeckNotFoundError):
await deck_service.get_deck(user2.id, deck.id)
# =============================================================================
# Get User Decks Tests
# =============================================================================
class TestGetUserDecks:
"""Tests for listing user's decks."""
@pytest.mark.asyncio
async def test_get_user_decks_returns_all(
self, db_session: AsyncSession, deck_service: DeckService
):
"""
Test that get_user_decks returns all user's decks.
Should return complete list of decks.
"""
user = await UserFactory.create(db_session)
await DeckFactory.create_for_user(db_session, user, name="Deck 1")
await DeckFactory.create_for_user(db_session, user, name="Deck 2")
await DeckFactory.create_for_user(db_session, user, name="Deck 3")
decks = await deck_service.get_user_decks(user.id)
assert len(decks) == 3
names = {d.name for d in decks}
assert names == {"Deck 1", "Deck 2", "Deck 3"}
@pytest.mark.asyncio
async def test_get_user_decks_empty(self, db_session: AsyncSession, deck_service: DeckService):
"""
Test that get_user_decks returns empty list for new user.
New users should have no decks.
"""
user = await UserFactory.create(db_session)
decks = await deck_service.get_user_decks(user.id)
assert decks == []
@pytest.mark.asyncio
async def test_get_user_decks_only_own_decks(
self, db_session: AsyncSession, deck_service: DeckService
):
"""
Test that get_user_decks only returns own decks.
Should not include other users' decks.
"""
user1 = await UserFactory.create(db_session)
user2 = await UserFactory.create(db_session)
await DeckFactory.create_for_user(db_session, user1, name="User1 Deck")
await DeckFactory.create_for_user(db_session, user2, name="User2 Deck")
decks = await deck_service.get_user_decks(user1.id)
assert len(decks) == 1
assert decks[0].name == "User1 Deck"
# =============================================================================
# Can Create Deck Tests
# =============================================================================
class TestCanCreateDeck:
"""Tests for deck limit checking."""
@pytest.mark.asyncio
async def test_can_create_deck_under_limit(
self, db_session: AsyncSession, deck_service: DeckService
):
"""
Test that can_create_deck returns True under limit.
User with fewer decks than max should be able to create more.
"""
user = await UserFactory.create(db_session)
await DeckFactory.create_for_user(db_session, user)
can_create = await deck_service.can_create_deck(user.id, max_decks=5)
assert can_create is True
@pytest.mark.asyncio
async def test_can_create_deck_at_limit(
self, db_session: AsyncSession, deck_service: DeckService
):
"""
Test that can_create_deck returns False at limit.
User at max decks should not be able to create more.
"""
user = await UserFactory.create(db_session)
for i in range(5):
await DeckFactory.create_for_user(db_session, user, name=f"Deck {i}")
can_create = await deck_service.can_create_deck(user.id, max_decks=5)
assert can_create is False
# =============================================================================
# Starter Deck Tests
# =============================================================================
class TestStarterDeck:
"""Tests for starter deck creation."""
@pytest.mark.asyncio
async def test_has_starter_deck_false_for_new_user(
self, db_session: AsyncSession, deck_service: DeckService
):
"""
Test that has_starter_deck returns False for new user.
New users haven't selected a starter yet.
"""
user = await UserFactory.create(db_session)
has_starter, starter_type = await deck_service.has_starter_deck(user.id)
assert has_starter is False
assert starter_type is None
@pytest.mark.asyncio
async def test_has_starter_deck_true_after_selection(
self, db_session: AsyncSession, deck_service: DeckService
):
"""
Test that has_starter_deck returns True after selecting starter.
Should return the type of starter selected.
"""
user = await UserFactory.create(db_session)
await DeckFactory.create_starter_deck(db_session, user, starter_type="grass")
has_starter, starter_type = await deck_service.has_starter_deck(user.id)
assert has_starter is True
assert starter_type == "grass"
@pytest.mark.asyncio
async def test_create_starter_deck(
self,
db_session: AsyncSession,
deck_service: DeckService,
default_config: DeckConfig,
):
"""
Test that create_starter_deck creates a proper starter deck.
Should create deck with is_starter=True and starter_type set.
"""
user = await UserFactory.create(db_session)
deck = await deck_service.create_starter_deck(
user_id=user.id,
starter_type="fire",
deck_config=default_config,
max_decks=5,
)
assert deck.is_starter is True
assert deck.starter_type == "fire"
# Starter deck names are defined in starter_decks.py and may vary
assert deck.name is not None and len(deck.name) > 0
@pytest.mark.asyncio
async def test_create_starter_deck_invalid_type(
self,
db_session: AsyncSession,
deck_service: DeckService,
default_config: DeckConfig,
):
"""
Test that create_starter_deck rejects invalid type.
Should raise ValueError for unknown starter type.
"""
user = await UserFactory.create(db_session)
with pytest.raises(ValueError, match="Invalid starter type"):
await deck_service.create_starter_deck(
user_id=user.id,
starter_type="invalid",
deck_config=default_config,
max_decks=5,
)
# =============================================================================
# Ownership Validation Tests
# =============================================================================
class TestOwnershipValidation:
"""Tests for card ownership validation in campaign mode."""
@pytest.mark.asyncio
async def test_validate_ownership_passes_when_owned(
self,
db_session: AsyncSession,
deck_service: DeckService,
collection_service: CollectionService,
default_config: DeckConfig,
):
"""
Test that ownership validation passes when user owns cards.
In campaign mode, user must own all cards in the deck.
"""
user = await UserFactory.create(db_session)
# Grant cards to user
await collection_service.add_cards(user.id, "a1-001-bulbasaur", 4, CardSource.BOOSTER)
cards = {"a1-001-bulbasaur": 4}
energy = {"grass": 20}
# Create deck with ownership validation
deck = await deck_service.create_deck(
user_id=user.id,
name="Owned Cards Deck",
cards=cards,
energy_cards=energy,
deck_config=default_config,
max_decks=5,
validate_ownership=True,
)
# Should have errors for card counts/IDs but NOT for ownership
ownership_errors = [e for e in (deck.validation_errors or []) if "Insufficient" in e]
assert len(ownership_errors) == 0
@pytest.mark.asyncio
async def test_validate_ownership_fails_when_not_owned(
self,
db_session: AsyncSession,
deck_service: DeckService,
default_config: DeckConfig,
):
"""
Test that ownership validation fails when user doesn't own cards.
Should include ownership errors in validation_errors.
"""
user = await UserFactory.create(db_session)
# User has no cards
cards = {"a1-001-bulbasaur": 4}
energy = {"grass": 20}
deck = await deck_service.create_deck(
user_id=user.id,
name="Unowned Cards Deck",
cards=cards,
energy_cards=energy,
deck_config=default_config,
max_decks=5,
validate_ownership=True,
)
assert deck.is_valid is False
ownership_errors = [e for e in (deck.validation_errors or []) if "Insufficient" in e]
assert len(ownership_errors) > 0
@pytest.mark.asyncio
async def test_skip_ownership_validation_freeplay(
self,
db_session: AsyncSession,
deck_service: DeckService,
default_config: DeckConfig,
):
"""
Test that ownership validation can be skipped for freeplay.
In freeplay mode, validate_ownership=False skips ownership checks.
"""
user = await UserFactory.create(db_session)
# User has no cards
cards = {"a1-001-bulbasaur": 4}
energy = {"grass": 20}
deck = await deck_service.create_deck(
user_id=user.id,
name="Freeplay Deck",
cards=cards,
energy_cards=energy,
deck_config=default_config,
max_decks=5,
validate_ownership=False, # Freeplay mode
)
# Should NOT have ownership errors
ownership_errors = [e for e in (deck.validation_errors or []) if "Insufficient" in e]
assert len(ownership_errors) == 0