8f67883be1
Updated Lineup model to support both leagues using the same pattern as RosterLink: - Made card_id nullable (PD league) - Added player_id nullable (SBA league) - Added XOR CHECK constraint to ensure exactly one ID is populated - Created league-specific methods: add_pd_lineup_card() and add_sba_lineup_player() - Replaced generic create_lineup_entry() with league-specific methods Database migration applied to convert existing schema. Bonus fix: Resolved Pendulum DateTime + asyncpg timezone compatibility issue by using .naive() on all DateTime defaults in Game, Play, and GameSession models. Updated tests to use league-specific lineup methods. Archived migration docs and script to .claude/archive/ for reference. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
338 lines
10 KiB
Python
338 lines
10 KiB
Python
"""
|
|
End-to-end integration tests for state persistence and recovery.
|
|
|
|
Tests the complete flow: StateManager → DatabaseOperations → PostgreSQL → Recovery
|
|
|
|
Author: Claude
|
|
Date: 2025-10-22
|
|
"""
|
|
|
|
import pytest
|
|
from uuid import uuid4
|
|
|
|
from app.core.state_manager import StateManager
|
|
from app.models.game_models import TeamLineupState, LineupPlayerState
|
|
from app.database.session import init_db
|
|
|
|
|
|
# Mark all tests in this module as integration tests
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
async def setup_database():
|
|
"""Set up test database schema"""
|
|
await init_db()
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_game_id():
|
|
"""Generate unique game ID for each test"""
|
|
return uuid4()
|
|
|
|
|
|
class TestStateManagerPersistence:
|
|
"""Tests for StateManager integration with database"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_game_and_persist(self, setup_database, sample_game_id):
|
|
"""Test creating game in StateManager and persisting to DB"""
|
|
state_manager = StateManager()
|
|
|
|
# Create in-memory state
|
|
state = await state_manager.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2
|
|
)
|
|
|
|
assert state.game_id == sample_game_id
|
|
|
|
# Persist to database
|
|
await state_manager.db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
# Verify in database
|
|
db_game = await state_manager.db_ops.get_game(sample_game_id)
|
|
assert db_game is not None
|
|
assert db_game.id == sample_game_id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_persist_and_recover(self, setup_database, sample_game_id):
|
|
"""Test complete flow: create → persist → remove → recover"""
|
|
state_manager = StateManager()
|
|
|
|
# Step 1: Create game in memory
|
|
await state_manager.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2
|
|
)
|
|
|
|
# Step 2: Persist to database
|
|
await state_manager.db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
# Step 3: Update game state
|
|
state = state_manager.get_state(sample_game_id)
|
|
state.inning = 5
|
|
state.half = "bottom"
|
|
state.home_score = 3
|
|
state.away_score = 2
|
|
state_manager.update_state(sample_game_id, state)
|
|
|
|
# Persist update
|
|
await state_manager.db_ops.update_game_state(
|
|
game_id=sample_game_id,
|
|
inning=5,
|
|
half="bottom",
|
|
home_score=3,
|
|
away_score=2
|
|
)
|
|
|
|
# Step 4: Remove from memory
|
|
state_manager.remove_game(sample_game_id)
|
|
assert state_manager.get_state(sample_game_id) is None
|
|
|
|
# Step 5: Recover from database
|
|
recovered = await state_manager.recover_game(sample_game_id)
|
|
|
|
assert recovered is not None
|
|
assert recovered.game_id == sample_game_id
|
|
assert recovered.inning == 5
|
|
assert recovered.half == "bottom"
|
|
assert recovered.home_score == 3
|
|
assert recovered.away_score == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_recover_nonexistent_game(self, setup_database):
|
|
"""Test recovering nonexistent game returns None"""
|
|
state_manager = StateManager()
|
|
fake_id = uuid4()
|
|
|
|
recovered = await state_manager.recover_game(fake_id)
|
|
|
|
assert recovered is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lineup_persistence(self, setup_database, sample_game_id):
|
|
"""Test lineup persistence and state"""
|
|
state_manager = StateManager()
|
|
|
|
# Create game
|
|
await state_manager.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2
|
|
)
|
|
|
|
# Persist game to DB
|
|
await state_manager.db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
# Create lineup in StateManager
|
|
lineup = TeamLineupState(
|
|
team_id=1,
|
|
players=[
|
|
LineupPlayerState(lineup_id=1, card_id=101, position="CF", batting_order=1),
|
|
LineupPlayerState(lineup_id=2, card_id=102, position="SS", batting_order=2),
|
|
LineupPlayerState(lineup_id=3, card_id=103, position="1B", batting_order=3),
|
|
]
|
|
)
|
|
state_manager.set_lineup(sample_game_id, team_id=1, lineup=lineup)
|
|
|
|
# Persist lineup entries to DB (SBA uses player_id)
|
|
for player in lineup.players:
|
|
await state_manager.db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
player_id=player.card_id, # Note: card_id here is used as player_id for SBA
|
|
position=player.position,
|
|
batting_order=player.batting_order
|
|
)
|
|
|
|
# Retrieve from DB
|
|
db_lineup = await state_manager.db_ops.get_active_lineup(sample_game_id, team_id=1)
|
|
|
|
assert len(db_lineup) == 3
|
|
assert db_lineup[0].player_id == 101
|
|
assert db_lineup[1].player_id == 102
|
|
assert db_lineup[2].player_id == 103
|
|
|
|
|
|
class TestCompleteGameFlow:
|
|
"""Tests simulating a complete game flow"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_game_with_plays(self, setup_database, sample_game_id):
|
|
"""Test game with plays - complete flow"""
|
|
state_manager = StateManager()
|
|
|
|
# Create and persist game
|
|
await state_manager.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2
|
|
)
|
|
|
|
await state_manager.db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
# Add some plays
|
|
for i in range(3):
|
|
await state_manager.db_ops.save_play({
|
|
"game_id": sample_game_id,
|
|
"play_number": i + 1,
|
|
"inning": 1,
|
|
"half": "top",
|
|
"outs_before": i,
|
|
"batting_order": i + 1,
|
|
"result_description": f"Play {i+1}",
|
|
"pa": 1,
|
|
"ab": 1
|
|
})
|
|
|
|
# Update game state
|
|
state = state_manager.get_state(sample_game_id)
|
|
state.play_count = 3
|
|
state_manager.update_state(sample_game_id, state)
|
|
|
|
# Persist state update
|
|
await state_manager.db_ops.update_game_state(
|
|
game_id=sample_game_id,
|
|
inning=1,
|
|
half="bottom",
|
|
home_score=0,
|
|
away_score=0
|
|
)
|
|
|
|
# Remove from memory and recover
|
|
state_manager.remove_game(sample_game_id)
|
|
recovered = await state_manager.recover_game(sample_game_id)
|
|
|
|
assert recovered is not None
|
|
assert recovered.play_count == 3 # Should reflect the plays
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_games_independence(self, setup_database):
|
|
"""Test that multiple games are independent"""
|
|
state_manager = StateManager()
|
|
|
|
game1 = uuid4()
|
|
game2 = uuid4()
|
|
|
|
# Create two games
|
|
await state_manager.create_game(game1, "sba", 1, 2)
|
|
await state_manager.create_game(game2, "pd", 3, 4)
|
|
|
|
# Persist both
|
|
await state_manager.db_ops.create_game(
|
|
game_id=game1,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
await state_manager.db_ops.create_game(
|
|
game_id=game2,
|
|
league_id="pd",
|
|
home_team_id=3,
|
|
away_team_id=4,
|
|
game_mode="ranked",
|
|
visibility="public"
|
|
)
|
|
|
|
# Update game1
|
|
state1 = state_manager.get_state(game1)
|
|
state1.home_score = 5
|
|
state_manager.update_state(game1, state1)
|
|
|
|
await state_manager.db_ops.update_game_state(
|
|
game_id=game1,
|
|
inning=1,
|
|
half="top",
|
|
home_score=5,
|
|
away_score=0
|
|
)
|
|
|
|
# Remove both from memory
|
|
state_manager.remove_game(game1)
|
|
state_manager.remove_game(game2)
|
|
|
|
# Recover both
|
|
recovered1 = await state_manager.recover_game(game1)
|
|
recovered2 = await state_manager.recover_game(game2)
|
|
|
|
# Verify independence
|
|
assert recovered1.home_score == 5
|
|
assert recovered2.home_score == 0
|
|
assert recovered1.league_id == "sba"
|
|
assert recovered2.league_id == "pd"
|
|
|
|
|
|
class TestStateManagerStatistics:
|
|
"""Tests for StateManager statistics with persistence"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stats_after_eviction_and_recovery(self, setup_database, sample_game_id):
|
|
"""Test stats are accurate after eviction and recovery"""
|
|
state_manager = StateManager()
|
|
|
|
# Create game
|
|
await state_manager.create_game(sample_game_id, "sba", 1, 2)
|
|
|
|
# Persist
|
|
await state_manager.db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
# Check stats
|
|
stats = state_manager.get_stats()
|
|
assert stats["active_games"] == 1
|
|
|
|
# Evict
|
|
state_manager.remove_game(sample_game_id)
|
|
stats = state_manager.get_stats()
|
|
assert stats["active_games"] == 0
|
|
|
|
# Recover
|
|
await state_manager.recover_game(sample_game_id)
|
|
stats = state_manager.get_stats()
|
|
assert stats["active_games"] == 1
|