cal/paper-dynasty-discord
1
0
Fork 0
Code Issues 13 Pull Requests 4 Actions Packages Projects Releases Wiki Activity

Add Player tests

Browse Source 
Continuing /new-game
This commit is contained in:
Cal Corum 2024-10-12 22:08:36 -05:00
parent 512efe98c4
commit 5ab2080f6d
9 changed files with 201 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
cogs/admins.py
View File

@ -1,15 +1,16 @@
import csv
import json
import api_calls
import db_calls_gameplay
from helpers import *
from api_calls import *
from discord import Member
from discord.ext import commands, tasks
from discord import app_commands
from api_calls import *
from helpers import *
import in_game
from in_game import ai_manager
from in_game.gameplay_models import Session, select, engine, Game, Cardset, Lineup, Team, Player
# date = f'{datetime.datetime.now().year}-{datetime.datetime.now().month}-{datetime.datetime.now().day}'
@ -559,6 +560,23 @@ class Admins(commands.Cog):
view = DropdownView([my_dropdown])
await ctx.send(f'Here is your dropdown:', view=view)
@commands.command(name='db-check', help='Mod: Check cached db volume')
@commands.is_owner()
async def db_check_command(self, ctx):
message = await ctx.send('I\'ll take a quick peek at the db and see what you\'ve got...')
with Session(engine) as session:
games = session.exec(select(Game.id)).all()
cardsets = session.exec(select(Cardset.id)).all()
lineups = session.exec(select(Lineup.id)).all()
teams = session.exec(select(Team.id)).all()
# players = session.exec(select(Player.id)).all()
players = []
output = f'## Database Counts\nGames: {len(games)}\nCardsets: {len(cardsets)}\nLineups: {len(lineups)}\nTeams: {len(teams)}\nPlayers: {len(players)}'
await message.edit(content=output)
async def setup(bot):
await bot.add_cog(Admins(bot))

33
cogs/gameplay.py
View File

@ -9,9 +9,9 @@ from discord.ext import commands
from api_calls import db_get
from helpers import PD_PLAYERS_ROLE_NAME, user_has_role
from in_game.game_helpers import PUBLIC_FIELDS_CATEGORY_NAME
from in_game.game_helpers import PUBLIC_FIELDS_CATEGORY_NAME, legal_check
from in_game.data_cache import get_pd_team
from in_game.gameplay_models import Session, engine, create_db_and_tables, select, Game, get_team
from in_game.gameplay_models import Lineup, Session, engine, create_db_and_tables, get_player, player_description, select, Game, get_team
from in_game.gameplay_queries import get_channel_game_or_none, get_games_by_team_id
@ -142,6 +142,35 @@ class Gameplay(commands.Cog):
await interaction.channel.send(content=game_info_log)
# Get Human SP card
human_sp_card = await db_get('cards', object_id=sp_card_id)
id_key = 'id'
if 'player_id' in human_sp_card['player']:
id_key = 'player_id'
human_sp_player = await get_player(session, human_sp_card['player'][id_key])
if human_sp_card['team']['id'] != human_team.id:
logging.error(f'Card_id {sp_card_id} does not belong to {human_team["abbrev"]} in Game {this_game.id}')
await interaction.channel.send(
f'Uh oh. Card ID {sp_card_id} is {human_sp_player.name} and belongs to {human_sp_card["team"]["sname"]}. Will you double check that before we get started?'
)
return
legal_data = await legal_check([sp_card_id], difficulty_name=league.name)
if not legal_data['legal']:
await interaction.edit_original_response(
content=f'It looks like this is a Ranked Legal game and {player_description(human_sp_player)} is not legal in {league.value} games. You can start a new game once you pick a new SP.'
)
return
human_sp_lineup = Lineup(
team_id=human_team.id,
player_id=human_sp_player.id,
card_id=sp_card_id,
position='P',
batting_order=10,
is_fatigued=False,
game=this_game
)
# Get AI SP

71
in_game/gameplay_models.py
View File

@ -10,6 +10,7 @@ from helpers import PD_SEASON
sqlite_url = 'sqlite:///storage/gameplay.db'
connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=False, connect_args=connect_args)
CACHE_LIMIT = 1209600 # in seconds
class GameCardsetLink(SQLModel, table=True):
@ -55,6 +56,7 @@ class CardsetBase(SQLModel):
class Cardset(CardsetBase, table=True):
game_links: list[GameCardsetLink] = Relationship(back_populates='cardset')
players: list['Player'] = Relationship(back_populates='cardset')
class Lineup(SQLModel, table=True):
@ -106,7 +108,7 @@ async def get_team(
session: Session, team_id: int | None = None, gm_id: int | None = None, team_abbrev: str | None = None, skip_cache: bool = False) -> Team:
if team_id is None and gm_id is None and team_abbrev is None:
err = 'One of "team_id", "gm_id", or "team_abbrev" must be included in search'
logging.error(f'gameplay_db - get_team - {err}')
logging.error(f'gameplay_models - get_team - {err}')
raise TypeError(err)
if not skip_cache:
@ -123,18 +125,18 @@ async def get_team(
logging.debug(f'we found a team: {this_team} / created: {this_team.created}')
tdelta = datetime.datetime.now() - this_team.created
logging.debug(f'tdelta: {tdelta}')
if tdelta.total_seconds() < 1209600:
if tdelta.total_seconds() < CACHE_LIMIT:
return this_team
else:
session.delete(this_team)
session.commit()
def cache_team(json_data: dict) -> Team:
# logging.info(f'gameplay_db - get_team - cache_team - writing a team to cache: {json_data}')
# logging.info(f'gameplay_models - get_team - cache_team - writing a team to cache: {json_data}')
valid_team = TeamBase.model_validate(json_data, from_attributes=True)
# logging.info(f'gameplay_db - get_team - cache_team - valid_team: {valid_team}')
# logging.info(f'gameplay_models - get_team - cache_team - valid_team: {valid_team}')
db_team = Team.model_validate(valid_team)
# logging.info(f'gameplay_db - get_team - cache_team - db_team: {db_team}')
# logging.info(f'gameplay_models - get_team - cache_team - db_team: {db_team}')
session.add(db_team)
session.commit()
session.refresh(db_team)
@ -158,7 +160,7 @@ async def get_team(
return cache_team(team)
err = 'Team not found'
logging.error(f'gameplay_db - get_team - {err}')
logging.error(f'gameplay_models - get_team - {err}')
raise LookupError(err)
@ -169,9 +171,9 @@ class PlayerBase(SQLModel):
image: str
mlbclub: str
franchise: str
cardset: dict
cardset_id: int | None = Field(default=None, foreign_key='cardset.id')
set_num: int
rarity: dict
rarity_id: int | None = Field(default=None)
pos_1: str
description: str
quantity: int | None = Field(default=999)
@ -193,7 +195,58 @@ class PlayerBase(SQLModel):
class Player(PlayerBase, table=True):
pass
cardset: Cardset = Relationship(back_populates='players')
def player_description(player: Player = None, player_dict: dict = None) -> str:
if player is None and player_dict is None:
err = 'One of "player" or "player_dict" must be included to get full description'
logging.error(f'gameplay_models - player_description - {err}')
raise TypeError(err)
if player is not None:
return f'{player.description} {player.name}'
r_val = f'{player_dict['description']}'
if 'name' in player_dict:
r_val += f' {player_dict["name"]}'
elif 'p_name' in player_dict:
r_val += f' {player_dict["p_name"]}'
return r_val
async def get_player(session: Session, player_id: int, skip_cache: bool = False) -> Player:
if not skip_cache:
this_player = session.get(Player, player_id)
if this_player is not None:
logging.info(f'we found a cached player: {this_player} / created: {this_player.created}')
tdelta = datetime.datetime.now() - this_player.created
logging.debug(f'tdelta: {tdelta}')
if tdelta.total_seconds() < CACHE_LIMIT:
return this_player
else:
session.delete(this_player)
session.commit()
def cache_player(json_data: dict) -> Player:
valid_player = PlayerBase.model_validate(json_data, from_attributes=True)
db_player = Player.model_validate(valid_player)
session.add(db_player)
session.commit()
session.refresh(db_player)
return db_player
p_query = await db_get('players', object_id=player_id, params=[('inc_dex', False)])
if p_query is not None:
if 'id' not in p_query:
p_query['id'] = p_query['player_id']
if 'name' not in p_query:
p_query['name'] = p_query['p_name']
return cache_player(p_query)
err = 'Player not found'
logging.error(f'gameplay_models - get_player - {err}')

2
pytest.ini
View File

@ -1,2 +1,4 @@
[pytest]
asyncio_mode = auto
filterwarnings =
ignore::DeprecationWarning

19
tests/factory.py
View File

@ -4,7 +4,7 @@ from sqlmodel import Session, SQLModel, create_engine
from sqlmodel.pool import StaticPool
from typing import Literal
from in_game.gameplay_models import Game, Lineup, Team
from in_game.gameplay_models import Game, Lineup, Team, Player
@pytest.fixture(name='session')
@ -60,4 +60,19 @@ def new_teams_fixture():
old_cache_team = Team(
id=3, abbrev='BAL', sname='Orioles', lname='Baltimore Orioles', gmid=181818, gmname='Brandon Hyde', gsheet='asdf1234', wallet=6969, team_value=69420, collection_value=169420, color='https://i.postimg.cc/8kLZCYXh/S10CLS.png', season=7, event=False, career=0, ranking=500, has_guide=False, is_ai=False, created=datetime.datetime.today() - datetime.timedelta(days=60))
return [team_1, team_2, incomplete_team, old_cache_team]
return [team_1, team_2, incomplete_team, old_cache_team]
@pytest.fixture(name='new_players')
def new_players_fixture():
player_1 = Player(
id=10139, name='Jihmih Tyranus', cost=316, image='https://pd.manticorum.com/api/v2/players/10139/battingcard?d=2024-5-07', mlbclub='Custom Ballplayers', franchise='Custom Ballplayers', set_num=420, pos_1='C', pos_2='3B', pos_3='LF', description='24 Custom', bbref_id='sandyman'
)
player_2 = Player(
id=9873, name='Finn Sturm', cost=146, image='https://pd.manticorum.com/api/v2/players/9873/battingcard?d=2024-5-06', mlbclub='Custom Ballplayers', franchise='Custom Ballplayers', set_num=69, pos_1='C', pos_2='1B', description='24 Custom', bbref_id='sandyman'
)
old_cache_player = Player(
id=10334, name='Ryan Ohearn', cost=91, image='https://pd.manticorum.com/api/v2/players/10334/battingcard?d=2024-8-18', mlbclub='Baltimore Orioles', franchise='Baltimore Orioles', set_num=420, pos_1='1B', pos_2='RF', pos_3='LF', description='2024', bbref_id='ohearry01', strat_code='656811', fangr_id='16442', mlbplayer_id=1200, created=datetime.datetime.today() - datetime.timedelta(days=60)
)
return [player_1, player_2, old_cache_player]

2
tests/test_gameplay_db_game.py → tests/test_gameplay_models_game.py
View File

@ -28,7 +28,7 @@ def test_create_game(session: Session, new_games: list[Game]):
assert game_1.ai_team == None
assert game_1.game_type == None
assert game_2.active == False
assert game_2.is_pd == False
assert game_2.is_pd == True
assert game_2.ranked == True
assert game_2.week_num == 6
assert game_2.game_num == 9

0
tests/test_gameplay_db_lineup.py → tests/test_gameplay_models_lineup.py
View File

61
tests/test_gameplay_models_player.py Normal file
View File

@ -0,0 +1,61 @@
import datetime
import pytest
from sqlmodel import Session
from in_game.gameplay_models import CACHE_LIMIT, Player, player_description, select, get_player
from factory import session_fixture, new_players_fixture
def test_create_player(session: Session, new_players: list[Player]):
player_1 = new_players[0]
player_2 = new_players[1]
session.add(player_1)
session.add(player_2)
session.commit()
assert player_1.id == new_players[0].id
assert player_1.name == new_players[0].name
assert player_1.cost == new_players[0].cost
assert player_1.bbref_id == new_players[0].bbref_id
assert player_1.image == new_players[0].image
assert player_1.bbref_id == new_players[0].bbref_id
assert player_1.description == new_players[0].description
assert player_1.fangr_id is None
assert player_2.fangr_id is None
assert player_2.pos_2 == new_players[1].pos_2
async def test_cached_players(session: Session, new_players: list[Player]):
player_1 = new_players[0]
player_2 = new_players[1]
player_3 = new_players[2]
session.add(player_1)
session.add(player_2)
session.add(player_3)
session.commit()
assert (datetime.datetime.now() - player_3.created).total_seconds() > CACHE_LIMIT
new_player_1 = await get_player(session, player_id=player_1.id)
new_player_2 = await get_player(session, player_id=player_2.id)
new_player_3 = await get_player(session, player_id=player_3.id)
new_player_4 = await get_player(session, player_id=69)
assert player_1.created == new_player_1.created
assert player_2.created == new_player_2.created
assert (datetime.datetime.now() - new_player_3.created).total_seconds() < CACHE_LIMIT
assert new_player_4 is not None
def test_player_description(session: Session, new_players: list[Player]):
player_1 = new_players[0]
session.add(player_1)
session.commit()
session.refresh(player_1)
player_dict = {'player_id': new_players[1].id, 'p_name': new_players[1].name, 'description': new_players[1].description}
assert player_description(player=player_1) == f'{new_players[0].description} {new_players[0].name}'
assert player_description(player_dict=player_dict) == f'{player_dict["description"]} {player_dict['p_name']}'

10
tests/test_gameplay_db_team.py → tests/test_gameplay_models_team.py
View File

@ -2,7 +2,7 @@ import datetime
from sqlmodel import Session, select
from sqlite3 import IntegrityError
from in_game.gameplay_models import Team, get_team
from in_game.gameplay_models import Team, get_team, CACHE_LIMIT
from factory import session_fixture, new_teams_fixture, pytest
def test_create_team(session: Session, new_teams: list[Team]):
@ -40,15 +40,17 @@ async def test_team_cache(session: Session, new_teams: list[Team]):
session.add(team_3)
session.commit()
assert (datetime.datetime.now() - team_3.created).total_seconds() > 1209600
assert (datetime.datetime.now() - team_3.created).total_seconds() > CACHE_LIMIT
new_team_1 = await get_team(session, team_id=team_1.id)
new_team_2 = await get_team(session, team_id=team_2.id)
new_team_3 = await get_team(session, team_abbrev='BAL')
new_team_3 = await get_team(session, team_id=team_3.id)
new_team_4 = await get_team(session, team_abbrev='NYY')
assert team_1.created == new_team_1.created
assert team_2.created == new_team_2.created
assert (datetime.datetime.now() - new_team_3.created).total_seconds() < 1209600
assert (datetime.datetime.now() - new_team_3.created).total_seconds() < CACHE_LIMIT
assert new_team_4.created is not None