From 64b42136eea01ce8d70be305720bea6848b8f3dc Mon Sep 17 00:00:00 2001 From: Mitch Rumler Date: Fri, 31 Jul 2026 14:13:32 +1000 Subject: [PATCH 1/2] Add anonymous multi-user vote admin command. Allow Admin/CM/Community Team to start timed anonymous polls so staff can approve or reject nominees without revealing voter identity or live tallies until close. Co-authored-by: Cursor --- .../e1a2b3c4d5e6_add_anonymous_vote_tables.py | 76 ++++ src/bot.py | 6 + src/cmds/core/admin.py | 132 ++++++ src/core/config.py | 13 + src/database/models/__init__.py | 1 + src/database/models/anonymous_vote.py | 73 ++++ src/views/anonymous_vote.py | 380 ++++++++++++++++++ tests/src/core/test_config.py | 4 + 8 files changed, 685 insertions(+) create mode 100644 alembic/versions/e1a2b3c4d5e6_add_anonymous_vote_tables.py create mode 100644 src/database/models/anonymous_vote.py create mode 100644 src/views/anonymous_vote.py diff --git a/alembic/versions/e1a2b3c4d5e6_add_anonymous_vote_tables.py b/alembic/versions/e1a2b3c4d5e6_add_anonymous_vote_tables.py new file mode 100644 index 0000000..f23d6e6 --- /dev/null +++ b/alembic/versions/e1a2b3c4d5e6_add_anonymous_vote_tables.py @@ -0,0 +1,76 @@ +"""Add anonymous vote tables + +Revision ID: e1a2b3c4d5e6 +Revises: d4f8c2a6e1b7 +Create Date: 2026-07-31 13:53:00.000000 + +""" +import sqlalchemy as sa +from sqlalchemy.dialects import mysql + +from alembic import op + +# revision identifiers, used by Alembic. +revision = "e1a2b3c4d5e6" +down_revision = "d4f8c2a6e1b7" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "anonymous_vote_session", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("guild_id", mysql.BIGINT(display_width=18), nullable=False), + sa.Column("channel_id", mysql.BIGINT(display_width=18), nullable=False), + sa.Column("message_id", mysql.BIGINT(display_width=18), nullable=True), + sa.Column("topic", mysql.TEXT(), nullable=True), + sa.Column("created_by_id", mysql.BIGINT(display_width=18), nullable=False), + sa.Column("closes_at", mysql.TIMESTAMP(), nullable=False), + sa.Column("closed", sa.Boolean(), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + op.create_table( + "anonymous_vote_candidate", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("session_id", sa.Integer(), nullable=False), + sa.Column("user_id", mysql.BIGINT(display_width=18), nullable=False), + sa.Column("display_name", mysql.TEXT(), nullable=False), + sa.ForeignKeyConstraint( + ["session_id"], + ["anonymous_vote_session.id"], + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id"), + ) + op.create_table( + "anonymous_vote_ballot", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("session_id", sa.Integer(), nullable=False), + sa.Column("candidate_id", sa.Integer(), nullable=False), + sa.Column("voter_id", mysql.BIGINT(display_width=18), nullable=False), + sa.Column("choice", sa.String(length=16), nullable=False), + sa.ForeignKeyConstraint( + ["candidate_id"], + ["anonymous_vote_candidate.id"], + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["session_id"], + ["anonymous_vote_session.id"], + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "session_id", + "candidate_id", + "voter_id", + name="uq_anonymous_vote_ballot_session_candidate_voter", + ), + ) + + +def downgrade() -> None: + op.drop_table("anonymous_vote_ballot") + op.drop_table("anonymous_vote_candidate") + op.drop_table("anonymous_vote_session") diff --git a/src/bot.py b/src/bot.py index 6f004f5..9d378c8 100644 --- a/src/bot.py +++ b/src/bot.py @@ -95,6 +95,7 @@ async def on_ready(self) -> None: async def _register_persistent_views(self) -> None: """Re-register persistent UI views so buttons survive bot restarts.""" + from src.views.anonymous_vote import register_anonymous_vote_views from src.views.bandecisionview import register_ban_views try: @@ -102,6 +103,11 @@ async def _register_persistent_views(self) -> None: except Exception: logger.exception("Failed to register persistent ban decision views") + try: + await register_anonymous_vote_views(self) + except Exception: + logger.exception("Failed to register persistent anonymous vote views") + async def on_application_command(self, ctx: ApplicationContext) -> None: """A global handler cog.""" logger.debug(f"Command '{ctx.command}' received.") diff --git a/src/cmds/core/admin.py b/src/cmds/core/admin.py index 791482d..f378209 100644 --- a/src/cmds/core/admin.py +++ b/src/cmds/core/admin.py @@ -1,19 +1,48 @@ """Admin command group for bot administration commands.""" import logging +import re +from datetime import datetime import discord from discord import ApplicationContext, Interaction, Option, WebhookMessage from discord.ext import commands from discord.ext.commands import has_any_role +from sqlalchemy import select +from sqlalchemy.orm import selectinload from src.bot import Bot from src.core import settings +from src.database.models import AnonymousVoteCandidate, AnonymousVoteSession from src.database.models.dynamic_role import RoleCategory +from src.database.session import AsyncSessionLocal +from src.helpers.duration import validate_duration +from src.views.anonymous_vote import ( + AnonymousVoteView, + build_poll_embed, + schedule_vote_close, +) logger = logging.getLogger(__name__) CATEGORY_CHOICES = [c.value for c in RoleCategory] +_MEMBER_TOKEN_RE = re.compile(r"<@!?(\d+)>|^(\d+)$") + + +def _parse_member_ids(raw: str) -> list[int]: + """Parse space/comma-separated mentions or snowflake IDs into unique IDs.""" + ids: list[int] = [] + for part in re.split(r"[\s,]+", raw.strip()): + if not part: + continue + match = _MEMBER_TOKEN_RE.fullmatch(part) + if not match: + raise ValueError( + f"Could not parse `{part}`. Use mentions or numeric user IDs." + ) + ids.append(int(match.group(1) or match.group(2))) + # Preserve order, drop duplicates + return list(dict.fromkeys(ids)) class AdminCog(commands.Cog): @@ -149,6 +178,109 @@ async def reload(self, ctx: ApplicationContext) -> Interaction | WebhookMessage: await self.bot.role_manager.reload() return await ctx.respond("Dynamic roles reloaded from database.", ephemeral=True) + @admin.command( + name="vote", + description="Start an anonymous timed vote on multiple members.", + ) + @has_any_role(*settings.role_groups.get("VOTE_STARTERS")) + async def vote( + self, + ctx: ApplicationContext, + members: Option( + str, + "Nominees as mentions or user IDs (space/comma separated, max 25)", + ), + duration: Option(str, "How long the vote stays open (e.g. 12h, 1d, 30m)"), + topic: Option(str, "Optional topic shown on the poll", required=False), + ) -> Interaction | WebhookMessage: + """Start an anonymous vote; tallies reveal automatically when duration ends.""" + closes_at_ts, error = validate_duration(duration) + if error: + return await ctx.respond(error, ephemeral=True) + + try: + member_ids = _parse_member_ids(members) + except ValueError as exc: + return await ctx.respond(str(exc), ephemeral=True) + + if not member_ids: + return await ctx.respond("Provide at least one nominee.", ephemeral=True) + if len(member_ids) > 25: + return await ctx.respond( + "Discord select menus support at most 25 nominees.", + ephemeral=True, + ) + + resolved: list[tuple[int, str]] = [] + missing: list[str] = [] + for user_id in member_ids: + member = ctx.guild.get_member(user_id) + if member is None: + try: + member = await ctx.guild.fetch_member(user_id) + except discord.HTTPException: + missing.append(str(user_id)) + continue + resolved.append((member.id, member.display_name)) + + if missing: + return await ctx.respond( + "Could not find member(s) in this server: " + ", ".join(f"`{m}`" for m in missing), + ephemeral=True, + ) + + closes_at = datetime.fromtimestamp(closes_at_ts) + await ctx.defer(ephemeral=True) + + async with AsyncSessionLocal() as session: + vote_session = AnonymousVoteSession( + guild_id=ctx.guild.id, + channel_id=ctx.channel.id, + message_id=None, + topic=topic, + created_by_id=ctx.author.id, + closes_at=closes_at, + closed=False, + ) + session.add(vote_session) + await session.flush() + + for user_id, display_name in resolved: + session.add( + AnonymousVoteCandidate( + session_id=vote_session.id, + user_id=user_id, + display_name=display_name, + ) + ) + await session.commit() + + loaded = await session.scalar( + select(AnonymousVoteSession) + .where(AnonymousVoteSession.id == vote_session.id) + .options(selectinload(AnonymousVoteSession.candidates)) + ) + session_id = loaded.id + candidates = list(loaded.candidates) + poll_embed = build_poll_embed(loaded, candidates) + + view = AnonymousVoteView(session_id, self.bot, candidates) + self.bot.add_view(view) + message = await ctx.channel.send(embed=poll_embed, view=view) + + async with AsyncSessionLocal() as session: + vote_session = await session.get(AnonymousVoteSession, session_id) + if vote_session: + vote_session.message_id = message.id + await session.commit() + + schedule_vote_close(self.bot, session_id, closes_at) + return await ctx.followup.send( + f"Anonymous vote #{session_id} started in {ctx.channel.mention}. " + f"Closes {discord.utils.format_dt(closes_at, style='R')}.", + ephemeral=True, + ) + def setup(bot: Bot) -> None: """Load the AdminCog.""" diff --git a/src/core/config.py b/src/core/config.py index b468ebc..1685f1b 100644 --- a/src/core/config.py +++ b/src/core/config.py @@ -259,6 +259,19 @@ def role_groups(self) -> dict[str, list[int]]: ], "ALL_HTB_STAFF": [self.roles.HTB_STAFF], "ALL_HTB_SUPPORT": [self.roles.HTB_SUPPORT], + "VOTE_STARTERS": [ + self.roles.ADMINISTRATOR, + self.roles.COMMUNITY_MANAGER, + self.roles.COMMUNITY_TEAM, + ], + "VOTE_CASTERS": [ + self.roles.ADMINISTRATOR, + self.roles.COMMUNITY_MANAGER, + self.roles.COMMUNITY_TEAM, + self.roles.SR_MODERATOR, + self.roles.MODERATOR, + self.roles.JR_MODERATOR, + ], } diff --git a/src/database/models/__init__.py b/src/database/models/__init__.py index 15ab602..0515ee2 100644 --- a/src/database/models/__init__.py +++ b/src/database/models/__init__.py @@ -1,6 +1,7 @@ # flake8: noqa from src.database.base_class import Base # noqa +from .anonymous_vote import AnonymousVoteBallot, AnonymousVoteCandidate, AnonymousVoteSession from .ban import Ban from .ctf import Ctf from .dynamic_role import DynamicRole, RoleCategory diff --git a/src/database/models/anonymous_vote.py b/src/database/models/anonymous_vote.py new file mode 100644 index 0000000..d11bf9c --- /dev/null +++ b/src/database/models/anonymous_vote.py @@ -0,0 +1,73 @@ +# flake8: noqa: D101 +from datetime import datetime + +from sqlalchemy import Boolean, ForeignKey, Integer, String, UniqueConstraint +from sqlalchemy.dialects.mysql import BIGINT, TEXT, TIMESTAMP +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from . import Base + + +class AnonymousVoteSession(Base): + """Timed anonymous vote session over one or more nominees.""" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + guild_id: Mapped[int] = mapped_column(BIGINT(18), nullable=False) + channel_id: Mapped[int] = mapped_column(BIGINT(18), nullable=False) + message_id: Mapped[int | None] = mapped_column(BIGINT(18), nullable=True) + topic: Mapped[str | None] = mapped_column(TEXT, nullable=True) + created_by_id: Mapped[int] = mapped_column(BIGINT(18), nullable=False) + closes_at: Mapped[datetime] = mapped_column(TIMESTAMP, nullable=False) + closed: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + + candidates: Mapped[list["AnonymousVoteCandidate"]] = relationship( + back_populates="session", + cascade="all, delete-orphan", + ) + ballots: Mapped[list["AnonymousVoteBallot"]] = relationship( + back_populates="session", + cascade="all, delete-orphan", + ) + + +class AnonymousVoteCandidate(Base): + """A nominee in an anonymous vote session.""" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + session_id: Mapped[int] = mapped_column( + Integer, ForeignKey("anonymous_vote_session.id", ondelete="CASCADE"), nullable=False + ) + user_id: Mapped[int] = mapped_column(BIGINT(18), nullable=False) + display_name: Mapped[str] = mapped_column(TEXT, nullable=False) + + session: Mapped["AnonymousVoteSession"] = relationship(back_populates="candidates") + ballots: Mapped[list["AnonymousVoteBallot"]] = relationship( + back_populates="candidate", + cascade="all, delete-orphan", + ) + + +class AnonymousVoteBallot(Base): + """A single voter's choice for one nominee. voter_id is never shown in Discord.""" + + __table_args__ = ( + UniqueConstraint( + "session_id", + "candidate_id", + "voter_id", + name="uq_anonymous_vote_ballot_session_candidate_voter", + ), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + session_id: Mapped[int] = mapped_column( + Integer, ForeignKey("anonymous_vote_session.id", ondelete="CASCADE"), nullable=False + ) + candidate_id: Mapped[int] = mapped_column( + Integer, ForeignKey("anonymous_vote_candidate.id", ondelete="CASCADE"), nullable=False + ) + voter_id: Mapped[int] = mapped_column(BIGINT(18), nullable=False) + choice: Mapped[str] = mapped_column(String(16), nullable=False) + + session: Mapped["AnonymousVoteSession"] = relationship(back_populates="ballots") + candidate: Mapped["AnonymousVoteCandidate"] = relationship(back_populates="ballots") diff --git a/src/views/anonymous_vote.py b/src/views/anonymous_vote.py new file mode 100644 index 0000000..0d1e585 --- /dev/null +++ b/src/views/anonymous_vote.py @@ -0,0 +1,380 @@ +"""Persistent UI and helpers for anonymous multi-user votes.""" + +from __future__ import annotations + +import logging +from collections import defaultdict +from datetime import datetime + +import discord +from discord import Interaction, SelectOption +from discord.ui import Button, Select, View +from sqlalchemy import select +from sqlalchemy.orm import selectinload + +from src.bot import Bot +from src.core import settings +from src.database.models import AnonymousVoteBallot, AnonymousVoteCandidate, AnonymousVoteSession +from src.database.session import AsyncSessionLocal +from src.helpers.schedule import schedule + +logger = logging.getLogger(__name__) + +CHOICE_APPROVE = "approve" +CHOICE_REJECT = "reject" +# Neutral marker shown per ballot cast (does not reveal approve vs reject). +VOTE_ACTIVITY_BOX = "⬜" +# Keep embed descriptions safely under Discord's 4096-char limit. +MAX_ACTIVITY_BOXES_PER_NOMINEE = 40 + +_pending_selection: dict[tuple[int, int], int] = {} + + +def _member_can_vote(member: discord.Member) -> bool: + voter_roles = set(settings.role_groups.get("VOTE_CASTERS", [])) + return bool(voter_roles.intersection({role.id for role in member.roles})) + + +def _ballot_counts_by_candidate(ballots: list[AnonymousVoteBallot]) -> dict[int, int]: + counts: dict[int, int] = defaultdict(int) + for ballot in ballots: + counts[ballot.candidate_id] += 1 + return counts + + +def _format_nominee_line(candidate: AnonymousVoteCandidate, vote_count: int) -> str: + """Format a nominee line with one activity box per cast ballot.""" + line = f"• **{candidate.display_name}** (`{candidate.user_id}`)" + if vote_count <= 0: + return line + shown = min(vote_count, MAX_ACTIVITY_BOXES_PER_NOMINEE) + boxes = VOTE_ACTIVITY_BOX * shown + if vote_count > MAX_ACTIVITY_BOXES_PER_NOMINEE: + boxes += "…" + return f"{line} {boxes}" + + +def build_poll_embed( + session: AnonymousVoteSession, + candidates: list[AnonymousVoteCandidate], + ballots: list[AnonymousVoteBallot] | None = None, +) -> discord.Embed: + """Build the public poll embed (no approve/reject tallies).""" + title = session.topic or "Anonymous vote" + counts = _ballot_counts_by_candidate(ballots or []) + lines = [_format_nominee_line(c, counts.get(c.id, 0)) for c in candidates] + embed = discord.Embed( + title=title, + description="\n".join(lines) if lines else "No nominees.", + color=0x9ACC14, + ) + embed.add_field( + name="How to vote", + value=( + "1. Select a nominee from the menu\n" + "2. Press **Approve** or **Reject**\n" + f"Each {VOTE_ACTIVITY_BOX} next to a name means someone voted on them " + "(not whether it was approve or reject). " + "Votes stay anonymous; exact tallies appear when the poll closes." + ), + inline=False, + ) + embed.add_field( + name="Closes", + value=discord.utils.format_dt(session.closes_at, style="F") + + f" ({discord.utils.format_dt(session.closes_at, style='R')})", + inline=False, + ) + embed.set_footer(text=f"Session #{session.id}") + return embed + + +def build_results_embed( + session: AnonymousVoteSession, + candidates: list[AnonymousVoteCandidate], + ballots: list[AnonymousVoteBallot], +) -> discord.Embed: + """Build results embed with totals only (no voter identities).""" + counts: dict[int, dict[str, int]] = defaultdict(lambda: {CHOICE_APPROVE: 0, CHOICE_REJECT: 0}) + for ballot in ballots: + if ballot.choice in (CHOICE_APPROVE, CHOICE_REJECT): + counts[ballot.candidate_id][ballot.choice] += 1 + + title = session.topic or "Anonymous vote" + lines = [] + for candidate in candidates: + tally = counts[candidate.id] + lines.append( + f"• **{candidate.display_name}** — ✓ {tally[CHOICE_APPROVE]} / ✗ {tally[CHOICE_REJECT]}" + ) + + embed = discord.Embed( + title=f"Results: {title}", + description="\n".join(lines) if lines else "No nominees.", + color=0x5865F2, + ) + embed.set_footer(text=f"Session #{session.id} • Closed • Voter identities are not shown") + return embed + + +class AnonymousVoteView(View): + """Persistent view: select a nominee, then approve/reject anonymously.""" + + def __init__( + self, + session_id: int, + bot: Bot, + candidates: list[AnonymousVoteCandidate] | None = None, + ): + super().__init__(timeout=None) + self.session_id = session_id + self.bot = bot + + options = [ + SelectOption( + label=(c.display_name[:100] or str(c.user_id)), + value=str(c.id), + description=f"ID {c.user_id}"[:100], + ) + for c in (candidates or []) + ] + if not options: + options = [SelectOption(label="No nominees", value="0", default=True)] + + select = Select( + placeholder="Select a nominee to vote on", + options=options[:25], + custom_id=f"anon_vote_select:{session_id}", + min_values=1, + max_values=1, + disabled=not candidates, + ) + + async def on_select(interaction: Interaction) -> None: + await self._handle_select(interaction, select.values) + + select.callback = on_select + self.add_item(select) + + approve_btn = Button( + label="Approve", + style=discord.ButtonStyle.success, + emoji="✅", + custom_id=f"anon_vote_approve:{session_id}", + ) + approve_btn.callback = self._on_approve + self.add_item(approve_btn) + + reject_btn = Button( + label="Reject", + style=discord.ButtonStyle.danger, + emoji="❌", + custom_id=f"anon_vote_reject:{session_id}", + ) + reject_btn.callback = self._on_reject + self.add_item(reject_btn) + + async def _handle_select(self, interaction: Interaction, values: list[str]) -> None: + if not isinstance(interaction.user, discord.Member) or not _member_can_vote(interaction.user): + await interaction.response.send_message( + "You are not authorized to vote in this poll.", ephemeral=True + ) + return + + if not values: + await interaction.response.send_message("No nominee selected.", ephemeral=True) + return + + candidate_id = int(values[0]) + if candidate_id <= 0: + await interaction.response.send_message("Invalid nominee.", ephemeral=True) + return + + async with AsyncSessionLocal() as session: + vote_session = await session.get(AnonymousVoteSession, self.session_id) + if not vote_session or vote_session.closed: + await interaction.response.send_message("This poll is closed.", ephemeral=True) + return + candidate = await session.get(AnonymousVoteCandidate, candidate_id) + if not candidate or candidate.session_id != self.session_id: + await interaction.response.send_message("Unknown nominee.", ephemeral=True) + return + display_name = candidate.display_name + + _pending_selection[(self.session_id, interaction.user.id)] = candidate_id + await interaction.response.send_message( + f"Selected **{display_name}**. Now press **Approve** or **Reject**.", + ephemeral=True, + ) + + async def _on_approve(self, interaction: Interaction) -> None: + await self._cast_vote(interaction, CHOICE_APPROVE) + + async def _on_reject(self, interaction: Interaction) -> None: + await self._cast_vote(interaction, CHOICE_REJECT) + + async def _cast_vote(self, interaction: Interaction, choice: str) -> None: + if not isinstance(interaction.user, discord.Member) or not _member_can_vote(interaction.user): + await interaction.response.send_message( + "You are not authorized to vote in this poll.", ephemeral=True + ) + return + + candidate_id = _pending_selection.get((self.session_id, interaction.user.id)) + if not candidate_id: + await interaction.response.send_message( + "Select a nominee from the menu first, then press Approve or Reject.", + ephemeral=True, + ) + return + + poll_embed: discord.Embed | None = None + async with AsyncSessionLocal() as session: + vote_session = await session.get(AnonymousVoteSession, self.session_id) + if not vote_session or vote_session.closed: + await interaction.response.send_message("This poll is closed.", ephemeral=True) + return + + candidate = await session.get(AnonymousVoteCandidate, candidate_id) + if not candidate or candidate.session_id != self.session_id: + await interaction.response.send_message("Unknown nominee.", ephemeral=True) + return + + stmt = select(AnonymousVoteBallot).where( + AnonymousVoteBallot.session_id == self.session_id, + AnonymousVoteBallot.candidate_id == candidate_id, + AnonymousVoteBallot.voter_id == interaction.user.id, + ) + ballot = await session.scalar(stmt) + if ballot: + ballot.choice = choice + action = "updated" + else: + session.add( + AnonymousVoteBallot( + session_id=self.session_id, + candidate_id=candidate_id, + voter_id=interaction.user.id, + choice=choice, + ) + ) + action = "recorded" + + display_name = candidate.display_name + await session.commit() + + loaded = await session.scalar( + select(AnonymousVoteSession) + .where(AnonymousVoteSession.id == self.session_id) + .options( + selectinload(AnonymousVoteSession.candidates), + selectinload(AnonymousVoteSession.ballots), + ) + ) + poll_embed = build_poll_embed( + loaded, + list(loaded.candidates), + list(loaded.ballots), + ) if loaded else None + + label = "Approve" if choice == CHOICE_APPROVE else "Reject" + await interaction.response.send_message( + f"Vote {action}: **{label}** for **{display_name}**. " + "Your choice is anonymous; only a neutral activity box is shown publicly.", + ephemeral=True, + ) + + if poll_embed is not None and interaction.message is not None: + try: + await interaction.message.edit(embed=poll_embed) + except discord.HTTPException: + logger.exception( + "Failed to refresh poll embed for session %s after vote.", + self.session_id, + ) + + +async def close_anonymous_vote(bot: Bot, session_id: int) -> None: + """Close a vote session, post totals, and disable controls.""" + async with AsyncSessionLocal() as session: + stmt = ( + select(AnonymousVoteSession) + .where(AnonymousVoteSession.id == session_id) + .options( + selectinload(AnonymousVoteSession.candidates), + selectinload(AnonymousVoteSession.ballots), + ) + ) + vote_session = await session.scalar(stmt) + if not vote_session: + logger.warning("Anonymous vote session %s not found for close.", session_id) + return + if vote_session.closed: + logger.debug("Anonymous vote session %s already closed.", session_id) + return + + vote_session.closed = True + candidates = list(vote_session.candidates) + ballots = list(vote_session.ballots) + channel_id = vote_session.channel_id + message_id = vote_session.message_id + results_embed = build_results_embed(vote_session, candidates, ballots) + await session.commit() + + channel = bot.get_channel(channel_id) + if channel is None: + try: + channel = await bot.fetch_channel(channel_id) + except discord.HTTPException: + logger.exception("Failed to fetch channel %s for vote session %s", channel_id, session_id) + return + + view = AnonymousVoteView(session_id, bot, candidates) + for item in view.children: + item.disabled = True + + if message_id: + try: + message = await channel.fetch_message(message_id) + await message.edit( + content="This anonymous vote is closed. Results below.", + embed=results_embed, + view=view, + ) + return + except discord.HTTPException: + logger.exception( + "Failed to edit poll message %s for session %s; posting results separately.", + message_id, + session_id, + ) + + await channel.send(embed=results_embed) + + +def schedule_vote_close(bot: Bot, session_id: int, closes_at: datetime) -> None: + """Schedule auto-close for a vote session on the bot event loop.""" + bot.loop.create_task(schedule(close_anonymous_vote(bot, session_id), closes_at)) + + +async def register_anonymous_vote_views(bot: Bot) -> None: + """Re-register open vote views and reschedule their closes after restart.""" + async with AsyncSessionLocal() as session: + stmt = ( + select(AnonymousVoteSession) + .where(AnonymousVoteSession.closed.is_(False)) + .options(selectinload(AnonymousVoteSession.candidates)) + ) + result = await session.scalars(stmt) + open_sessions = list(result.all()) + + now = datetime.now() + for vote_session in open_sessions: + bot.add_view(AnonymousVoteView(vote_session.id, bot, vote_session.candidates)) + if vote_session.closes_at <= now: + bot.loop.create_task(close_anonymous_vote(bot, vote_session.id)) + else: + schedule_vote_close(bot, vote_session.id, vote_session.closes_at) + + if open_sessions: + logger.info("Registered %d open anonymous vote session(s).", len(open_sessions)) diff --git a/tests/src/core/test_config.py b/tests/src/core/test_config.py index 0260bac..6730d84 100644 --- a/tests/src/core/test_config.py +++ b/tests/src/core/test_config.py @@ -81,6 +81,10 @@ def test_core_role_groups_present(self): self.assertIn("ALL_HTB_STAFF", settings.role_groups) self.assertIn("ALL_SR_MODS", settings.role_groups) self.assertIn("ALL_HTB_SUPPORT", settings.role_groups) + self.assertIn("VOTE_STARTERS", settings.role_groups) + self.assertIn("VOTE_CASTERS", settings.role_groups) + self.assertEqual(len(settings.role_groups["VOTE_STARTERS"]), 3) + self.assertEqual(len(settings.role_groups["VOTE_CASTERS"]), 6) def test_dynamic_role_groups_removed(self): """Test that dynamic role groups are no longer in settings.""" From 700467db1cf98e4ae7278f91474487cfa8c9192b Mon Sep 17 00:00:00 2001 From: Dimosthenis Schizas Date: Thu, 27 Aug 2026 23:21:06 +0300 Subject: [PATCH 2/2] fix(vote): harden the anonymous vote flow and add test coverage Concurrency and correctness fixes on top of the initial implementation, plus tests for the command and the view. Voting: - Read the chosen nominee from each interaction's own payload. py-cord shares the Select item across all voters and overwrites its state per interaction while callbacks run in later tasks, so two voters could cross nominees. - Move Approve/Reject onto a private per-voter ballot, replacing a module-level dict of pending selections that was lost on restart. - Re-check VOTE_CASTERS when a ballot is cast, not only when it is issued, so a voter whose role is revoked mid-poll cannot still cast. - Give the ballot an on_timeout. A finished view is evicted from the ViewStore, so a later click received no response at all and Discord showed 'This interaction failed' on buttons that still looked live. - Defer before database work in every callback; AsyncSessionLocal uses NullPool and Discord's initial-response deadline is 3 seconds. Persistence: - Write ballots as a single upsert so a double-click cannot race uq_anonymous_vote_ballot_session_candidate_voter. - Claim the close with a conditional UPDATE. The previous read-check-write straddled two awaits, and on_ready reschedules a close on every reconnect, so a long-running vote could publish its results more than once. - Store closes_at as BIGINT epoch seconds, matching Ban.unban_time, instead of a TIMESTAMP that caps at 2038 and round-trips through session timezones. - Delete the session row when the poll cannot be posted, rather than leaving it orphaned with no message id and no scheduled close. Input handling: - Bound topic to 200 characters. build_results_embed prefixes 'Results: ', so a longer topic exceeded the 256-character embed title limit and failed at close time, losing the tallies. - Cap vote duration at 30 days and require nominee IDs to be snowflake-shaped. Adds 72 tests covering the command and the view, including the approve/reject tally, the activity-box truncation boundary, and the concurrency guards. --- .../e1a2b3c4d5e6_add_anonymous_vote_tables.py | 2 +- src/cmds/core/admin.py | 205 +++-- src/database/models/anonymous_vote.py | 6 +- src/views/anonymous_vote.py | 391 +++++---- tests/src/cmds/core/test_admin.py | 238 +++++ tests/src/core/test_config.py | 25 +- tests/src/views/test_anonymous_vote.py | 826 ++++++++++++++++++ 7 files changed, 1452 insertions(+), 241 deletions(-) create mode 100644 tests/src/views/test_anonymous_vote.py diff --git a/alembic/versions/e1a2b3c4d5e6_add_anonymous_vote_tables.py b/alembic/versions/e1a2b3c4d5e6_add_anonymous_vote_tables.py index f23d6e6..3ee0ced 100644 --- a/alembic/versions/e1a2b3c4d5e6_add_anonymous_vote_tables.py +++ b/alembic/versions/e1a2b3c4d5e6_add_anonymous_vote_tables.py @@ -26,7 +26,7 @@ def upgrade() -> None: sa.Column("message_id", mysql.BIGINT(display_width=18), nullable=True), sa.Column("topic", mysql.TEXT(), nullable=True), sa.Column("created_by_id", mysql.BIGINT(display_width=18), nullable=False), - sa.Column("closes_at", mysql.TIMESTAMP(), nullable=False), + sa.Column("closes_at", mysql.BIGINT(display_width=18), nullable=False), sa.Column("closed", sa.Boolean(), nullable=False), sa.PrimaryKeyConstraint("id"), ) diff --git a/src/cmds/core/admin.py b/src/cmds/core/admin.py index f378209..7e0183a 100644 --- a/src/cmds/core/admin.py +++ b/src/cmds/core/admin.py @@ -2,13 +2,13 @@ import logging import re -from datetime import datetime +import time import discord from discord import ApplicationContext, Interaction, Option, WebhookMessage from discord.ext import commands from discord.ext.commands import has_any_role -from sqlalchemy import select +from sqlalchemy import delete, select from sqlalchemy.orm import selectinload from src.bot import Bot @@ -26,7 +26,10 @@ logger = logging.getLogger(__name__) CATEGORY_CHOICES = [c.value for c in RoleCategory] -_MEMBER_TOKEN_RE = re.compile(r"<@!?(\d+)>|^(\d+)$") +MAX_VOTE_DURATION_SECONDS = 30 * 24 * 60 * 60 +# Discord select menus carry at most 25 options. +MAX_VOTE_NOMINEES = 25 +_MEMBER_TOKEN_RE = re.compile(r"<@!?(\d{15,20})>|(\d{15,20})") def _parse_member_ids(raw: str) -> list[int]: @@ -45,6 +48,111 @@ def _parse_member_ids(raw: str) -> list[int]: return list(dict.fromkeys(ids)) +async def _fetch_member(ctx: ApplicationContext, user_id: int) -> discord.Member | None: + """Return the guild member for *user_id*, or None when Discord has no such member.""" + member = ctx.guild.get_member(user_id) + if member is not None: + return member + try: + return await ctx.guild.fetch_member(user_id) + except discord.HTTPException: + return None + + +async def _lookup_members( + ctx: ApplicationContext, member_ids: list[int] +) -> tuple[list[tuple[int, str]], list[str]]: + """Split nominee IDs into resolved (id, display name) pairs and unresolvable IDs.""" + resolved: list[tuple[int, str]] = [] + missing: list[str] = [] + for user_id in member_ids: + member = await _fetch_member(ctx, user_id) + if member is None: + missing.append(str(user_id)) + else: + resolved.append((member.id, member.display_name)) + return resolved, missing + + +async def _resolve_nominees(ctx: ApplicationContext, raw_members: str) -> list[tuple[int, str]]: + """Parse and resolve the nominee argument, raising ValueError with the user-facing reason.""" + member_ids = _parse_member_ids(raw_members) + if not member_ids: + raise ValueError("Provide at least one nominee.") + if len(member_ids) > MAX_VOTE_NOMINEES: + raise ValueError(f"Discord select menus support at most {MAX_VOTE_NOMINEES} nominees.") + + resolved, missing = await _lookup_members(ctx, member_ids) + if missing: + raise ValueError("Could not find member(s) in this server: " + ", ".join(f"`{m}`" for m in missing)) + return resolved + + +def _validate_vote_duration(duration: str) -> tuple[int, str]: + """Validate the requested duration and cap how far out a vote may close.""" + closes_at_ts, error = validate_duration(duration) + if error: + return 0, error + if closes_at_ts - int(time.time()) > MAX_VOTE_DURATION_SECONDS: + return 0, "A vote can stay open for at most 30 days." + return closes_at_ts, "" + + +async def _create_vote_session( + ctx: ApplicationContext, + topic: str | None, + closes_at_ts: int, + nominees: list[tuple[int, str]], +) -> tuple[int, discord.Embed, list[AnonymousVoteCandidate]]: + """Persist the session and its nominees; return the id, poll embed and candidates.""" + async with AsyncSessionLocal() as session: + vote_session = AnonymousVoteSession( + guild_id=ctx.guild.id, + channel_id=ctx.channel.id, + message_id=None, + topic=topic, + created_by_id=ctx.author.id, + closes_at=closes_at_ts, + closed=False, + ) + session.add(vote_session) + await session.flush() + + for user_id, display_name in nominees: + session.add( + AnonymousVoteCandidate( + session_id=vote_session.id, + user_id=user_id, + display_name=display_name, + ) + ) + await session.commit() + + loaded = await session.scalar( + select(AnonymousVoteSession) + .where(AnonymousVoteSession.id == vote_session.id) + .options(selectinload(AnonymousVoteSession.candidates)) + ) + candidates = list(loaded.candidates) + return loaded.id, build_poll_embed(loaded, candidates), candidates + + +async def _delete_vote_session(session_id: int) -> None: + """Drop a session that was never posted; its nominees and ballots cascade.""" + async with AsyncSessionLocal() as session: + await session.execute(delete(AnonymousVoteSession).where(AnonymousVoteSession.id == session_id)) + await session.commit() + + +async def _attach_poll_message(session_id: int, message_id: int) -> None: + """Record the posted message so the session can be edited and closed later.""" + async with AsyncSessionLocal() as session: + vote_session = await session.get(AnonymousVoteSession, session_id) + if vote_session: + vote_session.message_id = message_id + await session.commit() + + class AdminCog(commands.Cog): """Admin commands for bot administration.""" @@ -191,93 +299,38 @@ async def vote( "Nominees as mentions or user IDs (space/comma separated, max 25)", ), duration: Option(str, "How long the vote stays open (e.g. 12h, 1d, 30m)"), - topic: Option(str, "Optional topic shown on the poll", required=False), + topic: Option(str, "Optional topic shown on the poll", required=False, max_length=200), ) -> Interaction | WebhookMessage: """Start an anonymous vote; tallies reveal automatically when duration ends.""" - closes_at_ts, error = validate_duration(duration) + # Resolving up to 25 nominees can outlast Discord's 3s initial-response deadline, + # after which ctx.defer() itself fails with 10062 Unknown interaction. + await ctx.defer(ephemeral=True) + + closes_at_ts, error = _validate_vote_duration(duration) if error: return await ctx.respond(error, ephemeral=True) try: - member_ids = _parse_member_ids(members) + nominees = await _resolve_nominees(ctx, members) except ValueError as exc: return await ctx.respond(str(exc), ephemeral=True) - if not member_ids: - return await ctx.respond("Provide at least one nominee.", ephemeral=True) - if len(member_ids) > 25: - return await ctx.respond( - "Discord select menus support at most 25 nominees.", - ephemeral=True, - ) - - resolved: list[tuple[int, str]] = [] - missing: list[str] = [] - for user_id in member_ids: - member = ctx.guild.get_member(user_id) - if member is None: - try: - member = await ctx.guild.fetch_member(user_id) - except discord.HTTPException: - missing.append(str(user_id)) - continue - resolved.append((member.id, member.display_name)) - - if missing: - return await ctx.respond( - "Could not find member(s) in this server: " + ", ".join(f"`{m}`" for m in missing), - ephemeral=True, - ) - - closes_at = datetime.fromtimestamp(closes_at_ts) - await ctx.defer(ephemeral=True) - - async with AsyncSessionLocal() as session: - vote_session = AnonymousVoteSession( - guild_id=ctx.guild.id, - channel_id=ctx.channel.id, - message_id=None, - topic=topic, - created_by_id=ctx.author.id, - closes_at=closes_at, - closed=False, - ) - session.add(vote_session) - await session.flush() - - for user_id, display_name in resolved: - session.add( - AnonymousVoteCandidate( - session_id=vote_session.id, - user_id=user_id, - display_name=display_name, - ) - ) - await session.commit() - - loaded = await session.scalar( - select(AnonymousVoteSession) - .where(AnonymousVoteSession.id == vote_session.id) - .options(selectinload(AnonymousVoteSession.candidates)) - ) - session_id = loaded.id - candidates = list(loaded.candidates) - poll_embed = build_poll_embed(loaded, candidates) + session_id, poll_embed, candidates = await _create_vote_session(ctx, topic, closes_at_ts, nominees) view = AnonymousVoteView(session_id, self.bot, candidates) self.bot.add_view(view) - message = await ctx.channel.send(embed=poll_embed, view=view) - - async with AsyncSessionLocal() as session: - vote_session = await session.get(AnonymousVoteSession, session_id) - if vote_session: - vote_session.message_id = message.id - await session.commit() - - schedule_vote_close(self.bot, session_id, closes_at) + try: + message = await ctx.channel.send(embed=poll_embed, view=view) + except discord.HTTPException: + logger.exception("Failed to post anonymous vote %s; rolling back session.", session_id) + await _delete_vote_session(session_id) + return await ctx.followup.send("Could not post the poll in this channel.", ephemeral=True) + + await _attach_poll_message(session_id, message.id) + schedule_vote_close(self.bot, session_id, closes_at_ts) return await ctx.followup.send( f"Anonymous vote #{session_id} started in {ctx.channel.mention}. " - f"Closes {discord.utils.format_dt(closes_at, style='R')}.", + f"Closes .", ephemeral=True, ) diff --git a/src/database/models/anonymous_vote.py b/src/database/models/anonymous_vote.py index d11bf9c..c8db4bf 100644 --- a/src/database/models/anonymous_vote.py +++ b/src/database/models/anonymous_vote.py @@ -1,8 +1,6 @@ # flake8: noqa: D101 -from datetime import datetime - from sqlalchemy import Boolean, ForeignKey, Integer, String, UniqueConstraint -from sqlalchemy.dialects.mysql import BIGINT, TEXT, TIMESTAMP +from sqlalchemy.dialects.mysql import BIGINT, TEXT from sqlalchemy.orm import Mapped, mapped_column, relationship from . import Base @@ -17,7 +15,7 @@ class AnonymousVoteSession(Base): message_id: Mapped[int | None] = mapped_column(BIGINT(18), nullable=True) topic: Mapped[str | None] = mapped_column(TEXT, nullable=True) created_by_id: Mapped[int] = mapped_column(BIGINT(18), nullable=False) - closes_at: Mapped[datetime] = mapped_column(TIMESTAMP, nullable=False) + closes_at: Mapped[int] = mapped_column(BIGINT(18), nullable=False) closed: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) candidates: Mapped[list["AnonymousVoteCandidate"]] = relationship( diff --git a/src/views/anonymous_vote.py b/src/views/anonymous_vote.py index 0d1e585..f3915fd 100644 --- a/src/views/anonymous_vote.py +++ b/src/views/anonymous_vote.py @@ -2,14 +2,17 @@ from __future__ import annotations +import contextlib import logging +import time from collections import defaultdict from datetime import datetime import discord from discord import Interaction, SelectOption from discord.ui import Button, Select, View -from sqlalchemy import select +from sqlalchemy import select, update +from sqlalchemy.dialects.mysql import insert from sqlalchemy.orm import selectinload from src.bot import Bot @@ -26,8 +29,15 @@ VOTE_ACTIVITY_BOX = "⬜" # Keep embed descriptions safely under Discord's 4096-char limit. MAX_ACTIVITY_BOXES_PER_NOMINEE = 40 - -_pending_selection: dict[tuple[int, int], int] = {} +# Under the interaction token's 15-minute life, not equal to it: py-cord starts the +# timeout clock in ViewStore.add_view, which runs only once the followup send returns, +# so an exactly-900s ballot would expire after the token and fail its own cleanup edit. +BALLOT_TIMEOUT_SECONDS = 840 +# MariaDB reports 2 from ON DUPLICATE KEY UPDATE only when the stored value actually +# changed, and documents 0 for a no-op update. We never see that 0: SQLAlchemy's MySQL +# dialect ORs CLIENT_FOUND_ROWS into client_flag to get supports_sane_rowcount, so the +# server counts rows matched and 1 means "inserted or re-voted the same way". +UPSERT_ROWCOUNT_UPDATED = 2 def _member_can_vote(member: discord.Member) -> bool: @@ -72,7 +82,7 @@ def build_poll_embed( name="How to vote", value=( "1. Select a nominee from the menu\n" - "2. Press **Approve** or **Reject**\n" + "2. Press **Approve** or **Reject** on the private ballot you receive\n" f"Each {VOTE_ACTIVITY_BOX} next to a name means someone voted on them " "(not whether it was approve or reject). " "Votes stay anonymous; exact tallies appear when the poll closes." @@ -81,8 +91,7 @@ def build_poll_embed( ) embed.add_field( name="Closes", - value=discord.utils.format_dt(session.closes_at, style="F") - + f" ({discord.utils.format_dt(session.closes_at, style='R')})", + value=f" ()", inline=False, ) embed.set_footer(text=f"Session #{session.id}") @@ -117,8 +126,31 @@ def build_results_embed( return embed +def _ballot_confirmation(choice: str, display_name: str, updated: bool) -> str: + """Word the private vote confirmation.""" + label = "Approve" if choice == CHOICE_APPROVE else "Reject" + action = "updated" if updated else "recorded" + return ( + f"Vote {action}: **{label}** for **{display_name}**. " + "Your choice is anonymous; only a neutral activity box is shown publicly." + ) + + +def _candidate_options(candidates: list[AnonymousVoteCandidate] | None) -> list[SelectOption]: + """Build the nominee select options, with a disabled placeholder when there are none.""" + options = [ + SelectOption( + label=(c.display_name[:100] or str(c.user_id)), + value=str(c.id), + description=f"ID {c.user_id}"[:100], + ) + for c in (candidates or []) + ] + return options[:25] or [SelectOption(label="No nominees", value="0", default=True)] + + class AnonymousVoteView(View): - """Persistent view: select a nominee, then approve/reject anonymously.""" + """Persistent view: pick a nominee, then vote on a private ballot.""" def __init__( self, @@ -130,137 +162,140 @@ def __init__( self.session_id = session_id self.bot = bot - options = [ - SelectOption( - label=(c.display_name[:100] or str(c.user_id)), - value=str(c.id), - description=f"ID {c.user_id}"[:100], - ) - for c in (candidates or []) - ] - if not options: - options = [SelectOption(label="No nominees", value="0", default=True)] - - select = Select( + nominee_select = Select( placeholder="Select a nominee to vote on", - options=options[:25], + options=_candidate_options(candidates), custom_id=f"anon_vote_select:{session_id}", min_values=1, max_values=1, disabled=not candidates, ) + nominee_select.callback = self._on_select + self.add_item(nominee_select) - async def on_select(interaction: Interaction) -> None: - await self._handle_select(interaction, select.values) - - select.callback = on_select - self.add_item(select) - - approve_btn = Button( - label="Approve", - style=discord.ButtonStyle.success, - emoji="✅", - custom_id=f"anon_vote_approve:{session_id}", - ) - approve_btn.callback = self._on_approve - self.add_item(approve_btn) - - reject_btn = Button( - label="Reject", - style=discord.ButtonStyle.danger, - emoji="❌", - custom_id=f"anon_vote_reject:{session_id}", - ) - reject_btn.callback = self._on_reject - self.add_item(reject_btn) - - async def _handle_select(self, interaction: Interaction, values: list[str]) -> None: + async def _on_select(self, interaction: Interaction) -> None: + """Send the voter a private ballot for the nominee they picked.""" if not isinstance(interaction.user, discord.Member) or not _member_can_vote(interaction.user): - await interaction.response.send_message( - "You are not authorized to vote in this poll.", ephemeral=True - ) + await interaction.response.send_message("You are not authorized to vote in this poll.", ephemeral=True) return + # Read the pick off this interaction's own payload, never off the Select item: + # the item is shared by every voter and ViewStore.dispatch overwrites its state + # synchronously while callbacks run in later tasks, so voters would cross nominees. + values = interaction.data.get("values", []) if not values: await interaction.response.send_message("No nominee selected.", ephemeral=True) return + await interaction.response.defer(ephemeral=True) candidate_id = int(values[0]) - if candidate_id <= 0: - await interaction.response.send_message("Invalid nominee.", ephemeral=True) + display_name, error = await self._resolve_nominee(candidate_id) + if error: + await interaction.followup.send(error, ephemeral=True) return + await interaction.followup.send( + f"Vote on **{display_name}**:", + view=BallotView(self.session_id, candidate_id, interaction.message), + ephemeral=True, + ) + + async def _resolve_nominee(self, candidate_id: int) -> tuple[str | None, str | None]: + """Return the nominee's display name, or the error message to show the voter.""" async with AsyncSessionLocal() as session: vote_session = await session.get(AnonymousVoteSession, self.session_id) if not vote_session or vote_session.closed: - await interaction.response.send_message("This poll is closed.", ephemeral=True) - return + return None, "This poll is closed." + candidate = await session.get(AnonymousVoteCandidate, candidate_id) if not candidate or candidate.session_id != self.session_id: - await interaction.response.send_message("Unknown nominee.", ephemeral=True) - return - display_name = candidate.display_name + return None, "Unknown nominee." + + return candidate.display_name, None - _pending_selection[(self.session_id, interaction.user.id)] = candidate_id - await interaction.response.send_message( - f"Selected **{display_name}**. Now press **Approve** or **Reject**.", - ephemeral=True, - ) + +class BallotView(View): + """Private, per-voter ballot for a single nominee.""" + + def __init__(self, session_id: int, candidate_id: int, poll_message: discord.Message | None): + super().__init__(timeout=BALLOT_TIMEOUT_SECONDS) + self.session_id = session_id + self.candidate_id = candidate_id + self.poll_message = poll_message + + approve_btn = Button(label="Approve", style=discord.ButtonStyle.success, emoji="✅") + approve_btn.callback = self._on_approve + self.add_item(approve_btn) + + reject_btn = Button(label="Reject", style=discord.ButtonStyle.danger, emoji="❌") + reject_btn.callback = self._on_reject + self.add_item(reject_btn) async def _on_approve(self, interaction: Interaction) -> None: + """Record an approving ballot.""" await self._cast_vote(interaction, CHOICE_APPROVE) async def _on_reject(self, interaction: Interaction) -> None: + """Record a rejecting ballot.""" await self._cast_vote(interaction, CHOICE_REJECT) + async def on_timeout(self) -> None: + """ + Disable the ballot on expiry. + + py-cord evicts a finished view from the ViewStore, so a later click would get + no response at all rather than an error the voter can act on. + """ + self.disable_all_items() + if self.message is None: + return + with contextlib.suppress(discord.HTTPException): + await self.message.edit( + content="This ballot expired. Pick the nominee again to vote.", view=self + ) + async def _cast_vote(self, interaction: Interaction, choice: str) -> None: + """Persist the voter's choice, then refresh the public poll embed.""" if not isinstance(interaction.user, discord.Member) or not _member_can_vote(interaction.user): - await interaction.response.send_message( - "You are not authorized to vote in this poll.", ephemeral=True - ) + await interaction.response.send_message("You are not authorized to vote in this poll.", ephemeral=True) return - candidate_id = _pending_selection.get((self.session_id, interaction.user.id)) - if not candidate_id: - await interaction.response.send_message( - "Select a nominee from the menu first, then press Approve or Reject.", - ephemeral=True, - ) + await interaction.response.defer(ephemeral=True) + + confirmation, poll_embed, error = await self._record_ballot(interaction.user.id, choice) + if error: + await interaction.followup.send(error, ephemeral=True) return - poll_embed: discord.Embed | None = None + await interaction.followup.send(confirmation, ephemeral=True) + await self._refresh_poll_message(poll_embed) + + async def _record_ballot( + self, voter_id: int, choice: str + ) -> tuple[str | None, discord.Embed | None, str | None]: + """Upsert this voter's ballot and rebuild the poll embed, or return an error message.""" async with AsyncSessionLocal() as session: vote_session = await session.get(AnonymousVoteSession, self.session_id) if not vote_session or vote_session.closed: - await interaction.response.send_message("This poll is closed.", ephemeral=True) - return + return None, None, "This poll is closed." - candidate = await session.get(AnonymousVoteCandidate, candidate_id) + candidate = await session.get(AnonymousVoteCandidate, self.candidate_id) if not candidate or candidate.session_id != self.session_id: - await interaction.response.send_message("Unknown nominee.", ephemeral=True) - return - - stmt = select(AnonymousVoteBallot).where( - AnonymousVoteBallot.session_id == self.session_id, - AnonymousVoteBallot.candidate_id == candidate_id, - AnonymousVoteBallot.voter_id == interaction.user.id, - ) - ballot = await session.scalar(stmt) - if ballot: - ballot.choice = choice - action = "updated" - else: - session.add( - AnonymousVoteBallot( - session_id=self.session_id, - candidate_id=candidate_id, - voter_id=interaction.user.id, - choice=choice, - ) - ) - action = "recorded" + return None, None, "Unknown nominee." display_name = candidate.display_name + # One statement, so two clicks in flight cannot both insert and trip + # uq_anonymous_vote_ballot_session_candidate_voter. + result = await session.execute( + insert(AnonymousVoteBallot) + .values( + session_id=self.session_id, + candidate_id=self.candidate_id, + voter_id=voter_id, + choice=choice, + ) + .on_duplicate_key_update(choice=choice) + ) await session.commit() loaded = await session.scalar( @@ -271,33 +306,44 @@ async def _cast_vote(self, interaction: Interaction, choice: str) -> None: selectinload(AnonymousVoteSession.ballots), ) ) - poll_embed = build_poll_embed( - loaded, - list(loaded.candidates), - list(loaded.ballots), - ) if loaded else None - - label = "Approve" if choice == CHOICE_APPROVE else "Reject" - await interaction.response.send_message( - f"Vote {action}: **{label}** for **{display_name}**. " - "Your choice is anonymous; only a neutral activity box is shown publicly.", - ephemeral=True, + poll_embed = ( + build_poll_embed(loaded, list(loaded.candidates), list(loaded.ballots)) if loaded else None + ) + + confirmation = _ballot_confirmation( + choice, display_name, result.rowcount == UPSERT_ROWCOUNT_UPDATED ) + return confirmation, poll_embed, None - if poll_embed is not None and interaction.message is not None: - try: - await interaction.message.edit(embed=poll_embed) - except discord.HTTPException: - logger.exception( - "Failed to refresh poll embed for session %s after vote.", - self.session_id, - ) + async def _refresh_poll_message(self, poll_embed: discord.Embed | None) -> None: + """Redraw the public poll message so the new activity box shows.""" + if poll_embed is None or self.poll_message is None: + return + try: + await self.poll_message.edit(embed=poll_embed) + except discord.HTTPException: + logger.exception("Failed to refresh poll embed for session %s after vote.", self.session_id) -async def close_anonymous_vote(bot: Bot, session_id: int) -> None: - """Close a vote session, post totals, and disable controls.""" +async def _close_vote_session( + session_id: int, +) -> tuple[discord.Embed, list[AnonymousVoteCandidate], int, int | None] | None: + """Claim the close atomically and return what publishing its results needs.""" async with AsyncSessionLocal() as session: - stmt = ( + # Conditional UPDATE rather than read-then-write: on_ready re-schedules a close + # on every reconnect, so a long vote accumulates closes that all fire at once + # and would otherwise each publish a results message. + result = await session.execute( + update(AnonymousVoteSession) + .where(AnonymousVoteSession.id == session_id, AnonymousVoteSession.closed.is_(False)) + .values(closed=True) + ) + if result.rowcount == 0: + logger.debug("Anonymous vote session %s is already closed or gone.", session_id) + return None + await session.commit() + + vote_session = await session.scalar( select(AnonymousVoteSession) .where(AnonymousVoteSession.id == session_id) .options( @@ -305,56 +351,91 @@ async def close_anonymous_vote(bot: Bot, session_id: int) -> None: selectinload(AnonymousVoteSession.ballots), ) ) - vote_session = await session.scalar(stmt) - if not vote_session: - logger.warning("Anonymous vote session %s not found for close.", session_id) - return - if vote_session.closed: - logger.debug("Anonymous vote session %s already closed.", session_id) - return + if vote_session is None: + logger.warning("Anonymous vote session %s vanished after its close was claimed.", session_id) + return None - vote_session.closed = True candidates = list(vote_session.candidates) - ballots = list(vote_session.ballots) - channel_id = vote_session.channel_id - message_id = vote_session.message_id - results_embed = build_results_embed(vote_session, candidates, ballots) - await session.commit() + results_embed = build_results_embed(vote_session, candidates, list(vote_session.ballots)) + return results_embed, candidates, vote_session.channel_id, vote_session.message_id + +async def _resolve_poll_channel(bot: Bot, channel_id: int, session_id: int) -> discord.abc.Messageable | None: + """Return the channel the poll was posted in, fetching it when it is not cached.""" channel = bot.get_channel(channel_id) + if channel is not None: + return channel + try: + return await bot.fetch_channel(channel_id) + except discord.HTTPException: + logger.exception("Failed to fetch channel %s for vote session %s", channel_id, session_id) + return None + + +async def _edit_poll_with_results( + channel: discord.abc.Messageable, + message_id: int | None, + results_embed: discord.Embed, + view: AnonymousVoteView, + session_id: int, +) -> bool: + """Replace the poll message with its results; False when it could not be updated.""" + if not message_id: + return False + try: + message = await channel.fetch_message(message_id) + await message.edit( + content="This anonymous vote is closed. Results below.", + embed=results_embed, + view=view, + ) + return True + except discord.HTTPException: + logger.exception( + "Failed to edit poll message %s for session %s; posting results separately.", + message_id, + session_id, + ) + return False + + +async def _send_results( + channel: discord.abc.Messageable, results_embed: discord.Embed, session_id: int +) -> None: + """Post results as a new message. Swallowing here loses the tallies, so log loudly.""" + try: + await channel.send(embed=results_embed) + except discord.HTTPException: + logger.exception( + "Failed to post results for vote session %s; the tallies are now unrecoverable.", + session_id, + ) + + +async def close_anonymous_vote(bot: Bot, session_id: int) -> None: + """Close a vote session, post totals, and disable controls.""" + closed = await _close_vote_session(session_id) + if closed is None: + return + results_embed, candidates, channel_id, message_id = closed + + channel = await _resolve_poll_channel(bot, channel_id, session_id) if channel is None: - try: - channel = await bot.fetch_channel(channel_id) - except discord.HTTPException: - logger.exception("Failed to fetch channel %s for vote session %s", channel_id, session_id) - return + return view = AnonymousVoteView(session_id, bot, candidates) for item in view.children: item.disabled = True - if message_id: - try: - message = await channel.fetch_message(message_id) - await message.edit( - content="This anonymous vote is closed. Results below.", - embed=results_embed, - view=view, - ) - return - except discord.HTTPException: - logger.exception( - "Failed to edit poll message %s for session %s; posting results separately.", - message_id, - session_id, - ) - - await channel.send(embed=results_embed) + if not await _edit_poll_with_results(channel, message_id, results_embed, view, session_id): + await _send_results(channel, results_embed, session_id) -def schedule_vote_close(bot: Bot, session_id: int, closes_at: datetime) -> None: +def schedule_vote_close(bot: Bot, session_id: int, closes_at_ts: int) -> None: """Schedule auto-close for a vote session on the bot event loop.""" - bot.loop.create_task(schedule(close_anonymous_vote(bot, session_id), closes_at)) + bot.loop.create_task( + schedule(close_anonymous_vote(bot, session_id), datetime.fromtimestamp(closes_at_ts)) + ) async def register_anonymous_vote_views(bot: Bot) -> None: @@ -368,7 +449,7 @@ async def register_anonymous_vote_views(bot: Bot) -> None: result = await session.scalars(stmt) open_sessions = list(result.all()) - now = datetime.now() + now = int(time.time()) for vote_session in open_sessions: bot.add_view(AnonymousVoteView(vote_session.id, bot, vote_session.candidates)) if vote_session.closes_at <= now: diff --git a/tests/src/cmds/core/test_admin.py b/tests/src/cmds/core/test_admin.py index 6c37c90..7c4de22 100644 --- a/tests/src/cmds/core/test_admin.py +++ b/tests/src/cmds/core/test_admin.py @@ -2,6 +2,7 @@ from unittest.mock import AsyncMock, MagicMock, patch +import discord import pytest from src.cmds.core import admin @@ -9,6 +10,14 @@ from tests import helpers +def _session_ctx(session_mock: AsyncMock) -> MagicMock: + """Wrap an AsyncMock session so it works as ``async with AsyncSessionLocal() as s``.""" + ctx = MagicMock() + ctx.__aenter__ = AsyncMock(return_value=session_mock) + ctx.__aexit__ = AsyncMock(return_value=False) + return ctx + + class TestAdminCog: """Test the Admin cog.""" @@ -288,3 +297,232 @@ def test_setup(self, bot): """Test the setup function registers the cog.""" admin.setup(bot) bot.add_cog.assert_called_once() + + +class TestParseMemberIds: + """Test the nominee token parser.""" + + def test_parses_mentions_and_bare_snowflakes(self): + """Both mention forms and bare snowflakes resolve to ints, order preserved.""" + assert admin._parse_member_ids("<@123456789012345678> <@!234567890123456789> 345678901234567890") == [ + 123456789012345678, + 234567890123456789, + 345678901234567890, + ] + + def test_drops_duplicates(self): + """A repeated nominee is only counted once.""" + assert admin._parse_member_ids("123456789012345678, <@123456789012345678>") == [123456789012345678] + + def test_rejects_overlong_digit_run(self): + """A digit run longer than a snowflake is rejected at parse time.""" + with pytest.raises(ValueError, match="Could not parse"): + admin._parse_member_ids("9" * 600) + + def test_rejects_too_short_digit_run(self): + """A digit run shorter than a snowflake is rejected at parse time.""" + with pytest.raises(ValueError, match="Could not parse"): + admin._parse_member_ids("12345") + + +def _guild_ctx(get_member: MagicMock, fetch_member: AsyncMock) -> MagicMock: + """Build a context whose guild resolves members the two ways Discord offers.""" + ctx = MagicMock() + ctx.guild = MagicMock() + ctx.guild.get_member = get_member + ctx.guild.fetch_member = fetch_member + return ctx + + +class TestFetchMember: + """Test the member resolution fallback.""" + + @pytest.mark.asyncio + async def test_returns_the_cached_member_without_calling_the_api(self): + member = MagicMock() + fetch = AsyncMock() + ctx = _guild_ctx(MagicMock(return_value=member), fetch) + + assert await admin._fetch_member(ctx, 1) is member + fetch.assert_not_awaited() + + @pytest.mark.asyncio + async def test_falls_back_to_the_api_when_not_cached(self): + member = MagicMock() + ctx = _guild_ctx(MagicMock(return_value=None), AsyncMock(return_value=member)) + + assert await admin._fetch_member(ctx, 1) is member + + @pytest.mark.asyncio + async def test_returns_none_when_discord_has_no_such_member(self): + ctx = _guild_ctx( + MagicMock(return_value=None), + AsyncMock(side_effect=discord.HTTPException(MagicMock(), "unknown member")), + ) + + assert await admin._fetch_member(ctx, 1) is None + + +class TestResolveNominees: + """Test the nominee argument validation.""" + + @pytest.mark.asyncio + async def test_rejects_an_empty_nominee_list(self): + with pytest.raises(ValueError, match="at least one nominee"): + await admin._resolve_nominees(MagicMock(), " ") + + @pytest.mark.asyncio + async def test_rejects_more_nominees_than_a_select_menu_holds(self): + too_many = " ".join(str(123456789012345678 + i) for i in range(admin.MAX_VOTE_NOMINEES + 1)) + + with pytest.raises(ValueError, match="at most 25 nominees"): + await admin._resolve_nominees(MagicMock(), too_many) + + @pytest.mark.asyncio + async def test_names_the_members_it_could_not_find(self): + ctx = _guild_ctx( + MagicMock(return_value=None), + AsyncMock(side_effect=discord.HTTPException(MagicMock(), "unknown member")), + ) + + with pytest.raises(ValueError, match="123456789012345678"): + await admin._resolve_nominees(ctx, "123456789012345678") + + @pytest.mark.asyncio + async def test_returns_id_and_display_name_pairs(self): + member = MagicMock() + member.id = 123456789012345678 + member.display_name = "Nominee" + ctx = _guild_ctx(MagicMock(return_value=member), AsyncMock()) + + assert await admin._resolve_nominees(ctx, "123456789012345678") == [(123456789012345678, "Nominee")] + + +class TestVoteCommand: + """Test the /admin vote command.""" + + def test_topic_is_bounded_below_the_embed_title_limit(self, bot): + """Discord caps embed titles at 256; build_results_embed also prefixes 'Results: '.""" + cog = admin.AdminCog(bot) + topic = next(o for o in cog.vote.options if o.name == "topic") + + assert topic.max_length is not None + assert topic.max_length + len("Results: ") <= 256 + + @pytest.mark.asyncio + async def test_defers_before_resolving_members(self, ctx, bot): + """Up to 25 sequential fetch_member calls blow the 3s budget; defer must come first.""" + order: list[str] = [] + ctx.defer = AsyncMock(side_effect=lambda **kw: order.append("defer")) + ctx.guild.get_member = MagicMock(side_effect=lambda uid: order.append("resolve") or None) + ctx.guild.fetch_member = AsyncMock(side_effect=discord.HTTPException(MagicMock(), "unknown")) + + cog = admin.AdminCog(bot) + with patch("src.cmds.core.admin.validate_duration", return_value=(1800000000, "")), patch( + "src.cmds.core.admin.time.time", return_value=1799999000 + ): + await cog.vote.callback(cog, ctx, "<@123456789012345678>", "10m", None) + + assert order[0] == "defer" + assert "resolve" in order + + @pytest.mark.asyncio + async def test_rejects_duration_beyond_cap(self, ctx, bot): + """A duration validate_duration happily accepts is still capped at 30 days.""" + cog = admin.AdminCog(bot) + with patch("src.cmds.core.admin.AsyncSessionLocal") as session_local: + await cog.vote.callback(cog, ctx, "<@123456789012345678>", "50y", None) + + session_local.assert_not_called() + ctx.respond.assert_called_once() + assert "30 days" in ctx.respond.call_args[0][0] + + @pytest.mark.asyncio + async def test_stores_closes_at_as_epoch_seconds(self, ctx, bot): + """The validated epoch int is persisted and scheduled on unconverted.""" + member = MagicMock() + member.id = 123456789012345678 + member.display_name = "Nominee" + ctx.guild.get_member = MagicMock(return_value=member) + + loaded = MagicMock() + loaded.id = 42 + loaded.candidates = [] + + session = AsyncMock() + session.add = MagicMock() + session.scalar = AsyncMock(return_value=loaded) + + cog = admin.AdminCog(bot) + with ( + patch("src.cmds.core.admin.AsyncSessionLocal", return_value=_session_ctx(session)), + patch("src.cmds.core.admin.build_poll_embed", return_value=MagicMock()), + patch("src.cmds.core.admin.schedule_vote_close") as mock_schedule_close, + patch("src.cmds.core.admin.validate_duration", return_value=(1800000000, "")), + patch("src.cmds.core.admin.time.time", return_value=1799999000), + ): + await cog.vote.callback(cog, ctx, "<@123456789012345678>", "10m", "Topic") + + created = session.add.call_args_list[0].args[0] + assert created.closes_at == 1800000000 + mock_schedule_close.assert_called_once_with(bot, 42, 1800000000) + + @pytest.mark.asyncio + async def test_rolls_back_the_session_when_the_poll_cannot_be_posted(self, ctx, bot): + """A failed send must not leave an orphan open session that resurfaces on restart.""" + member = MagicMock() + member.id = 123456789012345678 + member.display_name = "Nominee" + ctx.guild.get_member = MagicMock(return_value=member) + ctx.channel.send = AsyncMock(side_effect=discord.HTTPException(MagicMock(), "no embed links")) + + loaded = MagicMock() + loaded.id = 42 + loaded.candidates = [] + + session = AsyncMock() + session.add = MagicMock() + session.scalar = AsyncMock(return_value=loaded) + + cog = admin.AdminCog(bot) + with ( + patch("src.cmds.core.admin.AsyncSessionLocal", return_value=_session_ctx(session)), + patch("src.cmds.core.admin.build_poll_embed", return_value=MagicMock()), + patch("src.cmds.core.admin.schedule_vote_close") as mock_schedule_close, + patch("src.cmds.core.admin.validate_duration", return_value=(1800000000, "")), + patch("src.cmds.core.admin.time.time", return_value=1799999000), + ): + await cog.vote.callback(cog, ctx, "<@123456789012345678>", "10m", "Topic") + + deletes = [ + call.args[0] for call in session.execute.await_args_list if getattr(call.args[0], "is_delete", False) + ] + assert len(deletes) == 1 + assert deletes[0].compile().params == {"id_1": 42} + + mock_schedule_close.assert_not_called() + ctx.followup.send.assert_awaited_once() + assert "Could not post the poll" in ctx.followup.send.await_args[0][0] + + @pytest.mark.asyncio + async def test_rejects_an_unparseable_duration(self, ctx, bot): + """An invalid duration is refused before any nominee lookup or DB work.""" + cog = admin.AdminCog(bot) + with patch("src.cmds.core.admin.AsyncSessionLocal") as session_local: + await cog.vote.callback(cog, ctx, "<@123456789012345678>", "banana", None) + + session_local.assert_not_called() + ctx.guild.get_member.assert_not_called() + ctx.respond.assert_called_once() + assert "could not parse" in ctx.respond.call_args[0][0].lower() + + @pytest.mark.asyncio + async def test_rejects_a_malformed_nominee_argument(self, ctx, bot): + """A nominee token that is neither a mention nor a snowflake is reported back.""" + cog = admin.AdminCog(bot) + with patch("src.cmds.core.admin.AsyncSessionLocal") as session_local: + await cog.vote.callback(cog, ctx, "not-a-mention", "10m", None) + + session_local.assert_not_called() + ctx.respond.assert_called_once() + assert "not-a-mention" in ctx.respond.call_args[0][0] diff --git a/tests/src/core/test_config.py b/tests/src/core/test_config.py index 6730d84..16175f9 100644 --- a/tests/src/core/test_config.py +++ b/tests/src/core/test_config.py @@ -2,8 +2,8 @@ from pydantic import ValidationError -from src.core.config import Global from src.core import settings +from src.core.config import Global class TestConfig(unittest.TestCase): @@ -81,10 +81,25 @@ def test_core_role_groups_present(self): self.assertIn("ALL_HTB_STAFF", settings.role_groups) self.assertIn("ALL_SR_MODS", settings.role_groups) self.assertIn("ALL_HTB_SUPPORT", settings.role_groups) - self.assertIn("VOTE_STARTERS", settings.role_groups) - self.assertIn("VOTE_CASTERS", settings.role_groups) - self.assertEqual(len(settings.role_groups["VOTE_STARTERS"]), 3) - self.assertEqual(len(settings.role_groups["VOTE_CASTERS"]), 6) + self.assertEqual( + settings.role_groups["VOTE_STARTERS"], + [ + settings.roles.ADMINISTRATOR, + settings.roles.COMMUNITY_MANAGER, + settings.roles.COMMUNITY_TEAM, + ], + ) + self.assertEqual( + settings.role_groups["VOTE_CASTERS"], + [ + settings.roles.ADMINISTRATOR, + settings.roles.COMMUNITY_MANAGER, + settings.roles.COMMUNITY_TEAM, + settings.roles.SR_MODERATOR, + settings.roles.MODERATOR, + settings.roles.JR_MODERATOR, + ], + ) def test_dynamic_role_groups_removed(self): """Test that dynamic role groups are no longer in settings.""" diff --git a/tests/src/views/test_anonymous_vote.py b/tests/src/views/test_anonymous_vote.py new file mode 100644 index 0000000..bca4bf4 --- /dev/null +++ b/tests/src/views/test_anonymous_vote.py @@ -0,0 +1,826 @@ +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock, patch + +import discord +import pytest +from discord.ui import Button, Select + +from src.core import settings +from src.views.anonymous_vote import ( + CHOICE_APPROVE, + CHOICE_REJECT, + MAX_ACTIVITY_BOXES_PER_NOMINEE, + VOTE_ACTIVITY_BOX, + AnonymousVoteView, + BallotView, + _ballot_counts_by_candidate, + _format_nominee_line, + build_poll_embed, + build_results_embed, + close_anonymous_vote, + register_anonymous_vote_views, + schedule_vote_close, +) +from tests import helpers + + +def _make_session( + session_id: int = 1, + topic: str | None = "Promotion round", + closes_at: int = 1800000000, + closed: bool = False, + message_id: int | None = 666, +) -> MagicMock: + """Build a mock AnonymousVoteSession model instance.""" + session = MagicMock() + session.id = session_id + session.topic = topic + session.closes_at = closes_at + session.closed = closed + session.channel_id = 555 + session.message_id = message_id + return session + + +def _make_candidate(candidate_id: int = 1, session_id: int = 1, name: str = "Nominee") -> MagicMock: + """Build a mock AnonymousVoteCandidate model instance.""" + candidate = MagicMock() + candidate.id = candidate_id + candidate.session_id = session_id + candidate.user_id = 100 + candidate_id + candidate.display_name = name + return candidate + + +def _make_ballot(candidate_id: int = 1, choice: str = CHOICE_APPROVE) -> MagicMock: + """Build a mock AnonymousVoteBallot model instance.""" + ballot = MagicMock() + ballot.candidate_id = candidate_id + ballot.choice = choice + return ballot + + +def _session_ctx(session_mock: AsyncMock) -> MagicMock: + """Wrap an AsyncMock session so it works as ``async with AsyncSessionLocal() as s``.""" + ctx = MagicMock() + ctx.__aenter__ = AsyncMock(return_value=session_mock) + ctx.__aexit__ = AsyncMock(return_value=False) + return ctx + + +def _make_voter(can_vote: bool = True) -> helpers.MockMember: + """Build a member who does or does not hold a VOTE_CASTERS role.""" + role_id = settings.role_groups["VOTE_CASTERS"][0] if can_vote else 999999 + return helpers.MockMember(roles=[helpers.MockRole(name="Voter", id=role_id)]) + + +def _make_interaction( + values: list[str] | None = None, + user: helpers.MockMember | None = None, + message: MagicMock | None = None, +) -> MagicMock: + """Build a lightweight mock Interaction with the attributes our callbacks use.""" + interaction = MagicMock() + interaction.user = user or _make_voter() + interaction.data = {"values": list(values or [])} + + interaction.response = MagicMock() + interaction.response.defer = AsyncMock() + interaction.response.send_message = AsyncMock() + + interaction.followup = MagicMock() + interaction.followup.send = AsyncMock() + + interaction.message = MagicMock() if message is None else message + interaction.message.edit = AsyncMock() + return interaction + + +def _loaded_session(session_id: int = 1, message_id: int | None = 666) -> MagicMock: + """Build the eagerly-loaded session the poll embed is rebuilt from.""" + loaded = _make_session(session_id=session_id, message_id=message_id) + loaded.candidates = [_make_candidate(session_id=session_id)] + loaded.ballots = [] + return loaded + + +def _make_poll_message() -> MagicMock: + """Build the public poll message a ballot writes its refreshed embed back to.""" + message = MagicMock() + message.edit = AsyncMock() + return message + + +def _mysql_dialect() -> object: + """The dialect the upsert is compiled against; ``on_duplicate_key_update`` is MySQL-only.""" + from sqlalchemy.dialects import mysql + + return mysql.dialect() + + +def _close_db(rowcount: int = 1, loaded: MagicMock | None = None) -> AsyncMock: + """Session mock for close; *rowcount* decides whether this coroutine claims the close.""" + session = AsyncMock() + session.execute = AsyncMock(return_value=MagicMock(rowcount=rowcount)) + session.scalar = AsyncMock(return_value=loaded) + return session + + +def _open_session_db(candidate: MagicMock, vote_session: MagicMock | None = None) -> AsyncMock: + """Build a session mock whose ``get`` resolves the vote session then the candidate.""" + session = AsyncMock() + session.add = MagicMock() + session.get = AsyncMock(side_effect=[vote_session or _make_session(), candidate]) + return session + + +class TestBuildPollEmbed: + def test_closes_field_renders_epoch_seconds_directly(self): + session = _make_session(closes_at=1800000000) + embed = build_poll_embed(session, [_make_candidate()]) + + closes_field = next(f for f in embed.fields if f.name == "Closes") + assert closes_field.value == " ()" + + +class TestAnonymousVoteViewShape: + @pytest.mark.asyncio + async def test_holds_only_the_select(self, bot): + view = AnonymousVoteView(9, bot, [_make_candidate(session_id=9)]) + + assert [type(c) for c in view.children] == [Select] + assert view.children[0].custom_id == "anon_vote_select:9" + assert view.timeout is None + + @pytest.mark.asyncio + async def test_module_keeps_no_pending_selection_state(self): + import src.views.anonymous_vote as module + + assert not hasattr(module, "_pending_selection") + + +class TestSelectCallback: + @pytest.mark.asyncio + async def test_reads_nominee_from_interaction_payload_not_shared_item(self, bot): + """Two concurrent voters must not cross nominees via the shared Select item.""" + candidates = [ + _make_candidate(candidate_id=1, session_id=9, name="Alice"), + _make_candidate(candidate_id=2, session_id=9, name="Bob"), + ] + view = AnonymousVoteView(9, bot, candidates) + + # py-cord's ViewStore.dispatch calls refresh_state on the shared item per + # interaction and the callback runs in a later task, so by the time Alice's + # callback runs the item can already hold Bob's selection. + bobs_interaction = _make_interaction(values=["2"]) + view.children[0]._selected_values = ["2"] + view.children[0]._interaction = bobs_interaction + assert view.children[0].values == ["2"] + + by_id = {c.id: c for c in candidates} + session = AsyncMock() + session.get = AsyncMock(side_effect=lambda model, pk: _make_session() if pk == 9 else by_id.get(pk)) + + interaction = _make_interaction(values=["1"]) + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await view.children[0].callback(interaction) + + ballot = interaction.followup.send.call_args.kwargs["view"] + assert ballot.candidate_id == 1 + assert "Alice" in interaction.followup.send.call_args[0][0] + + @pytest.mark.asyncio + async def test_sends_ephemeral_ballot_view_for_the_chosen_nominee(self, bot): + candidate = _make_candidate(candidate_id=5, session_id=9, name="Carol") + view = AnonymousVoteView(9, bot, [candidate]) + + poll_message = MagicMock() + interaction = _make_interaction(values=["5"], message=poll_message) + session = _open_session_db(candidate) + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await view.children[0].callback(interaction) + + kwargs = interaction.followup.send.call_args.kwargs + assert kwargs["ephemeral"] is True + ballot = kwargs["view"] + assert isinstance(ballot, BallotView) + assert (ballot.session_id, ballot.candidate_id) == (9, 5) + assert ballot.poll_message is poll_message + + @pytest.mark.asyncio + async def test_defers_before_touching_the_database(self, bot): + candidate = _make_candidate(candidate_id=5, session_id=9) + view = AnonymousVoteView(9, bot, [candidate]) + + order: list[str] = [] + interaction = _make_interaction(values=["5"]) + interaction.response.defer = AsyncMock(side_effect=lambda **kw: order.append("defer")) + + session = _open_session_db(candidate) + session_ctx = _session_ctx(session) + + def open_session() -> MagicMock: + order.append("db") + return session_ctx + + with patch("src.views.anonymous_vote.AsyncSessionLocal", side_effect=open_session): + await view.children[0].callback(interaction) + + assert order == ["defer", "db"] + interaction.response.send_message.assert_not_awaited() + + @pytest.mark.asyncio + async def test_rejects_a_member_without_a_voter_role(self, bot): + view = AnonymousVoteView(9, bot, [_make_candidate(session_id=9)]) + interaction = _make_interaction(values=["1"], user=_make_voter(can_vote=False)) + + with patch("src.views.anonymous_vote.AsyncSessionLocal") as session_local: + await view.children[0].callback(interaction) + + session_local.assert_not_called() + assert "not authorized" in interaction.response.send_message.call_args[0][0] + + @pytest.mark.asyncio + async def test_reports_a_closed_poll(self, bot): + view = AnonymousVoteView(9, bot, [_make_candidate(session_id=9)]) + interaction = _make_interaction(values=["1"]) + session = _open_session_db(None, vote_session=_make_session(closed=True)) + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await view.children[0].callback(interaction) + + assert interaction.followup.send.call_args[0][0] == "This poll is closed." + + @pytest.mark.asyncio + async def test_reports_a_nominee_from_another_session(self, bot): + view = AnonymousVoteView(9, bot, [_make_candidate(session_id=9)]) + interaction = _make_interaction(values=["1"]) + session = _open_session_db(_make_candidate(candidate_id=1, session_id=77)) + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await view.children[0].callback(interaction) + + assert interaction.followup.send.call_args[0][0] == "Unknown nominee." + + @pytest.mark.asyncio + async def test_reports_an_empty_selection_without_deferring(self, bot): + view = AnonymousVoteView(9, bot, [_make_candidate(session_id=9)]) + interaction = _make_interaction(values=[]) + + with patch("src.views.anonymous_vote.AsyncSessionLocal") as session_local: + await view.children[0].callback(interaction) + + session_local.assert_not_called() + interaction.response.defer.assert_not_awaited() + assert interaction.response.send_message.call_args[0][0] == "No nominee selected." + assert interaction.response.send_message.call_args.kwargs["ephemeral"] is True + + @pytest.mark.asyncio + async def test_reports_a_payload_carrying_no_values_key(self, bot): + view = AnonymousVoteView(9, bot, [_make_candidate(session_id=9)]) + interaction = _make_interaction() + interaction.data = {} + + with patch("src.views.anonymous_vote.AsyncSessionLocal") as session_local: + await view.children[0].callback(interaction) + + session_local.assert_not_called() + interaction.response.defer.assert_not_awaited() + assert interaction.response.send_message.call_args[0][0] == "No nominee selected." + + +class TestBallotView: + @pytest.mark.asyncio + async def test_holds_two_buttons_and_is_not_persistent(self): + ballot = BallotView(9, 5, _make_poll_message()) + + buttons = [c for c in ballot.children if isinstance(c, Button)] + assert [b.label for b in buttons] == ["Approve", "Reject"] + assert ballot.is_persistent() is False + + @pytest.mark.asyncio + async def test_upserts_the_ballot_instead_of_read_then_insert(self): + candidate = _make_candidate(candidate_id=5, session_id=9, name="Carol") + session = _open_session_db(candidate) + session.scalar = AsyncMock(return_value=_loaded_session()) + + ballot = BallotView(9, 5, _make_poll_message()) + interaction = _make_interaction() + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await ballot.children[0].callback(interaction) + + session.add.assert_not_called() + stmt = session.execute.await_args.args[0] + assert stmt.is_insert + compiled = str(stmt.compile(dialect=_mysql_dialect())) + assert "ON DUPLICATE KEY UPDATE" in compiled + + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("rowcount", "expected_wording"), + [(2, "Vote updated"), (1, "Vote recorded")], + ) + async def test_wording_reflects_whether_the_stored_choice_changed(self, rowcount, expected_wording): + """Only rowcount 2 proves a change; 1 is insert-or-same-value through asyncmy.""" + candidate = _make_candidate(candidate_id=5, session_id=9, name="Carol") + session = _open_session_db(candidate) + session.execute = AsyncMock(return_value=MagicMock(rowcount=rowcount)) + session.scalar = AsyncMock(return_value=_loaded_session()) + + ballot = BallotView(9, 5, _make_poll_message()) + interaction = _make_interaction() + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await ballot.children[0].callback(interaction) + + assert expected_wording in interaction.followup.send.call_args[0][0] + + @pytest.mark.asyncio + async def test_refuses_a_voter_who_lost_the_role_since_the_ballot_was_issued(self): + """The ballot outlives the role check that issued it, so re-check at cast time.""" + session = AsyncMock() + ballot = BallotView(9, 5, _make_poll_message()) + interaction = _make_interaction(user=_make_voter(can_vote=False)) + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)) as db: + await ballot.children[0].callback(interaction) + + db.assert_not_called() + session.execute.assert_not_awaited() + assert "not authorized" in interaction.response.send_message.call_args[0][0] + + @pytest.mark.asyncio + async def test_still_accepts_a_voter_who_kept_the_role(self): + candidate = _make_candidate(candidate_id=5, session_id=9, name="Carol") + session = _open_session_db(candidate) + session.scalar = AsyncMock(return_value=_loaded_session()) + + ballot = BallotView(9, 5, _make_poll_message()) + interaction = _make_interaction(user=_make_voter(can_vote=True)) + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await ballot.children[0].callback(interaction) + + session.execute.assert_awaited_once() + assert "Carol" in interaction.followup.send.call_args[0][0] + + @pytest.mark.asyncio + async def test_timeout_expires_before_the_interaction_token_does(self): + """The clock starts after the send returns, so 900 would fire past token expiry.""" + assert BallotView(9, 5, None).timeout == 840 + + @pytest.mark.asyncio + async def test_on_timeout_disables_the_ballot_and_says_it_lapsed(self): + message = _make_poll_message() + ballot = BallotView(9, 5, _make_poll_message()) + ballot.message = message + + await ballot.on_timeout() + + assert all(item.disabled for item in ballot.children) + assert "expired" in message.edit.await_args.kwargs["content"] + assert message.edit.await_args.kwargs["view"] is ballot + + @pytest.mark.asyncio + async def test_on_timeout_without_a_message_does_not_raise(self): + ballot = BallotView(9, 5, _make_poll_message()) + ballot.message = None + + await ballot.on_timeout() + + assert all(item.disabled for item in ballot.children) + + @pytest.mark.asyncio + async def test_on_timeout_survives_a_failed_edit(self): + message = _make_poll_message() + message.edit = AsyncMock(side_effect=discord.HTTPException(MagicMock(), "gone")) + ballot = BallotView(9, 5, _make_poll_message()) + ballot.message = message + + await ballot.on_timeout() + + message.edit.assert_awaited_once() + + @pytest.mark.asyncio + async def test_reports_a_vanished_session_instead_of_crashing_after_the_write(self): + """The ballot is already persisted here; an AttributeError would strand the voter.""" + candidate = _make_candidate(candidate_id=5, session_id=9, name="Carol") + session = _open_session_db(candidate) + session.scalar = AsyncMock(return_value=None) + + ballot = BallotView(9, 5, _make_poll_message()) + interaction = _make_interaction() + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await ballot.children[0].callback(interaction) + + interaction.followup.send.assert_awaited_once() + + @pytest.mark.asyncio + @pytest.mark.parametrize(("button_index", "expected"), [(0, CHOICE_APPROVE), (1, CHOICE_REJECT)]) + async def test_each_button_records_its_own_choice(self, button_index, expected): + candidate = _make_candidate(candidate_id=5, session_id=9, name="Carol") + session = _open_session_db(candidate) + session.scalar = AsyncMock(return_value=_loaded_session()) + + ballot = BallotView(9, 5, _make_poll_message()) + interaction = _make_interaction() + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await ballot.children[button_index].callback(interaction) + + params = session.execute.await_args.args[0].compile(dialect=_mysql_dialect()).params + assert params["choice"] == expected + assert expected.capitalize() in interaction.followup.send.call_args[0][0] + + @pytest.mark.asyncio + async def test_defers_before_touching_the_database(self): + candidate = _make_candidate(candidate_id=5, session_id=9) + session = _open_session_db(candidate) + session.scalar = AsyncMock(return_value=_loaded_session()) + + order: list[str] = [] + interaction = _make_interaction() + interaction.response.defer = AsyncMock(side_effect=lambda **kw: order.append("defer")) + session_ctx = _session_ctx(session) + + def open_session() -> MagicMock: + order.append("db") + return session_ctx + + ballot = BallotView(9, 5, _make_poll_message()) + with patch("src.views.anonymous_vote.AsyncSessionLocal", side_effect=open_session): + await ballot.children[0].callback(interaction) + + assert order == ["defer", "db"] + interaction.response.send_message.assert_not_awaited() + + @pytest.mark.asyncio + async def test_refreshes_the_public_poll_message(self): + candidate = _make_candidate(candidate_id=5, session_id=9) + session = _open_session_db(candidate) + session.scalar = AsyncMock(return_value=_loaded_session()) + + poll_message = MagicMock() + poll_message.edit = AsyncMock() + ballot = BallotView(9, 5, poll_message) + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await ballot.children[0].callback(_make_interaction()) + + poll_message.edit.assert_awaited_once() + assert "embed" in poll_message.edit.await_args.kwargs + + @pytest.mark.asyncio + async def test_records_the_vote_when_there_is_no_poll_message_to_refresh(self): + candidate = _make_candidate(candidate_id=5, session_id=9, name="Carol") + session = _open_session_db(candidate) + session.scalar = AsyncMock(return_value=_loaded_session()) + + ballot = BallotView(9, 5, None) + interaction = _make_interaction() + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await ballot.children[0].callback(interaction) + + session.execute.assert_awaited_once() + assert "Carol" in interaction.followup.send.call_args[0][0] + + @pytest.mark.asyncio + async def test_a_failed_refresh_does_not_lose_the_recorded_vote(self): + candidate = _make_candidate(candidate_id=5, session_id=9, name="Carol") + session = _open_session_db(candidate) + session.scalar = AsyncMock(return_value=_loaded_session()) + + poll_message = _make_poll_message() + poll_message.edit = AsyncMock(side_effect=discord.HTTPException(MagicMock(), "boom")) + + ballot = BallotView(9, 5, poll_message) + interaction = _make_interaction() + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await ballot.children[0].callback(interaction) + + session.execute.assert_awaited_once() + assert "Carol" in interaction.followup.send.call_args[0][0] + + @pytest.mark.asyncio + async def test_reports_a_closed_poll_without_writing(self): + session = _open_session_db(None, vote_session=_make_session(closed=True)) + poll_message = MagicMock() + poll_message.edit = AsyncMock() + + ballot = BallotView(9, 5, poll_message) + interaction = _make_interaction() + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await ballot.children[0].callback(interaction) + + session.execute.assert_not_awaited() + poll_message.edit.assert_not_awaited() + assert interaction.followup.send.call_args[0][0] == "This poll is closed." + + @pytest.mark.asyncio + async def test_reports_a_missing_nominee_without_writing(self): + session = _open_session_db(None) + poll_message = _make_poll_message() + + ballot = BallotView(9, 5, poll_message) + interaction = _make_interaction() + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await ballot.children[0].callback(interaction) + + session.execute.assert_not_awaited() + poll_message.edit.assert_not_awaited() + assert interaction.followup.send.call_args[0][0] == "Unknown nominee." + + @pytest.mark.asyncio + async def test_reports_a_nominee_from_another_session_without_writing(self): + """A ballot must not write against a candidate row belonging to a different poll.""" + session = _open_session_db(_make_candidate(candidate_id=5, session_id=77)) + poll_message = _make_poll_message() + + ballot = BallotView(9, 5, poll_message) + interaction = _make_interaction() + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await ballot.children[0].callback(interaction) + + session.execute.assert_not_awaited() + poll_message.edit.assert_not_awaited() + assert interaction.followup.send.call_args[0][0] == "Unknown nominee." + + +class TestCloseAnonymousVote: + @pytest.mark.asyncio + async def test_disables_the_select_and_posts_results(self, bot): + session = _close_db(loaded=_loaded_session(session_id=9)) + + message = _make_poll_message() + channel = MagicMock() + channel.fetch_message = AsyncMock(return_value=message) + bot.get_channel = MagicMock(return_value=channel) + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await close_anonymous_vote(bot, 9) + + session.execute.assert_awaited_once() + view = message.edit.await_args.kwargs["view"] + assert all(item.disabled for item in view.children) + + @pytest.mark.asyncio + async def test_a_failed_fallback_send_is_logged_not_raised(self, bot): + """close runs in a bare task, so an unhandled send failure would vanish silently.""" + session = _close_db(loaded=_loaded_session(session_id=9, message_id=None)) + + channel = MagicMock() + channel.send = AsyncMock(side_effect=discord.HTTPException(MagicMock(), "no perms")) + bot.get_channel = MagicMock(return_value=channel) + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await close_anonymous_vote(bot, 9) + + channel.send.assert_awaited_once() + + @pytest.mark.asyncio + async def test_a_failed_edit_falls_back_to_a_new_message(self, bot): + session = _close_db(loaded=_loaded_session(session_id=9, message_id=666)) + + channel = MagicMock() + channel.fetch_message = AsyncMock(side_effect=discord.HTTPException(MagicMock(), "gone")) + channel.send = AsyncMock() + bot.get_channel = MagicMock(return_value=channel) + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await close_anonymous_vote(bot, 9) + + channel.send.assert_awaited_once() + assert channel.send.await_args.kwargs["embed"] is not None + + @pytest.mark.asyncio + async def test_gives_up_when_the_channel_cannot_be_resolved(self, bot): + session = _close_db(loaded=_loaded_session(session_id=9)) + + bot.get_channel = MagicMock(return_value=None) + bot.fetch_channel = AsyncMock(side_effect=discord.HTTPException(MagicMock(), "nope")) + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await close_anonymous_vote(bot, 9) + + bot.fetch_channel.assert_awaited_once_with(555) + + @pytest.mark.asyncio + async def test_claims_the_close_with_a_conditional_update(self, bot): + """on_ready re-schedules a close on every reconnect, so closes pile up and race.""" + session = _close_db(loaded=_loaded_session(session_id=9)) + + message = _make_poll_message() + channel = MagicMock() + channel.fetch_message = AsyncMock(return_value=message) + bot.get_channel = MagicMock(return_value=channel) + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await close_anonymous_vote(bot, 9) + + stmt = session.execute.await_args.args[0] + assert stmt.is_update + compiled = str(stmt.compile(dialect=_mysql_dialect())) + assert "closed is false" in compiled.lower() + message.edit.assert_awaited_once() + + @pytest.mark.asyncio + async def test_a_losing_concurrent_close_publishes_nothing(self, bot): + """Second coroutine's UPDATE matches no row, so it must not post results twice.""" + session = _close_db(rowcount=0, loaded=_loaded_session(session_id=9)) + bot.get_channel = MagicMock() + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await close_anonymous_vote(bot, 9) + + bot.get_channel.assert_not_called() + + @pytest.mark.asyncio + async def test_ignores_a_close_it_did_not_claim(self, bot): + """Already closed and already deleted are one case: the UPDATE matched no row.""" + session = _close_db(rowcount=0) + bot.get_channel = MagicMock() + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await close_anonymous_vote(bot, 9) + + bot.get_channel.assert_not_called() + + @pytest.mark.asyncio + async def test_publishes_nothing_when_the_row_vanishes_after_the_claim(self, bot): + """Claimed the close, then the re-SELECT found nothing; must not crash the task.""" + session = _close_db(rowcount=1, loaded=None) + bot.get_channel = MagicMock() + + with patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)): + await close_anonymous_vote(bot, 9) + + bot.get_channel.assert_not_called() + + +class TestScheduleVoteClose: + @pytest.mark.asyncio + async def test_converts_epoch_seconds_to_datetime(self, bot): + with ( + patch("src.views.anonymous_vote.schedule") as mock_schedule, + patch("src.views.anonymous_vote.close_anonymous_vote", new_callable=MagicMock), + ): + schedule_vote_close(bot, 7, 1800000000) + + mock_schedule.assert_called_once() + assert mock_schedule.call_args[0][1] == datetime.fromtimestamp(1800000000) + + +class TestRegisterAnonymousVoteViews: + @pytest.mark.asyncio + async def test_schedules_close_for_future_session(self, bot): + vote_session = _make_session(session_id=3, closes_at=1800000000) + vote_session.candidates = [_make_candidate(session_id=3)] + + scalars_result = MagicMock() + scalars_result.all.return_value = [vote_session] + session = AsyncMock() + session.scalars = AsyncMock(return_value=scalars_result) + + with ( + patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)), + patch("src.views.anonymous_vote.schedule_vote_close") as mock_schedule_close, + ): + await register_anonymous_vote_views(bot) + + mock_schedule_close.assert_called_once_with(bot, 3, 1800000000) + bot.add_view.assert_called_once() + + @pytest.mark.asyncio + async def test_closes_expired_session_immediately(self, bot): + vote_session = _make_session(session_id=4, closes_at=1000000000) + vote_session.candidates = [] + + scalars_result = MagicMock() + scalars_result.all.return_value = [vote_session] + session = AsyncMock() + session.scalars = AsyncMock(return_value=scalars_result) + + with ( + patch("src.views.anonymous_vote.AsyncSessionLocal", return_value=_session_ctx(session)), + patch("src.views.anonymous_vote.schedule_vote_close") as mock_schedule_close, + patch("src.views.anonymous_vote.close_anonymous_vote") as mock_close, + ): + await register_anonymous_vote_views(bot) + + mock_schedule_close.assert_not_called() + mock_close.assert_called_once_with(bot, 4) + + +class TestBallotCountsByCandidate: + def test_counts_each_candidates_ballots_separately(self): + ballots = [ + _make_ballot(candidate_id=1), + _make_ballot(candidate_id=2), + _make_ballot(candidate_id=1), + _make_ballot(candidate_id=3), + _make_ballot(candidate_id=1), + _make_ballot(candidate_id=2), + ] + + counts = _ballot_counts_by_candidate(ballots) + + assert counts[1] == 3 + assert counts[2] == 2 + assert counts[3] == 1 + + def test_a_candidate_with_no_ballots_is_absent_and_reads_as_zero(self): + counts = _ballot_counts_by_candidate([_make_ballot(candidate_id=1)]) + + assert 2 not in counts + assert counts.get(2, 0) == 0 + assert counts.get(1, 0) == 1 + + def test_no_ballots_counts_nothing(self): + assert _ballot_counts_by_candidate([]) == {} + + +class TestFormatNomineeLine: + def test_no_ballots_renders_the_bare_nominee_line(self): + line = _format_nominee_line(_make_candidate(candidate_id=3, name="Carol"), 0) + + assert line == "• **Carol** (`103`)" + + def test_a_single_ballot_renders_one_box(self): + line = _format_nominee_line(_make_candidate(candidate_id=3, name="Carol"), 1) + + assert line == "• **Carol** (`103`) ⬜" + + def test_several_ballots_render_one_box_each(self): + line = _format_nominee_line(_make_candidate(candidate_id=3, name="Carol"), 4) + + assert line == "• **Carol** (`103`) ⬜⬜⬜⬜" + + def test_exactly_the_maximum_is_not_marked_as_truncated(self): + line = _format_nominee_line(_make_candidate(), MAX_ACTIVITY_BOXES_PER_NOMINEE) + + assert line.count(VOTE_ACTIVITY_BOX) == 40 + assert "…" not in line + + def test_one_over_the_maximum_caps_the_boxes_and_marks_truncation(self): + line = _format_nominee_line(_make_candidate(), MAX_ACTIVITY_BOXES_PER_NOMINEE + 1) + + assert line.count(VOTE_ACTIVITY_BOX) == 40 + assert line.endswith("…") + + def test_far_over_the_maximum_still_caps_at_the_maximum(self): + line = _format_nominee_line(_make_candidate(), 500) + + assert line.count(VOTE_ACTIVITY_BOX) == 40 + assert line.endswith("…") + + +class TestBuildResultsEmbed: + def test_tallies_approvals_and_rejections_per_nominee(self): + candidates = [ + _make_candidate(candidate_id=1, name="Alice"), + _make_candidate(candidate_id=2, name="Bob"), + ] + ballots = [ + _make_ballot(candidate_id=1, choice=CHOICE_APPROVE), + _make_ballot(candidate_id=1, choice=CHOICE_APPROVE), + _make_ballot(candidate_id=1, choice=CHOICE_REJECT), + _make_ballot(candidate_id=2, choice=CHOICE_REJECT), + _make_ballot(candidate_id=2, choice=CHOICE_REJECT), + ] + + embed = build_results_embed(_make_session(), candidates, ballots) + + assert "• **Alice** — ✓ 2 / ✗ 1" in embed.description + assert "• **Bob** — ✓ 0 / ✗ 2" in embed.description + + def test_a_nominee_with_no_ballots_tallies_zero_both_ways(self): + candidates = [ + _make_candidate(candidate_id=1, name="Alice"), + _make_candidate(candidate_id=2, name="Bob"), + ] + ballots = [_make_ballot(candidate_id=1, choice=CHOICE_APPROVE)] + + embed = build_results_embed(_make_session(), candidates, ballots) + + assert "• **Alice** — ✓ 1 / ✗ 0" in embed.description + assert "• **Bob** — ✓ 0 / ✗ 0" in embed.description + + def test_ignores_a_ballot_whose_choice_is_neither_approve_nor_reject(self): + candidate = _make_candidate(candidate_id=1, name="Alice") + ballots = [ + _make_ballot(candidate_id=1, choice=CHOICE_APPROVE), + _make_ballot(candidate_id=1, choice="abstain"), + _make_ballot(candidate_id=1, choice=""), + ] + + embed = build_results_embed(_make_session(), [candidate], ballots) + + assert "• **Alice** — ✓ 1 / ✗ 0" in embed.description