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

Implement GameService.end_game with history archival (GS-005)

Browse Source 
- Add GameEndResult dataclass with winner, loser, final views, duration
- Add _map_end_reason() to map core GameEndReason to DB EndReason
  (raises ValueError for unknown reasons to catch missing enum sync)
- Enhance end_game() to build replay data and return comprehensive result
- Add archive_to_history() to GameStateManager for complete game archival:
  - Creates GameHistory record with replay data
  - Deletes ActiveGame record
  - Clears Redis cache
  - All in single transaction
- Add ArchiveResult dataclass for archive operation metadata
- Add TODO for session_factory DI refactor in GameStateManager
- Update tests: 5 new end_game tests, 6 new archive_to_history tests

Phase 4 progress: 10/18 tasks complete

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Cal Corum 2026-01-29 20:10:27 -06:00
parent d5460ff418
commit 531d3e1e79
5 changed files with 770 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

161
backend/app/services/game_service.py
View File

@ -55,7 +55,7 @@ from app.core.models.card import CardInstance
from app.core.models.game_state import GameState
from app.core.rng import create_rng
from app.core.visibility import VisibleGameState, get_visible_state
from app.db.models.game import GameType
from app.db.models.game import EndReason, GameType
from app.services.card_service import CardService, get_card_service
from app.services.game_state_manager import GameStateManager, game_state_manager
@ -68,6 +68,41 @@ EngineFactory = Callable[[GameState], GameEngine]
CreationEngineFactory = Callable[[RulesConfig], GameEngine]
def _map_end_reason(core_reason: GameEndReason) -> EndReason:
"""Map core GameEndReason to database EndReason.
The core module uses its own enum for offline independence,
while the database has a separate enum for persistence. The DB enum
also includes service-layer reasons (like DISCONNECTION) that the
core engine doesn't know about.
Args:
core_reason: The GameEndReason from the core module.
Returns:
The corresponding EndReason for database storage.
Raises:
ValueError: If core_reason is not in the mapping (indicates missing enum sync).
"""
mapping = {
GameEndReason.PRIZES_TAKEN: EndReason.PRIZES_TAKEN,
GameEndReason.NO_POKEMON: EndReason.NO_POKEMON,
GameEndReason.DECK_EMPTY: EndReason.CANNOT_DRAW,
GameEndReason.RESIGNATION: EndReason.RESIGNATION,
GameEndReason.TIMEOUT: EndReason.TIMEOUT,
GameEndReason.TURN_LIMIT: EndReason.TURN_LIMIT,
GameEndReason.DRAW: EndReason.DRAW,
}
result = mapping.get(core_reason)
if result is None:
raise ValueError(
f"Unknown GameEndReason: {core_reason}. "
"Update _map_end_reason when adding new end reasons to core."
)
return result
# =============================================================================
# Exceptions
# =============================================================================
@ -256,6 +291,40 @@ class GameCreateResult:
message: str = ""
@dataclass
class GameEndResult:
"""Result of ending a game.
Contains all information needed to notify clients and record
the game in history.
Attributes:
success: Whether the game was ended successfully.
game_id: The game that ended.
winner_id: The winner's player ID (None for draws).
loser_id: The loser's player ID (None for draws).
end_reason: Why the game ended.
turn_count: Total number of turns played.
duration_seconds: Game duration in seconds.
player1_final_view: Final state visible to player 1.
player2_final_view: Final state visible to player 2.
history_id: The GameHistory record ID.
message: Description or error message.
"""
success: bool
game_id: str
winner_id: str | None = None
loser_id: str | None = None
end_reason: GameEndReason | None = None
turn_count: int = 0
duration_seconds: int = 0
player1_final_view: VisibleGameState | None = None
player2_final_view: VisibleGameState | None = None
history_id: str | None = None
message: str = ""
# =============================================================================
# GameService
# =============================================================================
@ -670,31 +739,103 @@ class GameService:
game_id: str,
winner_id: str | None,
end_reason: GameEndReason,
) -> None:
"""Forcibly end a game (e.g., due to timeout or disconnection).
) -> GameEndResult:
"""End a game and record it in history.
This should be called by the timeout system or when a player
disconnects without reconnecting within the grace period.
This method handles the complete game ending process:
1. Updates game state with winner and end reason
2. Creates a GameHistory record with replay data
3. Deletes the game from ActiveGame table and Redis cache
4. Returns final state for client notification
Called by:
- execute_action when a win condition is detected
- Timeout system when a player times out
- Disconnect handler when a player abandons
Args:
game_id: The game ID.
winner_id: The winner's player ID, or None for a draw.
end_reason: Why the game ended.
Returns:
GameEndResult with final state and history record ID.
Raises:
GameNotFoundError: If game doesn't exist.
"""
state = await self.get_game_state(game_id)
# Set winner and end reason
# Set winner and end reason on game state
state.winner_id = winner_id
state.end_reason = end_reason
# Persist to both cache and DB
await self._state_manager.save_to_cache(state)
await self._state_manager.persist_to_db(state)
# Get player IDs from state
player_ids = list(state.players.keys())
player1_id = player_ids[0] if len(player_ids) > 0 else None
player2_id = player_ids[1] if len(player_ids) > 1 else None
logger.info(f"Game {game_id} forcibly ended: winner={winner_id}, reason={end_reason}")
# Determine loser (opposite of winner)
loser_id: str | None = None
if winner_id is not None and len(player_ids) == 2:
loser_id = player_ids[1] if player_ids[0] == winner_id else player_ids[0]
# Get player views for final state
player1_view = get_visible_state(state, player1_id) if player1_id else None
player2_view = get_visible_state(state, player2_id) if player2_id else None
# Map core end reason to DB end reason
db_end_reason = _map_end_reason(end_reason)
# Determine if NPC won (for campaign games)
winner_is_npc = False
db_winner_id: UUID | None = None
if winner_id is not None:
# Check if winner is a player UUID or NPC
try:
db_winner_id = UUID(winner_id)
except ValueError:
# Winner ID is not a UUID, must be NPC
winner_is_npc = True
db_winner_id = None
# Build replay data from action log
replay_data = {
"version": 1,
"game_id": game_id,
"action_log": state.action_log,
"final_turn": state.turn_number,
"rules": state.rules.model_dump(mode="json"),
}
# Archive to history (creates GameHistory, deletes ActiveGame and cache)
archive_result = await self._state_manager.archive_to_history(
game_id=game_id,
state=state,
winner_id=db_winner_id,
winner_is_npc=winner_is_npc,
end_reason=db_end_reason,
replay_data=replay_data,
)
logger.info(
f"Game {game_id} ended: winner={winner_id}, reason={end_reason}, "
f"turns={archive_result.turn_count}, duration={archive_result.duration_seconds}s"
)
return GameEndResult(
success=True,
game_id=game_id,
winner_id=winner_id,
loser_id=loser_id,
end_reason=end_reason,
turn_count=archive_result.turn_count,
duration_seconds=archive_result.duration_seconds,
player1_final_view=player1_view,
player2_final_view=player2_view,
history_id=archive_result.history_id,
message=f"Game ended: {end_reason.value}",
)
# =========================================================================
# Game Creation

147
backend/app/services/game_state_manager.py
View File

@ -52,7 +52,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.core.models.game_state import GameState
from app.db.models import ActiveGame, GameType
from app.db.models import ActiveGame, EndReason, GameHistory, GameType
from app.db.redis import RedisHelper, redis_helper
from app.db.session import get_session
@ -74,6 +74,27 @@ class RecoveryResult:
total: int = 0
@dataclass(frozen=True)
class ArchiveResult:
"""Result of archiving a game to history.
Attributes:
success: Whether the archive operation succeeded.
game_id: The archived game's ID.
history_id: The GameHistory record ID (same as game_id).
duration_seconds: Game duration in seconds.
turn_count: Total turns played.
game_type: Type of game that was archived.
"""
success: bool
game_id: str
history_id: str
duration_seconds: int
turn_count: int
game_type: GameType
# Redis key patterns
GAME_KEY_PREFIX = "game:"
@ -87,6 +108,13 @@ class GameStateManager:
Attributes:
redis: RedisHelper instance for cache operations.
TODO: Add session_factory parameter for full DI support, similar to
ConnectionManager's redis_factory pattern. Currently methods accept
optional session parameter, but fallback to get_session() global.
This would eliminate need for patching in unit tests. See:
- ConnectionManager for reference implementation
- tests/unit/services/test_game_state_manager.py for affected tests
"""
def __init__(self, redis: RedisHelper | None = None) -> None:
@ -390,6 +418,123 @@ class GameStateManager:
return deleted
async def archive_to_history(
self,
game_id: str,
state: GameState,
winner_id: UUID | None,
winner_is_npc: bool,
end_reason: EndReason,
replay_data: dict | None = None,
session: AsyncSession | None = None,
) -> ArchiveResult:
"""Archive a completed game to history and cleanup active game.
This method handles the complete end-of-game persistence:
1. Loads ActiveGame to get metadata (started_at, player IDs, game type)
2. Creates GameHistory record with replay data
3. Deletes ActiveGame record
4. Deletes from Redis cache
All database operations happen in a single transaction.
Args:
game_id: Unique game identifier.
state: Final GameState to archive.
winner_id: UUID of the winner (None for draws or NPC wins).
winner_is_npc: True if NPC won (campaign games).
end_reason: Database EndReason enum value.
replay_data: Optional replay data dict (action log, rules, etc.).
session: Optional existing session for transaction control.
Returns:
ArchiveResult with archive metadata.
Example:
result = await manager.archive_to_history(
game_id=game.game_id,
state=game,
winner_id=UUID(winner_id) if winner_id else None,
winner_is_npc=False,
end_reason=EndReason.PRIZES_TAKEN,
replay_data={"action_log": game.action_log},
)
"""
now = datetime.now(UTC)
async def _archive(db: AsyncSession) -> ArchiveResult:
# Load ActiveGame to get metadata
result = await db.execute(select(ActiveGame).where(ActiveGame.id == UUID(game_id)))
active_game = result.scalar_one_or_none()
# Calculate duration and get metadata
if active_game is not None:
started_at = active_game.started_at
duration_seconds = int((now - started_at).total_seconds())
game_type = active_game.game_type
db_player1_id = active_game.player1_id
db_player2_id = active_game.player2_id
npc_id = active_game.npc_id
else:
# Fallback if ActiveGame not found (shouldn't happen normally)
# Player IDs in GameState are strings that may not be valid UUIDs
duration_seconds = 0
game_type = GameType.FREEPLAY
db_player1_id = None
db_player2_id = None
npc_id = None
logger.warning(f"ActiveGame not found for {game_id}, using defaults")
# Create GameHistory record
history = GameHistory(
id=UUID(game_id),
game_type=game_type,
player1_id=db_player1_id,
player2_id=db_player2_id,
npc_id=npc_id,
winner_id=winner_id,
winner_is_npc=winner_is_npc,
end_reason=end_reason,
turn_count=state.turn_number,
duration_seconds=duration_seconds,
replay_data=replay_data,
played_at=now,
)
db.add(history)
# Delete ActiveGame record
if active_game is not None:
await db.delete(active_game)
logger.debug(f"Deleted ActiveGame record: {game_id}")
await db.flush()
return ArchiveResult(
success=True,
game_id=game_id,
history_id=game_id,
duration_seconds=duration_seconds,
turn_count=state.turn_number,
game_type=game_type,
)
if session:
archive_result = await _archive(session)
else:
async with get_session() as db:
archive_result = await _archive(db)
await db.commit()
# Delete from Redis cache (outside transaction)
await self.delete_from_cache(game_id)
logger.info(
f"Archived game {game_id} to history: "
f"turns={archive_result.turn_count}, duration={archive_result.duration_seconds}s"
)
return archive_result
async def recover_active_games(
self,
session: AsyncSession | None = None,

6
backend/project_plans/PHASE_4_GAME_SERVICE.json
View File

@ -9,7 +9,7 @@
"description": "Real-time gameplay infrastructure - WebSocket communication, game lifecycle management, reconnection handling, and turn timeout system",
"totalEstimatedHours": 45,
"totalTasks": 18,
"completedTasks": 9,
"completedTasks": 10,
"status": "in_progress",
"masterPlan": "../PROJECT_PLAN_MASTER.json"
},
@ -297,8 +297,8 @@
"description": "Handle game completion - record history, cleanup active game",
"category": "services",
"priority": 9,
"completed": false,
"tested": false,
"completed": true,
"tested": true,
"dependencies": ["GS-003"],
"files": [
{"path": "app/services/game_service.py", "status": "modify"}

109
backend/tests/unit/services/test_game_service.py
View File

@ -40,11 +40,24 @@ def mock_state_manager() -> AsyncMock:
The state manager handles persistence to Redis (cache) and
Postgres (durable storage).
"""
from app.db.models.game import GameType
from app.services.game_state_manager import ArchiveResult
manager = AsyncMock()
manager.load_state = AsyncMock(return_value=None)
manager.save_to_cache = AsyncMock()
manager.persist_to_db = AsyncMock()
manager.cache_exists = AsyncMock(return_value=False)
manager.archive_to_history = AsyncMock(
return_value=ArchiveResult(
success=True,
game_id="game-123",
history_id="game-123",
duration_seconds=300,
turn_count=10,
game_type=GameType.FREEPLAY,
)
)
return manager
@ -998,23 +1011,29 @@ class TestResignGame:
class TestEndGame:
"""Tests for the end_game method (forced ending)."""
"""Tests for the end_game method.
The end_game method handles complete game ending:
- Sets winner and end reason on game state
- Archives game to history via state manager
- Returns GameEndResult with final state
"""
@pytest.mark.asyncio
async def test_end_game_sets_winner_and_reason(
async def test_end_game_returns_result_with_winner(
self,
game_service: GameService,
mock_state_manager: AsyncMock,
sample_game_state: GameState,
) -> None:
"""Test that end_game sets winner and end reason.
"""Test that end_game returns GameEndResult with winner info.
Used for timeout or disconnection scenarios where the game
needs to be forcibly ended.
The result should include winner, loser, end reason, and
game statistics from the archive operation.
"""
mock_state_manager.load_state.return_value = sample_game_state
await game_service.end_game(
result = await game_service.end_game(
"game-123",
winner_id="player-1",
end_reason=GameEndReason.TIMEOUT,
@ -1024,9 +1043,44 @@ class TestEndGame:
assert sample_game_state.winner_id == "player-1"
assert sample_game_state.end_reason == GameEndReason.TIMEOUT
# Verify persistence
mock_state_manager.save_to_cache.assert_called_once()
mock_state_manager.persist_to_db.assert_called_once()
# Verify result
assert result.success is True
assert result.game_id == "game-123"
assert result.winner_id == "player-1"
assert result.loser_id == "player-2"
assert result.end_reason == GameEndReason.TIMEOUT
assert result.duration_seconds == 300
assert result.turn_count == 10
assert result.history_id == "game-123"
# Verify archive was called
mock_state_manager.archive_to_history.assert_called_once()
@pytest.mark.asyncio
async def test_end_game_includes_final_player_views(
self,
game_service: GameService,
mock_state_manager: AsyncMock,
sample_game_state: GameState,
) -> None:
"""Test that end_game result includes visibility-filtered final states.
Each player should receive their own view of the final game state
for displaying the game-over screen.
"""
mock_state_manager.load_state.return_value = sample_game_state
result = await game_service.end_game(
"game-123",
winner_id="player-1",
end_reason=GameEndReason.PRIZES_TAKEN,
)
# Both players should have final views
assert result.player1_final_view is not None
assert result.player2_final_view is not None
assert result.player1_final_view.viewer_id == "player-1"
assert result.player2_final_view.viewer_id == "player-2"
@pytest.mark.asyncio
async def test_end_game_draw(
@ -1038,10 +1092,11 @@ class TestEndGame:
"""Test ending a game as a draw (no winner).
Some scenarios (timeout with equal scores) can result in a draw.
In this case, winner_id and loser_id should both be None.
"""
mock_state_manager.load_state.return_value = sample_game_state
await game_service.end_game(
result = await game_service.end_game(
"game-123",
winner_id=None,
end_reason=GameEndReason.DRAW,
@ -1049,6 +1104,9 @@ class TestEndGame:
assert sample_game_state.winner_id is None
assert sample_game_state.end_reason == GameEndReason.DRAW
assert result.winner_id is None
assert result.loser_id is None
assert result.end_reason == GameEndReason.DRAW
@pytest.mark.asyncio
async def test_end_game_not_found(
@ -1066,6 +1124,37 @@ class TestEndGame:
end_reason=GameEndReason.TIMEOUT,
)
@pytest.mark.asyncio
async def test_end_game_archives_with_replay_data(
self,
game_service: GameService,
mock_state_manager: AsyncMock,
sample_game_state: GameState,
) -> None:
"""Test that end_game passes replay data to archive_to_history.
Replay data includes the action log, final turn number, and
rules configuration for future replay functionality.
"""
from app.db.models.game import EndReason
sample_game_state.action_log = [{"type": "attack", "index": 0}]
mock_state_manager.load_state.return_value = sample_game_state
await game_service.end_game(
"game-123",
winner_id="player-1",
end_reason=GameEndReason.PRIZES_TAKEN,
)
# Verify archive was called with replay data
call_kwargs = mock_state_manager.archive_to_history.call_args.kwargs
assert call_kwargs["game_id"] == "game-123"
assert call_kwargs["end_reason"] == EndReason.PRIZES_TAKEN
assert call_kwargs["replay_data"] is not None
assert "action_log" in call_kwargs["replay_data"]
assert call_kwargs["replay_data"]["action_log"] == [{"type": "attack", "index": 0}]
class TestCreateGame:
"""Tests for the create_game method."""

371
backend/tests/unit/services/test_game_state_manager.py Normal file
View File

@ -0,0 +1,371 @@
"""Tests for GameStateManager.
This module tests the game state persistence layer that manages
the Redis-primary, Postgres-backup storage strategy.
"""
from datetime import UTC, datetime
from unittest.mock import AsyncMock, MagicMock, patch
from uuid import UUID, uuid4
import pytest
from app.core.enums import TurnPhase
from app.core.models.game_state import GameState, PlayerState
from app.db.models.game import EndReason, GameType
from app.services.game_state_manager import ArchiveResult, GameStateManager
@pytest.fixture
def mock_redis() -> AsyncMock:
"""Create a mock RedisHelper.
The RedisHelper handles cache operations for game state.
"""
redis = AsyncMock()
redis.set_json = AsyncMock()
redis.get_json = AsyncMock(return_value=None)
redis.delete = AsyncMock(return_value=True)
redis.exists = AsyncMock(return_value=False)
return redis
@pytest.fixture
def game_state_manager(mock_redis: AsyncMock) -> GameStateManager:
"""Create a GameStateManager with mock Redis."""
return GameStateManager(redis=mock_redis)
@pytest.fixture
def sample_game_state() -> GameState:
"""Create a sample game state for testing."""
player1 = PlayerState(player_id="player-1")
player2 = PlayerState(player_id="player-2")
return GameState(
game_id="game-123",
players={"player-1": player1, "player-2": player2},
current_player_id="player-1",
turn_number=10,
phase=TurnPhase.MAIN,
action_log=[{"type": "attack", "index": 0}],
)
class TestArchiveToHistory:
"""Tests for the archive_to_history method.
This method handles the complete game ending process:
- Creates GameHistory record
- Deletes ActiveGame record
- Clears Redis cache
"""
@pytest.mark.asyncio
async def test_archive_to_history_returns_result(
self,
game_state_manager: GameStateManager,
sample_game_state: GameState,
) -> None:
"""Test that archive_to_history returns ArchiveResult.
The result should contain game metadata and success status.
"""
from app.db.models.game import ActiveGame
game_id = str(uuid4())
sample_game_state.game_id = game_id
player1_id = uuid4()
player2_id = uuid4()
# Create mock ActiveGame
mock_active_game = MagicMock(spec=ActiveGame)
mock_active_game.id = UUID(game_id)
mock_active_game.game_type = GameType.FREEPLAY
mock_active_game.player1_id = player1_id
mock_active_game.player2_id = player2_id
mock_active_game.npc_id = None
mock_active_game.started_at = datetime.now(UTC)
# Mock database session
mock_session = AsyncMock()
mock_result = MagicMock()
mock_result.scalar_one_or_none.return_value = mock_active_game
mock_session.execute = AsyncMock(return_value=mock_result)
mock_session.add = MagicMock()
mock_session.delete = AsyncMock()
mock_session.flush = AsyncMock()
mock_session.commit = AsyncMock()
with patch("app.services.game_state_manager.get_session") as mock_get_session:
mock_get_session.return_value.__aenter__ = AsyncMock(return_value=mock_session)
mock_get_session.return_value.__aexit__ = AsyncMock(return_value=None)
result = await game_state_manager.archive_to_history(
game_id=game_id,
state=sample_game_state,
winner_id=player1_id,
winner_is_npc=False,
end_reason=EndReason.PRIZES_TAKEN,
replay_data={"action_log": sample_game_state.action_log},
)
assert isinstance(result, ArchiveResult)
assert result.success is True
assert result.game_id == game_id
assert result.history_id == game_id
assert result.turn_count == 10
assert result.game_type == GameType.FREEPLAY
@pytest.mark.asyncio
async def test_archive_to_history_creates_game_history(
self,
game_state_manager: GameStateManager,
sample_game_state: GameState,
) -> None:
"""Test that archive_to_history creates a GameHistory record.
The GameHistory record should be added to the session with
all the required fields populated.
"""
from app.db.models.game import ActiveGame, GameHistory
game_id = str(uuid4())
sample_game_state.game_id = game_id
player1_id = uuid4()
# Create mock ActiveGame
mock_active_game = MagicMock(spec=ActiveGame)
mock_active_game.id = UUID(game_id)
mock_active_game.game_type = GameType.CAMPAIGN
mock_active_game.player1_id = player1_id
mock_active_game.player2_id = None
mock_active_game.npc_id = "grass_trainer_1"
mock_active_game.started_at = datetime.now(UTC)
# Mock database session
mock_session = AsyncMock()
mock_result = MagicMock()
mock_result.scalar_one_or_none.return_value = mock_active_game
mock_session.execute = AsyncMock(return_value=mock_result)
mock_session.add = MagicMock()
mock_session.delete = AsyncMock()
mock_session.flush = AsyncMock()
mock_session.commit = AsyncMock()
with patch("app.services.game_state_manager.get_session") as mock_get_session:
mock_get_session.return_value.__aenter__ = AsyncMock(return_value=mock_session)
mock_get_session.return_value.__aexit__ = AsyncMock(return_value=None)
await game_state_manager.archive_to_history(
game_id=game_id,
state=sample_game_state,
winner_id=player1_id,
winner_is_npc=False,
end_reason=EndReason.RESIGNATION,
replay_data={"action_log": []},
)
# Verify GameHistory was added
mock_session.add.assert_called_once()
added_obj = mock_session.add.call_args[0][0]
assert isinstance(added_obj, GameHistory)
assert added_obj.id == UUID(game_id)
assert added_obj.game_type == GameType.CAMPAIGN
assert added_obj.npc_id == "grass_trainer_1"
assert added_obj.end_reason == EndReason.RESIGNATION
@pytest.mark.asyncio
async def test_archive_to_history_deletes_active_game(
self,
game_state_manager: GameStateManager,
sample_game_state: GameState,
) -> None:
"""Test that archive_to_history deletes the ActiveGame record.
After archiving to history, the game should be removed from
the active games table.
"""
from app.db.models.game import ActiveGame
game_id = str(uuid4())
sample_game_state.game_id = game_id
# Create mock ActiveGame
mock_active_game = MagicMock(spec=ActiveGame)
mock_active_game.id = UUID(game_id)
mock_active_game.game_type = GameType.FREEPLAY
mock_active_game.player1_id = uuid4()
mock_active_game.player2_id = uuid4()
mock_active_game.npc_id = None
mock_active_game.started_at = datetime.now(UTC)
# Mock database session
mock_session = AsyncMock()
mock_result = MagicMock()
mock_result.scalar_one_or_none.return_value = mock_active_game
mock_session.execute = AsyncMock(return_value=mock_result)
mock_session.add = MagicMock()
mock_session.delete = AsyncMock()
mock_session.flush = AsyncMock()
mock_session.commit = AsyncMock()
with patch("app.services.game_state_manager.get_session") as mock_get_session:
mock_get_session.return_value.__aenter__ = AsyncMock(return_value=mock_session)
mock_get_session.return_value.__aexit__ = AsyncMock(return_value=None)
await game_state_manager.archive_to_history(
game_id=game_id,
state=sample_game_state,
winner_id=uuid4(),
winner_is_npc=False,
end_reason=EndReason.PRIZES_TAKEN,
)
# Verify ActiveGame was deleted
mock_session.delete.assert_called_once_with(mock_active_game)
@pytest.mark.asyncio
async def test_archive_to_history_deletes_from_cache(
self,
game_state_manager: GameStateManager,
mock_redis: AsyncMock,
sample_game_state: GameState,
) -> None:
"""Test that archive_to_history deletes from Redis cache.
After archiving, the game state should be removed from the
Redis cache to free up memory.
"""
from app.db.models.game import ActiveGame
game_id = str(uuid4())
sample_game_state.game_id = game_id
# Create mock ActiveGame
mock_active_game = MagicMock(spec=ActiveGame)
mock_active_game.id = UUID(game_id)
mock_active_game.game_type = GameType.FREEPLAY
mock_active_game.player1_id = uuid4()
mock_active_game.player2_id = uuid4()
mock_active_game.npc_id = None
mock_active_game.started_at = datetime.now(UTC)
# Mock database session
mock_session = AsyncMock()
mock_result = MagicMock()
mock_result.scalar_one_or_none.return_value = mock_active_game
mock_session.execute = AsyncMock(return_value=mock_result)
mock_session.add = MagicMock()
mock_session.delete = AsyncMock()
mock_session.flush = AsyncMock()
mock_session.commit = AsyncMock()
with patch("app.services.game_state_manager.get_session") as mock_get_session:
mock_get_session.return_value.__aenter__ = AsyncMock(return_value=mock_session)
mock_get_session.return_value.__aexit__ = AsyncMock(return_value=None)
await game_state_manager.archive_to_history(
game_id=game_id,
state=sample_game_state,
winner_id=uuid4(),
winner_is_npc=False,
end_reason=EndReason.TIMEOUT,
)
# Verify Redis cache was deleted
mock_redis.delete.assert_called_once_with(f"game:{game_id}")
@pytest.mark.asyncio
async def test_archive_to_history_handles_missing_active_game(
self,
game_state_manager: GameStateManager,
sample_game_state: GameState,
) -> None:
"""Test that archive handles case when ActiveGame is not found.
This shouldn't normally happen, but the method should handle
it gracefully by using defaults from the game state.
"""
game_id = str(uuid4())
sample_game_state.game_id = game_id
# Mock database session - ActiveGame not found
mock_session = AsyncMock()
mock_result = MagicMock()
mock_result.scalar_one_or_none.return_value = None # Not found
mock_session.execute = AsyncMock(return_value=mock_result)
mock_session.add = MagicMock()
mock_session.flush = AsyncMock()
mock_session.commit = AsyncMock()
with patch("app.services.game_state_manager.get_session") as mock_get_session:
mock_get_session.return_value.__aenter__ = AsyncMock(return_value=mock_session)
mock_get_session.return_value.__aexit__ = AsyncMock(return_value=None)
result = await game_state_manager.archive_to_history(
game_id=game_id,
state=sample_game_state,
winner_id=None,
winner_is_npc=False,
end_reason=EndReason.DRAW,
)
# Should still succeed with defaults
assert result.success is True
assert result.game_type == GameType.FREEPLAY # Default
assert result.duration_seconds == 0 # Default when no started_at
@pytest.mark.asyncio
async def test_archive_to_history_npc_winner(
self,
game_state_manager: GameStateManager,
sample_game_state: GameState,
) -> None:
"""Test archiving when NPC is the winner.
In campaign mode, the NPC can win. The winner_is_npc flag
should be set correctly in GameHistory.
"""
from app.db.models.game import ActiveGame, GameHistory
game_id = str(uuid4())
sample_game_state.game_id = game_id
player1_id = uuid4()
# Create mock ActiveGame for campaign
mock_active_game = MagicMock(spec=ActiveGame)
mock_active_game.id = UUID(game_id)
mock_active_game.game_type = GameType.CAMPAIGN
mock_active_game.player1_id = player1_id
mock_active_game.player2_id = None
mock_active_game.npc_id = "fire_leader"
mock_active_game.started_at = datetime.now(UTC)
# Mock database session
mock_session = AsyncMock()
mock_result = MagicMock()
mock_result.scalar_one_or_none.return_value = mock_active_game
mock_session.execute = AsyncMock(return_value=mock_result)
mock_session.add = MagicMock()
mock_session.delete = AsyncMock()
mock_session.flush = AsyncMock()
mock_session.commit = AsyncMock()
with patch("app.services.game_state_manager.get_session") as mock_get_session:
mock_get_session.return_value.__aenter__ = AsyncMock(return_value=mock_session)
mock_get_session.return_value.__aexit__ = AsyncMock(return_value=None)
await game_state_manager.archive_to_history(
game_id=game_id,
state=sample_game_state,
winner_id=None, # NPC won, so no UUID
winner_is_npc=True,
end_reason=EndReason.NO_POKEMON,
)
# Verify GameHistory has NPC winner flag
added_obj = mock_session.add.call_args[0][0]
assert isinstance(added_obj, GameHistory)
assert added_obj.winner_is_npc is True
assert added_obj.winner_id is None