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

refactor: Consolidate scouting utilities, add test suite, use Discord timestamps

Browse Source 
- Consolidate SCOUT_TOKENS_PER_DAY and get_scout_tokens_used() into
  helpers/scouting.py (was duplicated across 3 files)
- Add midnight_timestamp() utility to helpers/utils.py
- Remove _build_scouted_ids() wrapper, use self.claims directly
- Fix build_scout_embed return type annotation
- Use Discord <t:UNIX:R> relative timestamps for scout window countdown
- Add 66-test suite covering helpers, ScoutView, and cog

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Cal Corum 2026-03-04 19:29:44 -06:00 committed by cal
parent 2d5bd86d52
commit 3c0fa133fd
10 changed files with 1987 additions and 174 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
cogs/economy_new/scouting.py
View File

@ -10,6 +10,7 @@ from discord import app_commands
from discord.ext import commands, tasks
from api_calls import db_get
from helpers.scouting import SCOUT_TOKENS_PER_DAY, get_scout_tokens_used
from helpers.utils import int_timestamp
from helpers.discord_utils import get_team_embed
from helpers.main import get_team_by_owner
@ -17,8 +18,6 @@ from helpers.constants import PD_SEASON, IMAGES
logger = logging.getLogger("discord_app")
SCOUT_TOKENS_PER_DAY = 2
class Scouting(commands.Cog):
"""Scout token tracking and expired opportunity cleanup."""
@ -45,20 +44,7 @@ class Scouting(commands.Cog):
)
return
now = datetime.datetime.now()
midnight = int_timestamp(
datetime.datetime(now.year, now.month, now.day, 0, 0, 0)
)
used_today = await db_get(
"rewards",
params=[
("name", "Scout Token"),
("team_id", team["id"]),
("created_after", midnight),
],
)
tokens_used = used_today["count"] if used_today else 0
tokens_used = await get_scout_tokens_used(team["id"])
tokens_remaining = max(0, SCOUT_TOKENS_PER_DAY - tokens_used)
embed = get_team_embed(title="Scout Tokens", team=team)

6
discord_ui/__init__.py
View File

@ -17,7 +17,10 @@ from .selectors import (
SelectView,
)
from .dropdowns import Dropdown, DropdownView
from .scout_view import ScoutView
# ScoutView intentionally NOT imported here to avoid circular import:
# helpers.main → discord_ui → scout_view → helpers.main
# Import directly: from discord_ui.scout_view import ScoutView
__all__ = [
"Question",
@ -34,5 +37,4 @@ __all__ = [
"SelectView",
"Dropdown",
"DropdownView",
"ScoutView",
]

137
discord_ui/scout_view.py
View File

@ -3,32 +3,30 @@ Scout View — Face-down card button UI for the Scouting feature.
When a player opens a pack, a ScoutView is posted with one button per card.
Other players can click a button to "scout" (blind-pick) one card, receiving
a copy. The opener keeps all their cards. Multiple players can scout different
cards from the same pack — each costs one scout token.
a copy. The opener keeps all their cards. Multiple players can scout the same
card — each gets their own copy.
"""
import datetime
import logging
import discord
from api_calls import db_get, db_post, db_patch
from api_calls import db_get, db_post
from helpers.main import get_team_by_owner, get_card_embeds
from helpers.scouting import SCOUT_TOKENS_PER_DAY, get_scout_tokens_used
from helpers.utils import int_timestamp
from helpers.discord_utils import get_team_embed
from helpers.constants import IMAGES, PD_SEASON
logger = logging.getLogger("discord_app")
SCOUT_TOKENS_PER_DAY = 2
class ScoutView(discord.ui.View):
"""Displays face-down card buttons for a scout opportunity.
- One button per card, labeled "Card 1" ... "Card N"
- Any player EXCEPT the pack opener can interact
- Each card can be scouted once; multiple players can scout different cards
- Any card can be scouted multiple times by different players
- One scout per player per pack
- Timeout: 30 minutes
"""
@ -40,6 +38,7 @@ class ScoutView(discord.ui.View):
opener_team: dict,
opener_user_id: int,
bot,
expires_unix: int = None,
):
super().__init__(timeout=1800.0)
self.scout_opp_id = scout_opp_id
@ -47,15 +46,18 @@ class ScoutView(discord.ui.View):
self.opener_team = opener_team
self.opener_user_id = opener_user_id
self.bot = bot
self.expires_unix = expires_unix
self.message: discord.Message | None = None
self.card_lines: list[tuple[int, str]] = []
# Per-card claim tracking: position -> scouter team name
self.claimed_positions: dict[int, str] = {}
# Per-card claim tracking: player_id -> list of scouter team names
self.claims: dict[int, list[str]] = {}
# Per-user lock: user IDs who have already scouted this pack
self.scouted_users: set[int] = set()
# Positions currently being processed (prevent double-click race)
self.processing: set[int] = set()
# Users currently being processed (prevent double-click race)
self.processing_users: set[int] = set()
# Total scout count
self.total_scouts = 0
for i, card in enumerate(cards):
button = ScoutButton(
@ -65,10 +67,6 @@ class ScoutView(discord.ui.View):
)
self.add_item(button)
@property
def all_claimed(self) -> bool:
return len(self.claimed_positions) >= len(self.cards)
async def update_message(self):
"""Refresh the embed with current claim state."""
if not self.message:
@ -76,34 +74,26 @@ class ScoutView(discord.ui.View):
from helpers.scouting import build_scouted_card_list
scouted_ids = {}
for pos, team_name in self.claimed_positions.items():
player_id = self.cards[pos]["player"]["player_id"]
scouted_ids[player_id] = team_name
card_list = build_scouted_card_list(self.card_lines, scouted_ids)
claim_count = len(self.claimed_positions)
if self.all_claimed:
title = "Fully Scouted!"
footer_text = f"Paper Dynasty Season {PD_SEASON} \u2022 All cards scouted"
else:
title = f"Scout Opportunity! ({claim_count}/{len(self.cards)} scouted)"
footer_text = (
f"Paper Dynasty Season {PD_SEASON} \u2022 One scout per player"
)
card_list = build_scouted_card_list(self.card_lines, self.claims)
title = f"Scout Opportunity! ({self.total_scouts} scouted)"
embed = get_team_embed(title=title, team=self.opener_team)
embed.description = (
f"**{self.opener_team['lname']}**'s pack\n\n" f"{card_list}\n\n"
)
if not self.all_claimed:
embed.description += (
"Pick a card — but which is which?\n"
"Costs 1 Scout Token (2 per day, resets at midnight Central)."
)
if self.expires_unix:
time_line = f"Scout window closes <t:{self.expires_unix}:R>."
else:
time_line = "Scout window closes in **30 minutes**."
embed.set_footer(text=footer_text, icon_url=IMAGES["logo"])
embed.description = (
f"**{self.opener_team['lname']}**'s pack\n\n"
f"{card_list}\n\n"
f"Pick a card — but which is which?\n"
f"Costs 1 Scout Token (2 per day, resets at midnight Central).\n"
f"{time_line}"
)
embed.set_footer(
text=f"Paper Dynasty Season {PD_SEASON} \u2022 One scout per player",
icon_url=IMAGES["logo"],
)
try:
await self.message.edit(embed=embed, view=self)
@ -119,18 +109,10 @@ class ScoutView(discord.ui.View):
try:
from helpers.scouting import build_scouted_card_list
scouted_ids = {}
for pos, team_name in self.claimed_positions.items():
player_id = self.cards[pos]["player"]["player_id"]
scouted_ids[player_id] = team_name
card_list = build_scouted_card_list(self.card_lines, self.claims)
card_list = build_scouted_card_list(self.card_lines, scouted_ids)
claim_count = len(self.claimed_positions)
if claim_count > 0:
title = (
f"Scout Window Closed ({claim_count}/{len(self.cards)} scouted)"
)
if self.total_scouts > 0:
title = f"Scout Window Closed ({self.total_scouts} scouted)"
else:
title = "Scout Window Closed"
@ -179,23 +161,11 @@ class ScoutButton(discord.ui.Button):
)
return
# This card already taken
if self.position in view.claimed_positions:
await interaction.response.send_message(
"This card was already scouted! Try a different one.",
ephemeral=True,
)
# Prevent double-click race for same user
if interaction.user.id in view.processing_users:
return
# Prevent double-click race on same card
if self.position in view.processing:
await interaction.response.send_message(
"Hold on, someone's claiming this card right now...",
ephemeral=True,
)
return
view.processing.add(self.position)
view.processing_users.add(interaction.user.id)
await interaction.response.defer(ephemeral=True)
try:
@ -209,19 +179,7 @@ class ScoutButton(discord.ui.Button):
return
# Check scout token balance
now = datetime.datetime.now()
midnight = int_timestamp(
datetime.datetime(now.year, now.month, now.day, 0, 0, 0)
)
used_today = await db_get(
"rewards",
params=[
("name", "Scout Token"),
("team_id", scouter_team["id"]),
("created_after", midnight),
],
)
tokens_used = used_today["count"] if used_today else 0
tokens_used = await get_scout_tokens_used(scouter_team["id"])
if tokens_used >= SCOUT_TOKENS_PER_DAY:
await interaction.followup.send(
@ -257,7 +215,7 @@ class ScoutButton(discord.ui.Button):
"team_id": scouter_team["id"],
"season": current["season"] if current else PD_SEASON,
"week": current["week"] if current else 1,
"created": int_timestamp(now),
"created": int_timestamp(),
},
)
@ -275,19 +233,12 @@ class ScoutButton(discord.ui.Button):
)
# Track the claim
view.claimed_positions[self.position] = scouter_team["lname"]
player_id = self.card["player"]["player_id"]
if player_id not in view.claims:
view.claims[player_id] = []
view.claims[player_id].append(scouter_team["lname"])
view.scouted_users.add(interaction.user.id)
# Update this button
self.disabled = True
self.style = discord.ButtonStyle.success
self.label = "Scouted!"
# If all cards claimed, disable remaining buttons and stop
if view.all_claimed:
for item in view.children:
item.disabled = True
view.stop()
view.total_scouts += 1
# Update the shared embed
await view.update_message()
@ -336,4 +287,4 @@ class ScoutButton(discord.ui.Button):
except Exception:
pass
finally:
view.processing.discard(self.position)
view.processing_users.discard(interaction.user.id)

53
helpers/scouting.py
View File

@ -12,13 +12,14 @@ import random
import discord
from api_calls import db_post
from helpers.utils import int_timestamp
from api_calls import db_get, db_post
from helpers.utils import int_timestamp, midnight_timestamp
from helpers.discord_utils import get_team_embed
from helpers.constants import IMAGES, PD_SEASON
logger = logging.getLogger("discord_app")
SCOUT_TOKENS_PER_DAY = 2
SCOUT_WINDOW_SECONDS = 1800 # 30 minutes
# Rarity value → display symbol
@ -32,6 +33,19 @@ RARITY_SYMBOLS = {
}
async def get_scout_tokens_used(team_id: int) -> int:
"""Return how many scout tokens a team has used today."""
used_today = await db_get(
"rewards",
params=[
("name", "Scout Token"),
("team_id", team_id),
("created_after", midnight_timestamp()),
],
)
return used_today["count"] if used_today else 0
def _build_card_lines(cards: list[dict]) -> list[tuple[int, str]]:
"""Build a shuffled list of (player_id, display_line) tuples."""
lines = []
@ -53,7 +67,8 @@ def build_scout_embed(
opener_team: dict,
cards: list[dict],
card_lines: list[tuple[int, str]] = None,
) -> discord.Embed:
expires_unix: int = None,
) -> tuple[discord.Embed, list[tuple[int, str]]]:
"""Build the embed shown above the scout buttons.
Shows a shuffled list of cards (rarity + player name) so scouters
@ -67,12 +82,17 @@ def build_scout_embed(
card_list = "\n".join(line for _, line in card_lines)
if expires_unix:
time_line = f"Scout window closes <t:{expires_unix}:R>."
else:
time_line = "Scout window closes in **30 minutes**."
embed.description = (
f"**{opener_team['lname']}** just opened a pack!\n\n"
f"**Cards in this pack:**\n{card_list}\n\n"
f"Pick a card — but which is which?\n"
f"Costs 1 Scout Token (2 per day, resets at midnight Central).\n"
f"This window closes in **30 minutes**."
f"{time_line}"
)
embed.set_footer(
text=f"Paper Dynasty Season {PD_SEASON} \u2022 One player per pack",
@ -83,20 +103,25 @@ def build_scout_embed(
def build_scouted_card_list(
card_lines: list[tuple[int, str]],
scouted_cards: dict[int, str],
scouted_cards: dict[int, list[str]],
) -> str:
"""Rebuild the card list marking scouted cards with the scouter's team name.
"""Rebuild the card list marking scouted cards with scouter team names.
Parameters
----------
card_lines : shuffled list of (player_id, display_line) tuples
scouted_cards : {player_id: scouter_team_name} for each claimed card
scouted_cards : {player_id: [team_name, ...]} for each claimed card
"""
result = []
for player_id, line in card_lines:
if player_id in scouted_cards:
team_name = scouted_cards[player_id]
result.append(f"{line} \u2014 \u2714\ufe0f *{team_name}*")
teams = scouted_cards.get(player_id)
if teams:
count = len(teams)
names = ", ".join(f"*{t}*" for t in teams)
if count == 1:
result.append(f"{line} \u2014 \u2714\ufe0f {names}")
else:
result.append(f"{line} \u2014 \u2714\ufe0f x{count} ({names})")
else:
result.append(line)
return "\n".join(result)
@ -152,7 +177,12 @@ async def create_scout_opportunity(
logger.error(f"Failed to create scout opportunity: {e}")
return
embed, card_lines = build_scout_embed(opener_team, pack_cards)
expires_unix = int(
(now + datetime.timedelta(seconds=SCOUT_WINDOW_SECONDS)).timestamp()
)
embed, card_lines = build_scout_embed(
opener_team, pack_cards, expires_unix=expires_unix
)
# Get bot reference from context
bot = getattr(context, "bot", None) or getattr(context, "client", None)
@ -163,6 +193,7 @@ async def create_scout_opportunity(
opener_team=opener_team,
opener_user_id=opener_user.id,
bot=bot,
expires_unix=expires_unix,
)
view.card_lines = card_lines

115
helpers/utils.py
View File

@ -4,6 +4,7 @@ General Utilities
This module contains standalone utility functions with minimal dependencies,
including timestamp conversion, position abbreviations, and simple helpers.
"""
import datetime
from typing import Optional
import discord
@ -16,48 +17,55 @@ def int_timestamp(datetime_obj: Optional[datetime.datetime] = None):
return int(datetime.datetime.now().timestamp())
def midnight_timestamp() -> int:
"""Return today's midnight (00:00:00) as an integer millisecond timestamp."""
now = datetime.datetime.now()
midnight = datetime.datetime(now.year, now.month, now.day, 0, 0, 0)
return int_timestamp(midnight)
def get_pos_abbrev(field_pos: str) -> str:
"""Convert position name to standard abbreviation."""
if field_pos.lower() == 'catcher':
return 'C'
elif field_pos.lower() == 'first baseman':
return '1B'
elif field_pos.lower() == 'second baseman':
return '2B'
elif field_pos.lower() == 'third baseman':
return '3B'
elif field_pos.lower() == 'shortstop':
return 'SS'
elif field_pos.lower() == 'left fielder':
return 'LF'
elif field_pos.lower() == 'center fielder':
return 'CF'
elif field_pos.lower() == 'right fielder':
return 'RF'
if field_pos.lower() == "catcher":
return "C"
elif field_pos.lower() == "first baseman":
return "1B"
elif field_pos.lower() == "second baseman":
return "2B"
elif field_pos.lower() == "third baseman":
return "3B"
elif field_pos.lower() == "shortstop":
return "SS"
elif field_pos.lower() == "left fielder":
return "LF"
elif field_pos.lower() == "center fielder":
return "CF"
elif field_pos.lower() == "right fielder":
return "RF"
else:
return 'P'
return "P"
def position_name_to_abbrev(position_name):
"""Convert position name to abbreviation (alternate format)."""
if position_name == 'Catcher':
return 'C'
elif position_name == 'First Base':
return '1B'
elif position_name == 'Second Base':
return '2B'
elif position_name == 'Third Base':
return '3B'
elif position_name == 'Shortstop':
return 'SS'
elif position_name == 'Left Field':
return 'LF'
elif position_name == 'Center Field':
return 'CF'
elif position_name == 'Right Field':
return 'RF'
elif position_name == 'Pitcher':
return 'P'
if position_name == "Catcher":
return "C"
elif position_name == "First Base":
return "1B"
elif position_name == "Second Base":
return "2B"
elif position_name == "Third Base":
return "3B"
elif position_name == "Shortstop":
return "SS"
elif position_name == "Left Field":
return "LF"
elif position_name == "Center Field":
return "CF"
elif position_name == "Right Field":
return "RF"
elif position_name == "Pitcher":
return "P"
else:
return position_name
@ -67,13 +75,13 @@ def user_has_role(user: discord.User | discord.Member, role_name: str) -> bool:
for x in user.roles:
if x.name == role_name:
return True
return False
def get_roster_sheet_legacy(team):
"""Get legacy roster sheet URL for a team."""
return f'https://docs.google.com/spreadsheets/d/{team.gsheet}/edit'
return f"https://docs.google.com/spreadsheets/d/{team.gsheet}/edit"
def get_roster_sheet(team):
@ -83,13 +91,15 @@ def get_roster_sheet(team):
Handles both dict and Team object formats.
"""
# Handle both dict (team["gsheet"]) and object (team.gsheet) formats
gsheet = team.get("gsheet") if isinstance(team, dict) else getattr(team, "gsheet", None)
return f'https://docs.google.com/spreadsheets/d/{gsheet}/edit'
gsheet = (
team.get("gsheet") if isinstance(team, dict) else getattr(team, "gsheet", None)
)
return f"https://docs.google.com/spreadsheets/d/{gsheet}/edit"
def get_player_url(team, player) -> str:
"""Generate player URL for SBA or Baseball Reference."""
if team.get('league') == 'SBA':
if team.get("league") == "SBA":
return f'https://statsplus.net/super-baseball-association/player/{player["player_id"]}'
else:
return f'https://www.baseball-reference.com/players/{player["bbref_id"][0]}/{player["bbref_id"]}.shtml'
@ -101,7 +111,7 @@ def owner_only(ctx) -> bool:
owners = [287463767924137994, 1087936030899347516]
# Handle both Context (has .author) and Interaction (has .user) objects
user = getattr(ctx, 'user', None) or getattr(ctx, 'author', None)
user = getattr(ctx, "user", None) or getattr(ctx, "author", None)
if user and user.id in owners:
return True
@ -121,35 +131,36 @@ def get_context_user(ctx):
discord.User or discord.Member: The user who invoked the command
"""
# Handle both Context (has .author) and Interaction (has .user) objects
return getattr(ctx, 'user', None) or getattr(ctx, 'author', None)
return getattr(ctx, "user", None) or getattr(ctx, "author", None)
def get_cal_user(ctx):
"""Get the Cal user from context. Always returns an object with .mention attribute."""
import logging
logger = logging.getLogger('discord_app')
logger = logging.getLogger("discord_app")
# Define placeholder user class first
class PlaceholderUser:
def __init__(self):
self.mention = "<@287463767924137994>"
self.id = 287463767924137994
# Handle both Context and Interaction objects
if hasattr(ctx, 'bot'): # Context object
if hasattr(ctx, "bot"): # Context object
bot = ctx.bot
logger.debug("get_cal_user: Using Context object")
elif hasattr(ctx, 'client'): # Interaction object
elif hasattr(ctx, "client"): # Interaction object
bot = ctx.client
logger.debug("get_cal_user: Using Interaction object")
else:
logger.error("get_cal_user: No bot or client found in context")
return PlaceholderUser()
if not bot:
logger.error("get_cal_user: bot is None")
return PlaceholderUser()
logger.debug(f"get_cal_user: Searching among members")
try:
for user in bot.get_all_members():
@ -158,7 +169,7 @@ def get_cal_user(ctx):
return user
except Exception as e:
logger.error(f"get_cal_user: Exception in get_all_members: {e}")
# Fallback: try to get user directly by ID
logger.debug("get_cal_user: User not found in get_all_members, trying get_user")
try:
@ -170,7 +181,7 @@ def get_cal_user(ctx):
logger.debug("get_cal_user: get_user returned None")
except Exception as e:
logger.error(f"get_cal_user: Exception in get_user: {e}")
# Last resort: return a placeholder user object with mention
logger.debug("get_cal_user: Using placeholder user")
return PlaceholderUser()
return PlaceholderUser()

0
tests/scouting/__init__.py Normal file
View File

160
tests/scouting/conftest.py Normal file
View File

@ -0,0 +1,160 @@
"""Shared fixtures for scouting feature tests."""
import pytest
import asyncio
from unittest.mock import AsyncMock, MagicMock, Mock
import discord
from discord.ext import commands
# ---------------------------------------------------------------------------
# Sample data factories
# ---------------------------------------------------------------------------
def _make_player(player_id, name, rarity_name, rarity_value, headshot=None):
"""Build a minimal player dict matching the API shape used by scouting."""
return {
"player_id": player_id,
"p_name": name,
"rarity": {"name": rarity_name, "value": rarity_value, "color": "ffffff"},
"headshot": headshot or "https://example.com/headshot.jpg",
}
def _make_card(card_id, player):
"""Wrap a player dict inside a card dict (as returned by the cards API)."""
return {"id": card_id, "player": player}
@pytest.fixture
def sample_players():
"""Five players spanning different rarities for a realistic pack."""
return [
_make_player(101, "Mike Trout", "MVP", 5),
_make_player(102, "Juan Soto", "All-Star", 3),
_make_player(103, "Marcus Semien", "Starter", 2),
_make_player(104, "Willy Adames", "Reserve", 1),
_make_player(105, "Generic Bench", "Replacement", 0),
]
@pytest.fixture
def sample_cards(sample_players):
"""Five card dicts wrapping the sample players."""
return [_make_card(i + 1, p) for i, p in enumerate(sample_players)]
@pytest.fixture
def opener_team():
"""Team dict for the pack opener."""
return {
"id": 10,
"abbrev": "OPN",
"sname": "Openers",
"lname": "Opening Squad",
"gm_id": 99999,
"gmname": "Opener GM",
"color": "a6ce39",
"logo": "https://example.com/logo.png",
"season": 4,
}
@pytest.fixture
def scouter_team():
"""Team dict for a player who scouts a card."""
return {
"id": 20,
"abbrev": "SCT",
"sname": "Scouts",
"lname": "Scouting Squad",
"gm_id": 88888,
"gmname": "Scout GM",
"color": "3498db",
"logo": "https://example.com/scout_logo.png",
"season": 4,
}
@pytest.fixture
def scouter_team_2():
"""Second scouter team for multi-scout tests."""
return {
"id": 30,
"abbrev": "SC2",
"sname": "Scouts2",
"lname": "Second Scouts",
"gm_id": 77777,
"gmname": "Scout GM 2",
"color": "e74c3c",
"logo": "https://example.com/scout2_logo.png",
"season": 4,
}
# ---------------------------------------------------------------------------
# Discord mocks
# ---------------------------------------------------------------------------
@pytest.fixture
def mock_bot():
"""Mock Discord bot."""
bot = AsyncMock(spec=commands.Bot)
bot.get_cog = Mock(return_value=None)
bot.add_cog = AsyncMock()
bot.wait_until_ready = AsyncMock()
# Mock guild / channel lookup for send_to_channel
channel_mock = AsyncMock(spec=discord.TextChannel)
channel_mock.send = AsyncMock()
guild_mock = Mock(spec=discord.Guild)
guild_mock.text_channels = [channel_mock]
channel_mock.name = "pd-network-news"
bot.guilds = [guild_mock]
return bot
@pytest.fixture
def mock_interaction():
"""Mock Discord interaction for slash commands."""
interaction = AsyncMock(spec=discord.Interaction)
interaction.response = AsyncMock()
interaction.response.defer = AsyncMock()
interaction.response.send_message = AsyncMock()
interaction.response.is_done = Mock(return_value=False)
interaction.followup = AsyncMock()
interaction.followup.send = AsyncMock()
interaction.user = Mock(spec=discord.Member)
interaction.user.id = 12345
interaction.user.mention = "<@12345>"
interaction.channel = Mock(spec=discord.TextChannel)
interaction.channel.name = "pack-openings"
interaction.channel.send = AsyncMock()
return interaction
@pytest.fixture
def mock_channel():
"""Mock #pack-openings channel."""
channel = AsyncMock(spec=discord.TextChannel)
channel.name = "pack-openings"
channel.send = AsyncMock()
return channel
# ---------------------------------------------------------------------------
# Logging suppression
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def silence_logging():
"""Suppress log noise during tests."""
import logging
logging.getLogger("discord_app").setLevel(logging.CRITICAL)

1029
tests/scouting/test_scout_view.py Normal file
View File

File diff suppressed because it is too large Load Diff

269
tests/scouting/test_scouting_cog.py Normal file
View File

@ -0,0 +1,269 @@
"""Tests for cogs/economy_new/scouting.py — the Scouting cog.
Covers the /scout-tokens command and the cleanup_expired background task.
Note: Scouting.__init__ calls self.cleanup_expired.start() which requires
a running event loop. All tests that instantiate the cog must be async.
"""
import datetime
import pytest
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import discord
from discord.ext import commands
from cogs.economy_new.scouting import Scouting, SCOUT_TOKENS_PER_DAY
def _make_team():
return {
"id": 1,
"lname": "Test Team",
"color": "a6ce39",
"logo": "https://example.com/logo.png",
"season": 4,
}
# ---------------------------------------------------------------------------
# Cog setup
# ---------------------------------------------------------------------------
class TestScoutingCogSetup:
"""Tests for cog initialization and lifecycle."""
@pytest.mark.asyncio
async def test_cog_initializes(self, mock_bot):
"""The Scouting cog should initialize without errors."""
cog = Scouting(mock_bot)
cog.cleanup_expired.cancel()
assert cog.bot is mock_bot
@pytest.mark.asyncio
async def test_cleanup_task_starts(self, mock_bot):
"""The cleanup_expired loop task should be started on init."""
cog = Scouting(mock_bot)
assert cog.cleanup_expired.is_running()
cog.cleanup_expired.cancel()
@pytest.mark.asyncio
async def test_cog_unload_calls_cancel(self, mock_bot):
"""Unloading the cog should call cancel on the cleanup task."""
cog = Scouting(mock_bot)
cog.cleanup_expired.cancel()
# Verify cog_unload runs without error
await cog.cog_unload()
# ---------------------------------------------------------------------------
# /scout-tokens command
# ---------------------------------------------------------------------------
class TestScoutTokensCommand:
"""Tests for the /scout-tokens slash command."""
@pytest.mark.asyncio
@patch("cogs.economy_new.scouting.get_scout_tokens_used", new_callable=AsyncMock)
@patch("cogs.economy_new.scouting.get_team_by_owner", new_callable=AsyncMock)
async def test_shows_remaining_tokens(
self, mock_get_team, mock_get_tokens, mock_bot
):
"""Should display the correct number of remaining tokens."""
cog = Scouting(mock_bot)
cog.cleanup_expired.cancel()
mock_get_team.return_value = _make_team()
mock_get_tokens.return_value = 1 # 1 used today
interaction = AsyncMock(spec=discord.Interaction)
interaction.response = AsyncMock()
interaction.response.defer = AsyncMock()
interaction.followup = AsyncMock()
interaction.followup.send = AsyncMock()
interaction.user = Mock()
interaction.user.id = 12345
await cog.scout_tokens_command.callback(cog, interaction)
interaction.response.defer.assert_called_once_with(ephemeral=True)
interaction.followup.send.assert_called_once()
call_kwargs = interaction.followup.send.call_args[1]
embed = call_kwargs["embed"]
remaining = SCOUT_TOKENS_PER_DAY - 1
assert str(remaining) in embed.description
@pytest.mark.asyncio
@patch("cogs.economy_new.scouting.get_scout_tokens_used", new_callable=AsyncMock)
@patch("cogs.economy_new.scouting.get_team_by_owner", new_callable=AsyncMock)
async def test_no_team_rejects(self, mock_get_team, mock_get_tokens, mock_bot):
"""A user without a PD team should get a rejection message."""
cog = Scouting(mock_bot)
cog.cleanup_expired.cancel()
mock_get_team.return_value = None
interaction = AsyncMock(spec=discord.Interaction)
interaction.response = AsyncMock()
interaction.response.defer = AsyncMock()
interaction.followup = AsyncMock()
interaction.followup.send = AsyncMock()
interaction.user = Mock()
interaction.user.id = 12345
await cog.scout_tokens_command.callback(cog, interaction)
msg = interaction.followup.send.call_args[0][0]
assert "team" in msg.lower()
mock_get_tokens.assert_not_called()
@pytest.mark.asyncio
@patch("cogs.economy_new.scouting.get_scout_tokens_used", new_callable=AsyncMock)
@patch("cogs.economy_new.scouting.get_team_by_owner", new_callable=AsyncMock)
async def test_all_tokens_used_shows_zero(
self, mock_get_team, mock_get_tokens, mock_bot
):
"""When all tokens are used, should show 0 remaining with extra message."""
cog = Scouting(mock_bot)
cog.cleanup_expired.cancel()
mock_get_team.return_value = _make_team()
mock_get_tokens.return_value = SCOUT_TOKENS_PER_DAY
interaction = AsyncMock(spec=discord.Interaction)
interaction.response = AsyncMock()
interaction.response.defer = AsyncMock()
interaction.followup = AsyncMock()
interaction.followup.send = AsyncMock()
interaction.user = Mock()
interaction.user.id = 12345
await cog.scout_tokens_command.callback(cog, interaction)
embed = interaction.followup.send.call_args[1]["embed"]
assert "0" in embed.description
assert (
"used all" in embed.description.lower()
or "tomorrow" in embed.description.lower()
)
@pytest.mark.asyncio
@patch("cogs.economy_new.scouting.get_scout_tokens_used", new_callable=AsyncMock)
@patch("cogs.economy_new.scouting.get_team_by_owner", new_callable=AsyncMock)
async def test_no_tokens_used_shows_full(
self, mock_get_team, mock_get_tokens, mock_bot
):
"""When no tokens have been used, should show the full daily allowance."""
cog = Scouting(mock_bot)
cog.cleanup_expired.cancel()
mock_get_team.return_value = _make_team()
mock_get_tokens.return_value = 0
interaction = AsyncMock(spec=discord.Interaction)
interaction.response = AsyncMock()
interaction.response.defer = AsyncMock()
interaction.followup = AsyncMock()
interaction.followup.send = AsyncMock()
interaction.user = Mock()
interaction.user.id = 12345
await cog.scout_tokens_command.callback(cog, interaction)
embed = interaction.followup.send.call_args[1]["embed"]
assert str(SCOUT_TOKENS_PER_DAY) in embed.description
@pytest.mark.asyncio
@patch("cogs.economy_new.scouting.get_scout_tokens_used", new_callable=AsyncMock)
@patch("cogs.economy_new.scouting.get_team_by_owner", new_callable=AsyncMock)
async def test_db_get_returns_none(self, mock_get_team, mock_get_tokens, mock_bot):
"""If get_scout_tokens_used returns 0 (API failure handled internally), should show full tokens."""
cog = Scouting(mock_bot)
cog.cleanup_expired.cancel()
mock_get_team.return_value = _make_team()
mock_get_tokens.return_value = (
0 # get_scout_tokens_used handles None internally
)
interaction = AsyncMock(spec=discord.Interaction)
interaction.response = AsyncMock()
interaction.response.defer = AsyncMock()
interaction.followup = AsyncMock()
interaction.followup.send = AsyncMock()
interaction.user = Mock()
interaction.user.id = 12345
await cog.scout_tokens_command.callback(cog, interaction)
embed = interaction.followup.send.call_args[1]["embed"]
assert str(SCOUT_TOKENS_PER_DAY) in embed.description
@pytest.mark.asyncio
@patch("cogs.economy_new.scouting.get_scout_tokens_used", new_callable=AsyncMock)
@patch("cogs.economy_new.scouting.get_team_by_owner", new_callable=AsyncMock)
async def test_over_limit_tokens_shows_zero(
self, mock_get_team, mock_get_tokens, mock_bot
):
"""If somehow more tokens than the daily limit were used, should show 0 not negative."""
cog = Scouting(mock_bot)
cog.cleanup_expired.cancel()
mock_get_team.return_value = _make_team()
mock_get_tokens.return_value = 5 # more than SCOUT_TOKENS_PER_DAY
interaction = AsyncMock(spec=discord.Interaction)
interaction.response = AsyncMock()
interaction.response.defer = AsyncMock()
interaction.followup = AsyncMock()
interaction.followup.send = AsyncMock()
interaction.user = Mock()
interaction.user.id = 12345
await cog.scout_tokens_command.callback(cog, interaction)
embed = interaction.followup.send.call_args[1]["embed"]
# Should show "0" not "-3"
assert "0" in embed.description
assert "-" not in embed.description.split("remaining")[0]
# ---------------------------------------------------------------------------
# cleanup_expired task
# ---------------------------------------------------------------------------
class TestCleanupExpired:
"""Tests for the background cleanup task."""
@pytest.mark.asyncio
@patch("cogs.economy_new.scouting.db_get", new_callable=AsyncMock)
async def test_cleanup_logs_expired_opportunities(self, mock_db_get, mock_bot):
"""The cleanup task should query for expired unclaimed opportunities."""
cog = Scouting(mock_bot)
cog.cleanup_expired.cancel()
mock_db_get.return_value = {"count": 3}
# Call the coroutine directly (not via the loop)
await cog.cleanup_expired.coro(cog)
mock_db_get.assert_called_once()
call_args = mock_db_get.call_args
assert call_args[0][0] == "scout_opportunities"
@pytest.mark.asyncio
@patch("cogs.economy_new.scouting.db_get", new_callable=AsyncMock)
async def test_cleanup_handles_api_failure(self, mock_db_get, mock_bot):
"""Cleanup should not crash if the API is unavailable."""
cog = Scouting(mock_bot)
cog.cleanup_expired.cancel()
mock_db_get.side_effect = Exception("API not ready")
# Should not raise
await cog.cleanup_expired.coro(cog)

374
tests/scouting/test_scouting_helpers.py Normal file
View File

@ -0,0 +1,374 @@
"""Tests for helpers/scouting.py — embed builders and scout opportunity creation.
Covers the pure functions (_build_card_lines, build_scout_embed,
build_scouted_card_list) and the async create_scout_opportunity flow.
"""
import pytest
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import discord
from helpers.scouting import (
_build_card_lines,
build_scout_embed,
build_scouted_card_list,
create_scout_opportunity,
RARITY_SYMBOLS,
)
# ---------------------------------------------------------------------------
# _build_card_lines
# ---------------------------------------------------------------------------
class TestBuildCardLines:
"""Tests for the shuffled card line builder."""
def test_returns_correct_count(self, sample_cards):
"""Should produce one line per card in the pack."""
lines = _build_card_lines(sample_cards)
assert len(lines) == len(sample_cards)
def test_each_line_contains_player_id(self, sample_cards):
"""Each tuple's first element should be the player_id from the card."""
lines = _build_card_lines(sample_cards)
ids = {pid for pid, _ in lines}
expected_ids = {c["player"]["player_id"] for c in sample_cards}
assert ids == expected_ids
def test_each_line_contains_player_name(self, sample_cards):
"""The display string should include the player's name."""
lines = _build_card_lines(sample_cards)
for pid, display in lines:
card = next(c for c in sample_cards if c["player"]["player_id"] == pid)
assert card["player"]["p_name"] in display
def test_each_line_contains_rarity_name(self, sample_cards):
"""The display string should include the rarity tier name."""
lines = _build_card_lines(sample_cards)
for pid, display in lines:
card = next(c for c in sample_cards if c["player"]["player_id"] == pid)
assert card["player"]["rarity"]["name"] in display
def test_rarity_symbol_present(self, sample_cards):
"""Each line should start with the appropriate rarity emoji."""
lines = _build_card_lines(sample_cards)
for pid, display in lines:
card = next(c for c in sample_cards if c["player"]["player_id"] == pid)
rarity_val = card["player"]["rarity"]["value"]
expected_symbol = RARITY_SYMBOLS.get(rarity_val, "\u26ab")
assert display.startswith(expected_symbol)
def test_output_is_shuffled(self, sample_cards):
"""Over many runs, the order should not always match the input order.
We run 20 iterations — if it comes out sorted every time, the shuffle
is broken (probability ~1/20! per run, effectively zero).
"""
input_order = [c["player"]["player_id"] for c in sample_cards]
saw_different = False
for _ in range(20):
lines = _build_card_lines(sample_cards)
output_order = [pid for pid, _ in lines]
if output_order != input_order:
saw_different = True
break
assert saw_different, "Card lines were never shuffled across 20 runs"
def test_empty_cards(self):
"""Empty input should produce an empty list."""
assert _build_card_lines([]) == []
def test_unknown_rarity_uses_fallback_symbol(self):
"""A rarity value not in RARITY_SYMBOLS should get the black circle fallback."""
card = {
"id": 99,
"player": {
"player_id": 999,
"p_name": "Unknown Rarity",
"rarity": {"name": "Legendary", "value": 99, "color": "gold"},
},
}
lines = _build_card_lines([card])
assert lines[0][1].startswith("\u26ab") # black circle fallback
# ---------------------------------------------------------------------------
# build_scout_embed
# ---------------------------------------------------------------------------
class TestBuildScoutEmbed:
"""Tests for the embed builder shown above scout buttons."""
def test_returns_embed_and_card_lines(self, opener_team, sample_cards):
"""Should return a (discord.Embed, list) tuple."""
embed, card_lines = build_scout_embed(opener_team, sample_cards)
assert isinstance(embed, discord.Embed)
assert isinstance(card_lines, list)
assert len(card_lines) == len(sample_cards)
def test_embed_description_contains_team_name(self, opener_team, sample_cards):
"""The embed body should mention the opener's team name."""
embed, _ = build_scout_embed(opener_team, sample_cards)
assert opener_team["lname"] in embed.description
def test_embed_description_contains_all_player_names(
self, opener_team, sample_cards
):
"""Every player name from the pack should appear in the embed."""
embed, _ = build_scout_embed(opener_team, sample_cards)
for card in sample_cards:
assert card["player"]["p_name"] in embed.description
def test_embed_mentions_token_cost(self, opener_team, sample_cards):
"""The embed should tell users about the scout token cost."""
embed, _ = build_scout_embed(opener_team, sample_cards)
assert "Scout Token" in embed.description
def test_embed_mentions_time_limit(self, opener_team, sample_cards):
"""The embed should mention the 30-minute window."""
embed, _ = build_scout_embed(opener_team, sample_cards)
assert "30 minutes" in embed.description
def test_prebuilt_card_lines_are_reused(self, opener_team, sample_cards):
"""When card_lines are passed in, they should be reused (not rebuilt)."""
prebuilt = [(101, "Custom Line 1"), (102, "Custom Line 2")]
embed, returned_lines = build_scout_embed(
opener_team, sample_cards, card_lines=prebuilt
)
assert returned_lines is prebuilt
assert "Custom Line 1" in embed.description
assert "Custom Line 2" in embed.description
# ---------------------------------------------------------------------------
# build_scouted_card_list
# ---------------------------------------------------------------------------
class TestBuildScoutedCardList:
"""Tests for the card list formatter that marks scouted cards."""
def test_no_scouts_returns_plain_lines(self):
"""With no scouts, output should match the raw card lines."""
card_lines = [
(101, "\U0001f7e3 MVP — Mike Trout"),
(102, "\U0001f535 All-Star — Juan Soto"),
]
result = build_scouted_card_list(card_lines, {})
assert result == "\U0001f7e3 MVP — Mike Trout\n\U0001f535 All-Star — Juan Soto"
def test_single_scout_shows_team_name(self):
"""A card scouted once should show a checkmark and the team name."""
card_lines = [
(101, "\U0001f7e3 MVP — Mike Trout"),
(102, "\U0001f535 All-Star — Juan Soto"),
]
scouted = {101: ["Scouting Squad"]}
result = build_scouted_card_list(card_lines, scouted)
assert "\u2714\ufe0f" in result # checkmark
assert "*Scouting Squad*" in result
# Unscouted card should appear plain
lines = result.split("\n")
assert "\u2714" not in lines[1]
def test_multiple_scouts_shows_count_and_names(self):
"""A card scouted multiple times should show the count and all team names."""
card_lines = [(101, "\U0001f7e3 MVP — Mike Trout")]
scouted = {101: ["Team A", "Team B", "Team C"]}
result = build_scouted_card_list(card_lines, scouted)
assert "x3" in result
assert "*Team A*" in result
assert "*Team B*" in result
assert "*Team C*" in result
def test_mixed_scouted_and_unscouted(self):
"""Only scouted cards should have marks; unscouted cards stay plain."""
card_lines = [
(101, "Line A"),
(102, "Line B"),
(103, "Line C"),
]
scouted = {102: ["Some Team"]}
result = build_scouted_card_list(card_lines, scouted)
lines = result.split("\n")
assert "\u2714" not in lines[0]
assert "\u2714" in lines[1]
assert "\u2714" not in lines[2]
def test_empty_input(self):
"""Empty card lines should produce an empty string."""
assert build_scouted_card_list([], {}) == ""
def test_two_scouts_shows_count(self):
"""Two scouts on the same card should show x2."""
card_lines = [(101, "Line A")]
scouted = {101: ["Team X", "Team Y"]}
result = build_scouted_card_list(card_lines, scouted)
assert "x2" in result
# ---------------------------------------------------------------------------
# create_scout_opportunity
# ---------------------------------------------------------------------------
class TestCreateScoutOpportunity:
"""Tests for the async scout opportunity creation flow."""
@pytest.mark.asyncio
@patch("helpers.scouting.db_post", new_callable=AsyncMock)
async def test_posts_to_api_and_sends_message(
self, mock_db_post, sample_cards, opener_team, mock_channel, mock_bot
):
"""Should POST to scout_opportunities and send a message to the channel."""
mock_db_post.return_value = {"id": 42}
opener_user = Mock()
opener_user.id = 99999
context = Mock()
context.bot = mock_bot
await create_scout_opportunity(
sample_cards, opener_team, mock_channel, opener_user, context
)
# API was called to create the opportunity
mock_db_post.assert_called_once()
call_args = mock_db_post.call_args
assert call_args[0][0] == "scout_opportunities"
assert call_args[1]["payload"]["opener_team_id"] == opener_team["id"]
# Message was sent to the channel
mock_channel.send.assert_called_once()
@pytest.mark.asyncio
@patch("helpers.scouting.db_post", new_callable=AsyncMock)
async def test_skips_wrong_channel(
self, mock_db_post, sample_cards, opener_team, mock_bot
):
"""Should silently return when the channel is not #pack-openings."""
channel = AsyncMock(spec=discord.TextChannel)
channel.name = "general"
opener_user = Mock()
opener_user.id = 99999
context = Mock()
context.bot = mock_bot
await create_scout_opportunity(
sample_cards, opener_team, channel, opener_user, context
)
mock_db_post.assert_not_called()
channel.send.assert_not_called()
@pytest.mark.asyncio
@patch("helpers.scouting.db_post", new_callable=AsyncMock)
async def test_skips_empty_pack(
self, mock_db_post, opener_team, mock_channel, mock_bot
):
"""Should silently return when pack_cards is empty."""
opener_user = Mock()
opener_user.id = 99999
context = Mock()
context.bot = mock_bot
await create_scout_opportunity(
[], opener_team, mock_channel, opener_user, context
)
mock_db_post.assert_not_called()
@pytest.mark.asyncio
@patch("helpers.scouting.db_post", new_callable=AsyncMock)
async def test_skips_none_channel(
self, mock_db_post, sample_cards, opener_team, mock_bot
):
"""Should handle None channel without crashing."""
opener_user = Mock()
opener_user.id = 99999
context = Mock()
context.bot = mock_bot
await create_scout_opportunity(
sample_cards, opener_team, None, opener_user, context
)
mock_db_post.assert_not_called()
@pytest.mark.asyncio
@patch("helpers.scouting.db_post", new_callable=AsyncMock)
async def test_api_failure_does_not_raise(
self, mock_db_post, sample_cards, opener_team, mock_channel, mock_bot
):
"""Scout creation failure must never crash the pack opening flow."""
mock_db_post.side_effect = Exception("API down")
opener_user = Mock()
opener_user.id = 99999
context = Mock()
context.bot = mock_bot
# Should not raise
await create_scout_opportunity(
sample_cards, opener_team, mock_channel, opener_user, context
)
@pytest.mark.asyncio
@patch("helpers.scouting.db_post", new_callable=AsyncMock)
async def test_channel_send_failure_does_not_raise(
self, mock_db_post, sample_cards, opener_team, mock_channel, mock_bot
):
"""If the channel.send fails, it should be caught gracefully."""
mock_db_post.return_value = {"id": 42}
mock_channel.send.side_effect = discord.HTTPException(
Mock(status=500), "Server error"
)
opener_user = Mock()
opener_user.id = 99999
context = Mock()
context.bot = mock_bot
# Should not raise
await create_scout_opportunity(
sample_cards, opener_team, mock_channel, opener_user, context
)
@pytest.mark.asyncio
@patch("helpers.scouting.db_post", new_callable=AsyncMock)
async def test_context_client_fallback(
self, mock_db_post, sample_cards, opener_team, mock_channel, mock_bot
):
"""When context.bot is None, should fall back to context.client for the bot ref."""
mock_db_post.return_value = {"id": 42}
opener_user = Mock()
opener_user.id = 99999
context = Mock(spec=[]) # empty spec — no .bot attribute
context.client = mock_bot
await create_scout_opportunity(
sample_cards, opener_team, mock_channel, opener_user, context
)
mock_channel.send.assert_called_once()
@pytest.mark.asyncio
@patch("helpers.scouting.db_post", new_callable=AsyncMock)
async def test_view_message_is_assigned(
self, mock_db_post, sample_cards, opener_team, mock_channel, mock_bot
):
"""The message returned by channel.send should be assigned to view.message.
This linkage is required for update_message and on_timeout to work.
"""
mock_db_post.return_value = {"id": 42}
sent_msg = AsyncMock(spec=discord.Message)
mock_channel.send.return_value = sent_msg
opener_user = Mock()
opener_user.id = 99999
context = Mock()
context.bot = mock_bot
await create_scout_opportunity(
sample_cards, opener_team, mock_channel, opener_user, context
)