Compare commits

..

4 Commits

Author SHA1 Message Date
Cal Corum
cc72827dad fix(gauntlet): fix loss_max=0 falsy-zero trap in recap marker logic
All checks were successful
Ruff Lint / lint (pull_request) Successful in 19s
`(loss_max or 99)` treats `loss_max=0` as 99, so 10-1 runs showed 
instead of  for perfect-run rewards. Fix uses explicit None check.
Tighten test to assert  presence rather than just absence of .

Addresses review feedback on PR #165.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-10 12:32:10 -05:00
Cal Corum
0b8beda8b5 feat(compare): add /compare slash command for side-by-side card comparison
All checks were successful
Ruff Lint / lint (pull_request) Successful in 13s
Implements Roadmap 2.5b: new /compare command lets players compare two
cards of the same type (batter vs batter or pitcher vs pitcher) in a
side-by-side embed with directional delta arrows (▲▼═).

- cogs/compare.py: new CompareCog with /compare slash command and
  player_autocomplete on both params; fetches battingcard/pitchingcard
  data from API; validates type compatibility; sends public embed
- tests/test_compare_command.py: 30 unit tests covering _delta_arrow,
  _is_pitcher, batter/pitcher embed builders, type mismatch error, and
  edge cases (None stats, tied values)
- paperdynasty.py: registers cogs.compare in COGS list

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-10 10:35:17 -05:00
Cal Corum
9228d2e66c feat(gauntlet): post completion recap embed on 10-win gauntlet finish
All checks were successful
Ruff Lint / lint (pull_request) Successful in 13s
Adds build_gauntlet_recap_embed (sync builder) and post_gauntlet_recap
(async sender) to gauntlets.py.  Called from post_result when wins == 10.
Embed shows champion name + user mention, final record, win-progression
ladder, and full prize-distribution table with earned/unearned markers.
Gracefully skips if channel is None or any step raises.

Roadmap 2.4a — closes growth-sweep finding 2026-04-10-003.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-04-10 10:31:57 -05:00
Cal Corum
f62e08889f fix(gameplay): replace bare except with NoResultFound in cache-miss paths
All checks were successful
Ruff Lint / lint (pull_request) Successful in 27s
In gameplay_queries.py, 10 try/except blocks used `except Exception:` to
detect SQLModel row-not-found cache misses (.one() returning no result).
This silently swallowed connection failures, attribute errors, and
programming bugs on the gameplay hot path.

Narrowed each handler to `except NoResultFound:` (sqlalchemy.exc).
Real errors now propagate instead of being misinterpreted as cache misses.

Refs: autonomous-pipeline finding analyst-2026-04-10-003
Category: stability / error_handling

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-10 10:28:40 -05:00
10 changed files with 1453 additions and 324 deletions

425
cogs/compare.py Normal file
View File

@ -0,0 +1,425 @@
"""
Compare cog /compare slash command.
Displays a side-by-side stat embed for two cards of the same type (batter
vs batter, pitcher vs pitcher) with directional delta arrows.
Card stats are derived from the battingcards / pitchingcards API endpoints
which carry the actual card data (running, steal range, pitcher ratings, etc.)
alongside the player's rarity and cost.
Batter stats shown:
Cost (Overall proxy), Rarity, Running, Steal Low, Steal High, Bunting,
Hit & Run
Pitcher stats shown:
Cost (Overall proxy), Rarity, Starter Rating, Relief Rating,
Closer Rating, Balk, Wild Pitch
Arrow semantics:
card2 is higher (better for -better stats)
card1 is higher (better for -better stats)
tied
"""
import logging
from typing import List, Optional, Tuple
import discord
from discord import app_commands
from discord.ext import commands
from api_calls import db_get
from constants import PD_PLAYERS_ROLE_NAME
from utilities.autocomplete import player_autocomplete
logger = logging.getLogger("discord_app")
# ----- helpers ----------------------------------------------------------------
GRADE_ORDER = ["A", "B", "C", "D", "E", "F"]
def _grade_to_int(grade: Optional[str]) -> Optional[int]:
"""Convert a letter grade (A-F) to a numeric rank for comparison.
Lower rank = better grade. Returns None when grade is None/empty.
"""
if grade is None:
return None
upper = grade.upper().strip()
try:
return GRADE_ORDER.index(upper)
except ValueError:
return None
def _delta_arrow(
val1,
val2,
higher_is_better: bool = True,
grade_field: bool = False,
) -> str:
"""Return a directional arrow showing which card has the better value.
Args:
val1: stat value for card1
val2: stat value for card2
higher_is_better: when True, a larger numeric value is preferred.
When False (e.g. balk, wild_pitch), a smaller value is preferred.
grade_field: when True, val1/val2 are letter grades (A-F) where A > B.
Returns:
'' if card2 wins, '' if card1 wins, '' if tied / not comparable.
"""
if val1 is None or val2 is None:
return ""
if grade_field:
n1 = _grade_to_int(val1)
n2 = _grade_to_int(val2)
if n1 is None or n2 is None or n1 == n2:
return ""
# Lower index = better grade; card2 wins when n2 < n1
return "" if n2 < n1 else ""
try:
n1 = float(val1)
n2 = float(val2)
except (TypeError, ValueError):
return ""
if n1 == n2:
return ""
if higher_is_better:
return "" if n2 > n1 else ""
else:
# lower is better (e.g. balk count)
return "" if n2 < n1 else ""
def _fmt(val) -> str:
"""Format a stat value for display. Falls back to '' when None."""
if val is None:
return ""
return str(val)
def _is_pitcher(player: dict) -> bool:
"""Return True if the player is a pitcher (pos_1 in SP, RP)."""
return player.get("pos_1", "").upper() in ("SP", "RP")
def _card_type_label(player: dict) -> str:
return "pitcher" if _is_pitcher(player) else "batter"
# ----- embed builder (pure function, testable without Discord state) ---------
_BATTER_STATS: List[Tuple[str, str, str, bool]] = [
# (label, key_in_card, key_in_player, higher_is_better)
("Cost (Overall)", "cost", "player", True),
("Rarity", "rarity_value", "player", True),
("Running", "running", "battingcard", True),
("Steal Low", "steal_low", "battingcard", True),
("Steal High", "steal_high", "battingcard", True),
("Bunting", "bunting", "battingcard", False), # grade: A>B>C...
("Hit & Run", "hit_and_run", "battingcard", False), # grade
]
_PITCHER_STATS: List[Tuple[str, str, str, bool]] = [
("Cost (Overall)", "cost", "player", True),
("Rarity", "rarity_value", "player", True),
("Starter Rating", "starter_rating", "pitchingcard", True),
("Relief Rating", "relief_rating", "pitchingcard", True),
("Closer Rating", "closer_rating", "pitchingcard", True),
("Balk", "balk", "pitchingcard", False), # lower is better
("Wild Pitch", "wild_pitch", "pitchingcard", False), # lower is better
]
_GRADE_FIELDS = {"bunting", "hit_and_run"}
class CompareMismatchError(ValueError):
"""Raised when two cards are not of the same type."""
def build_compare_embed(
card1: dict,
card2: dict,
card1_name: str,
card2_name: str,
) -> discord.Embed:
"""Build a side-by-side comparison embed for two cards.
Args:
card1: card data dict (player + battingcard OR pitchingcard).
Expects 'player', 'battingcard' or 'pitchingcard' keys.
card2: same shape as card1
card1_name: display name override (falls back to player p_name)
card2_name: display name override
Returns:
discord.Embed with inline stat rows
Raises:
CompareMismatchError: if card types differ (batter vs pitcher)
"""
p1 = card1.get("player", {})
p2 = card2.get("player", {})
type1 = _card_type_label(p1)
type2 = _card_type_label(p2)
if type1 != type2:
raise CompareMismatchError(
f"Card types differ: '{card1_name}' is a {type1}, "
f"'{card2_name}' is a {type2}."
)
color_hex = p1.get("rarity", {}).get("color", "3498DB")
try:
color = int(color_hex, 16)
except (TypeError, ValueError):
color = 0x3498DB
# Embed header
embed = discord.Embed(
title="Card Comparison",
description=(
f"**{card1_name}** vs **{card2_name}** — "
f"{'Pitchers' if type1 == 'pitcher' else 'Batters'}"
),
color=color,
)
# Thumbnail from card1 headshot if available
thumbnail = p1.get("headshot") or p2.get("headshot")
if thumbnail:
embed.set_thumbnail(url=thumbnail)
# Card name headers (inline, side-by-side feel)
embed.add_field(name="Card 1", value=f"**{card1_name}**", inline=True)
embed.add_field(name="Stat", value="\u200b", inline=True)
embed.add_field(name="Card 2", value=f"**{card2_name}**", inline=True)
# Choose stat spec
stats = _PITCHER_STATS if type1 == "pitcher" else _BATTER_STATS
for label, key, source, higher_is_better in stats:
# Extract values
if source == "player":
if key == "rarity_value":
v1 = p1.get("rarity", {}).get("value")
v2 = p2.get("rarity", {}).get("value")
# Display as rarity name + value
display1 = p1.get("rarity", {}).get("name", _fmt(v1))
display2 = p2.get("rarity", {}).get("name", _fmt(v2))
else:
v1 = p1.get(key)
v2 = p2.get(key)
display1 = _fmt(v1)
display2 = _fmt(v2)
elif source == "battingcard":
bc1 = card1.get("battingcard", {}) or {}
bc2 = card2.get("battingcard", {}) or {}
v1 = bc1.get(key)
v2 = bc2.get(key)
display1 = _fmt(v1)
display2 = _fmt(v2)
elif source == "pitchingcard":
pc1 = card1.get("pitchingcard", {}) or {}
pc2 = card2.get("pitchingcard", {}) or {}
v1 = pc1.get(key)
v2 = pc2.get(key)
display1 = _fmt(v1)
display2 = _fmt(v2)
else:
continue
is_grade = key in _GRADE_FIELDS
arrow = _delta_arrow(
v1,
v2,
higher_is_better=higher_is_better,
grade_field=is_grade,
)
embed.add_field(name="\u200b", value=display1, inline=True)
embed.add_field(name=label, value=arrow, inline=True)
embed.add_field(name="\u200b", value=display2, inline=True)
embed.set_footer(text="Paper Dynasty — /compare")
return embed
# ----- card fetch helpers -----------------------------------------------------
async def _fetch_player_by_name(name: str) -> Optional[dict]:
"""Search for a player by name and return the first match."""
result = await db_get(
"players/search",
params=[("q", name), ("limit", 1)],
timeout=5,
)
if not result or not result.get("players"):
return None
return result["players"][0]
async def _fetch_batting_card(player_id: int) -> Optional[dict]:
"""Fetch the variant-0 batting card for a player."""
result = await db_get(
"battingcards",
params=[("player_id", player_id), ("variant", 0)],
timeout=5,
)
if not result or not result.get("cards"):
# Fall back to any variant
result = await db_get(
"battingcards",
params=[("player_id", player_id)],
timeout=5,
)
if not result or not result.get("cards"):
return None
return result["cards"][0]
async def _fetch_pitching_card(player_id: int) -> Optional[dict]:
"""Fetch the variant-0 pitching card for a player."""
result = await db_get(
"pitchingcards",
params=[("player_id", player_id), ("variant", 0)],
timeout=5,
)
if not result or not result.get("cards"):
result = await db_get(
"pitchingcards",
params=[("player_id", player_id)],
timeout=5,
)
if not result or not result.get("cards"):
return None
return result["cards"][0]
async def _build_card_data(player: dict) -> dict:
"""Build a unified card data dict for use with build_compare_embed.
Returns a dict with 'player', 'battingcard', and 'pitchingcard' keys.
"""
pid = player.get("player_id") or player.get("id")
batting_card = None
pitching_card = None
if _is_pitcher(player):
pitching_card = await _fetch_pitching_card(pid)
else:
batting_card = await _fetch_batting_card(pid)
return {
"player": player,
"battingcard": batting_card,
"pitchingcard": pitching_card,
}
# ----- Cog --------------------------------------------------------------------
class CompareCog(commands.Cog, name="Compare"):
"""Slash command cog providing /compare for side-by-side card comparison."""
def __init__(self, bot: commands.Bot):
self.bot = bot
@app_commands.command(
name="compare",
description="Side-by-side stat comparison for two cards",
)
@app_commands.checks.has_any_role(PD_PLAYERS_ROLE_NAME)
@app_commands.describe(
card1="First player's card (type a name to search)",
card2="Second player's card (type a name to search)",
)
@app_commands.autocomplete(card1=player_autocomplete, card2=player_autocomplete)
async def compare_command(
self,
interaction: discord.Interaction,
card1: str,
card2: str,
):
"""Compare two cards side-by-side.
Fetches both players by name, validates that they are the same card
type (both batters or both pitchers), then builds and sends the
comparison embed.
"""
await interaction.response.defer()
# --- fetch player 1 ---------------------------------------------------
player1 = await _fetch_player_by_name(card1)
if not player1:
await interaction.edit_original_response(
content=f"Could not find a card for **{card1}**."
)
return
# --- fetch player 2 ---------------------------------------------------
player2 = await _fetch_player_by_name(card2)
if not player2:
await interaction.edit_original_response(
content=f"Could not find a card for **{card2}**."
)
return
# --- type-gate --------------------------------------------------------
type1 = _card_type_label(player1)
type2 = _card_type_label(player2)
if type1 != type2:
await interaction.edit_original_response(
content=(
"Can only compare cards of the same type "
"(batter vs batter, pitcher vs pitcher)."
),
)
return
# --- build card data --------------------------------------------------
card_data1 = await _build_card_data(player1)
card_data2 = await _build_card_data(player2)
name1 = player1.get("p_name", card1)
name2 = player2.get("p_name", card2)
# --- build embed ------------------------------------------------------
try:
embed = build_compare_embed(card_data1, card_data2, name1, name2)
except CompareMismatchError as exc:
logger.warning("CompareMismatchError (should not reach here): %s", exc)
await interaction.edit_original_response(
content=(
"Can only compare cards of the same type "
"(batter vs batter, pitcher vs pitcher)."
),
)
return
except Exception as exc:
logger.error(
"compare_command build_compare_embed error: %s", exc, exc_info=True
)
await interaction.edit_original_response(
content="Something went wrong building the comparison. Please contact Cal."
)
return
# Send publicly so players can share the result
await interaction.edit_original_response(embed=embed)
async def setup(bot: commands.Bot) -> None:
"""Discord.py cog loader entry point."""
await bot.add_cog(CompareCog(bot))

View File

@ -11,10 +11,9 @@ import discord
from discord import app_commands from discord import app_commands
from discord.ext import commands from discord.ext import commands
import aiohttp from api_calls import db_delete, db_get, db_post
from api_calls import AUTH_TOKEN, db_delete, db_get, db_post, get_req_url
from helpers.constants import PD_SEASON from helpers.constants import PD_SEASON
from helpers.main import get_team_by_owner
from helpers.refractor_constants import TIER_NAMES from helpers.refractor_constants import TIER_NAMES
from helpers.refractor_test_data import ( from helpers.refractor_test_data import (
build_batter_plays, build_batter_plays,
@ -26,9 +25,7 @@ from helpers.refractor_test_data import (
CURRENT_SEASON = PD_SEASON CURRENT_SEASON = PD_SEASON
SENTINEL_PITCHER_ID = 3 logger = logging.getLogger(__name__)
logger = logging.getLogger("discord_app")
class CleanupView(discord.ui.View): class CleanupView(discord.ui.View):
@ -108,45 +105,36 @@ class DevToolsCog(commands.Cog):
@group_dev.command( @group_dev.command(
name="refractor-test", description="Run refractor integration test on a card" name="refractor-test", description="Run refractor integration test on a card"
) )
@app_commands.describe( @app_commands.describe(card_id="The batting or pitching card ID to test")
card_id="Card-instance ID (from the unified cards table; discoverable via /refractor status)"
)
async def refractor_test(self, interaction: discord.Interaction, card_id: int): async def refractor_test(self, interaction: discord.Interaction, card_id: int):
await interaction.response.defer() await interaction.response.defer()
# --- Phase 1: Setup --- # --- Phase 1: Setup ---
# Look up card via the unified cards endpoint. cards.id is # Look up card (try batting first, then pitching)
# globally unique across all card instances, so there's no card = await db_get("battingcards", object_id=card_id)
# batting/pitching template-ID collision (spec card_type_key = "batting"
# 2026-04-11-refractor-test-unified-cards-lookup-design.md). if card is None:
card = await db_get("cards", object_id=card_id) card = await db_get("pitchingcards", object_id=card_id)
card_type_key = "pitching"
if card is None: if card is None:
await interaction.edit_original_response( await interaction.edit_original_response(
content=f"❌ Card #{card_id} not found." content=f"❌ Card #{card_id} not found (checked batting and pitching)."
) )
return return
player_id = card["player"]["player_id"] player_id = card["player"]["id"]
player_name = card["player"]["p_name"] player_name = card["player"]["p_name"]
team_id = card.get("team_id") or card["player"].get("team_id")
# The card instance's owning team is the authoritative context
# for this test — its refractor state will be read and synthetic
# plays will be credited against it.
team_id = (card.get("team") or {}).get("id")
if team_id is None: if team_id is None:
await interaction.edit_original_response( team = await get_team_by_owner(interaction.user.id)
content=f"❌ Card #{card_id} has no owning team." if team is None:
) await interaction.edit_original_response(
return content="❌ Could not determine team ID. You must own a team."
)
# Derive card type from the player's primary position. return
pos_1 = (card["player"].get("pos_1") or "").upper() team_id = team["id"]
if pos_1 == "SP":
card_type_key, card_type_default = "pitching", "sp"
elif pos_1 in ("RP", "CP"):
card_type_key, card_type_default = "pitching", "rp"
else:
card_type_key, card_type_default = "batting", "batter"
# Fetch refractor state # Fetch refractor state
refractor_data = await db_get( refractor_data = await db_get(
@ -154,17 +142,6 @@ class DevToolsCog(commands.Cog):
params=[("team_id", team_id), ("limit", 100)], params=[("team_id", team_id), ("limit", 100)],
) )
logger.info(
"dev_tools state-lookup: team_id=%s player_id=%s resp_keys=%s item_count=%s item_pids=%s",
team_id,
player_id,
list(refractor_data.keys()) if refractor_data else None,
len(refractor_data.get("items", [])) if refractor_data else 0,
[i.get("player_id") for i in refractor_data.get("items", [])]
if refractor_data
else [],
)
# Find this player's entry # Find this player's entry
card_state = None card_state = None
if refractor_data and refractor_data.get("items"): if refractor_data and refractor_data.get("items"):
@ -173,12 +150,6 @@ class DevToolsCog(commands.Cog):
card_state = item card_state = item
break break
logger.info(
"dev_tools state-lookup matched: player_id=%s card_state=%s",
player_id,
card_state,
)
# Determine current state and thresholds # Determine current state and thresholds
if card_state: if card_state:
current_tier = card_state["current_tier"] current_tier = card_state["current_tier"]
@ -188,7 +159,7 @@ class DevToolsCog(commands.Cog):
else: else:
current_tier = 0 current_tier = 0
current_value = 0 current_value = 0
card_type = card_type_default card_type = "batter" if card_type_key == "batting" else "sp"
next_threshold = ( next_threshold = (
37 if card_type == "batter" else (10 if card_type == "sp" else 3) 37 if card_type == "batter" else (10 if card_type == "sp" else 3)
) )
@ -203,7 +174,25 @@ class DevToolsCog(commands.Cog):
gap = max(0, next_threshold - current_value) gap = max(0, next_threshold - current_value)
plan = calculate_plays_needed(gap, card_type) plan = calculate_plays_needed(gap, card_type)
opposing_player_id = SENTINEL_PITCHER_ID # Find an opposing player
if card_type == "batter":
opposing_cards = await db_get(
"pitchingcards",
params=[("team_id", team_id), ("variant", 0)],
)
else:
opposing_cards = await db_get(
"battingcards",
params=[("team_id", team_id), ("variant", 0)],
)
if not opposing_cards or not opposing_cards.get("cards"):
await interaction.edit_original_response(
content=f"❌ No opposing {'pitcher' if card_type == 'batter' else 'batter'} cards found on team {team_id}."
)
return
opposing_player_id = opposing_cards["cards"][0]["player"]["id"]
# Build and send initial embed # Build and send initial embed
tier_name = TIER_NAMES.get(current_tier, f"T{current_tier}") tier_name = TIER_NAMES.get(current_tier, f"T{current_tier}")
@ -368,19 +357,27 @@ class DevToolsCog(commands.Cog):
continue continue
try: try:
today = date.today().isoformat() today = date.today().isoformat()
render_url = get_req_url( render_resp = await db_get(
f"players/{tu['player_id']}/{card_type_key}card/{today}/{variant}", f"players/{tu['player_id']}/{card_type_key}card/{today}/{variant}",
api_ver=2, none_okay=True,
) )
async with aiohttp.ClientSession(headers=AUTH_TOKEN) as sess: if render_resp:
async with sess.get(render_url) as r: results.append("✅ Card rendered + S3 upload triggered")
if r.status == 200: img_url = (
results.append("✅ Card rendered + S3 upload triggered") render_resp
else: if isinstance(render_resp, str)
body = await r.text() else render_resp.get("image_url")
results.append( )
f"⚠️ Card render non-200 ({r.status}): {body[:80]}" if (
) img_url
and isinstance(img_url, str)
and img_url.startswith("http")
):
embed.set_image(url=img_url)
else:
results.append(
"⚠️ Card render returned no data (may still be processing)"
)
except Exception as e: except Exception as e:
results.append(f"⚠️ Card render failed (non-fatal): {e}") results.append(f"⚠️ Card render failed (non-fatal): {e}")

View File

@ -4293,8 +4293,8 @@ async def _trigger_variant_renders(tier_ups: list) -> dict:
if variant is None: if variant is None:
continue continue
player_id = tier_up["player_id"] player_id = tier_up["player_id"]
ct = tier_up.get("card_type", "batter") track = tier_up.get("track_name", "Batter")
card_type = "pitching" if ct in ("sp", "rp") else "batting" card_type = "pitching" if track.lower() == "pitcher" else "batting"
try: try:
result = await db_get( result = await db_get(
f"players/{player_id}/{card_type}card/{today}/{variant}", f"players/{player_id}/{card_type}card/{today}/{variant}",

View File

@ -2393,6 +2393,156 @@ async def evolve_pokemon(this_team: Team, channel, responders):
await channel.send("All of your Pokemon are fully evolved!") await channel.send("All of your Pokemon are fully evolved!")
def build_gauntlet_recap_embed(
this_run: dict,
this_event: dict,
main_team: dict,
rewards: list[dict],
) -> discord.Embed:
"""Build a Discord embed summarising a completed gauntlet run.
Called when a player finishes a gauntlet (10 wins). This is a pure
synchronous builder so it can be unit-tested without a Discord connection.
Args:
this_run: gauntletruns API dict (must have wins/losses/team keys).
this_event: events API dict (name, url, short_desc).
main_team: teams API dict for the player's real team (gmid, lname, logo).
rewards: list of gauntletrewards API dicts for the event.
Returns:
A discord.Embed with champion highlight, run record, and prize table.
"""
# Gold/champion accent colour
GOLD = 0xFFD700
team_name = main_team.get("lname", "Unknown Team")
gmid = main_team.get("gmid")
wins = this_run.get("wins", 0)
losses = this_run.get("losses", 0)
event_name = this_event.get("name", "Gauntlet")
embed = discord.Embed(
title=f"Gauntlet Complete: {event_name}",
color=GOLD,
)
# Champion highlight
champion_value = f"**{team_name}**"
if gmid:
champion_value += f"\n<@{gmid}>"
if main_team.get("logo"):
embed.set_thumbnail(url=main_team["logo"])
embed.add_field(name="Champion", value=champion_value, inline=True)
# Run record
embed.add_field(
name="Final Record",
value=f"**{wins}-{losses}**",
inline=True,
)
# Bracket / progression — for the solo gauntlet format this is a
# milestone ladder (win-by-win), not a bracket tree.
if wins > 0:
bracket_lines = []
milestone_wins = sorted(
{r["win_num"] for r in rewards if r.get("win_num") is not None}
)
for mw in milestone_wins:
marker = "" if wins >= mw else ""
bracket_lines.append(f"{marker} Win {mw}")
if bracket_lines:
embed.add_field(
name="Progression",
value="\n".join(bracket_lines),
inline=False,
)
# Prize distribution table
if rewards:
prize_lines = []
for r in sorted(rewards, key=lambda x: x.get("win_num", 0)):
win_num = r.get("win_num", "?")
loss_max = r.get("loss_max")
label = f"{win_num}-0" if loss_max == 0 else f"{win_num} Wins"
earned = (
wins >= win_num and losses <= loss_max
if loss_max is not None
else wins >= win_num
)
marker = (
""
if earned
else ""
if losses > (loss_max if loss_max is not None else 99)
else ""
)
reward = r.get("reward", {})
if reward.get("money"):
prize_desc = f"{reward['money']}"
elif reward.get("player"):
prize_desc = reward["player"].get("description", "Special Card")
elif reward.get("pack_type"):
prize_desc = f"1x {reward['pack_type']['name']} Pack"
else:
prize_desc = "Reward"
prize_lines.append(f"{marker} {label}: {prize_desc}")
if prize_lines:
embed.add_field(
name="Prize Distribution",
value="\n".join(prize_lines),
inline=False,
)
embed.set_footer(text="Paper Dynasty Gauntlet")
return embed
async def post_gauntlet_recap(
this_run: dict,
this_event: dict,
main_team: dict,
channel,
) -> None:
"""Send a gauntlet completion recap embed to the given channel.
Fetches all rewards for the event so the prize table is complete, then
builds and posts the recap embed. Gracefully handles a missing or
unavailable channel by logging and returning without raising so the
gauntlet completion flow is never interrupted.
Args:
this_run: gauntletruns API dict.
this_event: events API dict.
main_team: player's main teams API dict.
channel: discord.TextChannel to post to, or None.
"""
if channel is None:
logger.warning(
"post_gauntlet_recap: no channel available — recap skipped "
f"(run_id={this_run.get('id')})"
)
return
try:
all_rewards_query = await db_get(
"gauntletrewards", params=[("gauntlet_id", this_event["id"])]
)
all_rewards = all_rewards_query.get("rewards", [])
embed = build_gauntlet_recap_embed(this_run, this_event, main_team, all_rewards)
await channel.send(embed=embed)
except Exception:
logger.warning(
"post_gauntlet_recap: failed to send recap embed "
f"(run_id={this_run.get('id')})",
exc_info=True,
)
async def post_result( async def post_result(
run_id: int, run_id: int,
is_win: bool, is_win: bool,
@ -2544,6 +2694,10 @@ async def post_result(
final_message += f"\n\nGo share the highlights in {get_channel(channel, 'pd-news-ticker').mention}!" final_message += f"\n\nGo share the highlights in {get_channel(channel, 'pd-news-ticker').mention}!"
await channel.send(content=final_message, embed=await get_embed(this_run)) await channel.send(content=final_message, embed=await get_embed(this_run))
# Post gauntlet completion recap embed when the run is finished (10 wins)
if this_run["wins"] == 10:
await post_gauntlet_recap(this_run, this_event, main_team, channel)
else: else:
# this_run = await db_patch( # this_run = await db_patch(
# 'gauntletruns', # 'gauntletruns',

View File

@ -59,12 +59,8 @@ async def get_player_headshot(player):
) )
try: try:
async with aiohttp.ClientSession() as session: resp = requests.get(req_url, timeout=2).text
async with session.get( soup = BeautifulSoup(resp, "html.parser")
req_url, timeout=aiohttp.ClientTimeout(total=2)
) as resp:
text = await resp.text()
soup = BeautifulSoup(text, "html.parser")
for item in soup.find_all("img"): for item in soup.find_all("img"):
if "headshot" in item["src"]: if "headshot" in item["src"]:
await db_patch( await db_patch(
@ -1765,14 +1761,12 @@ async def open_st_pr_packs(all_packs: list, team: dict, context):
logger.error(f"open_packs - unable to roll_for_cards for packs: {all_packs}") logger.error(f"open_packs - unable to roll_for_cards for packs: {all_packs}")
raise ValueError("I was not able to unpack these cards") raise ValueError("I was not able to unpack these cards")
async def _fetch_pack_cards(p_id): all_cards = []
result = await db_get("cards", params=[("pack_id", p_id)]) for p_id in pack_ids:
for card in result["cards"]: new_cards = await db_get("cards", params=[("pack_id", p_id)])
for card in new_cards["cards"]:
card.setdefault("pack_id", p_id) card.setdefault("pack_id", p_id)
return result["cards"] all_cards.extend(new_cards["cards"])
results = await asyncio.gather(*[_fetch_pack_cards(p_id) for p_id in pack_ids])
all_cards = [card for pack_cards in results for card in pack_cards]
if not all_cards: if not all_cards:
logger.error(f"open_packs - unable to get cards for packs: {pack_ids}") logger.error(f"open_packs - unable to get cards for packs: {pack_ids}")

View File

@ -54,6 +54,7 @@ COGS = [
"cogs.gameplay", "cogs.gameplay",
"cogs.economy_new.scouting", "cogs.economy_new.scouting",
"cogs.refractor", "cogs.refractor",
"cogs.compare",
] ]
intents = discord.Intents.default() intents = discord.Intents.default()

View File

@ -0,0 +1,395 @@
"""
Tests for the /compare slash command embed builder (cogs/compare.py).
What:
- build_compare_embed() is a pure function that takes two card-data dicts
and returns a discord.Embed.
- Tests verify field count, arrow directions, type-mismatch raises, and
tied stats.
Why:
- The embed builder has no Discord I/O so it can be tested synchronously
without a bot or API calls.
- Correct arrow direction is critical for usability: wrong arrows would
mislead players making trade/lineup decisions.
"""
import sys
import os
import pytest
import discord
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from cogs.compare import (
build_compare_embed,
CompareMismatchError,
_delta_arrow,
_is_pitcher,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
def _make_player(
name: str = "Test Player",
pos_1: str = "CF",
rarity_value: int = 3,
rarity_name: str = "All-Star",
cost: int = 300,
headshot: str = None,
) -> dict:
"""Build a minimal player dict that mirrors the API response shape."""
return {
"player_id": 1,
"p_name": name,
"pos_1": pos_1,
"cost": cost,
"rarity": {
"value": rarity_value,
"name": rarity_name,
"color": "FFD700",
},
"headshot": headshot,
}
def _make_batting_card(
running: int = 12,
steal_low: int = 7,
steal_high: int = 12,
bunting: str = "C",
hit_and_run: str = "B",
) -> dict:
"""Build a minimal batting card dict."""
return {
"running": running,
"steal_low": steal_low,
"steal_high": steal_high,
"steal_auto": False,
"steal_jump": 0.2,
"bunting": bunting,
"hit_and_run": hit_and_run,
"hand": "R",
"offense_col": 1,
}
def _make_pitching_card(
starter_rating: int = 7,
relief_rating: int = 4,
closer_rating: int = None,
balk: int = 2,
wild_pitch: int = 3,
) -> dict:
"""Build a minimal pitching card dict."""
return {
"starter_rating": starter_rating,
"relief_rating": relief_rating,
"closer_rating": closer_rating,
"balk": balk,
"wild_pitch": wild_pitch,
"hand": "R",
}
def _batter_card_data(player: dict, batting_card: dict) -> dict:
return {
"player": player,
"battingcard": batting_card,
"pitchingcard": None,
}
def _pitcher_card_data(player: dict, pitching_card: dict) -> dict:
return {
"player": player,
"battingcard": None,
"pitchingcard": pitching_card,
}
# ---------------------------------------------------------------------------
# _delta_arrow unit tests
# ---------------------------------------------------------------------------
class TestDeltaArrow:
"""_delta_arrow correctly indicates which side wins."""
def test_higher_wins_when_card2_greater(self):
"""▲ when card2 value is higher and higher_is_better=True."""
assert _delta_arrow(10, 15) == ""
def test_higher_wins_when_card1_greater(self):
"""▼ when card1 value is higher and higher_is_better=True."""
assert _delta_arrow(20, 10) == ""
def test_tied_returns_equals(self):
"""═ when both values are equal."""
assert _delta_arrow(10, 10) == ""
def test_lower_is_better_arrow_flipped(self):
"""▲ when card2 is lower and lower_is_better (e.g. balk count)."""
assert _delta_arrow(5, 2, higher_is_better=False) == ""
def test_lower_is_better_card1_wins(self):
"""▼ when card1 is lower and lower_is_better."""
assert _delta_arrow(2, 5, higher_is_better=False) == ""
def test_grade_field_a_beats_b(self):
"""▲ when card2 has grade A and card1 has grade B (grade_field=True)."""
assert _delta_arrow("B", "A", grade_field=True) == ""
def test_grade_field_b_beats_c(self):
"""▼ when card1 has grade B and card2 has grade C."""
assert _delta_arrow("B", "C", grade_field=True) == ""
def test_grade_field_tie(self):
"""═ when both cards have the same grade."""
assert _delta_arrow("C", "C", grade_field=True) == ""
def test_none_returns_equals(self):
"""═ when either value is None (missing stat)."""
assert _delta_arrow(None, 10) == ""
assert _delta_arrow(10, None) == ""
# ---------------------------------------------------------------------------
# _is_pitcher
# ---------------------------------------------------------------------------
class TestIsPitcher:
"""_is_pitcher correctly classifies positions."""
def test_sp_is_pitcher(self):
assert _is_pitcher({"pos_1": "SP"}) is True
def test_rp_is_pitcher(self):
assert _is_pitcher({"pos_1": "RP"}) is True
def test_cf_is_batter(self):
assert _is_pitcher({"pos_1": "CF"}) is False
def test_dh_is_batter(self):
assert _is_pitcher({"pos_1": "DH"}) is False
def test_lowercase_sp(self):
"""pos_1 comparison is case-insensitive."""
assert _is_pitcher({"pos_1": "sp"}) is True
# ---------------------------------------------------------------------------
# build_compare_embed — batter path
# ---------------------------------------------------------------------------
class TestBuildCompareEmbedBatters:
"""build_compare_embed works correctly for two batter cards."""
def setup_method(self):
"""Create two distinct batter cards for comparison."""
self.player1 = _make_player("Mike Trout", pos_1="CF", cost=500, rarity_value=5)
self.player2 = _make_player("Joe Batter", pos_1="LF", cost=200, rarity_value=2)
self.bc1 = _make_batting_card(
running=15, steal_low=9, steal_high=14, bunting="B", hit_and_run="A"
)
self.bc2 = _make_batting_card(
running=10, steal_low=5, steal_high=10, bunting="C", hit_and_run="C"
)
self.card1 = _batter_card_data(self.player1, self.bc1)
self.card2 = _batter_card_data(self.player2, self.bc2)
def test_embed_is_discord_embed(self):
"""Return value is a discord.Embed instance."""
embed = build_compare_embed(self.card1, self.card2, "Mike Trout", "Joe Batter")
assert isinstance(embed, discord.Embed)
def test_embed_title_contains_comparison(self):
"""Embed title identifies this as a card comparison."""
embed = build_compare_embed(self.card1, self.card2, "Mike Trout", "Joe Batter")
assert "Comparison" in embed.title
def test_embed_field_count(self):
"""
Batter embed has header row (3 fields) + 7 stat rows × 3 fields = 24
total inline fields.
"""
embed = build_compare_embed(self.card1, self.card2, "Mike Trout", "Joe Batter")
assert len(embed.fields) == 3 + 7 * 3 # 24
def test_higher_cost_gets_down_arrow_in_center_column(self):
"""
card1 has higher cost (500 vs 200). The center arrow field for
'Cost (Overall)' should be '' (card1 wins when higher_is_better).
"""
embed = build_compare_embed(self.card1, self.card2, "Mike Trout", "Joe Batter")
# First stat row starts at field index 3; center field of each row is idx+1
cost_arrow_field = embed.fields[4] # index 3=left, 4=center, 5=right
assert cost_arrow_field.value == ""
def test_higher_running_gets_down_arrow(self):
"""
card1 running=15 > card2 running=10 center arrow for Running is .
Running is the 3rd stat (header row at 0..2, cost at 3..5, rarity at
6..8, running at 9..11 center = index 10).
"""
embed = build_compare_embed(self.card1, self.card2, "Mike Trout", "Joe Batter")
running_arrow = embed.fields[10]
assert running_arrow.value == ""
def test_better_grade_card1_bunting_b_beats_c(self):
"""
card1 bunting='B' beats card2 bunting='C'.
Bunting is the 6th stat, center field at index 3 + 5*3 + 1 = 19.
Arrow should be (card1 wins).
"""
embed = build_compare_embed(self.card1, self.card2, "Mike Trout", "Joe Batter")
bunt_arrow = embed.fields[19]
assert bunt_arrow.value == ""
def test_type_mismatch_raises(self):
"""CompareMismatchError raised when one card is batter and other pitcher."""
pitcher_player = _make_player("Max Scherzer", pos_1="SP", cost=400)
pitcher_card = _make_pitching_card(starter_rating=9)
card_p = _pitcher_card_data(pitcher_player, pitcher_card)
with pytest.raises(CompareMismatchError, match="Card types differ"):
build_compare_embed(self.card1, card_p, "Mike Trout", "Max Scherzer")
# ---------------------------------------------------------------------------
# build_compare_embed — pitcher path
# ---------------------------------------------------------------------------
class TestBuildCompareEmbedPitchers:
"""build_compare_embed works correctly for two pitcher cards."""
def setup_method(self):
self.player1 = _make_player(
"Max Scherzer", pos_1="SP", cost=450, rarity_value=4
)
self.player2 = _make_player("Bullpen Bob", pos_1="RP", cost=150, rarity_value=2)
self.pc1 = _make_pitching_card(
starter_rating=9, relief_rating=5, closer_rating=None, balk=1, wild_pitch=2
)
self.pc2 = _make_pitching_card(
starter_rating=3, relief_rating=8, closer_rating=6, balk=3, wild_pitch=5
)
self.card1 = _pitcher_card_data(self.player1, self.pc1)
self.card2 = _pitcher_card_data(self.player2, self.pc2)
def test_embed_field_count_pitchers(self):
"""
Pitcher embed has header row (3 fields) + 7 stat rows × 3 fields = 24.
"""
embed = build_compare_embed(
self.card1, self.card2, "Max Scherzer", "Bullpen Bob"
)
assert len(embed.fields) == 3 + 7 * 3 # 24
def test_description_labels_pitchers(self):
"""Embed description identifies card type as Pitchers."""
embed = build_compare_embed(
self.card1, self.card2, "Max Scherzer", "Bullpen Bob"
)
assert "Pitchers" in embed.description
def test_starter_rating_card1_wins(self):
"""
card1 starter_rating=9 > card2 starter_rating=3 arrow (card1 wins).
Starter Rating is 3rd stat, center field at index 3 + 2*3 + 1 = 10.
"""
embed = build_compare_embed(
self.card1, self.card2, "Max Scherzer", "Bullpen Bob"
)
starter_arrow = embed.fields[10]
assert starter_arrow.value == ""
def test_relief_rating_card2_wins(self):
"""
card2 relief_rating=8 > card1 relief_rating=5 arrow (card2 wins).
Relief Rating is 4th stat, center field at index 3 + 3*3 + 1 = 13.
"""
embed = build_compare_embed(
self.card1, self.card2, "Max Scherzer", "Bullpen Bob"
)
relief_arrow = embed.fields[13]
assert relief_arrow.value == ""
def test_lower_balk_is_better(self):
"""
card1 balk=1 < card2 balk=3 lower is better arrow (card1 wins).
Balk is 6th stat, center field at index 3 + 5*3 + 1 = 19.
"""
embed = build_compare_embed(
self.card1, self.card2, "Max Scherzer", "Bullpen Bob"
)
balk_arrow = embed.fields[19]
assert balk_arrow.value == ""
def test_tied_stat_shows_equals(self):
"""═ when both pitchers have the same starter_rating."""
pc_tied = _make_pitching_card(starter_rating=9)
card_tied = _pitcher_card_data(self.player2, pc_tied)
embed = build_compare_embed(
self.card1, card_tied, "Max Scherzer", "Bullpen Bob"
)
starter_arrow = embed.fields[10]
assert starter_arrow.value == ""
def test_type_mismatch_pitcher_vs_batter_raises(self):
"""CompareMismatchError raised when pitcher compared to batter."""
batter_player = _make_player("Speedy Guy", pos_1="CF")
batter_card_data = _batter_card_data(batter_player, _make_batting_card())
with pytest.raises(CompareMismatchError):
build_compare_embed(
self.card1, batter_card_data, "Max Scherzer", "Speedy Guy"
)
# ---------------------------------------------------------------------------
# build_compare_embed — edge cases
# ---------------------------------------------------------------------------
class TestBuildCompareEmbedEdgeCases:
"""Edge cases: missing data, None stats, same card compared to itself."""
def test_missing_batting_card_graceful(self):
"""
When battingcard is None, stat values display as '' and arrows show .
No exception should be raised.
"""
player = _make_player("Player A", pos_1="1B")
card1 = {"player": player, "battingcard": None, "pitchingcard": None}
card2 = {"player": player, "battingcard": None, "pitchingcard": None}
# Should not raise
embed = build_compare_embed(card1, card2, "Player A", "Player A")
assert isinstance(embed, discord.Embed)
def test_same_card_all_tied(self):
"""Comparing a card against itself should show ═ on every arrow field."""
player = _make_player("Clone", pos_1="CF", cost=300)
bc = _make_batting_card(running=12)
card_data = _batter_card_data(player, bc)
embed = build_compare_embed(card_data, card_data, "Clone", "Clone")
# Arrow fields are at positions 4, 7, 10, 13, 16, 19, 22 (center of each row)
arrow_indices = [4, 7, 10, 13, 16, 19, 22]
for idx in arrow_indices:
assert embed.fields[idx].value == "", (
f"Expected ═ at field index {idx}, "
f"got {embed.fields[idx].value!r} (name={embed.fields[idx].name!r})"
)

View File

@ -100,34 +100,13 @@ class TestRefractorTestSetup:
return MagicMock(spec=commands.Bot) return MagicMock(spec=commands.Bot)
@pytest.fixture @pytest.fixture
def unified_card_response(self): def batting_card_response(self):
"""Factory for a response from GET /v2/cards/{id}. return {
"id": 1234,
The unified cards endpoint returns a card-instance record with "player": {"id": 100, "p_name": "Mike Trout"},
top-level player, team, pack, value, and variant fields. This "variant": 0,
replaces the separate battingcards/pitchingcards template endpoints "image_url": None,
that previously caused ID collisions (see spec }
2026-04-11-refractor-test-unified-cards-lookup-design.md).
The factory lets each test customize pos_1 and the IDs without
duplicating the full response shape.
"""
def _make(pos_1="CF", player_id=100, team_id=31, card_id=1234):
return {
"id": card_id,
"player": {
"player_id": player_id,
"p_name": "Mike Trout",
"pos_1": pos_1,
},
"team": {"id": team_id},
"variant": 0,
"pack": None,
"value": None,
}
return _make
@pytest.fixture @pytest.fixture
def refractor_cards_response(self): def refractor_cards_response(self):
@ -155,212 +134,112 @@ class TestRefractorTestSetup:
], ],
} }
async def test_unified_card_lookup( @pytest.fixture
def opposing_cards_response(self):
"""A valid pitching cards response with the 'cards' key."""
return {
"cards": [
{
"id": 9000,
"player": {"id": 200, "p_name": "Clayton Kershaw"},
"variant": 0,
}
]
}
async def test_batting_card_lookup(
self, self,
mock_interaction, mock_interaction,
mock_bot, mock_bot,
unified_card_response, batting_card_response,
refractor_cards_response, refractor_cards_response,
opposing_cards_response,
): ):
"""The setup phase should make a single db_get call targeting """Command should try the batting card endpoint first.
the unified 'cards' endpoint.
Regression guard for the previous two-step battingcards/pitchingcards Verifies that the first db_get call targets 'battingcards', not
fallback that caused ID collisions (e.g. card 494 resolving to 'pitchingcards', when looking up a card ID.
Cameron Maybin instead of the intended pitcher Grayson Rodriguez).
""" """
from cogs.dev_tools import DevToolsCog from cogs.dev_tools import DevToolsCog
cog = DevToolsCog(mock_bot) cog = DevToolsCog(mock_bot)
card = unified_card_response(pos_1="CF", player_id=100, card_id=1234) with (
patch("cogs.dev_tools.db_get", new_callable=AsyncMock) as mock_get,
patch("cogs.dev_tools.db_post", new_callable=AsyncMock),
):
mock_get.side_effect = [
batting_card_response, # GET battingcards/{id}
refractor_cards_response, # GET refractor/cards
opposing_cards_response, # GET pitchingcards (for opposing player)
]
with patch.object(cog, "_execute_refractor_test", new_callable=AsyncMock):
await cog.refractor_test.callback(cog, mock_interaction, card_id=1234)
first_call = mock_get.call_args_list[0]
assert "battingcards" in str(first_call)
async def test_pitching_card_fallback(
self,
mock_interaction,
mock_bot,
refractor_cards_response,
):
"""If batting card returns None, command should fall back to pitching card.
Ensures the two-step lookup: batting first, then pitching if batting
returns None. The second db_get call must target 'pitchingcards'.
"""
from cogs.dev_tools import DevToolsCog
cog = DevToolsCog(mock_bot)
pitching_card = {
"id": 5678,
"player": {"id": 200, "p_name": "Clayton Kershaw"},
"variant": 0,
"image_url": None,
}
refractor_cards_response["items"][0]["player_id"] = 200
refractor_cards_response["items"][0]["track"]["card_type"] = "sp"
refractor_cards_response["items"][0]["next_threshold"] = 10
opposing_batters = {
"cards": [
{"id": 7000, "player": {"id": 300, "p_name": "Babe Ruth"}, "variant": 0}
]
}
with ( with (
patch("cogs.dev_tools.db_get", new_callable=AsyncMock) as mock_get, patch("cogs.dev_tools.db_get", new_callable=AsyncMock) as mock_get,
patch("cogs.dev_tools.db_post", new_callable=AsyncMock), patch("cogs.dev_tools.db_post", new_callable=AsyncMock),
): ):
mock_get.side_effect = [card, refractor_cards_response] mock_get.side_effect = [
None, # batting card not found
pitching_card, # pitching card found
refractor_cards_response, # refractor/cards
opposing_batters, # battingcards for opposing player
]
with patch.object(cog, "_execute_refractor_test", new_callable=AsyncMock): with patch.object(cog, "_execute_refractor_test", new_callable=AsyncMock):
await cog.refractor_test.callback(cog, mock_interaction, 1234) await cog.refractor_test.callback(cog, mock_interaction, card_id=5678)
second_call = mock_get.call_args_list[1]
# First call: the unified cards endpoint assert "pitchingcards" in str(second_call)
first_call = mock_get.call_args_list[0]
assert first_call.args[0] == "cards"
assert first_call.kwargs.get("object_id") == 1234
# No secondary template-table fallback
endpoints_called = [c.args[0] for c in mock_get.call_args_list]
assert "battingcards" not in endpoints_called
assert "pitchingcards" not in endpoints_called
async def test_card_not_found_reports_error(self, mock_interaction, mock_bot): async def test_card_not_found_reports_error(self, mock_interaction, mock_bot):
"""If cards/{id} returns None, report 'not found' and never call """If neither batting nor pitching card exists, report an error and return.
_execute_refractor_test. The single unified endpoint means only
one db_get is made before the error path. The command should call edit_original_response with a message containing
'not found' and must NOT call _execute_refractor_test.
""" """
from cogs.dev_tools import DevToolsCog from cogs.dev_tools import DevToolsCog
cog = DevToolsCog(mock_bot) cog = DevToolsCog(mock_bot)
with patch( with patch("cogs.dev_tools.db_get", new_callable=AsyncMock, return_value=None):
"cogs.dev_tools.db_get", new_callable=AsyncMock, return_value=None
) as mock_get:
with patch.object( with patch.object(
cog, "_execute_refractor_test", new_callable=AsyncMock cog, "_execute_refractor_test", new_callable=AsyncMock
) as mock_exec: ) as mock_exec:
await cog.refractor_test.callback(cog, mock_interaction, 9999) await cog.refractor_test.callback(cog, mock_interaction, card_id=9999)
mock_exec.assert_not_called() mock_exec.assert_not_called()
# Exactly one db_get call — the unified lookup, no template fallback
assert mock_get.call_count == 1
call_kwargs = mock_interaction.edit_original_response.call_args[1] call_kwargs = mock_interaction.edit_original_response.call_args[1]
assert "not found" in call_kwargs["content"].lower() assert "not found" in call_kwargs["content"].lower()
async def test_pos_sp_derives_sp_type(
self,
mock_interaction,
mock_bot,
unified_card_response,
refractor_cards_response,
):
"""pos_1='SP' should derive card_type='sp', card_type_key='pitching'
and pass those into _execute_refractor_test.
"""
from cogs.dev_tools import DevToolsCog
cog = DevToolsCog(mock_bot)
card = unified_card_response(pos_1="SP", player_id=200)
# Make sure the refractor/cards lookup finds no matching entry,
# so the command falls through to the pos_1-derived defaults.
refractor_cards_response["items"] = []
with (
patch("cogs.dev_tools.db_get", new_callable=AsyncMock) as mock_get,
patch("cogs.dev_tools.db_post", new_callable=AsyncMock),
patch.object(
cog, "_execute_refractor_test", new_callable=AsyncMock
) as mock_exec,
):
mock_get.side_effect = [card, refractor_cards_response]
await cog.refractor_test.callback(cog, mock_interaction, 1234)
kwargs = mock_exec.call_args.kwargs
assert kwargs["card_type"] == "sp"
assert kwargs["card_type_key"] == "pitching"
async def test_pos_rp_derives_rp_type(
self,
mock_interaction,
mock_bot,
unified_card_response,
refractor_cards_response,
):
"""pos_1='RP' should derive card_type='rp', card_type_key='pitching'."""
from cogs.dev_tools import DevToolsCog
cog = DevToolsCog(mock_bot)
card = unified_card_response(pos_1="RP", player_id=201)
refractor_cards_response["items"] = []
with (
patch("cogs.dev_tools.db_get", new_callable=AsyncMock) as mock_get,
patch("cogs.dev_tools.db_post", new_callable=AsyncMock),
patch.object(
cog, "_execute_refractor_test", new_callable=AsyncMock
) as mock_exec,
):
mock_get.side_effect = [card, refractor_cards_response]
await cog.refractor_test.callback(cog, mock_interaction, 1234)
kwargs = mock_exec.call_args.kwargs
assert kwargs["card_type"] == "rp"
assert kwargs["card_type_key"] == "pitching"
async def test_pos_cp_derives_rp_type(
self,
mock_interaction,
mock_bot,
unified_card_response,
refractor_cards_response,
):
"""pos_1='CP' (closer) should also map to card_type='rp'."""
from cogs.dev_tools import DevToolsCog
cog = DevToolsCog(mock_bot)
card = unified_card_response(pos_1="CP", player_id=202)
refractor_cards_response["items"] = []
with (
patch("cogs.dev_tools.db_get", new_callable=AsyncMock) as mock_get,
patch("cogs.dev_tools.db_post", new_callable=AsyncMock),
patch.object(
cog, "_execute_refractor_test", new_callable=AsyncMock
) as mock_exec,
):
mock_get.side_effect = [card, refractor_cards_response]
await cog.refractor_test.callback(cog, mock_interaction, 1234)
kwargs = mock_exec.call_args.kwargs
assert kwargs["card_type"] == "rp"
assert kwargs["card_type_key"] == "pitching"
async def test_pos_batter_derives_batter_type(
self,
mock_interaction,
mock_bot,
unified_card_response,
refractor_cards_response,
):
"""pos_1='CF' (or any non-pitcher position) should derive
card_type='batter', card_type_key='batting'.
"""
from cogs.dev_tools import DevToolsCog
cog = DevToolsCog(mock_bot)
card = unified_card_response(pos_1="CF", player_id=203)
refractor_cards_response["items"] = []
with (
patch("cogs.dev_tools.db_get", new_callable=AsyncMock) as mock_get,
patch("cogs.dev_tools.db_post", new_callable=AsyncMock),
patch.object(
cog, "_execute_refractor_test", new_callable=AsyncMock
) as mock_exec,
):
mock_get.side_effect = [card, refractor_cards_response]
await cog.refractor_test.callback(cog, mock_interaction, 1234)
kwargs = mock_exec.call_args.kwargs
assert kwargs["card_type"] == "batter"
assert kwargs["card_type_key"] == "batting"
async def test_card_without_team_reports_error(
self,
mock_interaction,
mock_bot,
unified_card_response,
):
"""If the unified card response has team=None, the command should
report an error and not proceed to execute the refractor chain.
The card instance's owning team is now the authoritative team
context for the test (see spec option A: card's team is
authoritative, no get_team_by_owner fallback).
"""
from cogs.dev_tools import DevToolsCog
cog = DevToolsCog(mock_bot)
card = unified_card_response(pos_1="CF", player_id=100)
card["team"] = None
with patch("cogs.dev_tools.db_get", new_callable=AsyncMock, return_value=card):
with patch.object(
cog, "_execute_refractor_test", new_callable=AsyncMock
) as mock_exec:
await cog.refractor_test.callback(cog, mock_interaction, 1234)
mock_exec.assert_not_called()
call_kwargs = mock_interaction.edit_original_response.call_args[1]
assert "no owning team" in call_kwargs["content"].lower()
class TestRefractorTestExecute: class TestRefractorTestExecute:
"""Test the execution phase: API calls, step-by-step reporting, """Test the execution phase: API calls, step-by-step reporting,
@ -388,13 +267,7 @@ class TestRefractorTestExecute:
async def test_successful_batter_flow(self, mock_interaction, mock_bot, base_embed): async def test_successful_batter_flow(self, mock_interaction, mock_bot, base_embed):
"""Full happy path: game created, plays inserted, stats updated, """Full happy path: game created, plays inserted, stats updated,
tier-up detected, card rendered (200 OK from aiohttp render endpoint). tier-up detected, card rendered."""
Previously the render step was stubbed via a dead db_get mock; the
actual render call uses aiohttp.ClientSession directly, so the mock
was never consumed and the test hit the non-fatal except branch
( Card render failed) instead of the happy path.
"""
from cogs.dev_tools import DevToolsCog from cogs.dev_tools import DevToolsCog
cog = DevToolsCog(mock_bot) cog = DevToolsCog(mock_bot)
@ -419,23 +292,9 @@ class TestRefractorTestExecute:
], ],
} }
mock_render_resp = MagicMock()
mock_render_resp.status = 200
# sess.get(url) must return an async context manager (not a coroutine),
# so get_cm is a MagicMock with explicit __aenter__/__aexit__.
get_cm = MagicMock()
get_cm.__aenter__ = AsyncMock(return_value=mock_render_resp)
get_cm.__aexit__ = AsyncMock(return_value=None)
mock_sess = MagicMock()
mock_sess.__aenter__ = AsyncMock(return_value=mock_sess)
mock_sess.__aexit__ = AsyncMock(return_value=None)
mock_sess.get.return_value = get_cm
with ( with (
patch("cogs.dev_tools.db_post", new_callable=AsyncMock) as mock_post, patch("cogs.dev_tools.db_post", new_callable=AsyncMock) as mock_post,
patch("cogs.dev_tools.aiohttp.ClientSession", return_value=mock_sess), patch("cogs.dev_tools.db_get", new_callable=AsyncMock) as mock_get,
): ):
mock_post.side_effect = [ mock_post.side_effect = [
game_response, # POST games game_response, # POST games
@ -444,6 +303,7 @@ class TestRefractorTestExecute:
stats_response, # POST season-stats/update-game stats_response, # POST season-stats/update-game
eval_response, # POST refractor/evaluate-game eval_response, # POST refractor/evaluate-game
] ]
mock_get.return_value = {"image_url": "https://s3.example.com/card.png"}
await cog._execute_refractor_test( await cog._execute_refractor_test(
interaction=mock_interaction, interaction=mock_interaction,
@ -462,7 +322,6 @@ class TestRefractorTestExecute:
result_text = "\n".join(f.value for f in final_embed.fields if f.value) result_text = "\n".join(f.value for f in final_embed.fields if f.value)
assert "" in result_text assert "" in result_text
assert "game" in result_text.lower() assert "game" in result_text.lower()
assert "✅ Card rendered + S3 upload triggered" in result_text
async def test_stops_on_game_creation_failure( async def test_stops_on_game_creation_failure(
self, mock_interaction, mock_bot, base_embed self, mock_interaction, mock_bot, base_embed

View File

@ -0,0 +1,315 @@
"""
Tests for the gauntlet completion recap embed (Roadmap 2.4a).
These tests verify:
1. build_gauntlet_recap_embed produces an embed with the required fields
(title, champion, record, prize distribution) for a completed run.
2. The embed title always starts with "Gauntlet Complete:".
3. The champion field contains the team name and, when gmid is present,
a Discord user mention (<@gmid>).
4. The prize distribution marks earned rewards with and unearned with
or depending on whether losses were exceeded.
5. post_gauntlet_recap is a no-op when channel is None (graceful fallback).
6. post_gauntlet_recap calls channel.send exactly once on success.
7. post_gauntlet_recap does not raise when channel.send raises gauntlet
completion must never be interrupted by a recap failure.
The builder is a pure synchronous function so tests do not require an async
event loop; only the async sender tests use pytest-asyncio.
"""
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
import discord
from gauntlets import build_gauntlet_recap_embed, post_gauntlet_recap
# ---------------------------------------------------------------------------
# Test data helpers
# ---------------------------------------------------------------------------
def make_run(wins: int = 10, losses: int = 1, run_id: int = 42) -> dict:
"""Return a minimal gauntletruns API dict for a completed run."""
return {
"id": run_id,
"wins": wins,
"losses": losses,
"gauntlet": {"id": 9, "name": "2005 Live"},
"team": {"lname": "Gauntlet-NCB", "logo": None},
}
def make_event(event_id: int = 9, name: str = "2005 Live") -> dict:
"""Return a minimal events API dict."""
return {
"id": event_id,
"name": name,
"url": "https://example.com/gauntlet.png",
"short_desc": "Go 10-0!",
}
def make_main_team(gmid: int = 123456789, lname: str = "Normal CornBelters") -> dict:
"""Return a minimal teams API dict for the player's real team."""
return {
"id": 31,
"lname": lname,
"gmid": gmid,
"logo": "https://example.com/logo.png",
}
def make_rewards() -> list[dict]:
"""Return a representative gauntletrewards list for a gauntlet."""
return [
{
"win_num": 3,
"loss_max": 2,
"reward": {"money": 500, "player": None, "pack_type": None},
},
{
"win_num": 7,
"loss_max": 2,
"reward": {
"money": None,
"player": None,
"pack_type": {"id": 1, "name": "Standard"},
},
},
{
"win_num": 10,
"loss_max": 0,
"reward": {
"money": None,
"player": {
"player_id": 99,
"description": "Babe Ruth HoF",
},
"pack_type": None,
},
},
]
# ---------------------------------------------------------------------------
# Unit: build_gauntlet_recap_embed
# ---------------------------------------------------------------------------
class TestBuildGauntletRecapEmbed:
"""Verify the embed produced by the synchronous builder function."""
def test_title_starts_with_gauntlet_complete(self):
"""Title must start with 'Gauntlet Complete:' followed by the event name.
This is the canonical format expected by the PO spec and makes the
embed immediately recognisable in the channel feed.
"""
embed = build_gauntlet_recap_embed(
make_run(), make_event(), make_main_team(), make_rewards()
)
assert embed.title.startswith("Gauntlet Complete:")
assert "2005 Live" in embed.title
def test_embed_colour_is_gold(self):
"""Embed colour must be the gold/champion accent (0xFFD700).
Gold is the PO-specified accent for champion-level events.
"""
embed = build_gauntlet_recap_embed(
make_run(), make_event(), make_main_team(), make_rewards()
)
assert embed.color.value == 0xFFD700
def test_champion_field_contains_team_name(self):
"""The Champion field must display the team's long name.
Players identify with their team name, not the gauntlet draft copy.
"""
embed = build_gauntlet_recap_embed(
make_run(),
make_event(),
make_main_team(lname="Normal CornBelters"),
make_rewards(),
)
champion_field = next((f for f in embed.fields if f.name == "Champion"), None)
assert champion_field is not None, "Expected a 'Champion' embed field"
assert "Normal CornBelters" in champion_field.value
def test_champion_field_contains_user_mention_when_gmid_present(self):
"""When gmid is set, the Champion field must include a Discord user mention.
The mention (<@gmid>) creates social validation the winner is pinged
in the channel where they completed the gauntlet.
"""
embed = build_gauntlet_recap_embed(
make_run(), make_event(), make_main_team(gmid=987654321), make_rewards()
)
champion_field = next(f for f in embed.fields if f.name == "Champion")
assert "<@987654321>" in champion_field.value
def test_champion_field_omits_mention_when_gmid_is_none(self):
"""When gmid is None the embed must not include any mention syntax.
Some legacy records or AI teams may not have a Discord user ID.
The embed must still be valid without one.
"""
team = make_main_team()
team["gmid"] = None
embed = build_gauntlet_recap_embed(
make_run(), make_event(), team, make_rewards()
)
champion_field = next(f for f in embed.fields if f.name == "Champion")
assert "<@" not in champion_field.value
def test_final_record_field_present(self):
"""Final Record field must show wins-losses for the completed run."""
embed = build_gauntlet_recap_embed(
make_run(wins=10, losses=1), make_event(), make_main_team(), make_rewards()
)
record_field = next((f for f in embed.fields if f.name == "Final Record"), None)
assert record_field is not None, "Expected a 'Final Record' field"
assert "10" in record_field.value
assert "1" in record_field.value
def test_prize_distribution_field_present(self):
"""Prize Distribution field must be present when rewards are provided."""
embed = build_gauntlet_recap_embed(
make_run(), make_event(), make_main_team(), make_rewards()
)
prize_field = next(
(f for f in embed.fields if f.name == "Prize Distribution"), None
)
assert prize_field is not None, "Expected a 'Prize Distribution' field"
def test_earned_rewards_marked_with_checkmark(self):
"""Rewards that were earned (wins >= threshold and losses within limit)
must be marked with .
A 10-1 run earns the 3-win and 7-win milestones but not the 10-0 bonus.
"""
embed = build_gauntlet_recap_embed(
make_run(wins=10, losses=1), make_event(), make_main_team(), make_rewards()
)
prize_field = next(f for f in embed.fields if f.name == "Prize Distribution")
# 3-win (loss_max=2, losses=1) → earned
assert "" in prize_field.value
def test_unearned_perfect_bonus_marked_correctly(self):
"""The 10-0 bonus reward must NOT be marked earned when losses > 0."""
embed = build_gauntlet_recap_embed(
make_run(wins=10, losses=1), make_event(), make_main_team(), make_rewards()
)
prize_field = next(f for f in embed.fields if f.name == "Prize Distribution")
# The 10-0 bonus line must be marked ❌ — ineligible, not pending (⬜)
lines = prize_field.value.split("\n")
bonus_line = next((line for line in lines if "10-0" in line), None)
assert bonus_line is not None, "Expected a '10-0' line in prizes"
assert "" in bonus_line
def test_empty_rewards_list_omits_prize_field(self):
"""When rewards is an empty list the Prize Distribution field must be omitted.
Some event types may not have configured rewards; the embed must
still be valid and informative without a prize table.
"""
embed = build_gauntlet_recap_embed(
make_run(), make_event(), make_main_team(), []
)
prize_field = next(
(f for f in embed.fields if f.name == "Prize Distribution"), None
)
assert prize_field is None
# ---------------------------------------------------------------------------
# Async: post_gauntlet_recap
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_post_gauntlet_recap_sends_embed_on_success():
"""When all inputs are valid post_gauntlet_recap sends exactly one embed.
This confirms the function wires build_gauntlet_recap_embed channel.send
correctly and doesn't double-post.
"""
channel = AsyncMock()
all_rewards_response = {"rewards": make_rewards()}
with patch(
"gauntlets.db_get", new_callable=AsyncMock, return_value=all_rewards_response
):
await post_gauntlet_recap(make_run(), make_event(), make_main_team(), channel)
channel.send.assert_called_once()
# Verify an embed was passed (not just plain text)
kwargs = channel.send.call_args.kwargs
assert "embed" in kwargs
assert isinstance(kwargs["embed"], discord.Embed)
@pytest.mark.asyncio
async def test_post_gauntlet_recap_noop_when_channel_is_none():
"""When channel is None post_gauntlet_recap must return without raising.
The gauntlet channel may be unavailable (deleted, bot lost permissions,
or not set in the record). The completion flow must never fail due to
a missing recap channel.
"""
# No channel.send to assert on — just ensure no exception is raised
with patch(
"gauntlets.db_get", new_callable=AsyncMock, return_value={"rewards": []}
):
try:
await post_gauntlet_recap(make_run(), make_event(), make_main_team(), None)
except Exception as exc:
pytest.fail(f"post_gauntlet_recap raised with None channel: {exc}")
@pytest.mark.asyncio
async def test_post_gauntlet_recap_nonfatal_when_channel_send_raises():
"""A channel.send failure must not propagate out of post_gauntlet_recap.
The gauntlet run is already complete when the recap fires; a Discord API
error (rate limit, permissions revoked) must not corrupt the game state.
"""
channel = AsyncMock()
channel.send.side_effect = Exception("Discord API error")
all_rewards_response = {"rewards": make_rewards()}
with patch(
"gauntlets.db_get", new_callable=AsyncMock, return_value=all_rewards_response
):
try:
await post_gauntlet_recap(
make_run(), make_event(), make_main_team(), channel
)
except Exception as exc:
pytest.fail(f"post_gauntlet_recap raised when channel.send failed: {exc}")
@pytest.mark.asyncio
async def test_post_gauntlet_recap_nonfatal_when_db_get_raises():
"""A db_get failure inside post_gauntlet_recap must not propagate.
If the rewards endpoint is unavailable the recap silently skips rather
than crashing the completion flow.
"""
channel = AsyncMock()
with patch(
"gauntlets.db_get", new_callable=AsyncMock, side_effect=Exception("API down")
):
try:
await post_gauntlet_recap(
make_run(), make_event(), make_main_team(), channel
)
except Exception as exc:
pytest.fail(f"post_gauntlet_recap raised when db_get failed: {exc}")
# channel.send should NOT have been called since the error happened before it
channel.send.assert_not_called()

View File

@ -12,15 +12,10 @@ class TestTriggerVariantRenders:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_calls_render_url_for_each_tier_up(self): async def test_calls_render_url_for_each_tier_up(self):
"""Each tier-up with variant_created triggers a card render GET request. """Each tier-up with variant_created triggers a card render GET request."""
card_type field ('sp', 'rp', 'batter') determines the render URL segment
('pitchingcard' or 'battingcard'). This prevents silent wrong-type renders
when track_name strings vary ('Pitcher', 'Starting Pitcher', 'SP', etc.).
"""
tier_ups = [ tier_ups = [
{"player_id": 100, "variant_created": 7, "card_type": "batter"}, {"player_id": 100, "variant_created": 7, "track_name": "Batter"},
{"player_id": 200, "variant_created": 3, "card_type": "sp"}, {"player_id": 200, "variant_created": 3, "track_name": "Pitcher"},
] ]
with patch( with patch(
@ -31,20 +26,14 @@ class TestTriggerVariantRenders:
assert mock_get.call_count == 2 assert mock_get.call_count == 2
call_args_list = [call.args[0] for call in mock_get.call_args_list] call_args_list = [call.args[0] for call in mock_get.call_args_list]
assert any( assert any("100" in url and "7" in url for url in call_args_list)
"100" in url and "battingcard" in url and "7" in url assert any("200" in url and "3" in url for url in call_args_list)
for url in call_args_list
)
assert any(
"200" in url and "pitchingcard" in url and "3" in url
for url in call_args_list
)
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_skips_tier_ups_without_variant(self): async def test_skips_tier_ups_without_variant(self):
"""Tier-ups without variant_created are skipped.""" """Tier-ups without variant_created are skipped."""
tier_ups = [ tier_ups = [
{"player_id": 100, "card_type": "batter"}, {"player_id": 100, "track_name": "Batter"},
] ]
with patch( with patch(
@ -57,7 +46,7 @@ class TestTriggerVariantRenders:
async def test_api_failure_does_not_raise(self): async def test_api_failure_does_not_raise(self):
"""Render trigger failures are swallowed — fire-and-forget.""" """Render trigger failures are swallowed — fire-and-forget."""
tier_ups = [ tier_ups = [
{"player_id": 100, "variant_created": 7, "card_type": "batter"}, {"player_id": 100, "variant_created": 7, "track_name": "Batter"},
] ]
with patch( with patch(