aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--bot/bot.py12
-rw-r--r--bot/cogs/antimalware.py12
-rw-r--r--bot/cogs/error_handler.py6
-rw-r--r--bot/cogs/help_channels.py201
-rw-r--r--bot/cogs/information.py8
-rw-r--r--bot/cogs/snekbox.py10
-rw-r--r--bot/cogs/utils.py4
-rw-r--r--bot/cogs/verification.py12
-rw-r--r--bot/constants.py2
-rw-r--r--bot/decorators.py85
-rw-r--r--bot/resources/tags/free.md5
-rw-r--r--config-default.yml4
-rw-r--r--tests/bot/cogs/test_cogs.py4
-rw-r--r--tests/bot/cogs/test_information.py6
-rw-r--r--tests/bot/test_decorators.py147
-rw-r--r--tests/helpers.py23
16 files changed, 405 insertions, 136 deletions
diff --git a/bot/bot.py b/bot/bot.py
index 027d8d2a3..a85a22aa9 100644
--- a/bot/bot.py
+++ b/bot/bot.py
@@ -7,6 +7,7 @@ from typing import Optional
import aiohttp
import discord
from discord.ext import commands
+from sentry_sdk import push_scope
from bot import DEBUG_MODE, api, constants
from bot.async_stats import AsyncStatsClient
@@ -155,3 +156,14 @@ class Bot(commands.Bot):
gateway event before giving up and thus not populating the cache for unavailable guilds.
"""
await self._guild_available.wait()
+
+ async def on_error(self, event: str, *args, **kwargs) -> None:
+ """Log errors raised in event listeners rather than printing them to stderr."""
+ self.stats.incr(f"errors.event.{event}")
+
+ with push_scope() as scope:
+ scope.set_tag("event", event)
+ scope.set_extra("args", args)
+ scope.set_extra("kwargs", kwargs)
+
+ log.exception(f"Unhandled exception in {event}.")
diff --git a/bot/cogs/antimalware.py b/bot/cogs/antimalware.py
index 79bf486a4..66b5073e8 100644
--- a/bot/cogs/antimalware.py
+++ b/bot/cogs/antimalware.py
@@ -38,6 +38,18 @@ class AntiMalware(Cog):
"It looks like you tried to attach a Python file - "
f"please use a code-pasting service such as {URLs.site_schema}{URLs.site_paste}"
)
+ elif ".txt" in extensions_blocked:
+ # Work around Discord AutoConversion of messages longer than 2000 chars to .txt
+ cmd_channel = self.bot.get_channel(Channels.bot_commands)
+ embed.description = (
+ "**Uh-oh!** It looks like your message got zapped by our spam filter. "
+ "We currently don't allow `.txt` attachments, so here are some tips to help you travel safely: \n\n"
+ "• If you attempted to send a message longer than 2000 characters, try shortening your message "
+ "to fit within the character limit or use a pasting service (see below) \n\n"
+ "• If you tried to show someone your code, you can use codeblocks \n(run `!code-blocks` in "
+ f"{cmd_channel.mention} for more information) or use a pasting service like: "
+ f"\n\n{URLs.site_schema}{URLs.site_paste}"
+ )
elif extensions_blocked:
whitelisted_types = ', '.join(AntiMalwareConfig.whitelist)
meta_channel = self.bot.get_channel(Channels.meta)
diff --git a/bot/cogs/error_handler.py b/bot/cogs/error_handler.py
index dae283c6a..b2f4c59f6 100644
--- a/bot/cogs/error_handler.py
+++ b/bot/cogs/error_handler.py
@@ -9,7 +9,7 @@ from bot.api import ResponseCodeError
from bot.bot import Bot
from bot.constants import Channels
from bot.converters import TagNameConverter
-from bot.decorators import InChannelCheckFailure
+from bot.decorators import InWhitelistCheckFailure
log = logging.getLogger(__name__)
@@ -202,7 +202,7 @@ class ErrorHandler(Cog):
* BotMissingRole
* BotMissingAnyRole
* NoPrivateMessage
- * InChannelCheckFailure
+ * InWhitelistCheckFailure
"""
bot_missing_errors = (
errors.BotMissingPermissions,
@@ -215,7 +215,7 @@ class ErrorHandler(Cog):
await ctx.send(
f"Sorry, it looks like I don't have the permissions or roles I need to do that."
)
- elif isinstance(e, (InChannelCheckFailure, errors.NoPrivateMessage)):
+ elif isinstance(e, (InWhitelistCheckFailure, errors.NoPrivateMessage)):
ctx.bot.stats.incr("errors.wrong_channel_or_dm_error")
await ctx.send(e)
diff --git a/bot/cogs/help_channels.py b/bot/cogs/help_channels.py
index a61f30deb..b714a1642 100644
--- a/bot/cogs/help_channels.py
+++ b/bot/cogs/help_channels.py
@@ -10,6 +10,7 @@ from datetime import datetime
from pathlib import Path
import discord
+import discord.abc
from discord.ext import commands
from bot import constants
@@ -21,6 +22,7 @@ log = logging.getLogger(__name__)
ASKING_GUIDE_URL = "https://pythondiscord.com/pages/asking-good-questions/"
MAX_CHANNELS_PER_CATEGORY = 50
+EXCLUDED_CHANNELS = (constants.Channels.how_to_get_help,)
AVAILABLE_TOPIC = """
This channel is available. Feel free to ask a question in order to claim this channel!
@@ -39,8 +41,9 @@ channels in the Help: Available category.
AVAILABLE_MSG = f"""
This help channel is now **available**, which means that you can claim it by simply typing your \
question into it. Once claimed, the channel will move into the **Python Help: Occupied** category, \
-and will be yours until it has been inactive for {constants.HelpChannels.idle_minutes} minutes. When \
-that happens, it will be set to **dormant** and moved into the **Help: Dormant** category.
+and will be yours until it has been inactive for {constants.HelpChannels.idle_minutes} minutes or \
+is closed manually with `!close`. When that happens, it will be set to **dormant** and moved into \
+the **Help: Dormant** category.
You may claim a new channel once every {constants.HelpChannels.claim_minutes} minutes. If you \
currently cannot send a message in this channel, it means you are on cooldown and need to wait.
@@ -66,6 +69,8 @@ IN_USE_ANSWERED_EMOJI = "⌛"
IN_USE_UNANSWERED_EMOJI = "⏳"
NAME_SEPARATOR = "｜"
+CoroutineFunc = t.Callable[..., t.Coroutine]
+
class TaskData(t.NamedTuple):
"""Data for a scheduled task."""
@@ -89,12 +94,15 @@ class HelpChannels(Scheduler, commands.Cog):
* If there are no more dormant channels, the bot will automatically create a new one
* If there are no dormant channels to move, helpers will be notified (see `notify()`)
* When a channel becomes available, the dormant embed will be edited to show `AVAILABLE_MSG`
+ * User can only claim a channel at an interval `constants.HelpChannels.claim_minutes`
+ * To keep track of cooldowns, user which claimed a channel will have a temporary role
In Use Category
* Contains all channels which are occupied by someone needing help
* Channel moves to dormant category after `constants.HelpChannels.idle_minutes` of being idle
* Command can prematurely mark a channel as dormant
+ * Channel claimant is allowed to use the command
* Allowed roles for the command are configurable with `constants.HelpChannels.cmd_whitelist`
* When a channel becomes dormant, an embed with `DORMANT_MSG` will be sent
@@ -217,8 +225,8 @@ class HelpChannels(Scheduler, commands.Cog):
return role_check
- @commands.command(name="dormant", aliases=["close"], enabled=False)
- async def dormant_command(self, ctx: commands.Context) -> None:
+ @commands.command(name="close", aliases=["dormant"], enabled=False)
+ async def close_command(self, ctx: commands.Context) -> None:
"""
Make the current in-use help channel dormant.
@@ -226,14 +234,15 @@ class HelpChannels(Scheduler, commands.Cog):
delete the message that invoked this,
and reset the send permissions cooldown for the user who started the session.
"""
- log.trace("dormant command invoked; checking if the channel is in-use.")
+ log.trace("close command invoked; checking if the channel is in-use.")
if ctx.channel.category == self.in_use_category:
if await self.dormant_check(ctx):
with suppress(KeyError):
del self.help_channel_claimants[ctx.channel]
- with suppress(discord.errors.HTTPException, discord.errors.NotFound):
- await self.reset_claimant_send_permission(ctx.channel)
+ await self.remove_cooldown_role(ctx.author)
+ # Ignore missing task when cooldown has passed but the channel still isn't dormant.
+ self.cancel_task(ctx.author.id, ignore_missing=True)
await self.move_to_dormant(ctx.channel, "command")
self.cancel_task(ctx.channel.id)
@@ -271,18 +280,23 @@ class HelpChannels(Scheduler, commands.Cog):
log.trace(f"The clean name for `{channel}` is `{name}`")
except ValueError:
# If, for some reason, the channel name does not contain "help-" fall back gracefully
- log.info(f"Can't get clean name as `{channel}` does not follow the `{prefix}` naming convention.")
+ log.info(f"Can't get clean name because `{channel}` isn't prefixed by `{prefix}`.")
name = channel.name
return name
+ @staticmethod
+ def is_excluded_channel(channel: discord.abc.GuildChannel) -> bool:
+ """Check if a channel should be excluded from the help channel system."""
+ return not isinstance(channel, discord.TextChannel) or channel.id in EXCLUDED_CHANNELS
+
def get_category_channels(self, category: discord.CategoryChannel) -> t.Iterable[discord.TextChannel]:
"""Yield the text channels of the `category` in an unsorted manner."""
log.trace(f"Getting text channels in the category '{category}' ({category.id}).")
# This is faster than using category.channels because the latter sorts them.
for channel in self.bot.get_guild(constants.Guild.id).channels:
- if channel.category_id == category.id and isinstance(channel, discord.TextChannel):
+ if channel.category_id == category.id and not self.is_excluded_channel(channel):
yield channel
@staticmethod
@@ -400,7 +414,7 @@ class HelpChannels(Scheduler, commands.Cog):
# The ready event wasn't used because channels could change categories between the time
# the command is invoked and the cog is ready (e.g. if move_idle_channel wasn't called yet).
# This may confuse users. So would potentially long delays for the cog to become ready.
- self.dormant_command.enabled = True
+ self.close_command.enabled = True
await self.init_available()
@@ -419,6 +433,11 @@ class HelpChannels(Scheduler, commands.Cog):
self.bot.stats.gauge("help.total.available", total_available)
self.bot.stats.gauge("help.total.dormant", total_dormant)
+ @staticmethod
+ def is_claimant(member: discord.Member) -> bool:
+ """Return True if `member` has the 'Help Cooldown' role."""
+ return any(constants.Roles.help_cooldown == role.id for role in member.roles)
+
def is_dormant_message(self, message: t.Optional[discord.Message]) -> bool:
"""Return True if the contents of the `message` match `DORMANT_MSG`."""
if not message or not message.embeds:
@@ -466,6 +485,45 @@ class HelpChannels(Scheduler, commands.Cog):
self.schedule_task(channel.id, data)
+ async def move_to_bottom_position(self, channel: discord.TextChannel, category_id: int, **options) -> None:
+ """
+ Move the `channel` to the bottom position of `category` and edit channel attributes.
+
+ To ensure "stable sorting", we use the `bulk_channel_update` endpoint and provide the current
+ positions of the other channels in the category as-is. This should make sure that the channel
+ really ends up at the bottom of the category.
+
+ If `options` are provided, the channel will be edited after the move is completed. This is the
+ same order of operations that `discord.TextChannel.edit` uses. For information on available
+ options, see the documention on `discord.TextChannel.edit`. While possible, position-related
+ options should be avoided, as it may interfere with the category move we perform.
+ """
+ # Get a fresh copy of the category from the bot to avoid the cache mismatch issue we had.
+ category = await self.try_get_channel(category_id)
+
+ payload = [{"id": c.id, "position": c.position} for c in category.channels]
+
+ # Calculate the bottom position based on the current highest position in the category. If the
+ # category is currently empty, we simply use the current position of the channel to avoid making
+ # unnecessary changes to positions in the guild.
+ bottom_position = payload[-1]["position"] + 1 if payload else channel.position
+
+ payload.append(
+ {
+ "id": channel.id,
+ "position": bottom_position,
+ "parent_id": category.id,
+ "lock_permissions": True,
+ }
+ )
+
+ # We use d.py's method to ensure our request is processed by d.py's rate limit manager
+ await self.bot.http.bulk_channel_update(category.guild.id, payload)
+
+ # Now that the channel is moved, we can edit the other attributes
+ if options:
+ await channel.edit(**options)
+
async def move_to_available(self) -> None:
"""Make a channel available."""
log.trace("Making a channel available.")
@@ -477,18 +535,13 @@ class HelpChannels(Scheduler, commands.Cog):
log.trace(f"Moving #{channel} ({channel.id}) to the Available category.")
- await channel.edit(
+ await self.move_to_bottom_position(
+ channel=channel,
+ category_id=constants.Categories.help_available,
name=f"{AVAILABLE_EMOJI}{NAME_SEPARATOR}{self.get_clean_channel_name(channel)}",
- category=self.available_category,
- sync_permissions=True,
topic=AVAILABLE_TOPIC,
)
- log.trace(
- f"Ensuring that all channels in `{self.available_category}` have "
- f"synchronized permissions after moving `{channel}` into it."
- )
- await self.ensure_permissions_synchronization(self.available_category)
self.report_stats()
async def move_to_dormant(self, channel: discord.TextChannel, caller: str) -> None:
@@ -499,10 +552,10 @@ class HelpChannels(Scheduler, commands.Cog):
"""
log.info(f"Moving #{channel} ({channel.id}) to the Dormant category.")
- await channel.edit(
+ await self.move_to_bottom_position(
+ channel=channel,
+ category_id=constants.Categories.help_dormant,
name=self.get_clean_channel_name(channel),
- category=self.dormant_category,
- sync_permissions=True,
topic=DORMANT_TOPIC,
)
@@ -533,10 +586,10 @@ class HelpChannels(Scheduler, commands.Cog):
"""Make a channel in-use and schedule it to be made dormant."""
log.info(f"Moving #{channel} ({channel.id}) to the In Use category.")
- await channel.edit(
+ await self.move_to_bottom_position(
+ channel=channel,
+ category_id=constants.Categories.help_in_use,
name=f"{IN_USE_UNANSWERED_EMOJI}{NAME_SEPARATOR}{self.get_clean_channel_name(channel)}",
- category=self.in_use_category,
- sync_permissions=True,
topic=IN_USE_TOPIC,
)
@@ -596,10 +649,11 @@ class HelpChannels(Scheduler, commands.Cog):
async def check_for_answer(self, message: discord.Message) -> None:
"""Checks for whether new content in a help channel comes from non-claimants."""
channel = message.channel
- log.trace(f"Checking if #{channel} ({channel.id}) has been answered.")
# Confirm the channel is an in use help channel
if self.is_in_category(channel, constants.Categories.help_in_use):
+ log.trace(f"Checking if #{channel} ({channel.id}) has been answered.")
+
# Check if there is an entry in unanswered (does not persist across restarts)
if channel.id in self.unanswered:
claimant_id = self.help_channel_claimants[channel].id
@@ -624,8 +678,8 @@ class HelpChannels(Scheduler, commands.Cog):
await self.check_for_answer(message)
- if not self.is_in_category(channel, constants.Categories.help_available):
- return # Ignore messages outside the Available category.
+ if not self.is_in_category(channel, constants.Categories.help_available) or self.is_excluded_channel(channel):
+ return # Ignore messages outside the Available category or in excluded channels.
log.trace("Waiting for the cog to be ready before processing messages.")
await self.ready.wait()
@@ -641,6 +695,7 @@ class HelpChannels(Scheduler, commands.Cog):
)
return
+ log.info(f"Channel #{channel} was claimed by `{message.author.id}`.")
await self.move_to_in_use(channel)
await self.revoke_send_permissions(message.author)
# Add user with channel for dormant check.
@@ -658,67 +713,49 @@ class HelpChannels(Scheduler, commands.Cog):
# be put in the queue.
await self.move_to_available()
- @staticmethod
- async def ensure_permissions_synchronization(category: discord.CategoryChannel) -> None:
- """
- Ensure that all channels in the `category` have their permissions synchronized.
-
- This method mitigates an issue we have yet to find the cause for: Every so often, a channel in the
- `Help: Available` category gets in a state in which it will no longer synchronizes its permissions
- with the category. To prevent that, we iterate over the channels in the category and edit the channels
- that are observed to be in such a state. If no "out of sync" channels are observed, this method will
- not make API calls and should be fairly inexpensive to run.
- """
- for channel in category.channels:
- if not channel.permissions_synced:
- log.info(f"The permissions of channel `{channel}` were out of sync with category `{category}`.")
- await channel.edit(sync_permissions=True)
-
- async def update_category_permissions(
- self, category: discord.CategoryChannel, member: discord.Member, **permissions
- ) -> None:
- """
- Update the permissions of the given `member` for the given `category` with `permissions` passed.
-
- After updating the permissions for the member in the category, this helper function will call the
- `ensure_permissions_synchronization` method to ensure that all channels are still synchronizing their
- permissions with the category. It's currently unknown why some channels get "out of sync", but this
- hopefully mitigates the issue.
- """
- log.trace(f"Updating permissions for `{member}` in `{category}` with {permissions}.")
- await category.set_permissions(member, **permissions)
-
- log.trace(f"Ensuring that all channels in `{category}` are synchronized after permissions update.")
- await self.ensure_permissions_synchronization(category)
-
async def reset_send_permissions(self) -> None:
- """Reset send permissions for members with it set to False in the Available category."""
+ """Reset send permissions in the Available category for claimants."""
log.trace("Resetting send permissions in the Available category.")
+ guild = self.bot.get_guild(constants.Guild.id)
- for member, overwrite in self.available_category.overwrites.items():
- if isinstance(member, discord.Member) and overwrite.send_messages is False:
- log.trace(f"Resetting send permissions for {member} ({member.id}).")
+ # TODO: replace with a persistent cache cause checking every member is quite slow
+ for member in guild.members:
+ if self.is_claimant(member):
+ await self.remove_cooldown_role(member)
- # We don't use the permissions helper function here as we may have to reset multiple overwrites
- # and we don't want to enforce the permissions synchronization in each iteration.
- await self.available_category.set_permissions(member, overwrite=None)
+ async def add_cooldown_role(self, member: discord.Member) -> None:
+ """Add the help cooldown role to `member`."""
+ log.trace(f"Adding cooldown role for {member} ({member.id}).")
+ await self._change_cooldown_role(member, member.add_roles)
- log.trace(f"Ensuring channels in `Help: Available` are synchronized after permissions reset.")
- await self.ensure_permissions_synchronization(self.available_category)
+ async def remove_cooldown_role(self, member: discord.Member) -> None:
+ """Remove the help cooldown role from `member`."""
+ log.trace(f"Removing cooldown role for {member} ({member.id}).")
+ await self._change_cooldown_role(member, member.remove_roles)
- async def reset_claimant_send_permission(self, channel: discord.TextChannel) -> None:
- """Reset send permissions in the Available category for the help `channel` claimant."""
- log.trace(f"Attempting to find claimant for #{channel.name} ({channel.id}).")
- try:
- member = self.help_channel_claimants[channel]
- except KeyError:
- log.trace(f"Channel #{channel.name} ({channel.id}) not in claimant cache, permissions unchanged.")
+ async def _change_cooldown_role(self, member: discord.Member, coro_func: CoroutineFunc) -> None:
+ """
+ Change `member`'s cooldown role via awaiting `coro_func` and handle errors.
+
+ `coro_func` is intended to be `discord.Member.add_roles` or `discord.Member.remove_roles`.
+ """
+ guild = self.bot.get_guild(constants.Guild.id)
+ role = guild.get_role(constants.Roles.help_cooldown)
+ if role is None:
+ log.warning(f"Help cooldown role ({constants.Roles.help_cooldown}) could not be found!")
return
- log.trace(f"Resetting send permissions for {member} ({member.id}).")
- await self.update_category_permissions(self.available_category, member, overwrite=None)
- # Ignore missing task when claim cooldown has passed but the channel still isn't dormant.
- self.cancel_task(member.id, ignore_missing=True)
+ try:
+ await coro_func(role)
+ except discord.NotFound:
+ log.debug(f"Failed to change role for {member} ({member.id}): member not found")
+ except discord.Forbidden:
+ log.debug(
+ f"Forbidden to change role for {member} ({member.id}); "
+ f"possibly due to role hierarchy"
+ )
+ except discord.HTTPException as e:
+ log.error(f"Failed to change role for {member} ({member.id}): {e.status} {e.code}")
async def revoke_send_permissions(self, member: discord.Member) -> None:
"""
@@ -731,14 +768,14 @@ class HelpChannels(Scheduler, commands.Cog):
f"Revoking {member}'s ({member.id}) send message permissions in the Available category."
)
- await self.update_category_permissions(self.available_category, member, send_messages=False)
+ await self.add_cooldown_role(member)
# Cancel the existing task, if any.
# Would mean the user somehow bypassed the lack of permissions (e.g. user is guild owner).
self.cancel_task(member.id, ignore_missing=True)
timeout = constants.HelpChannels.claim_minutes * 60
- callback = self.update_category_permissions(self.available_category, member, overwrite=None)
+ callback = self.remove_cooldown_role(member)
log.trace(f"Scheduling {member}'s ({member.id}) send message permissions to be reinstated.")
self.schedule_task(member.id, TaskData(timeout, callback))
diff --git a/bot/cogs/information.py b/bot/cogs/information.py
index 7921a4932..ef2f308ca 100644
--- a/bot/cogs/information.py
+++ b/bot/cogs/information.py
@@ -12,7 +12,7 @@ from discord.utils import escape_markdown
from bot import constants
from bot.bot import Bot
-from bot.decorators import InChannelCheckFailure, in_channel, with_role
+from bot.decorators import InWhitelistCheckFailure, in_whitelist, with_role
from bot.pagination import LinePaginator
from bot.utils.checks import cooldown_with_role_bypass, with_role_check
from bot.utils.time import time_since
@@ -152,7 +152,7 @@ class Information(Cog):
# Non-staff may only do this in #bot-commands
if not with_role_check(ctx, *constants.STAFF_ROLES):
if not ctx.channel.id == constants.Channels.bot_commands:
- raise InChannelCheckFailure(constants.Channels.bot_commands)
+ raise InWhitelistCheckFailure(constants.Channels.bot_commands)
embed = await self.create_user_embed(ctx, user)
@@ -206,7 +206,7 @@ class Information(Cog):
description="\n\n".join(description)
)
- embed.set_thumbnail(url=user.avatar_url_as(format="png"))
+ embed.set_thumbnail(url=user.avatar_url_as(static_format="png"))
embed.colour = user.top_role.colour if roles else Colour.blurple()
return embed
@@ -331,7 +331,7 @@ class Information(Cog):
@cooldown_with_role_bypass(2, 60 * 3, BucketType.member, bypass_roles=constants.STAFF_ROLES)
@group(invoke_without_command=True)
- @in_channel(constants.Channels.bot_commands, bypass_roles=constants.STAFF_ROLES)
+ @in_whitelist(channels=(constants.Channels.bot_commands,), roles=constants.STAFF_ROLES)
async def raw(self, ctx: Context, *, message: Message, json: bool = False) -> None:
"""Shows information about the raw API response."""
# I *guess* it could be deleted right as the command is invoked but I felt like it wasn't worth handling
diff --git a/bot/cogs/snekbox.py b/bot/cogs/snekbox.py
index 315383b12..8d4688114 100644
--- a/bot/cogs/snekbox.py
+++ b/bot/cogs/snekbox.py
@@ -12,8 +12,8 @@ from discord import HTTPException, Message, NotFound, Reaction, User
from discord.ext.commands import Cog, Context, command, guild_only
from bot.bot import Bot
-from bot.constants import Channels, Roles, URLs
-from bot.decorators import in_channel
+from bot.constants import Categories, Channels, Roles, URLs
+from bot.decorators import in_whitelist
from bot.utils.messages import wait_for_deletion
log = logging.getLogger(__name__)
@@ -38,6 +38,10 @@ RAW_CODE_REGEX = re.compile(
)
MAX_PASTE_LEN = 1000
+
+# `!eval` command whitelists
+EVAL_CHANNELS = (Channels.bot_commands, Channels.esoteric)
+EVAL_CATEGORIES = (Categories.help_available, Categories.help_in_use)
EVAL_ROLES = (Roles.helpers, Roles.moderators, Roles.admins, Roles.owners, Roles.python_community, Roles.partners)
SIGKILL = 9
@@ -265,7 +269,7 @@ class Snekbox(Cog):
@command(name="eval", aliases=("e",))
@guild_only()
- @in_channel(Channels.bot_commands, hidden_channels=(Channels.esoteric,), bypass_roles=EVAL_ROLES)
+ @in_whitelist(channels=EVAL_CHANNELS, categories=EVAL_CATEGORIES, roles=EVAL_ROLES)
async def eval_command(self, ctx: Context, *, code: str = None) -> None:
"""
Run Python code and get the results.
diff --git a/bot/cogs/utils.py b/bot/cogs/utils.py
index 3ed471bbf..8023eb962 100644
--- a/bot/cogs/utils.py
+++ b/bot/cogs/utils.py
@@ -13,7 +13,7 @@ from discord.ext.commands import BadArgument, Cog, Context, command
from bot.bot import Bot
from bot.constants import Channels, MODERATION_ROLES, Mention, STAFF_ROLES
-from bot.decorators import in_channel, with_role
+from bot.decorators import in_whitelist, with_role
from bot.utils.time import humanize_delta
log = logging.getLogger(__name__)
@@ -118,7 +118,7 @@ class Utils(Cog):
await ctx.message.channel.send(embed=pep_embed)
@command()
- @in_channel(Channels.bot_commands, bypass_roles=STAFF_ROLES)
+ @in_whitelist(channels=(Channels.bot_commands,), roles=STAFF_ROLES)
async def charinfo(self, ctx: Context, *, characters: str) -> None:
"""Shows you information on up to 25 unicode characters."""
match = re.match(r"<(a?):(\w+):(\d+)>", characters)
diff --git a/bot/cogs/verification.py b/bot/cogs/verification.py
index b0a493e68..388b7a338 100644
--- a/bot/cogs/verification.py
+++ b/bot/cogs/verification.py
@@ -9,7 +9,7 @@ from discord.ext.commands import Cog, Context, command
from bot import constants
from bot.bot import Bot
from bot.cogs.moderation import ModLog
-from bot.decorators import InChannelCheckFailure, in_channel, without_role
+from bot.decorators import InWhitelistCheckFailure, in_whitelist, without_role
from bot.utils.checks import without_role_check
log = logging.getLogger(__name__)
@@ -122,7 +122,7 @@ class Verification(Cog):
@command(name='accept', aliases=('verify', 'verified', 'accepted'), hidden=True)
@without_role(constants.Roles.verified)
- @in_channel(constants.Channels.verification)
+ @in_whitelist(channels=(constants.Channels.verification,))
async def accept_command(self, ctx: Context, *_) -> None: # We don't actually care about the args
"""Accept our rules and gain access to the rest of the server."""
log.debug(f"{ctx.author} called !accept. Assigning the 'Developer' role.")
@@ -138,7 +138,7 @@ class Verification(Cog):
await ctx.message.delete()
@command(name='subscribe')
- @in_channel(constants.Channels.bot_commands)
+ @in_whitelist(channels=(constants.Channels.bot_commands,))
async def subscribe_command(self, ctx: Context, *_) -> None: # We don't actually care about the args
"""Subscribe to announcement notifications by assigning yourself the role."""
has_role = False
@@ -162,7 +162,7 @@ class Verification(Cog):
)
@command(name='unsubscribe')
- @in_channel(constants.Channels.bot_commands)
+ @in_whitelist(channels=(constants.Channels.bot_commands,))
async def unsubscribe_command(self, ctx: Context, *_) -> None: # We don't actually care about the args
"""Unsubscribe from announcement notifications by removing the role from yourself."""
has_role = False
@@ -187,8 +187,8 @@ class Verification(Cog):
# This cannot be static (must have a __func__ attribute).
async def cog_command_error(self, ctx: Context, error: Exception) -> None:
- """Check for & ignore any InChannelCheckFailure."""
- if isinstance(error, InChannelCheckFailure):
+ """Check for & ignore any InWhitelistCheckFailure."""
+ if isinstance(error, InWhitelistCheckFailure):
error.handled = True
@staticmethod
diff --git a/bot/constants.py b/bot/constants.py
index 2add028e7..a00b59505 100644
--- a/bot/constants.py
+++ b/bot/constants.py
@@ -383,6 +383,7 @@ class Channels(metaclass=YAMLGetter):
dev_log: int
esoteric: int
helpers: int
+ how_to_get_help: int
message_log: int
meta: int
mod_alerts: int
@@ -421,6 +422,7 @@ class Roles(metaclass=YAMLGetter):
announcements: int
contributors: int
core_developers: int
+ help_cooldown: int
helpers: int
jammers: int
moderators: int
diff --git a/bot/decorators.py b/bot/decorators.py
index 2d18eaa6a..2ee5879f2 100644
--- a/bot/decorators.py
+++ b/bot/decorators.py
@@ -3,7 +3,7 @@ import random
from asyncio import Lock, sleep
from contextlib import suppress
from functools import wraps
-from typing import Callable, Container, Union
+from typing import Callable, Container, Optional, Union
from weakref import WeakValueDictionary
from discord import Colour, Embed, Member
@@ -11,54 +11,79 @@ from discord.errors import NotFound
from discord.ext import commands
from discord.ext.commands import CheckFailure, Cog, Context
-from bot.constants import ERROR_REPLIES, RedirectOutput
+from bot.constants import Channels, ERROR_REPLIES, RedirectOutput
from bot.utils.checks import with_role_check, without_role_check
log = logging.getLogger(__name__)
-class InChannelCheckFailure(CheckFailure):
- """Raised when a check fails for a message being sent in a whitelisted channel."""
+class InWhitelistCheckFailure(CheckFailure):
+ """Raised when the `in_whitelist` check fails."""
- def __init__(self, *channels: int):
- self.channels = channels
- channels_str = ', '.join(f"<#{c_id}>" for c_id in channels)
+ def __init__(self, redirect_channel: Optional[int]) -> None:
+ self.redirect_channel = redirect_channel
- super().__init__(f"Sorry, but you may only use this command within {channels_str}.")
+ if redirect_channel:
+ redirect_message = f" here. Please use the <#{redirect_channel}> channel instead"
+ else:
+ redirect_message = ""
+ error_message = f"You are not allowed to use that command{redirect_message}."
+
+ super().__init__(error_message)
+
+
+def in_whitelist(
+ *,
+ channels: Container[int] = (),
+ categories: Container[int] = (),
+ roles: Container[int] = (),
+ redirect: Optional[int] = Channels.bot_commands,
-def in_channel(
- *channels: int,
- hidden_channels: Container[int] = None,
- bypass_roles: Container[int] = None
) -> Callable:
"""
- Checks that the message is in a whitelisted channel or optionally has a bypass role.
+ Check if a command was issued in a whitelisted context.
+
+ The whitelists that can be provided are:
- Hidden channels are channels which will not be displayed in the InChannelCheckFailure error
- message.
+ - `channels`: a container with channel ids for whitelisted channels
+ - `categories`: a container with category ids for whitelisted categories
+ - `roles`: a container with with role ids for whitelisted roles
+
+ If the command was invoked in a context that was not whitelisted, the member is either
+ redirected to the `redirect` channel that was passed (default: #bot-commands) or simply
+ told that they're not allowed to use this particular command (if `None` was passed).
"""
- hidden_channels = hidden_channels or []
- bypass_roles = bypass_roles or []
+ if redirect and redirect not in channels:
+ # It does not make sense for the channel whitelist to not contain the redirection
+ # channel (if applicable). That's why we add the redirection channel to the `channels`
+ # container if it's not already in it. As we allow any container type to be passed,
+ # we first create a tuple in order to safely add the redirection channel.
+ #
+ # Note: It's possible for the redirect channel to be in a whitelisted category, but
+ # there's no easy way to check that and as a channel can easily be moved in and out of
+ # categories, it's probably not wise to rely on its category in any case.
+ channels = tuple(channels) + (redirect,)
def predicate(ctx: Context) -> bool:
- """In-channel checker predicate."""
- if ctx.channel.id in channels or ctx.channel.id in hidden_channels:
- log.debug(f"{ctx.author} tried to call the '{ctx.command.name}' command. "
- f"The command was used in a whitelisted channel.")
+ """Check if a command was issued in a whitelisted context."""
+ if channels and ctx.channel.id in channels:
+ log.trace(f"{ctx.author} may use the `{ctx.command.name}` command as they are in a whitelisted channel.")
return True
- if bypass_roles:
- if any(r.id in bypass_roles for r in ctx.author.roles):
- log.debug(f"{ctx.author} tried to call the '{ctx.command.name}' command. "
- f"The command was not used in a whitelisted channel, "
- f"but the author had a role to bypass the in_channel check.")
- return True
+ # Only check the category id if we have a category whitelist and the channel has a `category_id`
+ if categories and hasattr(ctx.channel, "category_id") and ctx.channel.category_id in categories:
+ log.trace(f"{ctx.author} may use the `{ctx.command.name}` command as they are in a whitelisted category.")
+ return True
- log.debug(f"{ctx.author} tried to call the '{ctx.command.name}' command. "
- f"The in_channel check failed.")
+ # Only check the roles whitelist if we have one and ensure the author's roles attribute returns
+ # an iterable to prevent breakage in DM channels (for if we ever decide to enable commands there).
+ if roles and any(r.id in roles for r in getattr(ctx.author, "roles", ())):
+ log.trace(f"{ctx.author} may use the `{ctx.command.name}` command as they have a whitelisted role.")
+ return True
- raise InChannelCheckFailure(*channels)
+ log.trace(f"{ctx.author} may not use the `{ctx.command.name}` command within this context.")
+ raise InWhitelistCheckFailure(redirect)
return commands.check(predicate)
diff --git a/bot/resources/tags/free.md b/bot/resources/tags/free.md
new file mode 100644
index 000000000..582cca9da
--- /dev/null
+++ b/bot/resources/tags/free.md
@@ -0,0 +1,5 @@
+**We have a new help channel system!**
+
+We recently moved to a new help channel system. You can now use any channel in the **<#691405807388196926>** category to ask your question.
+
+For more information, check out [our website](https://pythondiscord.com/pages/resources/guides/help-channels/).
diff --git a/config-default.yml b/config-default.yml
index f2b0bfa9f..78a2ff853 100644
--- a/config-default.yml
+++ b/config-default.yml
@@ -132,6 +132,9 @@ guild:
meta: 429409067623251969
python_discussion: 267624335836053506
+ # Python Help: Available
+ how_to_get_help: 704250143020417084
+
# Logs
attachment_log: &ATTACH_LOG 649243850006855680
message_log: &MESSAGE_LOG 467752170159079424
@@ -201,6 +204,7 @@ guild:
roles:
announcements: 463658397560995840
contributors: 295488872404484098
+ help_cooldown: 699189276025421825
muted: &MUTED_ROLE 277914926603829249
partners: 323426753857191936
python_community: &PY_COMMUNITY_ROLE 458226413825294336
diff --git a/tests/bot/cogs/test_cogs.py b/tests/bot/cogs/test_cogs.py
index 39f6492cb..fdda59a8f 100644
--- a/tests/bot/cogs/test_cogs.py
+++ b/tests/bot/cogs/test_cogs.py
@@ -31,7 +31,7 @@ class CommandNameTests(unittest.TestCase):
def walk_modules() -> t.Iterator[ModuleType]:
"""Yield imported modules from the bot.cogs subpackage."""
def on_error(name: str) -> t.NoReturn:
- raise ImportError(name=name)
+ raise ImportError(name=name) # pragma: no cover
# The mock prevents asyncio.get_event_loop() from being called.
with mock.patch("discord.ext.tasks.loop"):
@@ -71,7 +71,7 @@ class CommandNameTests(unittest.TestCase):
for name in self.get_qualified_names(cmd):
with self.subTest(cmd=func_name, name=name):
- if name in all_names:
+ if name in all_names: # pragma: no cover
conflicts = ", ".join(all_names.get(name, ""))
self.fail(
f"Name '{name}' of the command {func_name} conflicts with {conflicts}."
diff --git a/tests/bot/cogs/test_information.py b/tests/bot/cogs/test_information.py
index 3c26374f5..b5f928dd6 100644
--- a/tests/bot/cogs/test_information.py
+++ b/tests/bot/cogs/test_information.py
@@ -7,7 +7,7 @@ import discord
from bot import constants
from bot.cogs import information
-from bot.decorators import InChannelCheckFailure
+from bot.decorators import InWhitelistCheckFailure
from tests import helpers
@@ -485,7 +485,7 @@ class UserEmbedTests(unittest.TestCase):
user.avatar_url_as.return_value = "avatar url"
embed = asyncio.run(self.cog.create_user_embed(ctx, user))
- user.avatar_url_as.assert_called_once_with(format="png")
+ user.avatar_url_as.assert_called_once_with(static_format="png")
self.assertEqual(embed.thumbnail.url, "avatar url")
@@ -525,7 +525,7 @@ class UserCommandTests(unittest.TestCase):
ctx = helpers.MockContext(author=self.author, channel=helpers.MockTextChannel(id=100))
msg = "Sorry, but you may only use this command within <#50>."
- with self.assertRaises(InChannelCheckFailure, msg=msg):
+ with self.assertRaises(InWhitelistCheckFailure, msg=msg):
asyncio.run(self.cog.user_info.callback(self.cog, ctx))
@unittest.mock.patch("bot.cogs.information.Information.create_user_embed", new_callable=unittest.mock.AsyncMock)
diff --git a/tests/bot/test_decorators.py b/tests/bot/test_decorators.py
new file mode 100644
index 000000000..a17dd3e16
--- /dev/null
+++ b/tests/bot/test_decorators.py
@@ -0,0 +1,147 @@
+import collections
+import unittest
+import unittest.mock
+
+from bot import constants
+from bot.decorators import InWhitelistCheckFailure, in_whitelist
+from tests import helpers
+
+
+InWhitelistTestCase = collections.namedtuple("WhitelistedContextTestCase", ("kwargs", "ctx", "description"))
+
+
+class InWhitelistTests(unittest.TestCase):
+ """Tests for the `in_whitelist` check."""
+
+ @classmethod
+ def setUpClass(cls):
+ """Set up helpers that only need to be defined once."""
+ cls.bot_commands = helpers.MockTextChannel(id=123456789, category_id=123456)
+ cls.help_channel = helpers.MockTextChannel(id=987654321, category_id=987654)
+ cls.non_whitelisted_channel = helpers.MockTextChannel(id=666666)
+ cls.dm_channel = helpers.MockDMChannel()
+
+ cls.non_staff_member = helpers.MockMember()
+ cls.staff_role = helpers.MockRole(id=121212)
+ cls.staff_member = helpers.MockMember(roles=(cls.staff_role,))
+
+ cls.channels = (cls.bot_commands.id,)
+ cls.categories = (cls.help_channel.category_id,)
+ cls.roles = (cls.staff_role.id,)
+
+ def test_predicate_returns_true_for_whitelisted_context(self):
+ """The predicate should return `True` if a whitelisted context was passed to it."""
+ test_cases = (
+ InWhitelistTestCase(
+ kwargs={"channels": self.channels},
+ ctx=helpers.MockContext(channel=self.bot_commands, author=self.non_staff_member),
+ description="In whitelisted channels by members without whitelisted roles",
+ ),
+ InWhitelistTestCase(
+ kwargs={"redirect": self.bot_commands.id},
+ ctx=helpers.MockContext(channel=self.bot_commands, author=self.non_staff_member),
+ description="`redirect` should be implicitly added to `channels`",
+ ),
+ InWhitelistTestCase(
+ kwargs={"categories": self.categories},
+ ctx=helpers.MockContext(channel=self.help_channel, author=self.non_staff_member),
+ description="Whitelisted category without whitelisted role",
+ ),
+ InWhitelistTestCase(
+ kwargs={"roles": self.roles},
+ ctx=helpers.MockContext(channel=self.non_whitelisted_channel, author=self.staff_member),
+ description="Whitelisted role outside of whitelisted channel/category"
+ ),
+ InWhitelistTestCase(
+ kwargs={
+ "channels": self.channels,
+ "categories": self.categories,
+ "roles": self.roles,
+ "redirect": self.bot_commands,
+ },
+ ctx=helpers.MockContext(channel=self.help_channel, author=self.staff_member),
+ description="Case with all whitelist kwargs used",
+ ),
+ )
+
+ for test_case in test_cases:
+ # patch `commands.check` with a no-op lambda that just returns the predicate passed to it
+ # so we can test the predicate that was generated from the specified kwargs.
+ with unittest.mock.patch("bot.decorators.commands.check", new=lambda predicate: predicate):
+ predicate = in_whitelist(**test_case.kwargs)
+
+ with self.subTest(test_description=test_case.description):
+ self.assertTrue(predicate(test_case.ctx))
+
+ def test_predicate_raises_exception_for_non_whitelisted_context(self):
+ """The predicate should raise `InWhitelistCheckFailure` for a non-whitelisted context."""
+ test_cases = (
+ # Failing check with explicit `redirect`
+ InWhitelistTestCase(
+ kwargs={
+ "categories": self.categories,
+ "channels": self.channels,
+ "roles": self.roles,
+ "redirect": self.bot_commands.id,
+ },
+ ctx=helpers.MockContext(channel=self.non_whitelisted_channel, author=self.non_staff_member),
+ description="Failing check with an explicit redirect channel",
+ ),
+
+ # Failing check with implicit `redirect`
+ InWhitelistTestCase(
+ kwargs={
+ "categories": self.categories,
+ "channels": self.channels,
+ "roles": self.roles,
+ },
+ ctx=helpers.MockContext(channel=self.non_whitelisted_channel, author=self.non_staff_member),
+ description="Failing check with an implicit redirect channel",
+ ),
+
+ # Failing check without `redirect`
+ InWhitelistTestCase(
+ kwargs={
+ "categories": self.categories,
+ "channels": self.channels,
+ "roles": self.roles,
+ "redirect": None,
+ },
+ ctx=helpers.MockContext(channel=self.non_whitelisted_channel, author=self.non_staff_member),
+ description="Failing check without a redirect channel",
+ ),
+
+ # Command issued in DM channel
+ InWhitelistTestCase(
+ kwargs={
+ "categories": self.categories,
+ "channels": self.channels,
+ "roles": self.roles,
+ "redirect": None,
+ },
+ ctx=helpers.MockContext(channel=self.dm_channel, author=self.dm_channel.me),
+ description="Commands issued in DM channel should be rejected",
+ ),
+ )
+
+ for test_case in test_cases:
+ if "redirect" not in test_case.kwargs or test_case.kwargs["redirect"] is not None:
+ # There are two cases in which we have a redirect channel:
+ # 1. No redirect channel was passed; the default value of `bot_commands` is used
+ # 2. An explicit `redirect` is set that is "not None"
+ redirect_channel = test_case.kwargs.get("redirect", constants.Channels.bot_commands)
+ redirect_message = f" here. Please use the <#{redirect_channel}> channel instead"
+ else:
+ # If an explicit `None` was passed for `redirect`, there is no redirect channel
+ redirect_message = ""
+
+ exception_message = f"You are not allowed to use that command{redirect_message}."
+
+ # patch `commands.check` with a no-op lambda that just returns the predicate passed to it
+ # so we can test the predicate that was generated from the specified kwargs.
+ with unittest.mock.patch("bot.decorators.commands.check", new=lambda predicate: predicate):
+ predicate = in_whitelist(**test_case.kwargs)
+
+ with self.subTest(test_description=test_case.description):
+ with self.assertRaisesRegex(InWhitelistCheckFailure, exception_message):
+ predicate(test_case.ctx)
diff --git a/tests/helpers.py b/tests/helpers.py
index 8e13f0f28..2b79a6c2a 100644
--- a/tests/helpers.py
+++ b/tests/helpers.py
@@ -315,7 +315,7 @@ class MockTextChannel(CustomMockMixin, unittest.mock.Mock, HashableMixin):
"""
spec_set = channel_instance
- def __init__(self, name: str = 'channel', channel_id: int = 1, **kwargs) -> None:
+ def __init__(self, **kwargs) -> None:
default_kwargs = {'id': next(self.discord_id), 'name': 'channel', 'guild': MockGuild()}
super().__init__(**collections.ChainMap(kwargs, default_kwargs))
@@ -323,6 +323,27 @@ class MockTextChannel(CustomMockMixin, unittest.mock.Mock, HashableMixin):
self.mention = f"#{self.name}"
+# Create data for the DMChannel instance
+state = unittest.mock.MagicMock()
+me = unittest.mock.MagicMock()
+dm_channel_data = {"id": 1, "recipients": [unittest.mock.MagicMock()]}
+dm_channel_instance = discord.DMChannel(me=me, state=state, data=dm_channel_data)
+
+
+class MockDMChannel(CustomMockMixin, unittest.mock.Mock, HashableMixin):
+ """
+ A MagicMock subclass to mock TextChannel objects.
+
+ Instances of this class will follow the specifications of `discord.TextChannel` instances. For
+ more information, see the `MockGuild` docstring.
+ """
+ spec_set = dm_channel_instance
+
+ def __init__(self, **kwargs) -> None:
+ default_kwargs = {'id': next(self.discord_id), 'recipient': MockUser(), "me": MockUser()}
+ super().__init__(**collections.ChainMap(kwargs, default_kwargs))
+
+
# Create a Message instance to get a realistic MagicMock of `discord.Message`
message_data = {
'id': 1,
generated by cgit v1.2.3 (git 2.39.1) at 2025-10-23 23:26:09 +0000