Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions bot/bot.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@

from twitchio.ext import commands

from bot.cooldown import GlobalCooldown
from bot.cooldown import CooldownManager
from game.state import Difficulty, GameState
from game.word_utils import load_word_list

Expand Down Expand Up @@ -92,7 +92,7 @@ def __init__(self, **kwargs: object) -> None:
)
scorer = kwargs.pop("scorer", None) # type: ignore[assignment]
self._command_prefix: str = initial_prefix
self._cooldown = GlobalCooldown(int(initial_cooldown))
self._cooldown = CooldownManager(int(initial_cooldown))
self._game_state = GameState(scorer=scorer) # type: ignore[arg-type]
self._on_state_change = on_state_change
self._next_difficulty: Difficulty = Difficulty.EASY
Expand Down Expand Up @@ -170,14 +170,14 @@ async def guess(self, ctx: commands.Context, word: str = "") -> None:

Usage: <prefix> guess <word>
"""
if self._cooldown.is_on_cooldown():
remaining = math.ceil(self._cooldown.remaining())
if self._cooldown.is_on_cooldown(ctx.author.name):
remaining = math.ceil(self._cooldown.remaining(ctx.author.name))
await ctx.send(
f"Please wait {remaining} seconds before guessing again."
)
return

self._cooldown.record()
self._cooldown.record(ctx.author.name)

if not word:
await ctx.send("Please provide a word to guess.")
Expand Down
33 changes: 7 additions & 26 deletions bot/cooldown.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,36 +20,17 @@ def is_on_cooldown(self, user: str) -> bool:
last = self._last_used.get(user, 0.0)
return (time.monotonic() - last) < self._cooldown

def remaining(self, user: str) -> float:
"""Return seconds remaining in the cooldown for *user* (0.0 if not active)."""
last = self._last_used.get(user, 0.0)
elapsed = time.monotonic() - last
return max(0.0, self._cooldown - elapsed)

def record(self, user: str) -> None:
"""Record that *user* just used a command."""
self._last_used[user] = time.monotonic()


class GlobalCooldown:
"""Global cooldown shared across all users for a single command."""

def __init__(self, cooldown_seconds: int) -> None:
self._cooldown = cooldown_seconds
self._last_used: float = 0.0

@property
def duration(self) -> int:
"""Return the current cooldown duration in seconds."""
return self._cooldown

def remaining(self) -> float:
"""Return seconds remaining in the cooldown (0.0 if not active)."""
elapsed = time.monotonic() - self._last_used
return max(0.0, self._cooldown - elapsed)

def is_on_cooldown(self) -> bool:
"""Return True if the cooldown is still active."""
return self.remaining() > 0

def record(self) -> None:
"""Record that the command was just used."""
self._last_used = time.monotonic()

def set_duration(self, seconds: int) -> None:
"""Update the cooldown duration."""
self._cooldown = seconds

31 changes: 25 additions & 6 deletions tests/test_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
from unittest.mock import AsyncMock, MagicMock, patch

from bot.bot import StreamantixBot, _validate_prefix, _validate_cooldown, _validate_difficulty
from bot.cooldown import GlobalCooldown
from bot.cooldown import CooldownManager
from game.state import Difficulty, GameState


Expand All @@ -16,7 +16,7 @@ def _make_bot(prefix: str = "!sx", cooldown: int = 5) -> StreamantixBot:
"""Return a StreamantixBot instance without connecting to Twitch."""
bot = object.__new__(StreamantixBot)
bot._command_prefix = prefix
bot._cooldown = GlobalCooldown(cooldown)
bot._cooldown = CooldownManager(cooldown)
bot._game_state = GameState()
bot._next_difficulty = Difficulty.EASY
return bot
Expand Down Expand Up @@ -284,34 +284,53 @@ async def test_confirmation_message_contains_value(self):
class TestGuessCooldownEnforcement:
async def test_guess_blocked_during_cooldown(self):
bot = _make_bot(cooldown=30)
bot._cooldown.record() # simulate a recent guess
ctx = _make_ctx()
ctx.author.name = "alice"
bot._cooldown.record("alice") # simulate a recent guess by this user
await _guess_fn(bot, ctx)
message = ctx.send.call_args[0][0]
assert "wait" in message.lower()

async def test_guess_allowed_when_not_on_cooldown(self):
bot = _make_bot(cooldown=0)
ctx = _make_ctx()
ctx.author.name = "alice"
await _guess_fn(bot, ctx)
message = ctx.send.call_args[0][0]
assert "wait" not in message.lower()

async def test_guess_records_cooldown(self):
bot = _make_bot(cooldown=30)
ctx = _make_ctx()
assert not bot._cooldown.is_on_cooldown()
ctx.author.name = "alice"
assert not bot._cooldown.is_on_cooldown("alice")
await _guess_fn(bot, ctx)
assert bot._cooldown.is_on_cooldown()
assert bot._cooldown.is_on_cooldown("alice")

async def test_blocked_guess_message_mentions_seconds(self):
bot = _make_bot(cooldown=30)
bot._cooldown.record()
ctx = _make_ctx()
ctx.author.name = "alice"
bot._cooldown.record("alice")
await _guess_fn(bot, ctx)
message = ctx.send.call_args[0][0]
assert "second" in message.lower()

async def test_different_users_have_independent_cooldowns(self):
"""One user being on cooldown must not block another user."""
bot = _make_bot(cooldown=30)
ctx_alice = _make_ctx()
ctx_alice.author.name = "alice"
ctx_bob = _make_ctx()
ctx_bob.author.name = "bob"
bot._cooldown.record("alice") # only alice is on cooldown
assert bot._cooldown.is_on_cooldown("alice")
assert not bot._cooldown.is_on_cooldown("bob")
# Bob should be allowed to guess despite alice being on cooldown
await _guess_fn(bot, ctx_bob)
message = ctx_bob.send.call_args[0][0]
assert "wait" not in message.lower()


# ---------------------------------------------------------------------------
# guess — game routing
Expand Down
79 changes: 30 additions & 49 deletions tests/test_cooldown.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,7 @@

import time

import pytest

from bot.cooldown import CooldownManager, GlobalCooldown
from bot.cooldown import CooldownManager


class TestCooldownManagerInit:
Expand Down Expand Up @@ -56,61 +54,44 @@ def test_record_new_user(self):
assert mgr.is_on_cooldown("newuser")


class TestGlobalCooldownInit:
def test_default_not_on_cooldown(self):
"""A freshly created GlobalCooldown should not be active."""
gc = GlobalCooldown(5)
assert not gc.is_on_cooldown()

def test_duration_stored(self):
"""The cooldown duration should be retrievable via the property."""
gc = GlobalCooldown(10)
assert gc.duration == 10

class TestCooldownManagerRemaining:
def test_remaining_zero_when_inactive(self):
"""remaining() should return 0.0 when no guess has been recorded."""
gc = GlobalCooldown(5)
assert gc.remaining() == 0.0


class TestGlobalCooldownIsOnCooldown:
def test_on_cooldown_after_record(self):
"""GlobalCooldown should be active immediately after record()."""
gc = GlobalCooldown(30)
gc.record()
assert gc.is_on_cooldown()

def test_not_on_cooldown_after_expiry(self):
"""GlobalCooldown should be inactive after the cooldown period expires."""
gc = GlobalCooldown(0)
gc.record()
time.sleep(0.01)
assert not gc.is_on_cooldown()
"""remaining() should return 0.0 when the user has not been seen."""
mgr = CooldownManager(5)
assert mgr.remaining("alice") == 0.0

def test_remaining_positive_during_cooldown(self):
"""remaining() should return a positive value while on cooldown."""
gc = GlobalCooldown(30)
gc.record()
assert gc.remaining() > 0
"""remaining() should return a positive value while user is on cooldown."""
mgr = CooldownManager(30)
mgr.record("alice")
assert mgr.remaining("alice") > 0

def test_remaining_never_negative(self):
"""remaining() should never return a negative value."""
gc = GlobalCooldown(0)
gc.record()
"""remaining() should never return a negative value after expiry."""
mgr = CooldownManager(0)
mgr.record("alice")
time.sleep(0.01)
assert gc.remaining() == 0.0
assert mgr.remaining("alice") == 0.0

def test_remaining_independent_per_user(self):
"""remaining() for one user should not affect another user."""
mgr = CooldownManager(30)
mgr.record("alice")
assert mgr.remaining("alice") > 0
assert mgr.remaining("bob") == 0.0

class TestGlobalCooldownSetDuration:

class TestCooldownManagerSetDuration:
def test_set_duration_updates_value(self):
"""set_duration() should update the cooldown duration."""
gc = GlobalCooldown(5)
gc.set_duration(15)
assert gc.duration == 15
mgr = CooldownManager(5)
mgr.set_duration(15)
assert mgr.duration == 15

def test_set_duration_zero_disables_cooldown(self):
"""Setting duration to 0 should disable the cooldown."""
gc = GlobalCooldown(30)
gc.record()
gc.set_duration(0)
assert not gc.is_on_cooldown()
"""Setting duration to 0 should disable the cooldown for a user."""
mgr = CooldownManager(30)
mgr.record("alice")
mgr.set_duration(0)
assert not mgr.is_on_cooldown("alice")

Loading