e058bc4a6c
- Add player_positions JSONB column to roster_links (migration 006) - Add player_data JSONB column to cache name/image/headshot (migration 007) - Add is_pitcher/is_batter computed properties for two-way player support - Update lineup submission to populate RosterLink with all players + positions - Update get_bench handler to use cached data (no runtime API calls) - Add BenchPlayer type to frontend with proper filtering - Add new Lineup components: InlineSubstitutionPanel, LineupSlotRow, PositionSelector, UnifiedLineupTab - Add integration tests for get_bench_players Bench players now load instantly without API dependency, and properly filter batters vs pitchers (including CP closer position). Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
1198 lines
37 KiB
Python
1198 lines
37 KiB
Python
"""
|
|
Integration tests for DatabaseOperations.
|
|
|
|
Tests actual database operations using the test database.
|
|
All fixtures are from tests/integration/conftest.py.
|
|
|
|
Key Features:
|
|
- Session injection pattern (db_ops fixture has injected session)
|
|
- Each test uses same session (no connection conflicts)
|
|
- Automatic rollback after each test (isolation)
|
|
|
|
Author: Claude
|
|
Date: 2025-10-22
|
|
"""
|
|
|
|
import pytest
|
|
from uuid import uuid4
|
|
|
|
from sqlalchemy import update, select
|
|
|
|
from app.database.operations import DatabaseOperations
|
|
from app.models.db_models import Lineup
|
|
|
|
|
|
# Mark all tests in this module as integration tests
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
class TestDatabaseOperationsGame:
|
|
"""Tests for game CRUD operations"""
|
|
|
|
async def test_create_game(self, db_ops, db_session):
|
|
"""Test creating a game in database"""
|
|
game_id = uuid4()
|
|
game = await db_ops.create_game(
|
|
game_id=game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
assert game.id == game_id
|
|
assert game.league_id == "sba"
|
|
assert game.status == "pending"
|
|
assert game.home_team_id == 1
|
|
assert game.away_team_id == 2
|
|
|
|
async def test_create_game_with_ai(self, db_ops, db_session):
|
|
"""Test creating a game with AI opponent"""
|
|
game_id = uuid4()
|
|
game = await db_ops.create_game(
|
|
game_id=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"
|
|
|
|
async def test_get_game(self, db_ops, db_session):
|
|
"""Test retrieving a game from database"""
|
|
game_id = uuid4()
|
|
# Create game
|
|
await db_ops.create_game(
|
|
game_id=game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Retrieve game
|
|
retrieved = await db_ops.get_game(game_id)
|
|
|
|
assert retrieved is not None
|
|
assert retrieved.id == game_id
|
|
assert retrieved.league_id == "sba"
|
|
|
|
async def test_get_game_nonexistent(self, db_ops):
|
|
"""Test retrieving nonexistent game returns None"""
|
|
fake_id = uuid4()
|
|
game = await db_ops.get_game(fake_id)
|
|
|
|
assert game is None
|
|
|
|
async def test_update_game_state(self, db_ops, db_session):
|
|
"""Test updating game state"""
|
|
game_id = uuid4()
|
|
# Create game
|
|
await db_ops.create_game(
|
|
game_id=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=game_id,
|
|
inning=5,
|
|
half="bottom",
|
|
home_score=3,
|
|
away_score=2,
|
|
status="active"
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Verify update
|
|
game = await db_ops.get_game(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"
|
|
|
|
async def test_update_game_state_nonexistent_raises_error(self, 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"""
|
|
|
|
async def test_add_sba_lineup_player(self, db_ops, db_session):
|
|
"""Test adding SBA player to lineup"""
|
|
game_id = uuid4()
|
|
# Create game first
|
|
await db_ops.create_game(
|
|
game_id=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=game_id,
|
|
team_id=1,
|
|
player_id=101,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
|
|
assert lineup.game_id == 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
|
|
|
|
async def test_add_pd_lineup_card(self, db_ops, db_session):
|
|
"""Test adding PD card to lineup"""
|
|
game_id = uuid4()
|
|
await db_ops.create_game(
|
|
game_id=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=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
|
|
|
|
async def test_get_active_lineup(self, db_ops, db_session):
|
|
"""Test retrieving active lineup for a team"""
|
|
game_id = uuid4()
|
|
await db_ops.create_game(
|
|
game_id=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=game_id,
|
|
team_id=1,
|
|
player_id=103,
|
|
position="1B",
|
|
batting_order=3
|
|
)
|
|
await db_ops.add_sba_lineup_player(
|
|
game_id=game_id,
|
|
team_id=1,
|
|
player_id=101,
|
|
position="CF",
|
|
batting_order=1
|
|
)
|
|
await db_ops.add_sba_lineup_player(
|
|
game_id=game_id,
|
|
team_id=1,
|
|
player_id=102,
|
|
position="SS",
|
|
batting_order=2
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Retrieve lineup
|
|
lineup = await db_ops.get_active_lineup(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
|
|
|
|
async def test_get_active_lineup_empty(self, db_ops, db_session):
|
|
"""Test retrieving lineup for team with no entries"""
|
|
game_id = uuid4()
|
|
await db_ops.create_game(
|
|
game_id=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(game_id, team_id=1)
|
|
|
|
assert lineup == []
|
|
|
|
|
|
class TestDatabaseOperationsPlays:
|
|
"""Tests for play operations"""
|
|
|
|
async def test_save_play(self, db_ops, db_session):
|
|
"""Test saving a play"""
|
|
game_id = uuid4()
|
|
await db_ops.create_game(
|
|
game_id=game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
# Create lineup entries for required foreign keys
|
|
batter = await db_ops.add_sba_lineup_player(
|
|
game_id=game_id, team_id=1, player_id=100, position="CF", batting_order=1
|
|
)
|
|
pitcher = await db_ops.add_sba_lineup_player(
|
|
game_id=game_id, team_id=2, player_id=200, position="P", batting_order=None
|
|
)
|
|
catcher = await db_ops.add_sba_lineup_player(
|
|
game_id=game_id, team_id=2, player_id=201, position="C", batting_order=1
|
|
)
|
|
await db_session.flush()
|
|
|
|
play_data = {
|
|
"game_id": game_id,
|
|
"play_number": 1,
|
|
"inning": 1,
|
|
"half": "top",
|
|
"outs_before": 0,
|
|
"batting_order": 1,
|
|
"result_description": "Single to left field",
|
|
"batter_id": batter.id,
|
|
"pitcher_id": pitcher.id,
|
|
"catcher_id": catcher.id,
|
|
"pa": 1,
|
|
"ab": 1,
|
|
"hit": 1
|
|
}
|
|
|
|
play_id = await db_ops.save_play(play_data)
|
|
await db_session.flush()
|
|
|
|
# Verify play was saved
|
|
plays = await db_ops.get_plays(game_id)
|
|
assert len(plays) == 1
|
|
assert plays[0].play_number == 1
|
|
assert plays[0].result_description == "Single to left field"
|
|
|
|
async def test_get_plays(self, db_ops, db_session):
|
|
"""Test retrieving plays for a game"""
|
|
game_id = uuid4()
|
|
await db_ops.create_game(
|
|
game_id=game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
# Create lineup entries for required foreign keys
|
|
batter = await db_ops.add_sba_lineup_player(
|
|
game_id=game_id, team_id=1, player_id=100, position="CF", batting_order=1
|
|
)
|
|
pitcher = await db_ops.add_sba_lineup_player(
|
|
game_id=game_id, team_id=2, player_id=200, position="P", batting_order=None
|
|
)
|
|
catcher = await db_ops.add_sba_lineup_player(
|
|
game_id=game_id, team_id=2, player_id=201, position="C", batting_order=1
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Save multiple plays
|
|
for i in range(3):
|
|
await db_ops.save_play({
|
|
"game_id": game_id,
|
|
"play_number": i + 1,
|
|
"inning": 1,
|
|
"half": "top",
|
|
"outs_before": i,
|
|
"batting_order": i + 1,
|
|
"result_description": f"Play {i+1}",
|
|
"batter_id": batter.id,
|
|
"pitcher_id": pitcher.id,
|
|
"catcher_id": catcher.id,
|
|
"pa": 1
|
|
})
|
|
await db_session.flush()
|
|
|
|
# Retrieve plays
|
|
plays = await db_ops.get_plays(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
|
|
|
|
async def test_get_plays_empty(self, db_ops, db_session):
|
|
"""Test retrieving plays for game with no plays"""
|
|
game_id = uuid4()
|
|
await db_ops.create_game(
|
|
game_id=game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
plays = await db_ops.get_plays(game_id)
|
|
|
|
assert plays == []
|
|
|
|
|
|
class TestDatabaseOperationsRecovery:
|
|
"""Tests for game state recovery"""
|
|
|
|
async def test_load_game_state_complete(self, db_ops, db_session):
|
|
"""Test loading complete game state"""
|
|
game_id = uuid4()
|
|
# Create game
|
|
await db_ops.create_game(
|
|
game_id=game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
# Add lineup entries (use add_sba_lineup_player instead of create_lineup_entry)
|
|
batter = await db_ops.add_sba_lineup_player(
|
|
game_id=game_id,
|
|
team_id=1,
|
|
player_id=101,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
pitcher = await db_ops.add_sba_lineup_player(
|
|
game_id=game_id,
|
|
team_id=2,
|
|
player_id=201,
|
|
position="P",
|
|
batting_order=None,
|
|
is_starter=True
|
|
)
|
|
catcher = await db_ops.add_sba_lineup_player(
|
|
game_id=game_id,
|
|
team_id=2,
|
|
player_id=202,
|
|
position="C",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Add play
|
|
await db_ops.save_play({
|
|
"game_id": game_id,
|
|
"play_number": 1,
|
|
"inning": 1,
|
|
"half": "top",
|
|
"outs_before": 0,
|
|
"batting_order": 1,
|
|
"result_description": "Single",
|
|
"batter_id": batter.id,
|
|
"pitcher_id": pitcher.id,
|
|
"catcher_id": catcher.id,
|
|
"pa": 1
|
|
})
|
|
|
|
# Update game state
|
|
await db_ops.update_game_state(
|
|
game_id=game_id,
|
|
inning=2,
|
|
half="bottom",
|
|
home_score=1,
|
|
away_score=0
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Load complete state
|
|
state = await db_ops.load_game_state(game_id)
|
|
|
|
assert state is not None
|
|
assert state["game"]["id"] == game_id
|
|
assert state["game"]["current_inning"] == 2
|
|
assert state["game"]["current_half"] == "bottom"
|
|
assert len(state["lineups"]) == 3 # batter, pitcher, catcher
|
|
assert len(state["plays"]) == 1
|
|
|
|
async def test_load_game_state_nonexistent(self, 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"""
|
|
|
|
async def test_create_game_session(self, db_ops, db_session):
|
|
"""Test creating a game session"""
|
|
game_id = uuid4()
|
|
await db_ops.create_game(
|
|
game_id=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(game_id)
|
|
|
|
assert session.game_id == game_id
|
|
assert session.state_snapshot == {} # Initially empty dict (DB default)
|
|
|
|
async def test_update_session_snapshot(self, db_ops, db_session):
|
|
"""Test updating session snapshot"""
|
|
game_id = uuid4()
|
|
await db_ops.create_game(
|
|
game_id=game_id,
|
|
league_id="sba",
|
|
home_team_id=1,
|
|
away_team_id=2,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
await db_ops.create_game_session(game_id)
|
|
|
|
snapshot = {
|
|
"inning": 3,
|
|
"outs": 2,
|
|
"runners": [1, 3]
|
|
}
|
|
|
|
await db_ops.update_session_snapshot(game_id, snapshot)
|
|
# No error = success
|
|
|
|
async def test_update_session_snapshot_nonexistent_raises_error(self, 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"""
|
|
|
|
async def test_add_pd_roster_card(self, db_ops, db_session):
|
|
"""Test adding a PD card to roster"""
|
|
game_id = uuid4()
|
|
# Create game first
|
|
await db_ops.create_game(
|
|
game_id=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=game_id,
|
|
card_id=123,
|
|
team_id=1
|
|
)
|
|
|
|
assert roster_data.id is not None
|
|
assert roster_data.game_id == game_id
|
|
assert roster_data.card_id == 123
|
|
assert roster_data.team_id == 1
|
|
|
|
async def test_add_sba_roster_player(self, db_ops, db_session):
|
|
"""Test adding an SBA player to roster"""
|
|
game_id = uuid4()
|
|
# Create game first
|
|
await db_ops.create_game(
|
|
game_id=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=game_id,
|
|
player_id=456,
|
|
team_id=10
|
|
)
|
|
|
|
assert roster_data.id is not None
|
|
assert roster_data.game_id == game_id
|
|
assert roster_data.player_id == 456
|
|
assert roster_data.team_id == 10
|
|
|
|
async def test_get_pd_roster_all_teams(self, db_ops, db_session):
|
|
"""Test getting all PD cards for a game"""
|
|
game_id = uuid4()
|
|
# Create game and add cards for both teams
|
|
await db_ops.create_game(
|
|
game_id=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, 101, 1)
|
|
await db_ops.add_pd_roster_card(game_id, 102, 1)
|
|
await db_ops.add_pd_roster_card(game_id, 201, 2)
|
|
await db_session.flush()
|
|
|
|
# Get all roster entries
|
|
roster = await db_ops.get_pd_roster(game_id)
|
|
|
|
assert len(roster) == 3
|
|
card_ids = {r.card_id for r in roster}
|
|
assert card_ids == {101, 102, 201}
|
|
|
|
async def test_get_pd_roster_filtered_by_team(self, db_ops, db_session):
|
|
"""Test getting PD cards filtered by team"""
|
|
game_id = uuid4()
|
|
# Create game and add cards for both teams
|
|
await db_ops.create_game(
|
|
game_id=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, 101, 1)
|
|
await db_ops.add_pd_roster_card(game_id, 102, 1)
|
|
await db_ops.add_pd_roster_card(game_id, 201, 2)
|
|
await db_session.flush()
|
|
|
|
# Get team 1 roster
|
|
team1_roster = await db_ops.get_pd_roster(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}
|
|
|
|
async def test_get_sba_roster_all_teams(self, db_ops, db_session):
|
|
"""Test getting all SBA players for a game"""
|
|
game_id = uuid4()
|
|
# Create game and add players for both teams
|
|
await db_ops.create_game(
|
|
game_id=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, 401, 10)
|
|
await db_ops.add_sba_roster_player(game_id, 402, 10)
|
|
await db_ops.add_sba_roster_player(game_id, 501, 20)
|
|
await db_session.flush()
|
|
|
|
# Get all roster entries
|
|
roster = await db_ops.get_sba_roster(game_id)
|
|
|
|
assert len(roster) == 3
|
|
player_ids = {r.player_id for r in roster}
|
|
assert player_ids == {401, 402, 501}
|
|
|
|
async def test_get_sba_roster_filtered_by_team(self, db_ops, db_session):
|
|
"""Test getting SBA players filtered by team"""
|
|
game_id = uuid4()
|
|
# Create game and add players for both teams
|
|
await db_ops.create_game(
|
|
game_id=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, 401, 10)
|
|
await db_ops.add_sba_roster_player(game_id, 402, 10)
|
|
await db_ops.add_sba_roster_player(game_id, 501, 20)
|
|
await db_session.flush()
|
|
|
|
# Get team 10 roster
|
|
team10_roster = await db_ops.get_sba_roster(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}
|
|
|
|
async def test_remove_roster_entry(self, db_ops, db_session):
|
|
"""Test removing a roster entry"""
|
|
game_id = uuid4()
|
|
# Create game and add card
|
|
await db_ops.create_game(
|
|
game_id=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=game_id,
|
|
card_id=123,
|
|
team_id=1
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Remove it
|
|
await db_ops.remove_roster_entry(roster_data.id)
|
|
await db_session.flush()
|
|
|
|
# Verify it's gone
|
|
roster = await db_ops.get_pd_roster(game_id)
|
|
assert len(roster) == 0
|
|
|
|
async def test_remove_nonexistent_roster_entry_raises_error(self, 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)
|
|
|
|
async def test_get_empty_pd_roster(self, db_ops, db_session):
|
|
"""Test getting PD roster for game with no cards"""
|
|
game_id = uuid4()
|
|
# Create game but don't add any cards
|
|
await db_ops.create_game(
|
|
game_id=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(game_id)
|
|
assert len(roster) == 0
|
|
|
|
async def test_get_empty_sba_roster(self, db_ops, db_session):
|
|
"""Test getting SBA roster for game with no players"""
|
|
game_id = uuid4()
|
|
# Create game but don't add any players
|
|
await db_ops.create_game(
|
|
game_id=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(game_id)
|
|
assert len(roster) == 0
|
|
|
|
async def test_get_bench_players(self, db_ops, db_session):
|
|
"""
|
|
Test get_bench_players returns roster players NOT in active lineup.
|
|
|
|
This verifies the RosterLink refactor where:
|
|
- RosterLink contains ALL eligible players with player_positions
|
|
- Lineup contains only ACTIVE players
|
|
- Bench = RosterLink players NOT IN Lineup
|
|
|
|
Also tests computed is_pitcher/is_batter properties.
|
|
"""
|
|
game_id = uuid4()
|
|
team_id = 10
|
|
|
|
# Create game
|
|
await db_ops.create_game(
|
|
game_id=game_id,
|
|
league_id="sba",
|
|
home_team_id=team_id,
|
|
away_team_id=20,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
# Add players to RosterLink with player_positions
|
|
# Player 101: Pitcher only
|
|
await db_ops.add_sba_roster_player(
|
|
game_id=game_id,
|
|
player_id=101,
|
|
team_id=team_id,
|
|
player_positions=["SP", "RP"]
|
|
)
|
|
# Player 102: Batter only (shortstop)
|
|
await db_ops.add_sba_roster_player(
|
|
game_id=game_id,
|
|
player_id=102,
|
|
team_id=team_id,
|
|
player_positions=["SS", "2B"]
|
|
)
|
|
# Player 103: Two-way player (pitcher and DH)
|
|
await db_ops.add_sba_roster_player(
|
|
game_id=game_id,
|
|
player_id=103,
|
|
team_id=team_id,
|
|
player_positions=["SP", "DH"]
|
|
)
|
|
# Player 104: Outfielder (will be in active lineup)
|
|
await db_ops.add_sba_roster_player(
|
|
game_id=game_id,
|
|
player_id=104,
|
|
team_id=team_id,
|
|
player_positions=["CF", "RF"]
|
|
)
|
|
|
|
# Add player 104 to ACTIVE lineup (not bench)
|
|
await db_ops.add_sba_lineup_player(
|
|
game_id=game_id,
|
|
team_id=team_id,
|
|
player_id=104,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Get bench players (should be 101, 102, 103 - NOT 104)
|
|
bench = await db_ops.get_bench_players(game_id, team_id)
|
|
|
|
# Verify count
|
|
assert len(bench) == 3
|
|
|
|
# Verify player IDs (104 should NOT be in bench)
|
|
bench_player_ids = {p.player_id for p in bench}
|
|
assert bench_player_ids == {101, 102, 103}
|
|
assert 104 not in bench_player_ids
|
|
|
|
# Verify computed properties for each player
|
|
bench_by_id = {p.player_id: p for p in bench}
|
|
|
|
# Player 101: Pitcher only
|
|
assert bench_by_id[101].is_pitcher is True
|
|
assert bench_by_id[101].is_batter is False
|
|
assert bench_by_id[101].player_positions == ["SP", "RP"]
|
|
|
|
# Player 102: Batter only
|
|
assert bench_by_id[102].is_pitcher is False
|
|
assert bench_by_id[102].is_batter is True
|
|
assert bench_by_id[102].player_positions == ["SS", "2B"]
|
|
|
|
# Player 103: Two-way player (BOTH is_pitcher AND is_batter)
|
|
assert bench_by_id[103].is_pitcher is True
|
|
assert bench_by_id[103].is_batter is True
|
|
assert bench_by_id[103].player_positions == ["SP", "DH"]
|
|
|
|
async def test_get_bench_players_empty(self, db_ops, db_session):
|
|
"""
|
|
Test get_bench_players returns empty list when all roster players are in lineup.
|
|
"""
|
|
game_id = uuid4()
|
|
team_id = 10
|
|
|
|
# Create game
|
|
await db_ops.create_game(
|
|
game_id=game_id,
|
|
league_id="sba",
|
|
home_team_id=team_id,
|
|
away_team_id=20,
|
|
game_mode="friendly",
|
|
visibility="public"
|
|
)
|
|
|
|
# Add player to roster
|
|
await db_ops.add_sba_roster_player(
|
|
game_id=game_id,
|
|
player_id=101,
|
|
team_id=team_id,
|
|
player_positions=["CF"]
|
|
)
|
|
|
|
# Add same player to active lineup
|
|
await db_ops.add_sba_lineup_player(
|
|
game_id=game_id,
|
|
team_id=team_id,
|
|
player_id=101,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Get bench players (should be empty)
|
|
bench = await db_ops.get_bench_players(game_id, team_id)
|
|
|
|
assert len(bench) == 0
|
|
|
|
|
|
class TestDatabaseOperationsRollback:
|
|
"""Tests for database rollback operations (delete_plays_after, etc.)"""
|
|
|
|
async def test_delete_plays_after(self, db_ops, db_session):
|
|
"""Test deleting plays after a specific play number"""
|
|
game_id = uuid4()
|
|
# Create game
|
|
await db_ops.create_game(
|
|
game_id=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=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=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=game_id,
|
|
team_id=2,
|
|
player_id=201,
|
|
position="C",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Create 5 plays
|
|
for play_num in range(1, 6):
|
|
await db_ops.save_play({
|
|
'game_id': 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
|
|
})
|
|
await db_session.flush()
|
|
|
|
# Delete plays after play 3
|
|
deleted_count = await db_ops.delete_plays_after(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(game_id)
|
|
assert len(remaining_plays) == 3
|
|
assert all(p.play_number <= 3 for p in remaining_plays)
|
|
|
|
async def test_delete_plays_after_with_no_plays_to_delete(self, db_ops, db_session):
|
|
"""Test deleting plays when none exist after the threshold"""
|
|
game_id = uuid4()
|
|
# Create game
|
|
await db_ops.create_game(
|
|
game_id=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=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=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=game_id,
|
|
team_id=2,
|
|
player_id=201,
|
|
position="C",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Create 3 plays
|
|
for play_num in range(1, 4):
|
|
await db_ops.save_play({
|
|
'game_id': 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
|
|
})
|
|
await db_session.flush()
|
|
|
|
# Delete plays after play 10 (none exist)
|
|
deleted_count = await db_ops.delete_plays_after(game_id, 10)
|
|
|
|
assert deleted_count == 0
|
|
|
|
# Verify all 3 plays remain
|
|
remaining_plays = await db_ops.get_plays(game_id)
|
|
assert len(remaining_plays) == 3
|
|
|
|
async def test_delete_substitutions_after(self, db_ops, db_session):
|
|
"""Test deleting substitutions after a specific play number"""
|
|
game_id = uuid4()
|
|
# Create game
|
|
await db_ops.create_game(
|
|
game_id=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=game_id,
|
|
team_id=1,
|
|
player_id=100,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
|
|
# Create substitutions
|
|
sub1 = await db_ops.add_sba_lineup_player(
|
|
game_id=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=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=game_id,
|
|
team_id=1,
|
|
player_id=103,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=False
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Manually set substitution fields using the test session
|
|
await db_session.execute(
|
|
update(Lineup)
|
|
.where(Lineup.id == starter.id)
|
|
.values(is_active=False, after_play=None)
|
|
)
|
|
await db_session.execute(
|
|
update(Lineup)
|
|
.where(Lineup.id == sub1.id)
|
|
.values(is_active=False, entered_inning=3, after_play=5, replacing_id=starter.id)
|
|
)
|
|
await db_session.execute(
|
|
update(Lineup)
|
|
.where(Lineup.id == sub2.id)
|
|
.values(is_active=False, entered_inning=5, after_play=10, replacing_id=sub1.id)
|
|
)
|
|
await db_session.execute(
|
|
update(Lineup)
|
|
.where(Lineup.id == sub3.id)
|
|
.values(is_active=True, entered_inning=7, after_play=15, replacing_id=sub2.id)
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Delete substitutions after play 10 (>= 10, so deletes sub2 and sub3)
|
|
deleted_count = await db_ops.delete_substitutions_after(game_id, 10)
|
|
|
|
assert deleted_count == 2 # sub2 (after play 10) and sub3 (after play 15) deleted
|
|
|
|
# Verify lineup state
|
|
result = await db_session.execute(
|
|
select(Lineup)
|
|
.where(
|
|
Lineup.game_id == 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
|
|
|
|
async def test_complete_rollback_scenario(self, db_ops, db_session):
|
|
"""Test complete rollback scenario: plays + substitutions"""
|
|
game_id = uuid4()
|
|
# Create game
|
|
await db_ops.create_game(
|
|
game_id=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=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=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=game_id,
|
|
team_id=2,
|
|
player_id=201,
|
|
position="C",
|
|
batting_order=1,
|
|
is_starter=True
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Create 10 plays
|
|
for play_num in range(1, 11):
|
|
await db_ops.save_play({
|
|
'game_id': 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
|
|
})
|
|
await db_session.flush()
|
|
|
|
# Create substitution at play 7
|
|
sub = await db_ops.add_sba_lineup_player(
|
|
game_id=game_id,
|
|
team_id=1,
|
|
player_id=101,
|
|
position="CF",
|
|
batting_order=1,
|
|
is_starter=False
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Manually set substitution fields
|
|
await db_session.execute(
|
|
update(Lineup)
|
|
.where(Lineup.id == sub.id)
|
|
.values(is_active=True, entered_inning=3, after_play=7, replacing_id=batter.id)
|
|
)
|
|
await db_session.flush()
|
|
|
|
# Rollback to play 5 (delete everything after play 5)
|
|
rollback_point = 5
|
|
|
|
plays_deleted = await db_ops.delete_plays_after(game_id, rollback_point)
|
|
subs_deleted = await db_ops.delete_substitutions_after(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(game_id)
|
|
assert len(remaining_plays) == 5
|
|
assert max(p.play_number for p in remaining_plays) == 5
|