Chewing_Bever
/
didier
mirror of https://github.com/stijndcl/didier
0
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #174 from stijndcl/remove-wordle 

Remove wordle
dinks
Stijn De Clercq 2023-07-07 15:14:23 +02:00 committed by GitHub
parent c3a7ff8e4c ee6013da5d
commit 9998236e03
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 61 additions and 22602 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
alembic/versions/1e3e7f4192c4_remove_wordle.py 100644
View File

@ -0,0 +1,57 @@
"""Remove wordle
Revision ID: 1e3e7f4192c4
Revises: 954ad804f057
Create Date: 2023-07-07 14:52:20.993687
"""
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision = "1e3e7f4192c4"
down_revision = "954ad804f057"
branch_labels = None
depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table("wordle_word")
op.drop_table("wordle_stats")
op.drop_table("wordle_guesses")
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
"wordle_guesses",
sa.Column("wordle_guess_id", sa.INTEGER(), autoincrement=True, nullable=False),
sa.Column("user_id", sa.BIGINT(), autoincrement=False, nullable=True),
sa.Column("guess", sa.TEXT(), autoincrement=False, nullable=False),
sa.ForeignKeyConstraint(["user_id"], ["users.user_id"], name="wordle_guesses_user_id_fkey"),
sa.PrimaryKeyConstraint("wordle_guess_id", name="wordle_guesses_pkey"),
)
op.create_table(
"wordle_stats",
sa.Column("wordle_stats_id", sa.INTEGER(), autoincrement=True, nullable=False),
sa.Column("user_id", sa.BIGINT(), autoincrement=False, nullable=True),
sa.Column("last_win", sa.DATE(), autoincrement=False, nullable=True),
sa.Column("games", sa.INTEGER(), server_default=sa.text("0"), autoincrement=False, nullable=False),
sa.Column("wins", sa.INTEGER(), server_default=sa.text("0"), autoincrement=False, nullable=False),
sa.Column("current_streak", sa.INTEGER(), server_default=sa.text("0"), autoincrement=False, nullable=False),
sa.Column("highest_streak", sa.INTEGER(), server_default=sa.text("0"), autoincrement=False, nullable=False),
sa.ForeignKeyConstraint(["user_id"], ["users.user_id"], name="wordle_stats_user_id_fkey"),
sa.PrimaryKeyConstraint("wordle_stats_id", name="wordle_stats_pkey"),
)
op.create_table(
"wordle_word",
sa.Column("word_id", sa.INTEGER(), autoincrement=True, nullable=False),
sa.Column("word", sa.TEXT(), autoincrement=False, nullable=False),
sa.Column("day", sa.DATE(), autoincrement=False, nullable=False),
sa.PrimaryKeyConstraint("word_id", name="wordle_word_pkey"),
sa.UniqueConstraint("day", name="wordle_word_day_key"),
)
# ### end Alembic commands ###

2
database/constants.py
View File

@ -1,2 +0,0 @@
WORDLE_GUESS_COUNT = 6
WORDLE_WORD_LENGTH = 5

85
database/crud/wordle.py
View File

@ -1,85 +0,0 @@
import datetime
from typing import Optional
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession
from database.crud.users import get_or_add_user
from database.schemas import WordleGuess, WordleWord
__all__ = [
"get_active_wordle_game",
"get_wordle_guesses",
"make_wordle_guess",
"set_daily_word",
"reset_wordle_games",
]
async def get_active_wordle_game(session: AsyncSession, user_id: int) -> list[WordleGuess]:
"""Find a player's active game"""
await get_or_add_user(session, user_id)
statement = select(WordleGuess).where(WordleGuess.user_id == user_id)
guesses = (await session.execute(statement)).scalars().all()
return guesses
async def get_wordle_guesses(session: AsyncSession, user_id: int) -> list[str]:
"""Get the strings of a player's guesses"""
active_game = await get_active_wordle_game(session, user_id)
return list(map(lambda g: g.guess.lower(), active_game))
async def make_wordle_guess(session: AsyncSession, user_id: int, guess: str):
"""Make a guess in your current game"""
guess_instance = WordleGuess(user_id=user_id, guess=guess)
session.add(guess_instance)
await session.commit()
async def get_daily_word(session: AsyncSession) -> Optional[WordleWord]:
"""Get the word of today"""
statement = select(WordleWord).where(WordleWord.day == datetime.date.today())
row = (await session.execute(statement)).scalar_one_or_none()
if row is None:
return None
return row
async def set_daily_word(session: AsyncSession, word: str, *, forced: bool = False) -> str:
"""Set the word of today
This does NOT overwrite the existing word if there is one, so that it can safely run
on startup every time.
In order to always overwrite the current word, set the "forced"-kwarg to True.
Returns the word that was chosen. If one already existed, return that instead.
"""
current_word = await get_daily_word(session)
if current_word is None:
current_word = WordleWord(word=word, day=datetime.date.today())
session.add(current_word)
await session.commit()
# Remove all active games
await reset_wordle_games(session)
elif forced:
current_word.word = word
session.add(current_word)
await session.commit()
# Remove all active games
await reset_wordle_games(session)
return current_word.word
async def reset_wordle_games(session: AsyncSession):
"""Reset all active games"""
statement = delete(WordleGuess)
await session.execute(statement)
await session.commit()

60
database/crud/wordle_stats.py
View File

@ -1,60 +0,0 @@
from datetime import date
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from database.crud.users import get_or_add_user
from database.schemas import WordleStats
__all__ = ["get_wordle_stats", "complete_wordle_game"]
async def get_wordle_stats(session: AsyncSession, user_id: int) -> WordleStats:
"""Get a user's wordle stats
If no entry is found, it is first created
"""
await get_or_add_user(session, user_id)
statement = select(WordleStats).where(WordleStats.user_id == user_id)
stats = (await session.execute(statement)).scalar_one_or_none()
if stats is not None:
return stats
stats = WordleStats(user_id=user_id)
session.add(stats)
await session.commit()
await session.refresh(stats)
return stats
async def complete_wordle_game(session: AsyncSession, user_id: int, win: bool):
"""Update the user's Wordle stats"""
stats = await get_wordle_stats(session, user_id)
stats.games += 1
if win:
stats.wins += 1
# Update streak
today = date.today()
last_win = stats.last_win
stats.last_win = today
if last_win is None or (today - last_win).days > 1:
# Never won a game before or streak is over
stats.current_streak = 1
else:
# On a streak: increase counter
stats.current_streak += 1
# Update max streak if necessary
if stats.current_streak > stats.highest_streak:
stats.highest_streak = stats.current_streak
else:
# Streak is over
stats.current_streak = 0
session.add(stats)
await session.commit()

47
database/schemas.py
View File

@ -45,9 +45,6 @@ __all__ = [
"UforaCourse",
"UforaCourseAlias",
"User",
"WordleGuess",
"WordleStats",
"WordleWord",
]
@ -343,47 +340,3 @@ class User(Base):
reminders: list[Reminder] = relationship(
"Reminder", back_populates="user", uselist=True, lazy="selectin", cascade="all, delete-orphan"
)
wordle_guesses: list[WordleGuess] = relationship(
"WordleGuess", back_populates="user", uselist=True, lazy="selectin", cascade="all, delete-orphan"
)
wordle_stats: WordleStats = relationship(
"WordleStats", back_populates="user", uselist=False, lazy="selectin", cascade="all, delete-orphan"
)
class WordleGuess(Base):
"""A user's Wordle guesses for today"""
__tablename__ = "wordle_guesses"
wordle_guess_id: int = Column(Integer, primary_key=True)
user_id: int = Column(BigInteger, ForeignKey("users.user_id"))
guess: str = Column(Text, nullable=False)
user: User = relationship("User", back_populates="wordle_guesses", uselist=False, lazy="selectin")
class WordleStats(Base):
"""Stats about a user's wordle performance"""
__tablename__ = "wordle_stats"
wordle_stats_id: int = Column(Integer, primary_key=True)
user_id: int = Column(BigInteger, ForeignKey("users.user_id"))
last_win: Optional[date] = Column(Date, nullable=True)
games: int = Column(Integer, server_default="0", nullable=False)
wins: int = Column(Integer, server_default="0", nullable=False)
current_streak: int = Column(Integer, server_default="0", nullable=False)
highest_streak: int = Column(Integer, server_default="0", nullable=False)
user: User = relationship("User", back_populates="wordle_stats", uselist=False, lazy="selectin")
class WordleWord(Base):
"""The current Wordle word"""
__tablename__ = "wordle_word"
word_id: int = Column(Integer, primary_key=True)
word: str = Column(Text, nullable=False)
day: date = Column(Date, nullable=False, unique=True)

18
database/utils/caches.py
View File

@ -4,8 +4,8 @@ from discord import app_commands
from overrides import overrides
from sqlalchemy.ext.asyncio import AsyncSession
from database.crud import easter_eggs, links, memes, ufora_courses, wordle
from database.schemas import EasterEgg, WordleWord
from database.crud import easter_eggs, links, memes, ufora_courses
from database.schemas import EasterEgg
__all__ = ["CacheManager", "EasterEggCache", "LinkCache", "UforaCourseCache"]
@ -132,17 +132,6 @@ class UforaCourseCache(DatabaseCache):
return [app_commands.Choice(name=suggestion, value=suggestion.lower()) for suggestion in suggestions]
class WordleCache(DatabaseCache):
"""Cache to store the current daily Wordle word"""
word: WordleWord
async def invalidate(self, database_session: AsyncSession):
word = await wordle.get_daily_word(database_session)
if word is not None:
self.word = word
class CacheManager:
"""Class that keeps track of all caches"""
@ -150,14 +139,12 @@ class CacheManager:
links: LinkCache
memes: MemeCache
ufora_courses: UforaCourseCache
wordle_word: WordleCache
def __init__(self):
self.easter_eggs = EasterEggCache()
self.links = LinkCache()
self.memes = MemeCache()
self.ufora_courses = UforaCourseCache()
self.wordle_word = WordleCache()
async def initialize_caches(self, postgres_session: AsyncSession):
"""Initialize the contents of all caches"""
@ -165,4 +152,3 @@ class CacheManager:
await self.links.invalidate(postgres_session)
await self.memes.invalidate(postgres_session)
await self.ufora_courses.invalidate(postgres_session)
await self.wordle_word.invalidate(postgres_session)

55
didier/cogs/games.py
View File

@ -1,14 +1,6 @@
from typing import Optional
import discord
from discord import app_commands
from discord.ext import commands
from database.constants import WORDLE_WORD_LENGTH
from database.crud.wordle import get_wordle_guesses, make_wordle_guess
from database.crud.wordle_stats import complete_wordle_game
from didier import Didier
from didier.data.embeds.wordle import WordleEmbed, WordleErrorEmbed, is_wordle_game_over
class Games(commands.Cog):
@ -19,53 +11,6 @@ class Games(commands.Cog):
def __init__(self, client: Didier):
self.client = client
@app_commands.command(name="wordle", description="Play Wordle!")
async def wordle(self, interaction: discord.Interaction, guess: Optional[str] = None):
"""View your active Wordle game
If an argument is provided, make a guess instead
"""
await interaction.response.defer(ephemeral=True)
# Guess is wrong length
if guess is not None and len(guess) != 0 and len(guess) != WORDLE_WORD_LENGTH:
embed = WordleErrorEmbed(message=f"Guess must be 5 characters, but `{guess}` is {len(guess)}.").to_embed()
return await interaction.followup.send(embed=embed)
word_instance = self.client.database_caches.wordle_word.word
async with self.client.postgres_session as session:
guesses = await get_wordle_guesses(session, interaction.user.id)
# Trying to guess with a complete game
if is_wordle_game_over(guesses, word_instance.word):
embed = WordleErrorEmbed(
message="You've already completed today's Wordle.\nTry again tomorrow!"
).to_embed()
return await interaction.followup.send(embed=embed)
# Make a guess
if guess:
# The guess is not a real word
if guess.lower() not in self.client.wordle_words:
embed = WordleErrorEmbed(message=f"`{guess}` is not a valid word.").to_embed()
return await interaction.followup.send(embed=embed)
guess = guess.lower()
await make_wordle_guess(session, interaction.user.id, guess)
# Don't re-request the game, we already have it
# just append locally
guesses.append(guess)
embed = WordleEmbed(guesses=guesses, word=word_instance).to_embed()
await interaction.followup.send(embed=embed)
# After responding to the interaction: update stats in the background
game_over = is_wordle_game_over(guesses, word_instance.word)
if game_over:
await complete_wordle_game(session, interaction.user.id, word_instance.word in guesses)
async def setup(client: Didier):
"""Load the cog"""

25
didier/cogs/tasks.py
View File

@ -10,7 +10,6 @@ from database import enums
from database.crud.birthdays import get_birthdays_on_day
from database.crud.reminders import get_all_reminders_for_category
from database.crud.ufora_announcements import remove_old_announcements
from database.crud.wordle import set_daily_word
from database.schemas import Reminder
from didier import Didier
from didier.data.embeds.schedules import (
@ -54,7 +53,6 @@ class Tasks(commands.Cog):
"reminders": self.reminders,
"ufora": self.pull_ufora_announcements,
"remove_ufora": self.remove_old_ufora_announcements,
"wordle": self.reset_wordle_word,
}
@overrides
@ -74,7 +72,6 @@ class Tasks(commands.Cog):
# Start other tasks
self.reminders.start()
self.reset_wordle_word.start()
@overrides
def cog_unload(self) -> None:
@ -266,34 +263,16 @@ class Tasks(commands.Cog):
async with self.client.postgres_session as session:
await remove_old_announcements(session)
@tasks.loop(time=DAILY_RESET_TIME)
async def reset_wordle_word(self, forced: bool = False):
"""Reset the daily Wordle word"""
async with self.client.postgres_session as session:
await set_daily_word(session, random.choice(tuple(self.client.wordle_words)), forced=forced)
await self.client.database_caches.wordle_word.invalidate(session)
@reset_wordle_word.before_loop
async def _before_reset_wordle_word(self):
await self.client.wait_until_ready()
@check_birthdays.error
@pull_schedules.error
@pull_ufora_announcements.error
@reminders.error
@remove_old_ufora_announcements.error
@reset_wordle_word.error
async def _on_tasks_error(self, error: BaseException):
"""Error handler for all tasks"""
self.client.dispatch("task_error", error)
async def setup(client: Didier):
"""Load the cog
Initially fetch the wordle word from the database, or reset it
if there hasn't been a reset yet today
"""
cog = Tasks(client)
await client.add_cog(cog)
await cog.reset_wordle_word()
"""Load the cog"""
await client.add_cog(Tasks(client))

142
didier/data/embeds/wordle.py
View File

@ -1,142 +0,0 @@
import enum
from dataclasses import dataclass
import discord
from overrides import overrides
from database.constants import WORDLE_GUESS_COUNT, WORDLE_WORD_LENGTH
from database.schemas import WordleWord
from didier.data.embeds.base import EmbedBaseModel
from didier.utils.types.datetime import int_to_weekday, tz_aware_now
__all__ = ["is_wordle_game_over", "WordleEmbed", "WordleErrorEmbed"]
def is_wordle_game_over(guesses: list[str], word: str) -> bool:
"""Check if the current game is over or not"""
if not guesses:
return False
if len(guesses) == WORDLE_GUESS_COUNT:
return True
return word.lower() in guesses
def footer() -> str:
"""Create the footer to put on the embed"""
today = tz_aware_now()
return f"{int_to_weekday(today.weekday())} {today.strftime('%d/%m/%Y')}"
class WordleColour(enum.IntEnum):
"""Colours for the Wordle embed"""
EMPTY = 0
WRONG_LETTER = 1
WRONG_POSITION = 2
CORRECT = 3
@dataclass
class WordleEmbed(EmbedBaseModel):
"""Embed for a Wordle game"""
guesses: list[str]
word: WordleWord
def _letter_colour(self, guess: str, index: int) -> WordleColour:
"""Get the colour for a guess at a given position"""
if guess[index] == self.word.word[index]:
return WordleColour.CORRECT
wrong_letter = 0
wrong_position = 0
for i, letter in enumerate(self.word.word):
if letter == guess[index] and guess[i] != guess[index]:
wrong_letter += 1
if i <= index and guess[i] == guess[index] and letter != guess[index]:
wrong_position += 1
if i >= index:
if wrong_position == 0:
break
if wrong_position <= wrong_letter:
return WordleColour.WRONG_POSITION
return WordleColour.WRONG_LETTER
def _guess_colours(self, guess: str) -> list[WordleColour]:
"""Create the colour codes for a specific guess"""
return [self._letter_colour(guess, i) for i in range(WORDLE_WORD_LENGTH)]
def colour_code_game(self) -> list[list[WordleColour]]:
"""Create the colour codes for an entire game"""
colours = []
# Add all the guesses
for guess in self.guesses:
colours.append(self._guess_colours(guess))
# Fill the rest with empty spots
for _ in range(WORDLE_GUESS_COUNT - len(colours)):
colours.append([WordleColour.EMPTY] * WORDLE_WORD_LENGTH)
return colours
def _colours_to_emojis(self, colours: list[list[WordleColour]]) -> list[list[str]]:
"""Turn the colours of the board into Discord emojis"""
colour_map = {
WordleColour.EMPTY: ":white_large_square:",
WordleColour.WRONG_LETTER: ":black_large_square:",
WordleColour.WRONG_POSITION: ":orange_square:",
WordleColour.CORRECT: ":green_square:",
}
emojis = []
for row in colours:
emojis.append(list(map(lambda char: colour_map[char], row)))
return emojis
@overrides
def to_embed(self, **kwargs) -> discord.Embed:
only_colours = kwargs.get("only_colours", False)
colours = self.colour_code_game()
embed = discord.Embed(colour=discord.Colour.blue(), title=f"Wordle #{self.word.word_id + 1}")
emojis = self._colours_to_emojis(colours)
rows = [" ".join(row) for row in emojis]
# Don't reveal anything if we only want to show the colours
if not only_colours and self.guesses:
for i, guess in enumerate(self.guesses):
rows[i] += f" ||{guess.upper()}||"
# If the game is over, reveal the word
if is_wordle_game_over(self.guesses, self.word.word):
rows.append(f"\n\nThe word was **{self.word.word.upper()}**!")
embed.description = "\n\n".join(rows)
embed.set_footer(text=footer())
return embed
@dataclass
class WordleErrorEmbed(EmbedBaseModel):
"""Embed to send error messages to the user"""
message: str
@overrides
def to_embed(self, **kwargs) -> discord.Embed:
embed = discord.Embed(colour=discord.Colour.red(), title="Wordle")
embed.description = self.message
embed.set_footer(text=footer())
return embed

10
didier/didier.py
View File

@ -38,7 +38,6 @@ class Didier(commands.Bot):
http_session: ClientSession
schedules: dict[settings.ScheduleType, Schedule] = {}
sniped: dict[int, tuple[discord.Message, Optional[discord.Message]]] = {}
wordle_words: set[str] = set()
def __init__(self):
activity = discord.Activity(type=discord.ActivityType.playing, name=settings.DISCORD_STATUS_MESSAGE)
@ -77,9 +76,6 @@ class Didier(commands.Bot):
# Create directories that are ignored on GitHub
self._create_ignored_directories()
# Load the Wordle dictionary
self._load_wordle_words()
# Initialize caches
self.database_caches = CacheManager()
async with self.postgres_session as session:
@ -137,12 +133,6 @@ class Didier(commands.Bot):
elif os.path.isdir(new_path := f"{path}/{file}"):
await self._load_directory_extensions(new_path)
def _load_wordle_words(self):
"""Load the dictionary of Wordle words"""
with open("files/dictionaries/words-english-wordle.txt", "r") as fp:
for line in fp:
self.wordle_words.add(line.strip())
async def load_schedules(self):
"""Parse & load all schedules into memory"""
self.schedules = {}

21952
files/dictionaries/words-english-wordle.txt
View File

File diff suppressed because it is too large Load Diff

138
tests/test_database/test_crud/test_wordle.py
View File

@ -1,138 +0,0 @@
from datetime import date, timedelta
import pytest
from freezegun import freeze_time
from sqlalchemy.ext.asyncio import AsyncSession
from database.crud import wordle as crud
from database.schemas import User, WordleGuess, WordleWord
@pytest.fixture
async def wordle_guesses(postgres: AsyncSession, user: User) -> list[WordleGuess]:
"""Fixture to generate some guesses"""
guesses = []
for guess in ["TEST", "WORDLE", "WORDS"]:
guess = WordleGuess(user_id=user.user_id, guess=guess)
postgres.add(guess)
await postgres.commit()
guesses.append(guess)
return guesses
@pytest.mark.postgres
async def test_get_active_wordle_game_none(postgres: AsyncSession, user: User):
"""Test getting an active game when there is none"""
result = await crud.get_active_wordle_game(postgres, user.user_id)
assert not result
@pytest.mark.postgres
async def test_get_active_wordle_game(postgres: AsyncSession, wordle_guesses: list[WordleGuess]):
"""Test getting an active game when there is one"""
result = await crud.get_active_wordle_game(postgres, wordle_guesses[0].user_id)
assert result == wordle_guesses
@pytest.mark.postgres
async def test_get_daily_word_none(postgres: AsyncSession):
"""Test getting the daily word when the database is empty"""
result = await crud.get_daily_word(postgres)
assert result is None
@pytest.mark.postgres
@freeze_time("2022-07-30")
async def test_get_daily_word_not_today(postgres: AsyncSession):
"""Test getting the daily word when there is an entry, but not for today"""
day = date.today() - timedelta(days=1)
word = "testword"
word_instance = WordleWord(word=word, day=day)
postgres.add(word_instance)
await postgres.commit()
assert await crud.get_daily_word(postgres) is None
@pytest.mark.postgres
@freeze_time("2022-07-30")
async def test_get_daily_word_present(postgres: AsyncSession):
"""Test getting the daily word when there is one for today"""
day = date.today()
word = "testword"
word_instance = WordleWord(word=word, day=day)
postgres.add(word_instance)
await postgres.commit()
daily_word = await crud.get_daily_word(postgres)
assert daily_word is not None
assert daily_word.word == word
@pytest.mark.postgres
@freeze_time("2022-07-30")
async def test_set_daily_word_none_present(postgres: AsyncSession):
"""Test setting the daily word when there is none"""
assert await crud.get_daily_word(postgres) is None
word = "testword"
await crud.set_daily_word(postgres, word)
daily_word = await crud.get_daily_word(postgres)
assert daily_word is not None
assert daily_word.word == word
@pytest.mark.postgres
@freeze_time("2022-07-30")
async def test_set_daily_word_present(postgres: AsyncSession):
"""Test setting the daily word when there already is one"""
word = "testword"
await crud.set_daily_word(postgres, word)
await crud.set_daily_word(postgres, "another word")
daily_word = await crud.get_daily_word(postgres)
assert daily_word is not None
assert daily_word.word == word
@pytest.mark.postgres
@freeze_time("2022-07-30")
async def test_set_daily_word_force_overwrite(postgres: AsyncSession):
"""Test setting the daily word when there already is one, but "forced" is set to True"""
word = "testword"
await crud.set_daily_word(postgres, word)
word = "anotherword"
await crud.set_daily_word(postgres, word, forced=True)
daily_word = await crud.get_daily_word(postgres)
assert daily_word is not None
assert daily_word.word == word
@pytest.mark.postgres
async def test_make_wordle_guess(postgres: AsyncSession, user: User):
"""Test making a guess in your current game"""
test_user_id = user.user_id
guess = "guess"
await crud.make_wordle_guess(postgres, test_user_id, guess)
assert await crud.get_wordle_guesses(postgres, test_user_id) == [guess]
other_guess = "otherguess"
await crud.make_wordle_guess(postgres, test_user_id, other_guess)
assert await crud.get_wordle_guesses(postgres, test_user_id) == [guess, other_guess]
@pytest.mark.postgres
async def test_reset_wordle_games(postgres: AsyncSession, wordle_guesses: list[WordleGuess], user: User):
"""Test dropping the collection of active games"""
test_user_id = user.user_id
assert await crud.get_active_wordle_game(postgres, test_user_id)
await crud.reset_wordle_games(postgres)
assert not await crud.get_active_wordle_game(postgres, test_user_id)

72
tests/test_database/test_crud/test_wordle_stats.py
View File

@ -1,72 +0,0 @@
import datetime
import pytest
from freezegun import freeze_time
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from database.crud import wordle_stats as crud
from database.schemas import User, WordleStats
async def insert_game_stats(postgres: AsyncSession, stats: WordleStats):
"""Helper function to insert some stats"""
postgres.add(stats)
await postgres.commit()
@pytest.mark.postgres
async def test_get_stats_non_existent_creates(postgres: AsyncSession, user: User):
"""Test getting a user's stats when the db is empty"""
test_user_id = user.user_id
statement = select(WordleStats).where(WordleStats.user_id == test_user_id)
assert (await postgres.execute(statement)).scalar_one_or_none() is None
await crud.get_wordle_stats(postgres, test_user_id)
assert (await postgres.execute(statement)).scalar_one_or_none() is not None
@pytest.mark.postgres
async def test_get_stats_existing_returns(postgres: AsyncSession, user: User):
"""Test getting a user's stats when there's already an entry present"""
test_user_id = user.user_id
stats = WordleStats(user_id=test_user_id)
stats.games = 20
await insert_game_stats(postgres, stats)
found_stats = await crud.get_wordle_stats(postgres, test_user_id)
assert found_stats.games == 20
@pytest.mark.postgres
@freeze_time("2022-07-30")
async def test_complete_wordle_game_won(postgres: AsyncSession, user: User):
"""Test completing a wordle game when you win"""
test_user_id = user.user_id
await crud.complete_wordle_game(postgres, test_user_id, win=True)
stats = await crud.get_wordle_stats(postgres, test_user_id)
assert stats.games == 1
assert stats.wins == 1
assert stats.current_streak == 1
assert stats.highest_streak == 1
assert stats.last_win == datetime.date.today()
@pytest.mark.postgres
@freeze_time("2022-07-30")
async def test_complete_wordle_game_lost(postgres: AsyncSession, user: User):
"""Test completing a wordle game when you lose"""
test_user_id = user.user_id
stats = WordleStats(user_id=test_user_id)
stats.current_streak = 10
await insert_game_stats(postgres, stats)
await crud.complete_wordle_game(postgres, test_user_id, win=False)
stats = await crud.get_wordle_stats(postgres, test_user_id)
# Check that streak was broken
assert stats.current_streak == 0
assert stats.games == 1