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..3ee0ced --- /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.BIGINT(display_width=18), 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..7e0183a 100644 --- a/src/cmds/core/admin.py +++ b/src/cmds/core/admin.py @@ -1,19 +1,156 @@ """Admin command group for bot administration commands.""" import logging +import re +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 delete, 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] +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]: + """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)) + + +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): @@ -149,6 +286,54 @@ 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, max_length=200), + ) -> Interaction | WebhookMessage: + """Start an anonymous vote; tallies reveal automatically when duration ends.""" + # 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: + nominees = await _resolve_nominees(ctx, members) + except ValueError as exc: + return await ctx.respond(str(exc), ephemeral=True) + + 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) + 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 .", + 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..c8db4bf --- /dev/null +++ b/src/database/models/anonymous_vote.py @@ -0,0 +1,71 @@ +# flake8: noqa: D101 +from sqlalchemy import Boolean, ForeignKey, Integer, String, UniqueConstraint +from sqlalchemy.dialects.mysql import BIGINT, TEXT +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[int] = mapped_column(BIGINT(18), 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..f3915fd --- /dev/null +++ b/src/views/anonymous_vote.py @@ -0,0 +1,461 @@ +"""Persistent UI and helpers for anonymous multi-user votes.""" + +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, update +from sqlalchemy.dialects.mysql import insert +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 +# 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: + 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** 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." + ), + inline=False, + ) + embed.add_field( + name="Closes", + value=f" ()", + 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 + + +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: pick a nominee, then vote on a private ballot.""" + + 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 + + nominee_select = Select( + placeholder="Select a nominee to vote on", + 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(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) + 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]) + 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: + return None, "This poll is closed." + + candidate = await session.get(AnonymousVoteCandidate, candidate_id) + if not candidate or candidate.session_id != self.session_id: + return None, "Unknown nominee." + + return candidate.display_name, None + + +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) + return + + 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 + + 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: + return None, None, "This poll is closed." + + candidate = await session.get(AnonymousVoteCandidate, self.candidate_id) + if not candidate or candidate.session_id != self.session_id: + 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( + 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 + ) + + confirmation = _ballot_confirmation( + choice, display_name, result.rowcount == UPSERT_ROWCOUNT_UPDATED + ) + return confirmation, poll_embed, None + + 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_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: + # 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( + selectinload(AnonymousVoteSession.candidates), + selectinload(AnonymousVoteSession.ballots), + ) + ) + if vote_session is None: + logger.warning("Anonymous vote session %s vanished after its close was claimed.", session_id) + return None + + candidates = list(vote_session.candidates) + 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: + return + + view = AnonymousVoteView(session_id, bot, candidates) + for item in view.children: + item.disabled = True + + 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_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), datetime.fromtimestamp(closes_at_ts)) + ) + + +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 = 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: + 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/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 0260bac..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,6 +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.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