cal/mantimon-tcg
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 3: Collections + Decks - Services and DI architecture

Browse Source 
Implemented with Repository Protocol pattern for offline fork support:
- CollectionService with PostgresCollectionRepository
- DeckService with PostgresDeckRepository
- DeckValidator with DeckConfig + CardService injection
- Starter deck definitions (5 types: grass, fire, water, psychic, lightning)
- Pydantic schemas for collection and deck APIs
- Unit tests for DeckValidator (32 tests passing)

Architecture follows pure dependency injection - no service locator patterns.
Added CLAUDE.md documenting DI requirements and patterns.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Cal Corum 2026-01-28 11:27:14 -06:00
parent 4859b2a9cb
commit 58349c126a
19 changed files with 3451 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

247
backend/CLAUDE.md Normal file
View File

@ -0,0 +1,247 @@
# Mantimon TCG Backend - AI Agent Guidelines
This document defines architecture requirements, patterns, and constraints for AI agents working on this codebase.
## Critical Architecture Requirement: Offline Fork Support
> **The `app/core/` module must remain extractable as a standalone offline game.**
This is a primary design goal. The core game engine should work without network, database, or authentication dependencies.
### Module Independence Rules
| Module | Can Import From | Cannot Import From |
|--------|-----------------|-------------------|
| `app/core/` | Python stdlib, pydantic | `app/services/`, `app/api/`, `app/db/`, sqlalchemy |
| `app/services/` | `app/core/`, `app/repositories/` | `app/api/` |
| `app/repositories/` | `app/core/`, `app/db/` | `app/services/`, `app/api/` |
| `app/api/` | All modules | - |
### Import Boundary Examples
```python
# ALLOWED in app/core/
from app.core.models import CardDefinition, GameState
from app.core.config import RulesConfig
from app.core.rng import RandomProvider
# FORBIDDEN in app/core/
from app.services import CardService # NO - service dependency
from app.api.deps import get_current_user # NO - auth dependency
from sqlalchemy.ext.asyncio import AsyncSession # NO - DB dependency
```
---
## Dependency Injection Pattern
**All services must use constructor-based dependency injection.** No service locator patterns.
### Required Pattern
```python
class DeckValidator:
"""Dependencies injected via constructor."""
def __init__(self, config: DeckConfig, card_service: CardService) -> None:
self._config = config
self._card_service = card_service
```
### Forbidden Patterns
```python
# WRONG - Service locator pattern
class DeckValidator:
def validate(self, cards):
service = get_card_service() # Hidden dependency!
...
# WRONG - Default instantiation hides dependency
class DeckValidator:
def __init__(self, config: DeckConfig | None = None):
self.config = config or DeckConfig() # Hidden creation!
```
### Why This Matters
1. **Testability**: Dependencies can be mocked without patching globals
2. **Offline Fork**: Services can be swapped for local implementations
3. **Explicit Dependencies**: Constructor shows all requirements
4. **Composition Root**: All wiring happens at application startup, not scattered
---
## Repository Protocol Pattern
Services access data through repository protocols, not directly through ORM models.
### Pattern
```python
# Protocol defines interface (app/repositories/protocols.py)
class CollectionRepository(Protocol):
async def get_all(self, user_id: UUID) -> list[CollectionEntry]: ...
async def upsert(self, ...) -> CollectionEntry: ...
# PostgreSQL implementation (app/repositories/postgres/)
class PostgresCollectionRepository:
def __init__(self, db: AsyncSession) -> None:
self._db = db
# Service uses protocol, not implementation
class CollectionService:
def __init__(self, repository: CollectionRepository, card_service: CardService):
self._repo = repository
self._card_service = card_service
```
### Benefits
- **Offline Fork**: Can implement `LocalCollectionRepository` using JSON files
- **Testing**: Can inject mock repositories
- **Decoupling**: Services don't know about SQLAlchemy
---
## Configuration Injection
Game rules come from `app/core/config.py` classes. These must be injected, not instantiated internally.
### Pattern
```python
# Inject config
class DeckValidator:
def __init__(self, config: DeckConfig, card_service: CardService):
self._config = config
def validate(self, cards):
if len(cards) != self._config.min_size: # Use injected config
...
# For validation functions that need config
def validate_starter_decks(config: DeckSizeConfig) -> dict[str, list[str]]:
"""Config is required parameter, not created internally."""
...
```
### Protocol for Minimal Interface
When a function only needs specific config values, use a Protocol:
```python
class DeckSizeConfig(Protocol):
"""Minimal interface for deck size validation."""
min_size: int
energy_deck_size: int
def validate_starter_decks(config: DeckSizeConfig) -> dict[str, list[str]]:
# Any object with min_size and energy_deck_size works
...
```
---
## Testing Requirements
### Test Docstrings Required
Every test must have a docstring explaining "what" and "why":
```python
def test_paralyzed_pokemon_cannot_attack():
"""
Test that paralyzed Pokemon are blocked from attacking.
Paralysis should prevent all attack actions until cleared
at the end of the affected player's turn.
"""
...
```
### Unit Tests vs Integration Tests
| Directory | Purpose | Database |
|-----------|---------|----------|
| `tests/unit/` | Pure unit tests, mocked dependencies | No |
| `tests/services/` | Integration tests with real DB | Yes |
| `tests/core/` | Core engine tests | No |
| `tests/api/` | API endpoint tests | Yes |
### Use Seeded RNG for Determinism
```python
from app.core import create_rng
def test_coin_flip():
rng = create_rng(seed=42)
results = [rng.coin_flip() for _ in range(5)]
assert results == [True, False, True, True, False] # Deterministic
```
---
## Code Organization
```
app/
├── core/ # Game engine (MUST be standalone-capable)
│ ├── models/ # Pydantic models for game state
│ ├── effects/ # Effect handler system
│ ├── config.py # RulesConfig and sub-configs
│ └── engine.py # GameEngine orchestrator
├── services/ # Business logic (uses repositories)
├── repositories/ # Data access layer
│ ├── protocols.py # Repository protocols (interfaces)
│ └── postgres/ # PostgreSQL implementations
├── schemas/ # Pydantic schemas for API
├── api/ # FastAPI routes
├── db/ # Database models and migrations
└── data/ # Static data (starter decks, etc.)
```
---
## Quick Reference
### Creating a New Service
1. Define constructor with all dependencies as parameters
2. Use repository protocols for data access
3. Inject CardService/DeckConfig rather than using `get_*` functions
4. Keep business logic separate from data access
### Creating a Repository
1. Define protocol in `app/repositories/protocols.py`
2. Create DTO dataclasses for protocol returns
3. Implement in `app/repositories/postgres/`
4. Use `_to_dto()` method to convert ORM -> DTO
### Adding Static Data
1. Place in `app/data/` module
2. Use Protocol for any config dependencies
3. Provide validation functions that accept config as parameter
---
## Common Mistakes to Avoid
| Mistake | Correct Approach |
|---------|------------------|
| `get_card_service()` in method body | Inject `CardService` via constructor |
| `config or DeckConfig()` default | Make config required parameter |
| Importing from `app.services` in `app.core` | Core must remain standalone |
| Hardcoded magic numbers | Use `DeckConfig` values |
| Tests without docstrings | Always explain what and why |
| Unit tests in `tests/services/` | Use `tests/unit/` for no-DB tests |
---
## See Also
- `app/core/AGENTS.md` - Detailed core engine guidelines
- `app/core/README.md` - Core module documentation
- `app/core/effects/README.md` - Effect system documentation

32
backend/PROJECT_PLAN_MASTER.json
View File

@ -2,13 +2,13 @@
"meta": {
"version": "1.0.0",
"created": "2026-01-27",
"lastUpdated": "2026-01-27",
"lastUpdated": "2026-01-28",
"planType": "master",
"projectName": "Mantimon TCG - Backend Services",
"description": "Live service backend for play.mantimon.com - server-authoritative multiplayer TCG with campaign mode, free-play, and F2P monetization",
"totalPhases": 6,
"completedPhases": 1,
"status": "Phase 0 complete, Phase 1 in progress"
"completedPhases": 3,
"status": "Phases 0-2 complete, Phase 3 (Collections + Decks) up next"
},
"architectureDecisions": {
@ -77,44 +77,46 @@
"Win condition checker",
"Visibility filter (hidden info security)",
"Card scraper with 372 cards + images",
"833 tests at 97% coverage"
"Engine validation script (29 test scenarios)",
"833 initial tests (now 1072 total with all phases)"
],
"completedDate": "2026-01-27"
},
{
"id": "PHASE_1",
"name": "Database + Infrastructure",
"status": "IN_PROGRESS",
"status": "COMPLETE",
"description": "PostgreSQL models, Redis caching, CardService, environment config",
"planFile": "project_plans/PHASE_1_DATABASE.json",
"estimatedWeeks": "2-3",
"dependencies": ["PHASE_0"],
"deliverables": [
"SQLAlchemy async models",
"SQLAlchemy async models (User, GameHistory, OAuthLinkedAccount)",
"Alembic migrations",
"Redis connection utilities",
"GameStateManager (Redis + Postgres write-behind)",
"CardService (JSON → CardDefinition)",
"Environment config (dev/staging/prod)",
"Docker compose for local dev"
]
],
"completedDate": "2026-01-27"
},
{
"id": "PHASE_2",
"name": "Authentication",
"status": "NOT_STARTED",
"status": "COMPLETE",
"description": "OAuth login, JWT sessions, user management",
"planFile": "project_plans/PHASE_2_AUTH.json",
"estimatedWeeks": "1-2",
"dependencies": ["PHASE_1"],
"deliverables": [
"OAuth integration (Google, Discord)",
"JWT token management",
"JWT token management with refresh tokens",
"User creation/login endpoints",
"Session middleware",
"Account linking (multiple providers)",
"Premium tier with expiration tracking"
]
"FastAPI auth dependencies (CurrentUser, DbSession)",
"Account linking (multiple providers per user)",
"Premium tier with expiration tracking",
"98 new tests, 1072 total tests"
],
"completedDate": "2026-01-28"
},
{
"id": "PHASE_3",

4
backend/app/data/__init__.py Normal file
View File

@ -0,0 +1,4 @@
"""Data definitions for Mantimon TCG.
This package contains static data definitions like starter decks.
"""

296
backend/app/data/starter_decks.py Normal file
View File

@ -0,0 +1,296 @@
"""Starter deck definitions for Mantimon TCG.
This module defines the 5 starter decks available to new players:
- Grass: Bulbasaur, Caterpie, and Bellsprout evolution lines
- Fire: Charmander, Growlithe, and Ponyta lines
- Water: Squirtle, Poliwag, and Horsea lines
- Psychic: Abra, Gastly, and Drowzee lines
- Lightning: Pikachu, Magnemite, and Voltorb lines
Each deck contains exactly 40 Pokemon/Trainer cards + 20 energy cards,
following Mantimon TCG house rules.
Deck composition philosophy:
- 3 evolution lines (4 cards each for basics, 3-4 for evolutions)
- Basic support trainers for card draw and healing
- Type-specific energy (14) + colorless (6)
Usage:
from app.data.starter_decks import get_starter_deck, STARTER_TYPES
deck = get_starter_deck("grass")
cards = deck["cards"] # {card_id: quantity}
energy = deck["energy_cards"] # {type: quantity}
"""
from typing import Protocol, TypedDict
class DeckSizeConfig(Protocol):
"""Protocol for deck size configuration.
Defines the minimal interface needed for deck validation.
Any object with these attributes (like DeckConfig) satisfies this protocol.
"""
min_size: int
energy_deck_size: int
class StarterDeckDefinition(TypedDict):
"""Type definition for starter deck structure."""
name: str
description: str
cards: dict[str, int]
energy_cards: dict[str, int]
# Available starter deck types
STARTER_TYPES: list[str] = ["grass", "fire", "water", "psychic", "lightning"]
# Classic starter types (always available)
CLASSIC_STARTERS: list[str] = ["grass", "fire", "water"]
# Rotating starter types (may change seasonally)
ROTATING_STARTERS: list[str] = ["psychic", "lightning"]
STARTER_DECKS: dict[str, StarterDeckDefinition] = {
"grass": {
"name": "Forest Guardian",
"description": "A balanced Grass deck featuring Bulbasaur, Caterpie, and Bellsprout evolution lines.",
"cards": {
# Bulbasaur line (11 cards)
"a1-001-bulbasaur": 4,
"a1-002-ivysaur": 3,
"a1-003-venusaur": 2,
# Caterpie line (9 cards)
"a1-005-caterpie": 4,
"a1-006-metapod": 3,
"a1-007-butterfree": 2,
# Bellsprout line (9 cards)
"a1-018-bellsprout": 4,
"a1-019-weepinbell": 3,
"a1-020-victreebel": 2,
# Tangela (2 cards) - extra basics
"a1-024-tangela": 2,
# Trainers (11 cards)
"a1-219-erika": 2,
"a1-216-helix-fossil": 2,
"a1-217-dome-fossil": 2,
"a1-218-old-amber": 2,
"a1-224-brock": 3,
},
"energy_cards": {
"grass": 14,
"colorless": 6,
},
},
"fire": {
"name": "Inferno Blaze",
"description": "An aggressive Fire deck featuring Charmander, Growlithe, and Ponyta evolution lines.",
"cards": {
# Charmander line (9 cards)
"a1-033-charmander": 4,
"a1-034-charmeleon": 3,
"a1-035-charizard": 2,
# Growlithe line (6 cards)
"a1-039-growlithe": 4,
"a1-040-arcanine": 2,
# Ponyta line (7 cards)
"a1-042-ponyta": 4,
"a1-043-rapidash": 3,
# Vulpix (3 cards) - extra basics
"a1-037-vulpix": 3,
# Magmar (4 cards) - extra basics
"a1-044-magmar": 4,
# Trainers (11 cards)
"a1-221-blaine": 3,
"a1-216-helix-fossil": 2,
"a1-217-dome-fossil": 2,
"a1-218-old-amber": 2,
"a1-224-brock": 2,
},
"energy_cards": {
"fire": 14,
"colorless": 6,
},
},
"water": {
"name": "Tidal Wave",
"description": "A versatile Water deck featuring Squirtle, Poliwag, and Horsea evolution lines.",
"cards": {
# Squirtle line (9 cards)
"a1-053-squirtle": 4,
"a1-054-wartortle": 3,
"a1-055-blastoise": 2,
# Poliwag line (9 cards)
"a1-059-poliwag": 4,
"a1-060-poliwhirl": 3,
"a1-061-poliwrath": 2,
# Horsea line (6 cards)
"a1-070-horsea": 4,
"a1-071-seadra": 2,
# Seel line (5 cards) - extra
"a1-064-seel": 3,
"a1-065-dewgong": 2,
# Trainers (11 cards)
"a1-220-misty": 3,
"a1-216-helix-fossil": 2,
"a1-217-dome-fossil": 2,
"a1-218-old-amber": 2,
"a1-224-brock": 2,
},
"energy_cards": {
"water": 14,
"colorless": 6,
},
},
"psychic": {
"name": "Mind Over Matter",
"description": "A control-focused Psychic deck featuring Abra, Gastly, and Drowzee evolution lines.",
"cards": {
# Abra line (9 cards)
"a1-115-abra": 4,
"a1-116-kadabra": 3,
"a1-117-alakazam": 2,
# Gastly line (9 cards)
"a1-120-gastly": 4,
"a1-121-haunter": 3,
"a1-122-gengar": 2,
# Drowzee line (6 cards)
"a1-124-drowzee": 4,
"a1-125-hypno": 2,
# Slowpoke line (5 cards) - extra
"a1-118-slowpoke": 3,
"a1-119-slowbro": 2,
# Trainers (11 cards)
"a1-225-sabrina": 3,
"a1-216-helix-fossil": 2,
"a1-217-dome-fossil": 2,
"a1-218-old-amber": 2,
"a1-224-brock": 2,
},
"energy_cards": {
"psychic": 14,
"colorless": 6,
},
},
"lightning": {
"name": "Thunder Strike",
"description": "A fast Lightning deck featuring Pikachu, Magnemite, and Voltorb evolution lines.",
"cards": {
# Pikachu line (7 cards)
"a1-094-pikachu": 4,
"a1-095-raichu": 3,
# Magnemite line (7 cards)
"a1-097-magnemite": 4,
"a1-098-magneton": 3,
# Voltorb line (7 cards)
"a1-099-voltorb": 4,
"a1-100-electrode": 3,
# Electabuzz (4 cards) - extra basics
"a1-101-electabuzz": 4,
# Blitzle line (4 cards)
"a1-105-blitzle": 2,
"a1-106-zebstrika": 2,
# Trainers (11 cards)
"a1-226-lt-surge": 3,
"a1-216-helix-fossil": 2,
"a1-217-dome-fossil": 2,
"a1-218-old-amber": 2,
"a1-224-brock": 2,
},
"energy_cards": {
"lightning": 14,
"colorless": 6,
},
},
}
def get_starter_deck(starter_type: str) -> StarterDeckDefinition:
"""Get a starter deck definition by type.
Args:
starter_type: One of grass, fire, water, psychic, lightning.
Returns:
StarterDeckDefinition with name, description, cards, and energy_cards.
Raises:
ValueError: If starter_type is not valid.
Example:
deck = get_starter_deck("grass")
print(f"Deck: {deck['name']}")
print(f"Total cards: {sum(deck['cards'].values())}")
"""
if starter_type not in STARTER_DECKS:
raise ValueError(
f"Invalid starter type: {starter_type}. " f"Must be one of: {', '.join(STARTER_TYPES)}"
)
return STARTER_DECKS[starter_type]
def validate_starter_decks(config: DeckSizeConfig) -> dict[str, list[str]]:
"""Validate all starter deck definitions against config rules.
Checks that each deck has the correct number of cards and energy
as defined by the provided configuration.
Args:
config: Configuration providing min_size and energy_deck_size.
Typically a DeckConfig instance, but any object satisfying
the DeckSizeConfig protocol works.
Returns:
Dictionary mapping deck type to list of validation errors.
Empty dict if all decks are valid.
Example:
from app.core.config import DeckConfig
config = DeckConfig()
errors = validate_starter_decks(config)
if errors:
for deck_type, deck_errors in errors.items():
print(f"{deck_type}: {deck_errors}")
"""
errors: dict[str, list[str]] = {}
for deck_type, deck in STARTER_DECKS.items():
deck_errors: list[str] = []
# Check card count against config
total_cards = sum(deck["cards"].values())
if total_cards != config.min_size:
deck_errors.append(f"Expected {config.min_size} cards, got {total_cards}")
# Check energy count against config
total_energy = sum(deck["energy_cards"].values())
if total_energy != config.energy_deck_size:
deck_errors.append(f"Expected {config.energy_deck_size} energy, got {total_energy}")
if deck_errors:
errors[deck_type] = deck_errors
return errors
def get_starter_card_ids(starter_type: str) -> list[str]:
"""Get list of all card IDs in a starter deck.
Args:
starter_type: One of grass, fire, water, psychic, lightning.
Returns:
List of card IDs (without quantities).
Example:
card_ids = get_starter_card_ids("grass")
# ["a1-001-bulbasaur", "a1-002-ivysaur", ...]
"""
deck = get_starter_deck(starter_type)
return list(deck["cards"].keys())

31
backend/app/repositories/__init__.py Normal file
View File

@ -0,0 +1,31 @@
"""Repository layer for Mantimon TCG.
This package defines repository protocols (interfaces) and their implementations.
Repositories handle pure data access (CRUD operations), while services contain
business logic.
The protocol pattern enables:
- Easy testing with mock repositories
- Multiple storage backends (PostgreSQL, SQLite, JSON files)
- Offline fork support without rewriting service layer
Usage:
from app.repositories import CollectionRepository, DeckRepository
from app.repositories.postgres import PostgresCollectionRepository
# In production (dependency injection)
repo = PostgresCollectionRepository(db_session)
# In tests
repo = MockCollectionRepository()
"""
from app.repositories.protocols import (
CollectionRepository,
DeckRepository,
)
__all__ = [
"CollectionRepository",
"DeckRepository",
]

27
backend/app/repositories/postgres/__init__.py Normal file
View File

@ -0,0 +1,27 @@
"""PostgreSQL repository implementations for Mantimon TCG.
This package contains PostgreSQL-specific implementations of the repository
protocols. These implementations use SQLAlchemy async sessions for database
access.
Usage:
from app.repositories.postgres import (
PostgresCollectionRepository,
PostgresDeckRepository,
)
# Create repository with database session
collection_repo = PostgresCollectionRepository(db_session)
deck_repo = PostgresDeckRepository(db_session)
# Use via service layer
service = CollectionService(collection_repo)
"""
from app.repositories.postgres.collection import PostgresCollectionRepository
from app.repositories.postgres.deck import PostgresDeckRepository
__all__ = [
"PostgresCollectionRepository",
"PostgresDeckRepository",
]

221
backend/app/repositories/postgres/collection.py Normal file
View File

@ -0,0 +1,221 @@
"""PostgreSQL implementation of CollectionRepository.
This module provides the PostgreSQL-specific implementation of the
CollectionRepository protocol using SQLAlchemy async sessions.
The implementation uses PostgreSQL's ON CONFLICT for efficient upserts.
Example:
async with get_db_session() as db:
repo = PostgresCollectionRepository(db)
entries = await repo.get_all(user_id)
"""
from datetime import UTC, datetime
from uuid import UUID
from sqlalchemy import delete, select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models.collection import CardSource, Collection
from app.repositories.protocols import CollectionEntry
def _to_dto(model: Collection) -> CollectionEntry:
"""Convert ORM model to DTO."""
return CollectionEntry(
id=model.id,
user_id=model.user_id,
card_definition_id=model.card_definition_id,
quantity=model.quantity,
source=model.source,
obtained_at=model.obtained_at,
created_at=model.created_at,
updated_at=model.updated_at,
)
class PostgresCollectionRepository:
"""PostgreSQL implementation of CollectionRepository.
Uses SQLAlchemy async sessions for database access. All operations
commit immediately for simplicity - transaction management should
be handled at the service layer if needed.
Attributes:
_db: The async database session.
"""
def __init__(self, db: AsyncSession) -> None:
"""Initialize with database session.
Args:
db: SQLAlchemy async session.
"""
self._db = db
async def get_all(self, user_id: UUID) -> list[CollectionEntry]:
"""Get all collection entries for a user.
Args:
user_id: The user's UUID.
Returns:
List of all collection entries, ordered by card_definition_id.
"""
result = await self._db.execute(
select(Collection)
.where(Collection.user_id == user_id)
.order_by(Collection.card_definition_id)
)
return [_to_dto(model) for model in result.scalars().all()]
async def get_by_card(self, user_id: UUID, card_definition_id: str) -> CollectionEntry | None:
"""Get a specific collection entry.
Args:
user_id: The user's UUID.
card_definition_id: The card ID to look up.
Returns:
CollectionEntry if exists, None otherwise.
"""
result = await self._db.execute(
select(Collection).where(
Collection.user_id == user_id,
Collection.card_definition_id == card_definition_id,
)
)
model = result.scalar_one_or_none()
return _to_dto(model) if model else None
async def get_quantity(self, user_id: UUID, card_definition_id: str) -> int:
"""Get quantity of a specific card owned by user.
Args:
user_id: The user's UUID.
card_definition_id: Card ID to check.
Returns:
Number of copies owned (0 if not owned).
"""
result = await self._db.execute(
select(Collection.quantity).where(
Collection.user_id == user_id,
Collection.card_definition_id == card_definition_id,
)
)
quantity = result.scalar_one_or_none()
return quantity if quantity is not None else 0
async def upsert(
self,
user_id: UUID,
card_definition_id: str,
quantity: int,
source: CardSource,
) -> CollectionEntry:
"""Add or update a collection entry using PostgreSQL ON CONFLICT.
If entry exists, increments quantity. Otherwise creates new entry.
Args:
user_id: The user's UUID.
card_definition_id: Card ID to add.
quantity: Number of copies to add.
source: How the cards were obtained.
Returns:
The created or updated CollectionEntry.
"""
now = datetime.now(UTC)
stmt = pg_insert(Collection).values(
user_id=user_id,
card_definition_id=card_definition_id,
quantity=quantity,
source=source,
obtained_at=now,
)
stmt = stmt.on_conflict_do_update(
constraint="uq_collection_user_card",
set_={
"quantity": Collection.quantity + quantity,
"updated_at": now,
},
)
await self._db.execute(stmt)
await self._db.commit()
# Fetch and return the updated entry
entry = await self.get_by_card(user_id, card_definition_id)
return entry # type: ignore[return-value]
async def decrement(
self,
user_id: UUID,
card_definition_id: str,
quantity: int,
) -> CollectionEntry | None:
"""Decrement quantity of a collection entry.
If quantity reaches 0 or below, deletes the entry.
Args:
user_id: The user's UUID.
card_definition_id: Card ID to decrement.
quantity: Number of copies to remove.
Returns:
Updated entry, or None if entry was deleted or didn't exist.
"""
# Get current entry
result = await self._db.execute(
select(Collection).where(
Collection.user_id == user_id,
Collection.card_definition_id == card_definition_id,
)
)
model = result.scalar_one_or_none()
if model is None:
return None
new_quantity = model.quantity - quantity
if new_quantity <= 0:
# Delete the entry
await self._db.execute(
delete(Collection).where(
Collection.user_id == user_id,
Collection.card_definition_id == card_definition_id,
)
)
await self._db.commit()
return None
# Update quantity
model.quantity = new_quantity
await self._db.commit()
await self._db.refresh(model)
return _to_dto(model)
async def exists_with_source(self, user_id: UUID, source: CardSource) -> bool:
"""Check if user has any entries with the given source.
Args:
user_id: The user's UUID.
source: The CardSource to check for.
Returns:
True if any entries exist with that source.
"""
result = await self._db.execute(
select(Collection.id)
.where(
Collection.user_id == user_id,
Collection.source == source,
)
.limit(1)
)
return result.scalar_one_or_none() is not None

246
backend/app/repositories/postgres/deck.py Normal file
View File

@ -0,0 +1,246 @@
"""PostgreSQL implementation of DeckRepository.
This module provides the PostgreSQL-specific implementation of the
DeckRepository protocol using SQLAlchemy async sessions.
Example:
async with get_db_session() as db:
repo = PostgresDeckRepository(db)
decks = await repo.get_by_user(user_id)
"""
from uuid import UUID
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models.deck import Deck
from app.repositories.protocols import DeckEntry
def _to_dto(model: Deck) -> DeckEntry:
"""Convert ORM model to DTO."""
return DeckEntry(
id=model.id,
user_id=model.user_id,
name=model.name,
cards=model.cards or {},
energy_cards=model.energy_cards or {},
is_valid=model.is_valid,
validation_errors=model.validation_errors,
is_starter=model.is_starter,
starter_type=model.starter_type,
description=model.description,
created_at=model.created_at,
updated_at=model.updated_at,
)
class PostgresDeckRepository:
"""PostgreSQL implementation of DeckRepository.
Uses SQLAlchemy async sessions for database access.
Attributes:
_db: The async database session.
"""
def __init__(self, db: AsyncSession) -> None:
"""Initialize with database session.
Args:
db: SQLAlchemy async session.
"""
self._db = db
async def get_by_id(self, deck_id: UUID) -> DeckEntry | None:
"""Get a deck by its ID.
Args:
deck_id: The deck's UUID.
Returns:
DeckEntry if found, None otherwise.
"""
result = await self._db.execute(select(Deck).where(Deck.id == deck_id))
model = result.scalar_one_or_none()
return _to_dto(model) if model else None
async def get_by_user(self, user_id: UUID) -> list[DeckEntry]:
"""Get all decks for a user.
Args:
user_id: The user's UUID.
Returns:
List of all user's decks, ordered by name.
"""
result = await self._db.execute(
select(Deck).where(Deck.user_id == user_id).order_by(Deck.name)
)
return [_to_dto(model) for model in result.scalars().all()]
async def get_user_deck(self, user_id: UUID, deck_id: UUID) -> DeckEntry | None:
"""Get a specific deck owned by a user.
Combines ownership check with retrieval.
Args:
user_id: The user's UUID.
deck_id: The deck's UUID.
Returns:
DeckEntry if found and owned by user, None otherwise.
"""
result = await self._db.execute(
select(Deck).where(
Deck.id == deck_id,
Deck.user_id == user_id,
)
)
model = result.scalar_one_or_none()
return _to_dto(model) if model else None
async def count_by_user(self, user_id: UUID) -> int:
"""Count how many decks a user has.
Args:
user_id: The user's UUID.
Returns:
Number of decks owned by user.
"""
result = await self._db.execute(select(func.count(Deck.id)).where(Deck.user_id == user_id))
return result.scalar_one()
async def create(
self,
user_id: UUID,
name: str,
cards: dict[str, int],
energy_cards: dict[str, int],
is_valid: bool,
validation_errors: list[str] | None,
is_starter: bool = False,
starter_type: str | None = None,
description: str | None = None,
) -> DeckEntry:
"""Create a new deck.
Args:
user_id: The user's UUID.
name: Display name for the deck.
cards: Card ID to quantity mapping.
energy_cards: Energy type to quantity mapping.
is_valid: Whether deck passes validation.
validation_errors: List of validation error messages.
is_starter: Whether this is a starter deck.
starter_type: Type of starter deck if applicable.
description: Optional deck description.
Returns:
The created DeckEntry.
"""
deck = Deck(
user_id=user_id,
name=name,
cards=cards,
energy_cards=energy_cards,
is_valid=is_valid,
validation_errors=validation_errors,
is_starter=is_starter,
starter_type=starter_type,
description=description,
)
self._db.add(deck)
await self._db.commit()
await self._db.refresh(deck)
return _to_dto(deck)
async def update(
self,
deck_id: UUID,
name: str | None = None,
cards: dict[str, int] | None = None,
energy_cards: dict[str, int] | None = None,
is_valid: bool | None = None,
validation_errors: list[str] | None = None,
description: str | None = None,
) -> DeckEntry | None:
"""Update an existing deck.
Only provided (non-None) fields are updated.
Args:
deck_id: The deck's UUID.
name: New name (optional).
cards: New card composition (optional).
energy_cards: New energy composition (optional).
is_valid: New validation status (optional).
validation_errors: New validation errors (optional).
description: New description (optional).
Returns:
Updated DeckEntry, or None if deck not found.
"""
result = await self._db.execute(select(Deck).where(Deck.id == deck_id))
deck = result.scalar_one_or_none()
if deck is None:
return None
if name is not None:
deck.name = name
if cards is not None:
deck.cards = cards
if energy_cards is not None:
deck.energy_cards = energy_cards
if is_valid is not None:
deck.is_valid = is_valid
if validation_errors is not None:
deck.validation_errors = validation_errors
if description is not None:
deck.description = description
await self._db.commit()
await self._db.refresh(deck)
return _to_dto(deck)
async def delete(self, deck_id: UUID) -> bool:
"""Delete a deck.
Args:
deck_id: The deck's UUID.
Returns:
True if deleted, False if not found.
"""
result = await self._db.execute(select(Deck).where(Deck.id == deck_id))
deck = result.scalar_one_or_none()
if deck is None:
return False
await self._db.delete(deck)
await self._db.commit()
return True
async def has_starter(self, user_id: UUID) -> tuple[bool, str | None]:
"""Check if user has a starter deck.
Args:
user_id: The user's UUID.
Returns:
Tuple of (has_starter, starter_type).
"""
result = await self._db.execute(
select(Deck.starter_type)
.where(
Deck.user_id == user_id,
Deck.is_starter == True, # noqa: E712
)
.limit(1)
)
starter_type = result.scalar_one_or_none()
return (starter_type is not None, starter_type)

316
backend/app/repositories/protocols.py Normal file
View File

@ -0,0 +1,316 @@
"""Repository protocol definitions for Mantimon TCG.
This module defines the abstract interfaces (Protocols) for data access.
Concrete implementations can be PostgreSQL, SQLite, or in-memory storage.
The protocols define WHAT operations are available, not HOW they're implemented.
This enables the offline fork to implement LocalCollectionRepository while
the backend uses PostgresCollectionRepository.
Example:
class CollectionRepository(Protocol):
async def get_all(self, user_id: UUID) -> list[CollectionEntry]: ...
# PostgreSQL implementation
class PostgresCollectionRepository:
def __init__(self, db: AsyncSession): ...
async def get_all(self, user_id: UUID) -> list[CollectionEntry]: ...
# Local/offline implementation
class LocalCollectionRepository:
def __init__(self, storage_path: Path): ...
async def get_all(self, user_id: UUID) -> list[CollectionEntry]: ...
"""
from dataclasses import dataclass
from datetime import datetime
from typing import Protocol
from uuid import UUID
from app.db.models.collection import CardSource
# =============================================================================
# Data Transfer Objects (DTOs)
# =============================================================================
# These are storage-agnostic representations used by protocols.
# They decouple the service layer from ORM models.
@dataclass
class CollectionEntry:
"""Storage-agnostic representation of a collection entry.
This DTO decouples the service layer from the ORM model,
allowing different storage backends to return the same structure.
"""
id: UUID
user_id: UUID
card_definition_id: str
quantity: int
source: CardSource
obtained_at: datetime
created_at: datetime
updated_at: datetime
@dataclass
class DeckEntry:
"""Storage-agnostic representation of a deck.
This DTO decouples the service layer from the ORM model.
"""
id: UUID
user_id: UUID
name: str
cards: dict[str, int]
energy_cards: dict[str, int]
is_valid: bool
validation_errors: list[str] | None
is_starter: bool
starter_type: str | None
description: str | None
created_at: datetime
updated_at: datetime
# =============================================================================
# Repository Protocols
# =============================================================================
class CollectionRepository(Protocol):
"""Protocol for card collection data access.
Implementations handle storage-specific details (PostgreSQL, SQLite, JSON).
Services use this protocol for business logic without knowing storage details.
All methods are async to support both database and file-based storage.
"""
async def get_all(self, user_id: UUID) -> list[CollectionEntry]:
"""Get all collection entries for a user.
Args:
user_id: The user's UUID.
Returns:
List of all collection entries, ordered by card_definition_id.
"""
...
async def get_by_card(self, user_id: UUID, card_definition_id: str) -> CollectionEntry | None:
"""Get a specific collection entry.
Args:
user_id: The user's UUID.
card_definition_id: The card ID to look up.
Returns:
CollectionEntry if exists, None otherwise.
"""
...
async def get_quantity(self, user_id: UUID, card_definition_id: str) -> int:
"""Get quantity of a specific card owned by user.
Args:
user_id: The user's UUID.
card_definition_id: Card ID to check.
Returns:
Number of copies owned (0 if not owned).
"""
...
async def upsert(
self,
user_id: UUID,
card_definition_id: str,
quantity: int,
source: CardSource,
) -> CollectionEntry:
"""Add or update a collection entry.
If entry exists, increments quantity. Otherwise creates new entry.
Args:
user_id: The user's UUID.
card_definition_id: Card ID to add.
quantity: Number of copies to add.
source: How the cards were obtained.
Returns:
The created or updated CollectionEntry.
"""
...
async def decrement(
self,
user_id: UUID,
card_definition_id: str,
quantity: int,
) -> CollectionEntry | None:
"""Decrement quantity of a collection entry.
If quantity reaches 0, deletes the entry.
Args:
user_id: The user's UUID.
card_definition_id: Card ID to decrement.
quantity: Number of copies to remove.
Returns:
Updated entry, or None if entry was deleted or didn't exist.
"""
...
async def exists_with_source(self, user_id: UUID, source: CardSource) -> bool:
"""Check if user has any entries with the given source.
Useful for checking if user has received a starter deck.
Args:
user_id: The user's UUID.
source: The CardSource to check for.
Returns:
True if any entries exist with that source.
"""
...
class DeckRepository(Protocol):
"""Protocol for deck data access.
Implementations handle storage-specific details (PostgreSQL, SQLite, JSON).
Services use this protocol for business logic without knowing storage details.
"""
async def get_by_id(self, deck_id: UUID) -> DeckEntry | None:
"""Get a deck by its ID.
Args:
deck_id: The deck's UUID.
Returns:
DeckEntry if found, None otherwise.
"""
...
async def get_by_user(self, user_id: UUID) -> list[DeckEntry]:
"""Get all decks for a user.
Args:
user_id: The user's UUID.
Returns:
List of all user's decks, ordered by name.
"""
...
async def get_user_deck(self, user_id: UUID, deck_id: UUID) -> DeckEntry | None:
"""Get a specific deck owned by a user.
Combines ownership check with retrieval.
Args:
user_id: The user's UUID.
deck_id: The deck's UUID.
Returns:
DeckEntry if found and owned by user, None otherwise.
"""
...
async def count_by_user(self, user_id: UUID) -> int:
"""Count how many decks a user has.
Args:
user_id: The user's UUID.
Returns:
Number of decks owned by user.
"""
...
async def create(
self,
user_id: UUID,
name: str,
cards: dict[str, int],
energy_cards: dict[str, int],
is_valid: bool,
validation_errors: list[str] | None,
is_starter: bool = False,
starter_type: str | None = None,
description: str | None = None,
) -> DeckEntry:
"""Create a new deck.
Args:
user_id: The user's UUID.
name: Display name for the deck.
cards: Card ID to quantity mapping.
energy_cards: Energy type to quantity mapping.
is_valid: Whether deck passes validation.
validation_errors: List of validation error messages.
is_starter: Whether this is a starter deck.
starter_type: Type of starter deck if applicable.
description: Optional deck description.
Returns:
The created DeckEntry.
"""
...
async def update(
self,
deck_id: UUID,
name: str | None = None,
cards: dict[str, int] | None = None,
energy_cards: dict[str, int] | None = None,
is_valid: bool | None = None,
validation_errors: list[str] | None = None,
description: str | None = None,
) -> DeckEntry | None:
"""Update an existing deck.
Only provided (non-None) fields are updated.
Args:
deck_id: The deck's UUID.
name: New name (optional).
cards: New card composition (optional).
energy_cards: New energy composition (optional).
is_valid: New validation status (optional).
validation_errors: New validation errors (optional).
description: New description (optional).
Returns:
Updated DeckEntry, or None if deck not found.
"""
...
async def delete(self, deck_id: UUID) -> bool:
"""Delete a deck.
Args:
deck_id: The deck's UUID.
Returns:
True if deleted, False if not found.
"""
...
async def has_starter(self, user_id: UUID) -> tuple[bool, str | None]:
"""Check if user has a starter deck.
Args:
user_id: The user's UUID.
Returns:
Tuple of (has_starter, starter_type).
"""
...

30
backend/app/schemas/__init__.py
View File

@ -10,6 +10,22 @@ from app.schemas.auth import (
TokenResponse,
TokenType,
)
from app.schemas.collection import (
CollectionAddRequest,
CollectionCardResponse,
CollectionEntryResponse,
CollectionResponse,
)
from app.schemas.deck import (
DeckCreateRequest,
DeckListResponse,
DeckResponse,
DeckUpdateRequest,
DeckValidateRequest,
DeckValidationResponse,
StarterDeckSelectRequest,
StarterStatusResponse,
)
from app.schemas.user import (
OAuthUserInfo,
UserCreate,
@ -24,6 +40,20 @@ __all__ = [
"TokenResponse",
"RefreshTokenRequest",
"OAuthState",
# Collection schemas
"CollectionEntryResponse",
"CollectionResponse",
"CollectionAddRequest",
"CollectionCardResponse",
# Deck schemas
"DeckCreateRequest",
"DeckUpdateRequest",
"DeckResponse",
"DeckListResponse",
"DeckValidateRequest",
"DeckValidationResponse",
"StarterDeckSelectRequest",
"StarterStatusResponse",
# User schemas
"UserResponse",
"UserCreate",

87
backend/app/schemas/collection.py Normal file
View File

@ -0,0 +1,87 @@
"""Collection schemas for Mantimon TCG.
This module defines Pydantic models for collection-related API requests
and responses. Collections track which cards a user owns.
Example:
entry = CollectionEntryResponse(
card_definition_id="a1-001-bulbasaur",
quantity=3,
source=CardSource.BOOSTER,
obtained_at=datetime.now(UTC)
)
"""
from datetime import datetime
from pydantic import BaseModel, Field
from app.db.models.collection import CardSource
class CollectionEntryResponse(BaseModel):
"""Response model for a single collection entry.
Represents one card type in a user's collection with quantity.
Attributes:
card_definition_id: ID of the card definition (e.g., "a1-001-bulbasaur").
quantity: Number of copies owned.
source: How the first copy was obtained.
obtained_at: When the card was first added to collection.
"""
card_definition_id: str = Field(..., description="Card definition ID")
quantity: int = Field(..., ge=1, description="Number of copies owned")
source: CardSource = Field(..., description="How the card was obtained")
obtained_at: datetime = Field(..., description="When first obtained")
model_config = {"from_attributes": True}
class CollectionResponse(BaseModel):
"""Response model for a user's full collection.
Contains aggregate statistics and list of all owned cards.
Attributes:
total_unique_cards: Number of distinct card types owned.
total_card_count: Total number of cards (sum of all quantities).
entries: List of all collection entries.
"""
total_unique_cards: int = Field(..., ge=0, description="Distinct card types owned")
total_card_count: int = Field(..., ge=0, description="Total cards owned")
entries: list[CollectionEntryResponse] = Field(
default_factory=list, description="All collection entries"
)
class CollectionAddRequest(BaseModel):
"""Request model for adding cards to a collection.
Used by admin endpoints to grant cards to users.
Attributes:
card_definition_id: ID of the card to add.
quantity: Number of copies to add (default 1).
source: How the card was obtained.
"""
card_definition_id: str = Field(..., description="Card definition ID to add")
quantity: int = Field(default=1, ge=1, le=99, description="Number of copies to add")
source: CardSource = Field(..., description="Source of the card")
class CollectionCardResponse(BaseModel):
"""Response model for a single card lookup in collection.
Returns quantity for a specific card in a user's collection.
Attributes:
card_definition_id: ID of the card.
quantity: Number of copies owned (0 if not owned).
"""
card_definition_id: str = Field(..., description="Card definition ID")
quantity: int = Field(..., ge=0, description="Number of copies owned")

155
backend/app/schemas/deck.py Normal file
View File

@ -0,0 +1,155 @@
"""Deck schemas for Mantimon TCG.
This module defines Pydantic models for deck-related API requests
and responses. Decks contain card compositions for gameplay.
Example:
deck = DeckResponse(
id=uuid4(),
name="Electric Storm",
cards={"a1-094-pikachu": 4, "a1-095-raichu": 2},
energy_cards={"lightning": 14, "colorless": 6},
is_valid=True,
validation_errors=None,
is_starter=False,
starter_type=None,
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC)
)
"""
from datetime import datetime
from uuid import UUID
from pydantic import BaseModel, Field
class DeckCreateRequest(BaseModel):
"""Request model for creating a new deck.
Attributes:
name: Display name for the deck.
cards: Mapping of card IDs to quantities (40 cards total).
energy_cards: Mapping of energy types to quantities (20 total).
"""
name: str = Field(..., min_length=1, max_length=100, description="Deck name")
cards: dict[str, int] = Field(..., description="Card ID to quantity mapping")
energy_cards: dict[str, int] = Field(..., description="Energy type to quantity mapping")
class DeckUpdateRequest(BaseModel):
"""Request model for updating a deck.
All fields are optional - only provided fields are updated.
Attributes:
name: New display name for the deck.
cards: New card composition.
energy_cards: New energy composition.
"""
name: str | None = Field(
default=None, min_length=1, max_length=100, description="New deck name"
)
cards: dict[str, int] | None = Field(default=None, description="New card composition")
energy_cards: dict[str, int] | None = Field(default=None, description="New energy composition")
class DeckResponse(BaseModel):
"""Response model for a deck.
Includes the full deck composition and validation state.
Attributes:
id: Unique deck identifier.
name: Display name of the deck.
cards: Mapping of card IDs to quantities.
energy_cards: Mapping of energy types to quantities.
is_valid: Whether deck passes all validation rules.
validation_errors: List of validation error messages (if any).
is_starter: Whether this is a starter deck.
starter_type: Type of starter deck (grass, fire, etc.) if applicable.
created_at: When the deck was created.
updated_at: When the deck was last modified.
"""
id: UUID = Field(..., description="Deck ID")
name: str = Field(..., description="Deck name")
cards: dict[str, int] = Field(..., description="Card ID to quantity mapping")
energy_cards: dict[str, int] = Field(..., description="Energy type to quantity mapping")
is_valid: bool = Field(..., description="Whether deck is valid")
validation_errors: list[str] | None = Field(
default=None, description="Validation error messages"
)
is_starter: bool = Field(default=False, description="Is starter deck")
starter_type: str | None = Field(default=None, description="Starter deck type")
created_at: datetime = Field(..., description="Creation timestamp")
updated_at: datetime = Field(..., description="Last update timestamp")
model_config = {"from_attributes": True}
class DeckListResponse(BaseModel):
"""Response model for listing user's decks.
Attributes:
decks: List of user's decks.
deck_count: Number of decks the user has.
deck_limit: Maximum decks allowed (None for unlimited/premium).
"""
decks: list[DeckResponse] = Field(default_factory=list, description="User's decks")
deck_count: int = Field(..., ge=0, description="Number of decks")
deck_limit: int | None = Field(default=None, description="Max decks (None = unlimited)")
class DeckValidateRequest(BaseModel):
"""Request model for validating a deck without saving.
Used to check if a deck composition is valid before creating it.
Attributes:
cards: Card ID to quantity mapping to validate.
energy_cards: Energy type to quantity mapping to validate.
"""
cards: dict[str, int] = Field(..., description="Card ID to quantity mapping")
energy_cards: dict[str, int] = Field(..., description="Energy type to quantity mapping")
class DeckValidationResponse(BaseModel):
"""Response model for deck validation results.
Attributes:
is_valid: Whether the deck passes all validation rules.
errors: List of validation error messages.
"""
is_valid: bool = Field(..., description="Whether deck is valid")
errors: list[str] = Field(default_factory=list, description="Validation errors")
class StarterDeckSelectRequest(BaseModel):
"""Request model for selecting a starter deck.
Attributes:
starter_type: Type of starter deck to select.
"""
starter_type: str = Field(
...,
description="Starter deck type (grass, fire, water, psychic, lightning)",
)
class StarterStatusResponse(BaseModel):
"""Response model for starter deck status.
Attributes:
has_starter: Whether user has selected a starter deck.
starter_type: Type of starter deck selected (if any).
"""
has_starter: bool = Field(..., description="Has starter been selected")
starter_type: str | None = Field(default=None, description="Selected starter type")

272
backend/app/services/collection_service.py Normal file
View File

@ -0,0 +1,272 @@
"""Collection service for Mantimon TCG.
This module provides business logic for user card collections. It uses
the CollectionRepository protocol for data access, enabling easy testing
and multiple storage backends (PostgreSQL, SQLite, local files).
The service layer handles:
- Card ID validation via CardService
- Starter deck granting logic
- Collection statistics
Example:
from app.services.card_service import CardService
from app.services.collection_service import CollectionService
from app.repositories.postgres import PostgresCollectionRepository
# Create dependencies
card_service = CardService()
card_service.load_all()
# Create repository and service
repo = PostgresCollectionRepository(db_session)
service = CollectionService(repo, card_service)
# Add cards to collection
entry = await service.add_cards(
user_id, "a1-001-bulbasaur", quantity=2, source=CardSource.BOOSTER
)
"""
from uuid import UUID
from app.db.models.collection import CardSource
from app.repositories.protocols import CollectionEntry, CollectionRepository
from app.services.card_service import CardService
class CollectionService:
"""Service for card collection business logic.
Uses repository pattern for data access, enabling:
- Easy unit testing with mock repositories
- Multiple storage backends
- Offline fork support
Attributes:
_repo: The collection repository implementation.
_card_service: The card service for card validation.
"""
def __init__(self, repository: CollectionRepository, card_service: CardService) -> None:
"""Initialize with dependencies.
Args:
repository: Implementation of CollectionRepository protocol.
card_service: Card service for validating card IDs.
"""
self._repo = repository
self._card_service = card_service
async def get_collection(self, user_id: UUID) -> list[CollectionEntry]:
"""Get all cards in a user's collection.
Args:
user_id: The user's UUID.
Returns:
List of CollectionEntry for the user.
Example:
collection = await service.get_collection(user_id)
for entry in collection:
print(f"{entry.card_definition_id}: {entry.quantity}")
"""
return await self._repo.get_all(user_id)
async def get_card_quantity(self, user_id: UUID, card_definition_id: str) -> int:
"""Get quantity of a specific card owned by user.
Args:
user_id: The user's UUID.
card_definition_id: Card ID to check.
Returns:
Number of copies owned (0 if not owned).
"""
return await self._repo.get_quantity(user_id, card_definition_id)
async def get_collection_entry(
self, user_id: UUID, card_definition_id: str
) -> CollectionEntry | None:
"""Get a specific collection entry.
Args:
user_id: The user's UUID.
card_definition_id: Card ID to look up.
Returns:
CollectionEntry if exists, None otherwise.
"""
return await self._repo.get_by_card(user_id, card_definition_id)
async def add_cards(
self,
user_id: UUID,
card_definition_id: str,
quantity: int,
source: CardSource,
) -> CollectionEntry:
"""Add cards to a user's collection.
Validates that the card ID exists before adding. Uses upsert
pattern: creates new entry if card not owned, or increments
quantity if already owned.
Args:
user_id: The user's UUID.
card_definition_id: Card ID to add.
quantity: Number of copies to add.
source: How the cards were obtained.
Returns:
The created or updated CollectionEntry.
Raises:
ValueError: If card_definition_id doesn't exist.
Example:
entry = await service.add_cards(
user_id, "a1-001-bulbasaur",
quantity=2, source=CardSource.BOOSTER
)
"""
# Validate card exists
if self._card_service.get_card(card_definition_id) is None:
raise ValueError(f"Invalid card ID: {card_definition_id}")
return await self._repo.upsert(user_id, card_definition_id, quantity, source)
async def remove_cards(
self,
user_id: UUID,
card_definition_id: str,
quantity: int,
) -> CollectionEntry | None:
"""Remove cards from a user's collection.
Decrements quantity. If quantity reaches 0, deletes the entry.
Args:
user_id: The user's UUID.
card_definition_id: Card ID to remove.
quantity: Number of copies to remove.
Returns:
Updated CollectionEntry, or None if card not owned
or all copies were removed.
"""
return await self._repo.decrement(user_id, card_definition_id, quantity)
async def has_cards(
self,
user_id: UUID,
card_requirements: dict[str, int],
) -> bool:
"""Check if user owns at least the required quantity of each card.
Args:
user_id: The user's UUID.
card_requirements: Mapping of card IDs to required quantities.
Returns:
True if user owns enough of all cards, False otherwise.
Example:
can_build = await service.has_cards(
user_id, {"a1-001-bulbasaur": 4, "a1-002-ivysaur": 2}
)
"""
for card_id, required_qty in card_requirements.items():
owned_qty = await self._repo.get_quantity(user_id, card_id)
if owned_qty < required_qty:
return False
return True
async def get_owned_cards_dict(self, user_id: UUID) -> dict[str, int]:
"""Get user's collection as a card_id -> quantity mapping.
Useful for deck validation in campaign mode.
Args:
user_id: The user's UUID.
Returns:
Dictionary mapping card IDs to quantities.
Example:
owned = await service.get_owned_cards_dict(user_id)
# Pass to DeckValidator
result = validator.validate_deck(cards, energy, owned_cards=owned)
"""
collection = await self._repo.get_all(user_id)
return {entry.card_definition_id: entry.quantity for entry in collection}
async def grant_starter_deck(
self,
user_id: UUID,
starter_type: str,
) -> list[CollectionEntry]:
"""Grant all cards from a starter deck to user's collection.
Uses CardSource.STARTER for all granted cards.
Args:
user_id: The user's UUID.
starter_type: Type of starter deck (grass, fire, water, etc.).
Returns:
List of CollectionEntry created/updated.
Raises:
ValueError: If starter_type is invalid.
Example:
entries = await service.grant_starter_deck(user_id, "grass")
"""
# Import here to avoid circular dependency
from app.data.starter_decks import STARTER_TYPES, get_starter_deck
if starter_type not in STARTER_TYPES:
raise ValueError(
f"Invalid starter type: {starter_type}. "
f"Must be one of: {', '.join(STARTER_TYPES)}"
)
starter_deck = get_starter_deck(starter_type)
entries: list[CollectionEntry] = []
# Add all cards from the deck
for card_id, quantity in starter_deck["cards"].items():
entry = await self.add_cards(user_id, card_id, quantity, CardSource.STARTER)
entries.append(entry)
return entries
async def has_starter_deck(self, user_id: UUID) -> bool:
"""Check if user has already received a starter deck.
Checks for any cards with STARTER source in collection.
Args:
user_id: The user's UUID.
Returns:
True if user has starter cards, False otherwise.
"""
return await self._repo.exists_with_source(user_id, CardSource.STARTER)
async def get_collection_stats(self, user_id: UUID) -> dict[str, int]:
"""Get aggregate statistics for user's collection.
Args:
user_id: The user's UUID.
Returns:
Dictionary with total_unique_cards and total_card_count.
"""
collection = await self._repo.get_all(user_id)
return {
"total_unique_cards": len(collection),
"total_card_count": sum(entry.quantity for entry in collection),
}

414
backend/app/services/deck_service.py Normal file
View File

@ -0,0 +1,414 @@
"""Deck service for Mantimon TCG.
This module provides business logic for deck management. It uses
the DeckRepository protocol for data access and DeckValidator for
validation logic.
The service layer handles:
- Deck slot limits (free vs premium users)
- Deck validation with optional ownership checking
- Starter deck creation
Example:
from app.core.config import DeckConfig
from app.services.card_service import CardService
from app.services.deck_service import DeckService
from app.services.deck_validator import DeckValidator
from app.repositories.postgres import PostgresDeckRepository, PostgresCollectionRepository
# Create dependencies
card_service = CardService()
card_service.load_all()
deck_validator = DeckValidator(DeckConfig(), card_service)
# Create repositories
deck_repo = PostgresDeckRepository(db_session)
collection_repo = PostgresCollectionRepository(db_session)
# Create service with all dependencies
service = DeckService(deck_repo, deck_validator, card_service, collection_repo)
# Create a deck
deck = await service.create_deck(
user_id=user_id,
name="My Deck",
cards={"a1-001-bulbasaur": 4, ...},
energy_cards={"grass": 14, "colorless": 6},
max_decks=5, # From user.max_decks
)
"""
from uuid import UUID
from app.core.models.card import CardDefinition
from app.repositories.protocols import (
CollectionRepository,
DeckEntry,
DeckRepository,
)
from app.services.card_service import CardService
from app.services.deck_validator import DeckValidationResult, DeckValidator
class DeckLimitExceededError(Exception):
"""Raised when user tries to create more decks than allowed."""
pass
class DeckNotFoundError(Exception):
"""Raised when deck is not found or not owned by user."""
pass
class DeckService:
"""Service for deck business logic.
Uses repository pattern for data access, enabling:
- Easy unit testing with mock repositories
- Multiple storage backends
- Offline fork support
Attributes:
_deck_repo: The deck repository implementation.
_collection_repo: The collection repository (for ownership checks).
_deck_validator: The deck validator for validation logic.
_card_service: The card service for card lookups.
"""
def __init__(
self,
deck_repository: DeckRepository,
deck_validator: DeckValidator,
card_service: CardService,
collection_repository: CollectionRepository | None = None,
) -> None:
"""Initialize with dependencies.
Args:
deck_repository: Implementation of DeckRepository protocol.
deck_validator: Validator for deck compositions.
card_service: Card service for looking up card definitions.
collection_repository: Implementation of CollectionRepository protocol.
Required for ownership validation in campaign mode.
"""
self._deck_repo = deck_repository
self._deck_validator = deck_validator
self._card_service = card_service
self._collection_repo = collection_repository
async def create_deck(
self,
user_id: UUID,
name: str,
cards: dict[str, int],
energy_cards: dict[str, int],
max_decks: int,
validate_ownership: bool = True,
is_starter: bool = False,
starter_type: str | None = None,
description: str | None = None,
) -> DeckEntry:
"""Create a new deck.
Validates the deck and stores validation results. Invalid decks
CAN be saved (with errors) to support work-in-progress decks.
Args:
user_id: The user's UUID.
name: Display name for the deck.
cards: Card ID to quantity mapping.
energy_cards: Energy type to quantity mapping.
max_decks: Maximum decks allowed (from user.max_decks).
validate_ownership: If True, checks card ownership (campaign mode).
is_starter: Whether this is a starter deck.
starter_type: Type of starter deck if applicable.
description: Optional deck description.
Returns:
The created DeckEntry.
Raises:
DeckLimitExceededError: If user has reached deck limit.
Example:
deck = await service.create_deck(
user_id=user_id,
name="Grass Power",
cards={"a1-001-bulbasaur": 4, ...},
energy_cards={"grass": 14, "colorless": 6},
max_decks=5,
)
"""
# Check deck limit
current_count = await self._deck_repo.count_by_user(user_id)
if current_count >= max_decks:
raise DeckLimitExceededError(
f"Deck limit reached ({current_count}/{max_decks}). "
"Upgrade to premium for unlimited decks."
)
# Validate deck
validation = await self.validate_deck(
cards, energy_cards, user_id if validate_ownership else None
)
return await self._deck_repo.create(
user_id=user_id,
name=name,
cards=cards,
energy_cards=energy_cards,
is_valid=validation.is_valid,
validation_errors=validation.errors if validation.errors else None,
is_starter=is_starter,
starter_type=starter_type,
description=description,
)
async def update_deck(
self,
user_id: UUID,
deck_id: UUID,
name: str | None = None,
cards: dict[str, int] | None = None,
energy_cards: dict[str, int] | None = None,
validate_ownership: bool = True,
description: str | None = None,
) -> DeckEntry:
"""Update an existing deck.
Re-validates if cards or energy_cards change.
Args:
user_id: The user's UUID (for ownership verification).
deck_id: The deck's UUID.
name: New name (optional).
cards: New card composition (optional).
energy_cards: New energy composition (optional).
validate_ownership: If True, checks card ownership (campaign mode).
description: New description (optional).
Returns:
The updated DeckEntry.
Raises:
DeckNotFoundError: If deck not found or not owned by user.
"""
# Verify ownership
deck = await self._deck_repo.get_user_deck(user_id, deck_id)
if deck is None:
raise DeckNotFoundError(f"Deck {deck_id} not found or not owned by user")
# Determine if we need to re-validate
needs_revalidation = cards is not None or energy_cards is not None
# Use existing values if not provided
final_cards = cards if cards is not None else deck.cards
final_energy = energy_cards if energy_cards is not None else deck.energy_cards
# Re-validate if needed
is_valid = deck.is_valid
validation_errors = deck.validation_errors
if needs_revalidation:
validation = await self.validate_deck(
final_cards, final_energy, user_id if validate_ownership else None
)
is_valid = validation.is_valid
validation_errors = validation.errors if validation.errors else None
result = await self._deck_repo.update(
deck_id=deck_id,
name=name,
cards=cards,
energy_cards=energy_cards,
is_valid=is_valid,
validation_errors=validation_errors,
description=description,
)
if result is None:
raise DeckNotFoundError(f"Deck {deck_id} not found")
return result
async def delete_deck(self, user_id: UUID, deck_id: UUID) -> bool:
"""Delete a deck.
Args:
user_id: The user's UUID (for ownership verification).
deck_id: The deck's UUID.
Returns:
True if deleted.
Raises:
DeckNotFoundError: If deck not found or not owned by user.
"""
# Verify ownership
deck = await self._deck_repo.get_user_deck(user_id, deck_id)
if deck is None:
raise DeckNotFoundError(f"Deck {deck_id} not found or not owned by user")
return await self._deck_repo.delete(deck_id)
async def get_deck(self, user_id: UUID, deck_id: UUID) -> DeckEntry:
"""Get a deck owned by user.
Args:
user_id: The user's UUID.
deck_id: The deck's UUID.
Returns:
The DeckEntry.
Raises:
DeckNotFoundError: If deck not found or not owned by user.
"""
deck = await self._deck_repo.get_user_deck(user_id, deck_id)
if deck is None:
raise DeckNotFoundError(f"Deck {deck_id} not found or not owned by user")
return deck
async def get_user_decks(self, user_id: UUID) -> list[DeckEntry]:
"""Get all decks for a user.
Args:
user_id: The user's UUID.
Returns:
List of all user's decks.
"""
return await self._deck_repo.get_by_user(user_id)
async def can_create_deck(self, user_id: UUID, max_decks: int) -> bool:
"""Check if user can create another deck.
Args:
user_id: The user's UUID.
max_decks: Maximum decks allowed (from user.max_decks).
Returns:
True if user can create another deck.
"""
current_count = await self._deck_repo.count_by_user(user_id)
return current_count < max_decks
async def get_deck_count(self, user_id: UUID) -> int:
"""Get number of decks user has.
Args:
user_id: The user's UUID.
Returns:
Number of decks.
"""
return await self._deck_repo.count_by_user(user_id)
async def validate_deck(
self,
cards: dict[str, int],
energy_cards: dict[str, int],
user_id: UUID | None = None,
) -> DeckValidationResult:
"""Validate a deck composition.
Args:
cards: Card ID to quantity mapping.
energy_cards: Energy type to quantity mapping.
user_id: If provided, validates card ownership (campaign mode).
Pass None for freeplay mode.
Returns:
DeckValidationResult with is_valid and errors.
"""
owned_cards: dict[str, int] | None = None
if user_id is not None and self._collection_repo is not None:
# Get user's collection for ownership validation
collection = await self._collection_repo.get_all(user_id)
owned_cards = {entry.card_definition_id: entry.quantity for entry in collection}
return self._deck_validator.validate_deck(cards, energy_cards, owned_cards)
async def get_deck_for_game(self, user_id: UUID, deck_id: UUID) -> list[CardDefinition]:
"""Expand a deck to a list of CardDefinitions for game use.
Used by GameEngine to create a game from a deck.
Args:
user_id: The user's UUID.
deck_id: The deck's UUID.
Returns:
List of CardDefinition objects for each card in the deck
(duplicates included based on quantity).
Raises:
DeckNotFoundError: If deck not found or not owned by user.
"""
deck = await self.get_deck(user_id, deck_id)
cards: list[CardDefinition] = []
for card_id, quantity in deck.cards.items():
card_def = self._card_service.get_card(card_id)
if card_def is not None:
cards.extend([card_def] * quantity)
return cards
async def has_starter_deck(self, user_id: UUID) -> tuple[bool, str | None]:
"""Check if user has a starter deck.
Args:
user_id: The user's UUID.
Returns:
Tuple of (has_starter, starter_type).
"""
return await self._deck_repo.has_starter(user_id)
async def create_starter_deck(
self,
user_id: UUID,
starter_type: str,
max_decks: int,
) -> DeckEntry:
"""Create a starter deck for a user.
This creates the deck but does NOT grant the cards to collection.
Use CollectionService.grant_starter_deck() to grant the cards.
Args:
user_id: The user's UUID.
starter_type: Type of starter deck (grass, fire, water, etc.).
max_decks: Maximum decks allowed (from user.max_decks).
Returns:
The created starter DeckEntry.
Raises:
ValueError: If starter_type is invalid.
DeckLimitExceededError: If user has reached deck limit.
"""
from app.data.starter_decks import STARTER_TYPES, get_starter_deck
if starter_type not in STARTER_TYPES:
raise ValueError(
f"Invalid starter type: {starter_type}. "
f"Must be one of: {', '.join(STARTER_TYPES)}"
)
starter = get_starter_deck(starter_type)
return await self.create_deck(
user_id=user_id,
name=starter["name"],
cards=starter["cards"],
energy_cards=starter["energy_cards"],
max_decks=max_decks,
validate_ownership=False, # Starter decks skip ownership check
is_starter=True,
starter_type=starter_type,
description=starter["description"],
)

229
backend/app/services/deck_validator.py Normal file
View File

@ -0,0 +1,229 @@
"""Deck validation service for Mantimon TCG.
This module provides standalone deck validation logic that can be used
without database dependencies. It validates deck compositions against
the game rules defined in DeckConfig.
The validator is separate from DeckService to allow:
- Unit testing without database
- Validation before saving (API /validate endpoint)
- Reuse across different contexts (import/export, AI deck building)
Usage:
from app.core.config import DeckConfig
from app.services.card_service import CardService
from app.services.deck_validator import DeckValidator, DeckValidationResult
card_service = CardService()
card_service.load_all()
validator = DeckValidator(DeckConfig(), card_service)
# Validate without ownership check (freeplay mode)
result = validator.validate_deck(cards, energy_cards)
# Validate with ownership check (campaign mode)
result = validator.validate_deck(cards, energy_cards, owned_cards=user_collection)
"""
from dataclasses import dataclass, field
from app.core.config import DeckConfig
from app.services.card_service import CardService
@dataclass
class DeckValidationResult:
"""Result of deck validation.
Contains validation status and all errors found. Multiple errors
can be returned to help the user fix all issues at once.
Attributes:
is_valid: Whether the deck passes all validation rules.
errors: List of human-readable error messages.
"""
is_valid: bool = True
errors: list[str] = field(default_factory=list)
def add_error(self, error: str) -> None:
"""Add an error and mark as invalid.
Args:
error: Human-readable error message.
"""
self.is_valid = False
self.errors.append(error)
class DeckValidator:
"""Validates deck compositions against game rules.
This validator checks:
1. Total card count (40 cards in main deck)
2. Total energy count (20 energy cards)
3. Maximum copies per card (4)
4. Minimum Basic Pokemon requirement (1)
5. Card ID validity (card must exist)
6. Card ownership (optional, for campaign mode)
The validator uses DeckConfig for rule values, allowing different
game modes to have different rules if needed.
Attributes:
_config: The deck configuration with validation rules.
_card_service: The card service for card lookups.
"""
def __init__(self, config: DeckConfig, card_service: CardService) -> None:
"""Initialize the validator with dependencies.
Args:
config: Deck configuration with validation rules.
card_service: Card service for looking up card definitions.
"""
self._config = config
self._card_service = card_service
@property
def config(self) -> DeckConfig:
"""Get the deck configuration."""
return self._config
def validate_deck(
self,
cards: dict[str, int],
energy_cards: dict[str, int],
owned_cards: dict[str, int] | None = None,
) -> DeckValidationResult:
"""Validate a deck composition.
Checks all validation rules and returns all errors found (not just
the first one). This helps users fix all issues at once.
Args:
cards: Mapping of card IDs to quantities for the main deck.
energy_cards: Mapping of energy type names to quantities.
owned_cards: If provided, validates that the user owns enough
copies of each card. Pass None to skip ownership validation
(for freeplay mode).
Returns:
DeckValidationResult with is_valid status and list of errors.
Example:
result = validator.validate_deck(
cards={"a1-001-bulbasaur": 4, "a1-002-ivysaur": 4, ...},
energy_cards={"grass": 14, "colorless": 6},
owned_cards={"a1-001-bulbasaur": 10, ...}
)
if not result.is_valid:
for error in result.errors:
print(error)
"""
result = DeckValidationResult()
# 1. Validate total card count
total_cards = sum(cards.values())
if total_cards != self._config.min_size:
result.add_error(
f"Main deck must have exactly {self._config.min_size} cards, " f"got {total_cards}"
)
# 2. Validate total energy count
total_energy = sum(energy_cards.values())
if total_energy != self._config.energy_deck_size:
result.add_error(
f"Energy deck must have exactly {self._config.energy_deck_size} cards, "
f"got {total_energy}"
)
# 3. Validate max copies per card
for card_id, quantity in cards.items():
if quantity > self._config.max_copies_per_card:
result.add_error(
f"Card '{card_id}' has {quantity} copies, "
f"max allowed is {self._config.max_copies_per_card}"
)
# 4 & 5. Validate card IDs exist and count Basic Pokemon
basic_pokemon_count = 0
invalid_card_ids: list[str] = []
for card_id in cards:
card_def = self._card_service.get_card(card_id)
if card_def is None:
invalid_card_ids.append(card_id)
elif card_def.is_basic_pokemon():
basic_pokemon_count += cards[card_id]
if invalid_card_ids:
# Limit displayed invalid IDs to avoid huge error messages
display_ids = invalid_card_ids[:5]
more = len(invalid_card_ids) - 5
error_msg = f"Invalid card IDs: {', '.join(display_ids)}"
if more > 0:
error_msg += f" (and {more} more)"
result.add_error(error_msg)
# Check minimum Basic Pokemon requirement
if basic_pokemon_count < self._config.min_basic_pokemon:
result.add_error(
f"Deck must have at least {self._config.min_basic_pokemon} Basic Pokemon, "
f"got {basic_pokemon_count}"
)
# 6. Validate ownership if owned_cards provided (campaign mode)
if owned_cards is not None:
insufficient_cards: list[tuple[str, int, int]] = []
for card_id, required_qty in cards.items():
owned_qty = owned_cards.get(card_id, 0)
if owned_qty < required_qty:
insufficient_cards.append((card_id, required_qty, owned_qty))
if insufficient_cards:
# Limit displayed insufficient cards
display_cards = insufficient_cards[:5]
more = len(insufficient_cards) - 5
error_parts = [f"'{c[0]}' (need {c[1]}, own {c[2]})" for c in display_cards]
error_msg = f"Insufficient cards: {', '.join(error_parts)}"
if more > 0:
error_msg += f" (and {more} more)"
result.add_error(error_msg)
return result
def validate_cards_exist(self, card_ids: list[str]) -> list[str]:
"""Check which card IDs are invalid.
Utility method to check card ID validity without full deck validation.
Args:
card_ids: List of card IDs to check.
Returns:
List of invalid card IDs (empty if all valid).
"""
invalid = []
for card_id in card_ids:
if self._card_service.get_card(card_id) is None:
invalid.append(card_id)
return invalid
def count_basic_pokemon(self, cards: dict[str, int]) -> int:
"""Count Basic Pokemon in a deck.
Utility method to count Basic Pokemon without full validation.
Args:
cards: Mapping of card IDs to quantities.
Returns:
Total number of Basic Pokemon cards in the deck.
"""
count = 0
for card_id, quantity in cards.items():
card_def = self._card_service.get_card(card_id)
if card_def and card_def.is_basic_pokemon():
count += quantity
return count

18
backend/project_plans/PHASE_3_COLLECTION_DECKS.json
View File

@ -9,8 +9,8 @@
"description": "Card ownership (collections), deck building, validation, and starter deck selection",
"totalEstimatedHours": 29,
"totalTasks": 14,
"completedTasks": 0,
"status": "not_started",
"completedTasks": 4,
"status": "in_progress",
"masterPlan": "../PROJECT_PLAN_MASTER.json"
},
@ -74,7 +74,7 @@
"description": "Define request/response models for collection operations",
"category": "critical",
"priority": 1,
"completed": false,
"completed": true,
"tested": false,
"dependencies": [],
"files": [
@ -96,7 +96,7 @@
"description": "Define request/response models for deck operations",
"category": "critical",
"priority": 2,
"completed": false,
"completed": true,
"tested": false,
"dependencies": [],
"files": [
@ -120,8 +120,8 @@
"description": "Standalone validation logic for deck rules - no DB dependency",
"category": "critical",
"priority": 3,
"completed": false,
"tested": false,
"completed": true,
"tested": true,
"dependencies": [],
"files": [
{"path": "app/services/deck_validator.py", "status": "create"}
@ -413,11 +413,11 @@
"description": "Unit tests for deck validation logic",
"category": "high",
"priority": 11,
"completed": false,
"tested": false,
"completed": true,
"tested": true,
"dependencies": ["COLL-003"],
"files": [
{"path": "tests/services/test_deck_validator.py", "status": "create"}
{"path": "tests/unit/services/test_deck_validator.py", "status": "create"}
],
"details": [
"Test valid deck passes all checks",

5
backend/tests/unit/__init__.py Normal file
View File

@ -0,0 +1,5 @@
"""Unit tests for Mantimon TCG.
This package contains pure unit tests that don't require database
or external services. Tests here use mocks and fixtures only.
"""

4
backend/tests/unit/services/__init__.py Normal file
View File

@ -0,0 +1,4 @@
"""Unit tests for Mantimon TCG services.
Tests in this package don't require database connections.
"""

841
backend/tests/unit/services/test_deck_validator.py Normal file
View File

@ -0,0 +1,841 @@
"""Tests for the DeckValidator service.
These tests verify deck validation logic without database dependencies.
CardService is mocked and injected to isolate the validation logic and
enable fast, deterministic unit tests.
The DeckValidator enforces Mantimon TCG house rules:
- 40 cards in main deck
- 20 energy cards in separate energy deck
- Max 4 copies of any single card
- At least 1 Basic Pokemon required
- Campaign mode: must own all cards in deck
- Freeplay mode: ownership validation skipped
"""
from unittest.mock import MagicMock
import pytest
from app.core.config import DeckConfig
from app.core.enums import CardType, EnergyType, PokemonStage
from app.core.models.card import CardDefinition
from app.services.deck_validator import (
DeckValidationResult,
DeckValidator,
)
# =============================================================================
# Test Fixtures
# =============================================================================
@pytest.fixture
def mock_card_service():
"""Create a mock CardService with test cards.
Provides a set of test cards:
- 3 Basic Pokemon (pikachu, bulbasaur, charmander)
- 2 Stage 1 Pokemon (raichu, ivysaur)
- 2 Trainer cards (potion, professor-oak)
This allows testing deck validation without loading real card data.
"""
service = MagicMock()
# Define test cards
cards = {
"test-001-pikachu": CardDefinition(
id="test-001-pikachu",
name="Pikachu",
card_type=CardType.POKEMON,
hp=60,
pokemon_type=EnergyType.LIGHTNING,
stage=PokemonStage.BASIC,
),
"test-002-bulbasaur": CardDefinition(
id="test-002-bulbasaur",
name="Bulbasaur",
card_type=CardType.POKEMON,
hp=70,
pokemon_type=EnergyType.GRASS,
stage=PokemonStage.BASIC,
),
"test-003-charmander": CardDefinition(
id="test-003-charmander",
name="Charmander",
card_type=CardType.POKEMON,
hp=60,
pokemon_type=EnergyType.FIRE,
stage=PokemonStage.BASIC,
),
"test-004-raichu": CardDefinition(
id="test-004-raichu",
name="Raichu",
card_type=CardType.POKEMON,
hp=100,
pokemon_type=EnergyType.LIGHTNING,
stage=PokemonStage.STAGE_1,
evolves_from="Pikachu",
),
"test-005-ivysaur": CardDefinition(
id="test-005-ivysaur",
name="Ivysaur",
card_type=CardType.POKEMON,
hp=90,
pokemon_type=EnergyType.GRASS,
stage=PokemonStage.STAGE_1,
evolves_from="Bulbasaur",
),
"test-101-potion": CardDefinition(
id="test-101-potion",
name="Potion",
card_type=CardType.TRAINER,
trainer_type="item",
),
"test-102-professor-oak": CardDefinition(
id="test-102-professor-oak",
name="Professor Oak",
card_type=CardType.TRAINER,
trainer_type="supporter",
),
}
service.get_card = lambda card_id: cards.get(card_id)
return service
@pytest.fixture
def valid_energy_deck() -> dict[str, int]:
"""Create a valid 20-card energy deck."""
return {
"lightning": 10,
"grass": 6,
"colorless": 4,
}
@pytest.fixture
def default_config() -> DeckConfig:
"""Create a default DeckConfig for testing."""
return DeckConfig()
@pytest.fixture
def validator(default_config, mock_card_service) -> DeckValidator:
"""Create a DeckValidator with default config and mock card service."""
return DeckValidator(default_config, mock_card_service)
def create_basic_pokemon_mock():
"""Create a mock that returns Basic Pokemon for any card ID.
Useful for tests that need all cards to be valid Basic Pokemon.
"""
mock = MagicMock()
mock.get_card = lambda cid: CardDefinition(
id=cid,
name=f"Card {cid}",
card_type=CardType.POKEMON,
hp=60,
pokemon_type=EnergyType.LIGHTNING,
stage=PokemonStage.BASIC,
)
return mock
# =============================================================================
# DeckValidationResult Tests
# =============================================================================
class TestDeckValidationResult:
"""Tests for the DeckValidationResult dataclass."""
def test_default_is_valid(self):
"""Test that a new result starts as valid with no errors.
A fresh DeckValidationResult should indicate validity until
errors are explicitly added.
"""
result = DeckValidationResult()
assert result.is_valid is True
assert result.errors == []
def test_add_error_marks_invalid(self):
"""Test that adding an error marks the result as invalid.
Once any error is added, is_valid should be False.
"""
result = DeckValidationResult()
result.add_error("Test error")
assert result.is_valid is False
assert "Test error" in result.errors
def test_add_multiple_errors(self):
"""Test that multiple errors can be accumulated.
All errors should be collected, not just the first one,
to give users complete feedback on what needs fixing.
"""
result = DeckValidationResult()
result.add_error("Error 1")
result.add_error("Error 2")
result.add_error("Error 3")
assert result.is_valid is False
assert len(result.errors) == 3
assert "Error 1" in result.errors
assert "Error 2" in result.errors
assert "Error 3" in result.errors
# =============================================================================
# Card Count Validation Tests
# =============================================================================
class TestCardCountValidation:
"""Tests for main deck card count validation (40 cards required)."""
def test_valid_card_count_passes(self, default_config):
"""Test that exactly 40 cards passes validation.
The main deck must have exactly 40 cards per Mantimon house rules.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)} # 40 cards
energy = {"lightning": 20}
result = validator.validate_deck(cards, energy)
# Should pass card count check
assert "must have exactly 40 cards" not in str(result.errors)
def test_39_cards_fails(self, default_config):
"""Test that 39 cards fails validation.
One card short of the required 40 should produce an error.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(9)} # 36 cards
cards["card-extra"] = 3 # 39 total
energy = {"lightning": 20}
result = validator.validate_deck(cards, energy)
assert result.is_valid is False
assert any("must have exactly 40 cards" in e and "got 39" in e for e in result.errors)
def test_41_cards_fails(self, default_config):
"""Test that 41 cards fails validation.
One card over the required 40 should produce an error.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)} # 40 cards
cards["card-extra"] = 1 # 41 total
energy = {"lightning": 20}
result = validator.validate_deck(cards, energy)
assert result.is_valid is False
assert any("must have exactly 40 cards" in e and "got 41" in e for e in result.errors)
def test_empty_deck_fails(self, default_config):
"""Test that an empty deck fails validation.
A deck with no cards should fail the card count check.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
result = validator.validate_deck({}, {"lightning": 20})
assert result.is_valid is False
assert any("must have exactly 40 cards" in e and "got 0" in e for e in result.errors)
# =============================================================================
# Energy Count Validation Tests
# =============================================================================
class TestEnergyCountValidation:
"""Tests for energy deck card count validation (20 energy required)."""
def test_valid_energy_count_passes(self, default_config):
"""Test that exactly 20 energy cards passes validation.
The energy deck must have exactly 20 cards per Mantimon house rules.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)}
energy = {"lightning": 14, "colorless": 6} # 20 total
result = validator.validate_deck(cards, energy)
assert "Energy deck must have exactly 20" not in str(result.errors)
def test_19_energy_fails(self, default_config):
"""Test that 19 energy cards fails validation.
One energy short of the required 20 should produce an error.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)}
energy = {"lightning": 19}
result = validator.validate_deck(cards, energy)
assert result.is_valid is False
assert any("Energy deck must have exactly 20" in e and "got 19" in e for e in result.errors)
def test_21_energy_fails(self, default_config):
"""Test that 21 energy cards fails validation.
One energy over the required 20 should produce an error.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)}
energy = {"lightning": 21}
result = validator.validate_deck(cards, energy)
assert result.is_valid is False
assert any("Energy deck must have exactly 20" in e and "got 21" in e for e in result.errors)
def test_empty_energy_deck_fails(self, default_config):
"""Test that an empty energy deck fails validation."""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)}
result = validator.validate_deck(cards, {})
assert result.is_valid is False
assert any("Energy deck must have exactly 20" in e and "got 0" in e for e in result.errors)
# =============================================================================
# Max Copies Per Card Tests
# =============================================================================
class TestMaxCopiesValidation:
"""Tests for maximum copies per card validation (4 max)."""
def test_4_copies_allowed(self, default_config):
"""Test that 4 copies of a card is allowed.
The maximum of 4 copies per card should pass validation.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {
"test-001-pikachu": 4, # Max allowed
}
# Pad to 40 cards
for i in range(9):
cards[f"filler-{i:03d}"] = 4
energy = {"lightning": 20}
result = validator.validate_deck(cards, energy)
assert "max allowed is 4" not in str(result.errors)
def test_5_copies_fails(self, default_config):
"""Test that 5 copies of a card fails validation.
One copy over the maximum should produce an error identifying the card.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {
"test-001-pikachu": 5, # One over max
}
# Pad to 40 (5 + 35)
for i in range(7):
cards[f"filler-{i:03d}"] = 5
energy = {"lightning": 20}
result = validator.validate_deck(cards, energy)
assert result.is_valid is False
assert any(
"test-001-pikachu" in e and "5 copies" in e and "max allowed is 4" in e
for e in result.errors
)
def test_multiple_cards_over_limit(self, default_config):
"""Test that multiple cards over limit all get reported.
Each card exceeding the limit should generate its own error.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {
"card-a": 5,
"card-b": 6,
"card-c": 4, # OK
}
# Pad to 40
cards["filler"] = 25
energy = {"lightning": 20}
result = validator.validate_deck(cards, energy)
assert result.is_valid is False
# Both card-a and card-b should be reported
error_str = str(result.errors)
assert "card-a" in error_str
assert "card-b" in error_str
# =============================================================================
# Basic Pokemon Requirement Tests
# =============================================================================
class TestBasicPokemonRequirement:
"""Tests for minimum Basic Pokemon requirement (at least 1)."""
def test_deck_with_basic_pokemon_passes(self, default_config):
"""Test that a deck with Basic Pokemon passes validation.
Having at least 1 Basic Pokemon satisfies this requirement.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)}
energy = {"lightning": 20}
result = validator.validate_deck(cards, energy)
assert "at least 1 Basic Pokemon" not in str(result.errors)
def test_deck_without_basic_pokemon_fails(self, default_config):
"""Test that a deck without Basic Pokemon fails validation.
A deck composed entirely of Stage 1/2 Pokemon and Trainers
cannot start a game and should fail validation.
"""
# Create a mock that returns Stage 1 pokemon for all IDs
mock_service = MagicMock()
mock_service.get_card = lambda cid: CardDefinition(
id=cid,
name=f"Card {cid}",
card_type=CardType.POKEMON,
hp=60,
pokemon_type=EnergyType.LIGHTNING,
stage=PokemonStage.STAGE_1,
evolves_from="SomeBasic",
)
validator = DeckValidator(default_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)}
energy = {"lightning": 20}
result = validator.validate_deck(cards, energy)
assert result.is_valid is False
assert any("at least 1 Basic Pokemon" in e for e in result.errors)
def test_deck_with_only_trainers_fails(self, default_config):
"""Test that a deck with only Trainers fails Basic Pokemon check.
Trainers don't count toward the Basic Pokemon requirement.
"""
# Create a mock that returns Trainers for all IDs
mock_service = MagicMock()
mock_service.get_card = lambda cid: CardDefinition(
id=cid,
name=f"Trainer {cid}",
card_type=CardType.TRAINER,
trainer_type="item",
)
validator = DeckValidator(default_config, mock_service)
cards = {f"trainer-{i:03d}": 4 for i in range(10)}
energy = {"lightning": 20}
result = validator.validate_deck(cards, energy)
assert result.is_valid is False
assert any("at least 1 Basic Pokemon" in e for e in result.errors)
# =============================================================================
# Card ID Validation Tests
# =============================================================================
class TestCardIdValidation:
"""Tests for card ID existence validation."""
def test_valid_card_ids_pass(self, default_config):
"""Test that valid card IDs pass validation.
All card IDs in the deck should exist in the CardService.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)}
energy = {"lightning": 20}
result = validator.validate_deck(cards, energy)
assert "Invalid card IDs" not in str(result.errors)
def test_invalid_card_id_fails(self, default_config):
"""Test that an invalid card ID fails validation.
Card IDs not found in CardService should produce an error.
"""
# Create a mock that returns None for specific cards
mock_service = MagicMock()
def mock_get_card(cid):
if cid == "nonexistent-card":
return None
return CardDefinition(
id=cid,
name=f"Card {cid}",
card_type=CardType.POKEMON,
hp=60,
pokemon_type=EnergyType.LIGHTNING,
stage=PokemonStage.BASIC,
)
mock_service.get_card = mock_get_card
validator = DeckValidator(default_config, mock_service)
cards = {
"valid-card": 4,
"nonexistent-card": 4,
}
# Pad to 40
for i in range(8):
cards[f"card-{i:03d}"] = 4
energy = {"lightning": 20}
result = validator.validate_deck(cards, energy)
assert result.is_valid is False
assert any("Invalid card IDs" in e and "nonexistent-card" in e for e in result.errors)
def test_multiple_invalid_ids_reported(self, default_config):
"""Test that multiple invalid IDs are reported together.
The error message should list multiple invalid IDs (up to a limit).
"""
# Create a mock that returns None for "bad-*" cards
mock_service = MagicMock()
def mock_get_card(cid):
if cid.startswith("bad"):
return None
return CardDefinition(
id=cid,
name=f"Card {cid}",
card_type=CardType.POKEMON,
hp=60,
pokemon_type=EnergyType.LIGHTNING,
stage=PokemonStage.BASIC,
)
mock_service.get_card = mock_get_card
validator = DeckValidator(default_config, mock_service)
cards = {
"bad-1": 4,
"bad-2": 4,
"bad-3": 4,
"good": 28,
}
energy = {"lightning": 20}
result = validator.validate_deck(cards, energy)
assert result.is_valid is False
error_str = str(result.errors)
assert "bad-1" in error_str
assert "bad-2" in error_str
assert "bad-3" in error_str
# =============================================================================
# Ownership Validation Tests
# =============================================================================
class TestOwnershipValidation:
"""Tests for card ownership validation (campaign mode)."""
def test_owned_cards_pass(self, default_config):
"""Test that deck passes when user owns all cards.
In campaign mode, user must own sufficient copies of each card.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)}
energy = {"lightning": 20}
# User owns 10 copies of each card
owned = {f"card-{i:03d}": 10 for i in range(10)}
result = validator.validate_deck(cards, energy, owned_cards=owned)
assert "Insufficient cards" not in str(result.errors)
def test_insufficient_ownership_fails(self, default_config):
"""Test that deck fails when user doesn't own enough copies.
Needing 4 copies but only owning 2 should produce an error.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)}
energy = {"lightning": 20}
# User owns only 2 copies of first card
owned = {f"card-{i:03d}": 10 for i in range(10)}
owned["card-000"] = 2 # Need 4, only have 2
result = validator.validate_deck(cards, energy, owned_cards=owned)
assert result.is_valid is False
assert any(
"Insufficient cards" in e and "card-000" in e and "need 4" in e and "own 2" in e
for e in result.errors
)
def test_unowned_card_fails(self, default_config):
"""Test that deck fails when user doesn't own a card at all.
A card with 0 owned copies should fail ownership validation.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {"owned-card": 20, "unowned-card": 20}
energy = {"lightning": 20}
owned = {"owned-card": 20} # Missing unowned-card entirely
result = validator.validate_deck(cards, energy, owned_cards=owned)
assert result.is_valid is False
assert any(
"Insufficient cards" in e and "unowned-card" in e and "own 0" in e
for e in result.errors
)
def test_freeplay_skips_ownership(self, default_config):
"""Test that passing None for owned_cards skips ownership validation.
In freeplay mode, users have access to all cards regardless of
their actual collection.
"""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)}
energy = {"lightning": 20}
# owned_cards=None means freeplay mode
result = validator.validate_deck(cards, energy, owned_cards=None)
assert "Insufficient cards" not in str(result.errors)
# =============================================================================
# Multiple Errors Tests
# =============================================================================
class TestMultipleErrors:
"""Tests for returning all errors at once."""
def test_multiple_errors_returned_together(self, default_config):
"""Test that multiple validation errors are all returned.
When a deck has multiple issues, all should be reported so the
user can fix everything at once rather than iteratively.
"""
# Create a mock that returns None for all cards
mock_service = MagicMock()
mock_service.get_card = lambda cid: None
validator = DeckValidator(default_config, mock_service)
cards = {
"bad-card": 5, # Invalid ID + over copy limit
}
# Total is 5 (not 40)
energy = {"lightning": 10} # Only 10 (not 20)
owned = {} # Empty ownership
result = validator.validate_deck(cards, energy, owned_cards=owned)
assert result.is_valid is False
# Should have multiple errors
assert len(result.errors) >= 3
error_str = str(result.errors)
# Card count error
assert "40 cards" in error_str
# Energy count error
assert "20" in error_str
# Invalid card ID or max copies
assert "bad-card" in error_str
# =============================================================================
# Custom Config Tests
# =============================================================================
class TestCustomConfig:
"""Tests for using custom DeckConfig values."""
def test_custom_deck_size(self):
"""Test that custom deck size is respected.
Different game modes might use different deck sizes (e.g., 60-card).
"""
custom_config = DeckConfig(min_size=60, max_size=60)
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(custom_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)} # 40 cards
energy = {"lightning": 20}
result = validator.validate_deck(cards, energy)
assert result.is_valid is False
assert any("must have exactly 60 cards" in e for e in result.errors)
def test_custom_energy_size(self):
"""Test that custom energy deck size is respected."""
custom_config = DeckConfig(energy_deck_size=30)
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(custom_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)} # 40 cards
energy = {"lightning": 20} # 20, but need 30
result = validator.validate_deck(cards, energy)
assert result.is_valid is False
assert any("must have exactly 30" in e for e in result.errors)
def test_custom_max_copies(self):
"""Test that custom max copies per card is respected."""
custom_config = DeckConfig(max_copies_per_card=2)
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(custom_config, mock_service)
cards = {f"card-{i:03d}": 4 for i in range(10)} # 4 copies each
energy = {"lightning": 20}
result = validator.validate_deck(cards, energy)
assert result.is_valid is False
assert any("max allowed is 2" in e for e in result.errors)
# =============================================================================
# Utility Method Tests
# =============================================================================
class TestUtilityMethods:
"""Tests for utility methods on DeckValidator."""
def test_validate_cards_exist_all_valid(self, default_config):
"""Test validate_cards_exist returns empty list when all valid."""
mock_service = create_basic_pokemon_mock()
validator = DeckValidator(default_config, mock_service)
card_ids = ["card-1", "card-2", "card-3"]
invalid = validator.validate_cards_exist(card_ids)
assert invalid == []
def test_validate_cards_exist_some_invalid(self, default_config):
"""Test validate_cards_exist returns invalid IDs."""
mock_service = MagicMock()
def mock_get(cid):
if cid.startswith("bad"):
return None
return CardDefinition(
id=cid,
name=f"Card {cid}",
card_type=CardType.POKEMON,
hp=60,
pokemon_type=EnergyType.LIGHTNING,
stage=PokemonStage.BASIC,
)
mock_service.get_card = mock_get
validator = DeckValidator(default_config, mock_service)
card_ids = ["good-1", "bad-1", "good-2", "bad-2"]
invalid = validator.validate_cards_exist(card_ids)
assert set(invalid) == {"bad-1", "bad-2"}
def test_count_basic_pokemon(self, default_config):
"""Test count_basic_pokemon returns correct count."""
mock_service = MagicMock()
def mock_get(cid):
if cid.startswith("basic"):
return CardDefinition(
id=cid,
name=f"Card {cid}",
card_type=CardType.POKEMON,
hp=60,
pokemon_type=EnergyType.LIGHTNING,
stage=PokemonStage.BASIC,
)
elif cid.startswith("stage1"):
return CardDefinition(
id=cid,
name=f"Card {cid}",
card_type=CardType.POKEMON,
hp=90,
pokemon_type=EnergyType.LIGHTNING,
stage=PokemonStage.STAGE_1,
evolves_from="SomeBasic",
)
else:
return CardDefinition(
id=cid,
name=f"Trainer {cid}",
card_type=CardType.TRAINER,
trainer_type="item",
)
mock_service.get_card = mock_get
validator = DeckValidator(default_config, mock_service)
cards = {
"basic-1": 4,
"basic-2": 3,
"stage1-1": 4,
"trainer-1": 4,
}
count = validator.count_basic_pokemon(cards)
# basic-1: 4 + basic-2: 3 = 7
assert count == 7
def test_config_property(self, default_config, mock_card_service):
"""Test that config property returns the injected DeckConfig."""
validator = DeckValidator(default_config, mock_card_service)
assert validator.config is default_config
assert validator.config.min_size == 40
assert validator.config.energy_deck_size == 20