0ebe72c09d
This commit completes all Phase 3 work with comprehensive test coverage: Test Coverage: - 31 unit tests for SubstitutionRules (all validation paths) - 10 integration tests for SubstitutionManager (DB + state sync) - 679 total tests in test suite (609/609 unit tests passing - 100%) Testing Scope: - Pinch hitter validation and execution - Defensive replacement validation and execution - Pitching change validation and execution (min batters, force changes) - Double switch validation - Multiple substitutions in sequence - Batting order preservation - Database persistence verification - State sync verification - Lineup cache updates All substitution system components are now production-ready: ✅ Core validation logic (SubstitutionRules) ✅ Orchestration layer (SubstitutionManager) ✅ Database operations ✅ WebSocket event handlers ✅ Comprehensive test coverage ✅ Complete documentation Phase 3 Overall: 100% Complete - Phase 3A-D (X-Check Core): 100% - Phase 3E (Position Ratings + Redis): 100% - Phase 3F (Substitutions): 100% 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
1089 lines
34 KiB
Python
1089 lines
34 KiB
Python
"""
|
|
Integration tests for DatabaseOperations.
|
|
|
|
Tests actual database operations using the test database.
|
|
These tests are slower than unit tests but verify real DB interactions.
|
|
|
|
Author: Claude
|
|
Date: 2025-10-22
|
|
"""
|
|
|
|
import pytest
|
|
from uuid import uuid4
|
|
|
|
from app.database.operations import DatabaseOperations
|
|
from app.database.session import init_db, engine
|
|
|
|
|
|
# Mark all tests in this module as integration tests
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
async def setup_database():
|
|
"""
|
|
Set up test database schema.
|
|
|
|
Runs once per test function (noop if tables exist).
|
|
"""
|
|
# Create all tables (will skip if they exist)
|
|
await init_db()
|
|
yield
|
|
# Teardown if needed (tables persist between test runs)
|
|
|
|
|
|
@pytest.fixture
|
|
async def db_ops():
|
|
"""Create DatabaseOperations instance for each test"""
|
|
return DatabaseOperations()
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_game_id():
|
|
"""Generate a unique game ID for each test"""
|
|
return uuid4()
|
|
|
|
|
|
class TestDatabaseOperationsGame:
|
|
"""Tests for game CRUD operations"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_game(self, setup_database, db_ops, sample_game_id):
|
|
"""Test creating a game in database"""
|
|
game = await 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"
|
|
)
|
|
|
|
assert game.id == sample_game_id
|
|
assert game.league_id == "sba"
|
|
assert game.status == "pending"
|
|
assert game.home_team_id == 1
|
|
assert game.away_team_id == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_game_with_ai(self, setup_database, db_ops, sample_game_id):
|
|
"""Test creating a game with AI opponent"""
|
|
game = await db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="pd",
|
|
home_team_id=10,
|
|
away_team_id=20,
|
|
game_mode="practice",
|
|
visibility="private",
|
|
home_team_is_ai=False,
|
|
away_team_is_ai=True,
|
|
ai_difficulty="balanced"
|
|
)
|
|
|
|
assert game.away_team_is_ai is True
|
|
assert game.home_team_is_ai is False
|
|
assert game.ai_difficulty == "balanced"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_game(self, setup_database, db_ops, sample_game_id):
|
|
"""Test retrieving a game from database"""
|
|
# Create game
|
|
await 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"
|
|
)
|
|
|
|
# Retrieve game
|
|
retrieved = await db_ops.get_game(sample_game_id)
|
|
|
|
assert retrieved is not None
|
|
assert retrieved.id == sample_game_id
|
|
assert retrieved.league_id == "sba"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_game_nonexistent(self, setup_database, db_ops):
|
|
"""Test retrieving nonexistent game returns None"""
|
|
fake_id = uuid4()
|
|
game = await db_ops.get_game(fake_id)
|
|
|
|
assert game is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_game_state(self, setup_database, db_ops, sample_game_id):
|
|
"""Test updating game state"""
|
|
# Create game
|
|
await 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"
|
|
)
|
|
|
|
# Update state
|
|
await db_ops.update_game_state(
|
|
game_id=sample_game_id,
|
|
inning=5,
|
|
half="bottom",
|
|
home_score=3,
|
|
away_score=2,
|
|
status="active"
|
|
)
|
|
|
|
# Verify update
|
|
game = await db_ops.get_game(sample_game_id)
|
|
assert game.current_inning == 5
|
|
assert game.current_half == "bottom"
|
|
assert game.home_score == 3
|
|
assert game.away_score == 2
|
|
assert game.status == "active"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_game_state_nonexistent_raises_error(self, setup_database, db_ops):
|
|
"""Test updating nonexistent game raises error"""
|
|
fake_id = uuid4()
|
|
|
|
with pytest.raises(ValueError, match="not found"):
|
|
await db_ops.update_game_state(
|
|
game_id=fake_id,
|
|
inning=1,
|
|
half="top",
|
|
home_score=0,
|
|
away_score=0
|
|
)
|
|
|
|
|
|
class TestDatabaseOperationsLineup:
|
|
"""Tests for lineup operations"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_sba_lineup_player(self, setup_database, db_ops, sample_game_id):
|
|
"""Test adding SBA player to lineup"""
|
|
# Create game first
|
|
await 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 SBA player to lineup
|
|
lineup = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
player_id=101,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
|
|
assert lineup.game_id == sample_game_id
|
|
assert lineup.team_id == 1
|
|
assert lineup.player_id == 101
|
|
assert lineup.card_id is None
|
|
assert lineup.position == "CF"
|
|
assert lineup.batting_order == 1
|
|
assert lineup.is_active is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_pd_lineup_card(self, setup_database, db_ops, sample_game_id):
|
|
"""Test adding PD card to lineup"""
|
|
await db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="pd",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
lineup = await db_ops.add_pd_lineup_card(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
card_id=200,
|
|
position="P",
|
|
batting_order=None, # Pitcher, no batting order (AL rules)
|
|
is_starter=True
|
|
)
|
|
|
|
assert lineup.position == "P"
|
|
assert lineup.batting_order is None
|
|
assert lineup.card_id == 200
|
|
assert lineup.player_id is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_active_lineup(self, setup_database, db_ops, sample_game_id):
|
|
"""Test retrieving active lineup for a team"""
|
|
await 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 multiple SBA players to lineup
|
|
await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
player_id=103,
|
|
position="1B",
|
|
batting_order=3
|
|
)
|
|
await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
player_id=101,
|
|
position="CF",
|
|
batting_order=1
|
|
)
|
|
await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
player_id=102,
|
|
position="SS",
|
|
batting_order=2
|
|
)
|
|
|
|
# Retrieve lineup
|
|
lineup = await db_ops.get_active_lineup(sample_game_id, team_id=1)
|
|
|
|
assert len(lineup) == 3
|
|
# Should be sorted by batting order
|
|
assert lineup[0].batting_order == 1
|
|
assert lineup[1].batting_order == 2
|
|
assert lineup[2].batting_order == 3
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_active_lineup_empty(self, setup_database, db_ops, sample_game_id):
|
|
"""Test retrieving lineup for team with no entries"""
|
|
await 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"
|
|
)
|
|
|
|
lineup = await db_ops.get_active_lineup(sample_game_id, team_id=1)
|
|
|
|
assert lineup == []
|
|
|
|
|
|
class TestDatabaseOperationsPlays:
|
|
"""Tests for play operations"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_save_play(self, setup_database, db_ops, sample_game_id):
|
|
"""Test saving a play"""
|
|
await 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"
|
|
)
|
|
|
|
play_data = {
|
|
"game_id": sample_game_id,
|
|
"play_number": 1,
|
|
"inning": 1,
|
|
"half": "top",
|
|
"outs_before": 0,
|
|
"batting_order": 1,
|
|
"result_description": "Single to left field",
|
|
"pa": 1,
|
|
"ab": 1,
|
|
"hit": 1
|
|
}
|
|
|
|
play = await db_ops.save_play(play_data)
|
|
|
|
assert play.game_id == sample_game_id
|
|
assert play.play_number == 1
|
|
assert play.result_description == "Single to left field"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_plays(self, setup_database, db_ops, sample_game_id):
|
|
"""Test retrieving plays for a game"""
|
|
await 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"
|
|
)
|
|
|
|
# Save multiple plays
|
|
for i in range(3):
|
|
await 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
|
|
})
|
|
|
|
# Retrieve plays
|
|
plays = await db_ops.get_plays(sample_game_id)
|
|
|
|
assert len(plays) == 3
|
|
# Should be ordered by play_number
|
|
assert plays[0].play_number == 1
|
|
assert plays[1].play_number == 2
|
|
assert plays[2].play_number == 3
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_plays_empty(self, setup_database, db_ops, sample_game_id):
|
|
"""Test retrieving plays for game with no plays"""
|
|
await 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"
|
|
)
|
|
|
|
plays = await db_ops.get_plays(sample_game_id)
|
|
|
|
assert plays == []
|
|
|
|
|
|
class TestDatabaseOperationsRecovery:
|
|
"""Tests for game state recovery"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_game_state_complete(self, setup_database, db_ops, sample_game_id):
|
|
"""Test loading complete game state"""
|
|
# Create game
|
|
await 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 lineups
|
|
await db_ops.create_lineup_entry(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
card_id=101,
|
|
position="CF",
|
|
batting_order=1
|
|
)
|
|
|
|
# Add play
|
|
await db_ops.save_play({
|
|
"game_id": sample_game_id,
|
|
"play_number": 1,
|
|
"inning": 1,
|
|
"half": "top",
|
|
"outs_before": 0,
|
|
"batting_order": 1,
|
|
"result_description": "Single",
|
|
"pa": 1
|
|
})
|
|
|
|
# Update game state
|
|
await db_ops.update_game_state(
|
|
game_id=sample_game_id,
|
|
inning=2,
|
|
half="bottom",
|
|
home_score=1,
|
|
away_score=0
|
|
)
|
|
|
|
# Load complete state
|
|
state = await db_ops.load_game_state(sample_game_id)
|
|
|
|
assert state is not None
|
|
assert state["game"]["id"] == sample_game_id
|
|
assert state["game"]["current_inning"] == 2
|
|
assert state["game"]["current_half"] == "bottom"
|
|
assert len(state["lineups"]) == 1
|
|
assert len(state["plays"]) == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_game_state_nonexistent(self, setup_database, db_ops):
|
|
"""Test loading nonexistent game returns None"""
|
|
fake_id = uuid4()
|
|
state = await db_ops.load_game_state(fake_id)
|
|
|
|
assert state is None
|
|
|
|
|
|
class TestDatabaseOperationsGameSession:
|
|
"""Tests for game session operations"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_game_session(self, setup_database, db_ops, sample_game_id):
|
|
"""Test creating a game session"""
|
|
await 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"
|
|
)
|
|
|
|
session = await db_ops.create_game_session(sample_game_id)
|
|
|
|
assert session.game_id == sample_game_id
|
|
assert session.state_snapshot is None # Initially null
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_session_snapshot(self, setup_database, db_ops, sample_game_id):
|
|
"""Test updating session snapshot"""
|
|
await 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"
|
|
)
|
|
await db_ops.create_game_session(sample_game_id)
|
|
|
|
snapshot = {
|
|
"inning": 3,
|
|
"outs": 2,
|
|
"runners": [1, 3]
|
|
}
|
|
|
|
await db_ops.update_session_snapshot(sample_game_id, snapshot)
|
|
|
|
# Note: Would need to query session to verify, but this tests no errors
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_session_snapshot_nonexistent_raises_error(self, setup_database, db_ops):
|
|
"""Test updating nonexistent session raises error"""
|
|
fake_id = uuid4()
|
|
|
|
with pytest.raises(ValueError, match="not found"):
|
|
await db_ops.update_session_snapshot(fake_id, {})
|
|
|
|
|
|
class TestDatabaseOperationsRoster:
|
|
"""Tests for roster link operations"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_pd_roster_card(self, setup_database, db_ops, sample_game_id):
|
|
"""Test adding a PD card to roster"""
|
|
# Create game first
|
|
await db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="pd",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
# Add roster card
|
|
roster_data = await db_ops.add_pd_roster_card(
|
|
game_id=sample_game_id,
|
|
card_id=123,
|
|
team_id=1
|
|
)
|
|
|
|
assert roster_data.id is not None
|
|
assert roster_data.game_id == sample_game_id
|
|
assert roster_data.card_id == 123
|
|
assert roster_data.team_id == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_sba_roster_player(self, setup_database, db_ops, sample_game_id):
|
|
"""Test adding an SBA player to roster"""
|
|
# Create game first
|
|
await db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=10,
|
|
away_team_id=20,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
# Add roster player
|
|
roster_data = await db_ops.add_sba_roster_player(
|
|
game_id=sample_game_id,
|
|
player_id=456,
|
|
team_id=10
|
|
)
|
|
|
|
assert roster_data.id is not None
|
|
assert roster_data.game_id == sample_game_id
|
|
assert roster_data.player_id == 456
|
|
assert roster_data.team_id == 10
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_duplicate_pd_card_raises_error(self, setup_database, db_ops, sample_game_id):
|
|
"""Test adding duplicate PD card to roster fails"""
|
|
# Create game and add card
|
|
await db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="pd",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
await db_ops.add_pd_roster_card(
|
|
game_id=sample_game_id,
|
|
card_id=123,
|
|
team_id=1
|
|
)
|
|
|
|
# Try to add same card again - should fail
|
|
with pytest.raises(ValueError, match="Could not add card to roster"):
|
|
await db_ops.add_pd_roster_card(
|
|
game_id=sample_game_id,
|
|
card_id=123,
|
|
team_id=1
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_duplicate_sba_player_raises_error(self, setup_database, db_ops, sample_game_id):
|
|
"""Test adding duplicate SBA player to roster fails"""
|
|
# Create game and add player
|
|
await db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=10,
|
|
away_team_id=20,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
await db_ops.add_sba_roster_player(
|
|
game_id=sample_game_id,
|
|
player_id=456,
|
|
team_id=10
|
|
)
|
|
|
|
# Try to add same player again - should fail
|
|
with pytest.raises(ValueError, match="Could not add player to roster"):
|
|
await db_ops.add_sba_roster_player(
|
|
game_id=sample_game_id,
|
|
player_id=456,
|
|
team_id=10
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_pd_roster_all_teams(self, setup_database, db_ops, sample_game_id):
|
|
"""Test getting all PD cards for a game"""
|
|
# Create game and add cards for both teams
|
|
await db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="pd",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
await db_ops.add_pd_roster_card(sample_game_id, 101, 1)
|
|
await db_ops.add_pd_roster_card(sample_game_id, 102, 1)
|
|
await db_ops.add_pd_roster_card(sample_game_id, 201, 2)
|
|
|
|
# Get all roster entries
|
|
roster = await db_ops.get_pd_roster(sample_game_id)
|
|
|
|
assert len(roster) == 3
|
|
card_ids = {r.card_id for r in roster}
|
|
assert card_ids == {101, 102, 201}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_pd_roster_filtered_by_team(self, setup_database, db_ops, sample_game_id):
|
|
"""Test getting PD cards filtered by team"""
|
|
# Create game and add cards for both teams
|
|
await db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="pd",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
await db_ops.add_pd_roster_card(sample_game_id, 101, 1)
|
|
await db_ops.add_pd_roster_card(sample_game_id, 102, 1)
|
|
await db_ops.add_pd_roster_card(sample_game_id, 201, 2)
|
|
|
|
# Get team 1 roster
|
|
team1_roster = await db_ops.get_pd_roster(sample_game_id, team_id=1)
|
|
|
|
assert len(team1_roster) == 2
|
|
card_ids = {r.card_id for r in team1_roster}
|
|
assert card_ids == {101, 102}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_sba_roster_all_teams(self, setup_database, db_ops, sample_game_id):
|
|
"""Test getting all SBA players for a game"""
|
|
# Create game and add players for both teams
|
|
await db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=10,
|
|
away_team_id=20,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
await db_ops.add_sba_roster_player(sample_game_id, 401, 10)
|
|
await db_ops.add_sba_roster_player(sample_game_id, 402, 10)
|
|
await db_ops.add_sba_roster_player(sample_game_id, 501, 20)
|
|
|
|
# Get all roster entries
|
|
roster = await db_ops.get_sba_roster(sample_game_id)
|
|
|
|
assert len(roster) == 3
|
|
player_ids = {r.player_id for r in roster}
|
|
assert player_ids == {401, 402, 501}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_sba_roster_filtered_by_team(self, setup_database, db_ops, sample_game_id):
|
|
"""Test getting SBA players filtered by team"""
|
|
# Create game and add players for both teams
|
|
await db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=10,
|
|
away_team_id=20,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
await db_ops.add_sba_roster_player(sample_game_id, 401, 10)
|
|
await db_ops.add_sba_roster_player(sample_game_id, 402, 10)
|
|
await db_ops.add_sba_roster_player(sample_game_id, 501, 20)
|
|
|
|
# Get team 10 roster
|
|
team10_roster = await db_ops.get_sba_roster(sample_game_id, team_id=10)
|
|
|
|
assert len(team10_roster) == 2
|
|
player_ids = {r.player_id for r in team10_roster}
|
|
assert player_ids == {401, 402}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_roster_entry(self, setup_database, db_ops, sample_game_id):
|
|
"""Test removing a roster entry"""
|
|
# Create game and add card
|
|
await db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="pd",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
roster_data = await db_ops.add_pd_roster_card(
|
|
game_id=sample_game_id,
|
|
card_id=123,
|
|
team_id=1
|
|
)
|
|
|
|
# Remove it
|
|
await db_ops.remove_roster_entry(roster_data.id)
|
|
|
|
# Verify it's gone
|
|
roster = await db_ops.get_pd_roster(sample_game_id)
|
|
assert len(roster) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_nonexistent_roster_entry_raises_error(self, setup_database, db_ops):
|
|
"""Test removing nonexistent roster entry fails"""
|
|
fake_id = 999999
|
|
|
|
with pytest.raises(ValueError, match="not found"):
|
|
await db_ops.remove_roster_entry(fake_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_empty_pd_roster(self, setup_database, db_ops, sample_game_id):
|
|
"""Test getting PD roster for game with no cards"""
|
|
# Create game but don't add any cards
|
|
await db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="pd",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
roster = await db_ops.get_pd_roster(sample_game_id)
|
|
assert len(roster) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_empty_sba_roster(self, setup_database, db_ops, sample_game_id):
|
|
"""Test getting SBA roster for game with no players"""
|
|
# Create game but don't add any players
|
|
await db_ops.create_game(
|
|
game_id=sample_game_id,
|
|
league_id="sba",
|
|
home_team_id=10,
|
|
away_team_id=20,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
roster = await db_ops.get_sba_roster(sample_game_id)
|
|
assert len(roster) == 0
|
|
|
|
|
|
class TestDatabaseOperationsRollback:
|
|
"""Tests for database rollback operations (delete_plays_after, etc.)"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_plays_after(self, setup_database, db_ops, sample_game_id):
|
|
"""Test deleting plays after a specific play number"""
|
|
# Create game
|
|
await 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 entries for batter, pitcher, and catcher
|
|
batter = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
player_id=100,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
pitcher = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=2,
|
|
player_id=200,
|
|
position="P",
|
|
batting_order=None,
|
|
is_starter=True
|
|
)
|
|
catcher = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=2,
|
|
player_id=201,
|
|
position="C",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
|
|
# Create 5 plays
|
|
for play_num in range(1, 6):
|
|
await db_ops.save_play({
|
|
'game_id': sample_game_id,
|
|
'play_number': play_num,
|
|
'inning': 1,
|
|
'half': 'top',
|
|
'outs_before': 0,
|
|
'batter_id': batter.id,
|
|
'pitcher_id': pitcher.id,
|
|
'catcher_id': catcher.id,
|
|
'dice_roll': f'10+{play_num}',
|
|
'result_description': f'Play {play_num}',
|
|
'pa': 1,
|
|
'complete': True
|
|
})
|
|
|
|
# Delete plays after play 3
|
|
deleted_count = await db_ops.delete_plays_after(sample_game_id, 3)
|
|
|
|
assert deleted_count == 2 # Plays 4 and 5 deleted
|
|
|
|
# Verify only plays 1-3 remain
|
|
remaining_plays = await db_ops.get_plays(sample_game_id)
|
|
assert len(remaining_plays) == 3
|
|
assert all(p.play_number <= 3 for p in remaining_plays)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_plays_after_with_no_plays_to_delete(self, setup_database, db_ops, sample_game_id):
|
|
"""Test deleting plays when none exist after the threshold"""
|
|
# Create game
|
|
await 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 for play
|
|
batter = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
player_id=100,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
pitcher = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=2,
|
|
player_id=200,
|
|
position="P",
|
|
batting_order=None,
|
|
is_starter=True
|
|
)
|
|
catcher = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=2,
|
|
player_id=201,
|
|
position="C",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
|
|
# Create 3 plays
|
|
for play_num in range(1, 4):
|
|
await db_ops.save_play({
|
|
'game_id': sample_game_id,
|
|
'play_number': play_num,
|
|
'inning': 1,
|
|
'half': 'top',
|
|
'outs_before': 0,
|
|
'batter_id': batter.id,
|
|
'pitcher_id': pitcher.id,
|
|
'catcher_id': catcher.id,
|
|
'dice_roll': f'10+{play_num}',
|
|
'result_description': f'Play {play_num}',
|
|
'pa': 1,
|
|
'complete': True
|
|
})
|
|
|
|
# Delete plays after play 10 (none exist)
|
|
deleted_count = await db_ops.delete_plays_after(sample_game_id, 10)
|
|
|
|
assert deleted_count == 0
|
|
|
|
# Verify all 3 plays remain
|
|
remaining_plays = await db_ops.get_plays(sample_game_id)
|
|
assert len(remaining_plays) == 3
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_substitutions_after(self, setup_database, db_ops, sample_game_id):
|
|
"""Test deleting substitutions after a specific play number"""
|
|
# Create game
|
|
await 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 starter
|
|
starter = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
player_id=100,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
|
|
# Create substitutions - need to manually set substitution fields
|
|
sub1 = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
player_id=101,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=False
|
|
)
|
|
sub2 = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
player_id=102,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=False
|
|
)
|
|
sub3 = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
player_id=103,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=False
|
|
)
|
|
|
|
# Manually set substitution fields using SQLAlchemy
|
|
from app.database.session import AsyncSessionLocal
|
|
from app.models.db_models import Lineup
|
|
from sqlalchemy import select, update
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
# Update starter - mark as inactive
|
|
await session.execute(
|
|
update(Lineup)
|
|
.where(Lineup.id == starter.id)
|
|
.values(is_active=False, after_play=None)
|
|
)
|
|
|
|
# Update sub1 - substituted at play 5
|
|
await session.execute(
|
|
update(Lineup)
|
|
.where(Lineup.id == sub1.id)
|
|
.values(is_active=False, entered_inning=3, after_play=5, replacing_id=starter.id)
|
|
)
|
|
|
|
# Update sub2 - substituted at play 10
|
|
await session.execute(
|
|
update(Lineup)
|
|
.where(Lineup.id == sub2.id)
|
|
.values(is_active=False, entered_inning=5, after_play=10, replacing_id=sub1.id)
|
|
)
|
|
|
|
# Update sub3 - substituted at play 15
|
|
await session.execute(
|
|
update(Lineup)
|
|
.where(Lineup.id == sub3.id)
|
|
.values(is_active=True, entered_inning=7, after_play=15, replacing_id=sub2.id)
|
|
)
|
|
|
|
await session.commit()
|
|
|
|
# Delete substitutions after play 10 (>= 10, so deletes sub2 and sub3)
|
|
deleted_count = await db_ops.delete_substitutions_after(sample_game_id, 10)
|
|
|
|
assert deleted_count == 2 # sub2 (after play 10) and sub3 (after play 15) deleted
|
|
|
|
# Verify lineup state - need to get ALL lineup entries, not just active
|
|
from app.database.session import AsyncSessionLocal
|
|
from app.models.db_models import Lineup
|
|
from sqlalchemy import select
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
result = await session.execute(
|
|
select(Lineup)
|
|
.where(
|
|
Lineup.game_id == sample_game_id,
|
|
Lineup.team_id == 1
|
|
)
|
|
)
|
|
all_lineup = list(result.scalars().all())
|
|
|
|
# Should have starter + 1 sub (sub1 only)
|
|
assert len([p for p in all_lineup if p.after_play is not None]) == 1
|
|
# The remaining sub should be sub1 (after_play=5)
|
|
remaining_sub = [p for p in all_lineup if p.after_play is not None][0]
|
|
assert remaining_sub.after_play == 5
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_complete_rollback_scenario(self, setup_database, db_ops, sample_game_id):
|
|
"""Test complete rollback scenario: plays + substitutions"""
|
|
# Create game
|
|
await 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
|
|
batter = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
player_id=100,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
pitcher = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=2,
|
|
player_id=200,
|
|
position="P",
|
|
batting_order=None,
|
|
is_starter=True
|
|
)
|
|
catcher = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=2,
|
|
player_id=201,
|
|
position="C",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
|
|
# Create 10 plays
|
|
for play_num in range(1, 11):
|
|
await db_ops.save_play({
|
|
'game_id': sample_game_id,
|
|
'play_number': play_num,
|
|
'inning': (play_num - 1) // 3 + 1,
|
|
'half': 'top' if play_num % 2 == 1 else 'bot',
|
|
'outs_before': 0,
|
|
'batter_id': batter.id,
|
|
'pitcher_id': pitcher.id,
|
|
'catcher_id': catcher.id,
|
|
'dice_roll': f'10+{play_num}',
|
|
'result_description': f'Play {play_num}',
|
|
'pa': 1,
|
|
'complete': True
|
|
})
|
|
|
|
# Create substitution at play 7
|
|
sub = await db_ops.add_sba_lineup_player(
|
|
game_id=sample_game_id,
|
|
team_id=1,
|
|
player_id=101,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=False
|
|
)
|
|
|
|
# Manually set substitution fields
|
|
from app.database.session import AsyncSessionLocal
|
|
from app.models.db_models import Lineup
|
|
from sqlalchemy import update
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
await session.execute(
|
|
update(Lineup)
|
|
.where(Lineup.id == sub.id)
|
|
.values(is_active=True, entered_inning=3, after_play=7, replacing_id=batter.id)
|
|
)
|
|
await session.commit()
|
|
|
|
# Rollback to play 5 (delete everything after play 5)
|
|
rollback_point = 5
|
|
|
|
plays_deleted = await db_ops.delete_plays_after(sample_game_id, rollback_point)
|
|
subs_deleted = await db_ops.delete_substitutions_after(sample_game_id, rollback_point)
|
|
|
|
# Verify deletions
|
|
assert plays_deleted == 5 # Plays 6-10 deleted
|
|
assert subs_deleted == 1 # Substitution at play 7 deleted
|
|
|
|
# Verify remaining data
|
|
remaining_plays = await db_ops.get_plays(sample_game_id)
|
|
assert len(remaining_plays) == 5
|
|
assert max(p.play_number for p in remaining_plays) == 5
|