diff options
| -rw-r--r-- | bot/exts/backend/error_handler.py | 22 | ||||
| -rw-r--r-- | bot/exts/info/tags.py | 160 | ||||
| -rw-r--r-- | bot/pagination.py | 20 | ||||
| -rw-r--r-- | tests/bot/exts/backend/test_error_handler.py | 44 | ||||
| -rw-r--r-- | tests/helpers.py | 20 |
5 files changed, 132 insertions, 134 deletions
diff --git a/bot/exts/backend/error_handler.py b/bot/exts/backend/error_handler.py index cc2b5ef56..561bf8068 100644 --- a/bot/exts/backend/error_handler.py +++ b/bot/exts/backend/error_handler.py @@ -1,7 +1,8 @@ import copy import difflib +import typing as t -from discord import Embed +from discord import Embed, Interaction from discord.ext.commands import ChannelNotFound, Cog, Context, TextChannelConverter, VoiceChannelConverter, errors from pydis_core.site_api import ResponseCodeError from sentry_sdk import push_scope @@ -21,6 +22,10 @@ class ErrorHandler(Cog): def __init__(self, bot: Bot): self.bot = bot + @staticmethod + async def _can_run(_: Interaction) -> bool: + return False + def _get_error_embed(self, title: str, body: str) -> Embed: """Return an embed that contains the exception.""" return Embed( @@ -159,7 +164,7 @@ class ErrorHandler(Cog): return True return False - async def try_get_tag(self, ctx: Context) -> None: + async def try_get_tag(self, interaction: Interaction, can_run: t.Callable[[Interaction], bool] = False) -> None: """ Attempt to display a tag by interpreting the command name as a tag name. @@ -168,27 +173,28 @@ class ErrorHandler(Cog): the context to prevent infinite recursion in the case of a CommandNotFound exception. """ tags_get_command = self.bot.get_command("tags get") + tags_get_command.can_run = can_run if can_run else self._can_run if not tags_get_command: log.debug("Not attempting to parse message as a tag as could not find `tags get` command.") return - ctx.invoked_from_error_handler = True + interaction.invoked_from_error_handler = True log_msg = "Cancelling attempt to fall back to a tag due to failed checks." try: - if not await tags_get_command.can_run(ctx): + if not await tags_get_command.can_run(interaction): log.debug(log_msg) return except errors.CommandError as tag_error: log.debug(log_msg) - await self.on_command_error(ctx, tag_error) + await self.on_command_error(interaction, tag_error) return - if await ctx.invoke(tags_get_command, argument_string=ctx.message.content): + if await interaction.invoke(tags_get_command, tag_name=interaction.message.content): return - if not any(role.id in MODERATION_ROLES for role in ctx.author.roles): - await self.send_command_suggestion(ctx, ctx.invoked_with) + if not any(role.id in MODERATION_ROLES for role in interaction.user.roles): + await self.send_command_suggestion(interaction, interaction.invoked_with) async def try_run_eval(self, ctx: Context) -> bool: """ diff --git a/bot/exts/info/tags.py b/bot/exts/info/tags.py index 83d3a9d93..25c51def9 100644 --- a/bot/exts/info/tags.py +++ b/bot/exts/info/tags.py @@ -4,12 +4,12 @@ import enum import re import time from pathlib import Path -from typing import Callable, Iterable, Literal, NamedTuple, Optional, Union +from typing import Literal, NamedTuple, Optional, Union import discord import frontmatter -from discord import Embed, Member -from discord.ext.commands import Cog, Context, group +from discord import Embed, Member, app_commands +from discord.ext.commands import Cog from bot import constants from bot.bot import Bot @@ -27,6 +27,8 @@ TEST_CHANNELS = ( REGEX_NON_ALPHABET = re.compile(r"[^a-z]", re.MULTILINE & re.IGNORECASE) FOOTER_TEXT = f"To show a tag, type {constants.Bot.prefix}tags <tagname>." +GUILD_ID = constants.Guild.id + class COOLDOWN(enum.Enum): """Sentinel value to signal that a tag is on cooldown.""" @@ -138,6 +140,15 @@ class Tags(Cog): self.bot = bot self.tags: dict[TagIdentifier, Tag] = {} self.initialize_tags() + self.bot.tree.copy_global_to(guild=discord.Object(id=GUILD_ID)) + + tag_group = app_commands.Group(name="tag", description="...") + # search_tag = app_commands.Group(name="search", description="...", parent=tag_group) + + @Cog.listener() + async def on_ready(self) -> None: + """Called when the cog is ready.""" + await self.bot.tree.sync(guild=discord.Object(id=GUILD_ID)) def initialize_tags(self) -> None: """Load all tags from resources into `self.tags`.""" @@ -182,90 +193,9 @@ class Tags(Cog): return suggestions - def _get_tags_via_content( - self, - check: Callable[[Iterable], bool], - keywords: str, - user: Member, - ) -> list[tuple[TagIdentifier, Tag]]: - """ - Search for tags via contents. - - `predicate` will be the built-in any, all, or a custom callable. Must return a bool. - """ - keywords_processed = [] - for keyword in keywords.split(","): - keyword_sanitized = keyword.strip().casefold() - if not keyword_sanitized: - # this happens when there are leading / trailing / consecutive comma. - continue - keywords_processed.append(keyword_sanitized) - - if not keywords_processed: - # after sanitizing, we can end up with an empty list, for example when keywords is "," - # in that case, we simply want to search for such keywords directly instead. - keywords_processed = [keywords] - - matching_tags = [] - for identifier, tag in self.tags.items(): - matches = (query in tag.content.casefold() for query in keywords_processed) - if tag.accessible_by(user) and check(matches): - matching_tags.append((identifier, tag)) - - return matching_tags - - async def _send_matching_tags( - self, - ctx: Context, - keywords: str, - matching_tags: list[tuple[TagIdentifier, Tag]], - ) -> None: - """Send the result of matching tags to user.""" - if len(matching_tags) == 1: - await ctx.send(embed=matching_tags[0][1].embed) - elif matching_tags: - is_plural = keywords.strip().count(" ") > 0 or keywords.strip().count(",") > 0 - embed = Embed( - title=f"Here are the tags containing the given keyword{'s' * is_plural}:", - ) - await LinePaginator.paginate( - sorted( - f"**\N{RIGHT-POINTING DOUBLE ANGLE QUOTATION MARK}** {identifier.name}" - for identifier, _ in matching_tags - ), - ctx, - embed, - **self.PAGINATOR_DEFAULTS, - ) - - @group(name="tags", aliases=("tag", "t"), invoke_without_command=True, usage="[tag_group] [tag_name]") - async def tags_group(self, ctx: Context, *, argument_string: Optional[str]) -> None: - """Show all known tags, a single tag, or run a subcommand.""" - await self.get_command(ctx, argument_string=argument_string) - - @tags_group.group(name="search", invoke_without_command=True) - async def search_tag_content(self, ctx: Context, *, keywords: str) -> None: - """ - Search inside tags' contents for tags. Allow searching for multiple keywords separated by comma. - - Only search for tags that has ALL the keywords. - """ - matching_tags = self._get_tags_via_content(all, keywords, ctx.author) - await self._send_matching_tags(ctx, keywords, matching_tags) - - @search_tag_content.command(name="any") - async def search_tag_content_any_keyword(self, ctx: Context, *, keywords: Optional[str] = "any") -> None: - """ - Search inside tags' contents for tags. Allow searching for multiple keywords separated by comma. - - Search for tags that has ANY of the keywords. - """ - matching_tags = self._get_tags_via_content(any, keywords or "any", ctx.author) - await self._send_matching_tags(ctx, keywords, matching_tags) - async def get_tag_embed( self, - ctx: Context, + interaction: discord.Interaction, tag_identifier: TagIdentifier, ) -> Optional[Union[Embed, Literal[COOLDOWN.obj]]]: """ @@ -276,7 +206,7 @@ class Tags(Cog): filtered_tags = [ (ident, tag) for ident, tag in self.get_fuzzy_matches(tag_identifier)[:10] - if tag.accessible_by(ctx.author) + if tag.accessible_by(interaction.user) ] # Try exact match, includes checking through alt names @@ -295,10 +225,10 @@ class Tags(Cog): tag = filtered_tags[0][1] if tag is not None: - if tag.on_cooldown_in(ctx.channel): + if tag.on_cooldown_in(interaction.channel): log.debug(f"Tag {str(tag_identifier)!r} is on cooldown.") return COOLDOWN.obj - tag.set_cooldown_for(ctx.channel) + tag.set_cooldown_for(interaction.channel) self.bot.stats.incr( f"tags.usages" @@ -313,7 +243,7 @@ class Tags(Cog): suggested_tags_text = "\n".join( f"**\N{RIGHT-POINTING DOUBLE ANGLE QUOTATION MARK}** {identifier}" for identifier, tag in filtered_tags - if not tag.on_cooldown_in(ctx.channel) + if not tag.on_cooldown_in(interaction.channel) ) return Embed( title="Did you mean ...", @@ -362,8 +292,8 @@ class Tags(Cog): if identifier.group == group and tag.accessible_by(user) ) - @tags_group.command(name="get", aliases=("show", "g"), usage="[tag_group] [tag_name]") - async def get_command(self, ctx: Context, *, argument_string: Optional[str]) -> bool: + @tag_group.command(name="get") + async def get_command(self, interaction: discord.Interaction, *, tag_name: Optional[str]) -> bool: """ If a single argument matching a group name is given, list all accessible tags from that group Otherwise display the tag if one was found for the given arguments, or try to display suggestions for that name. @@ -373,38 +303,68 @@ class Tags(Cog): Returns True if a message was sent, or if the tag is on cooldown. Returns False if no message was sent. """ # noqa: D205, D415 - if not argument_string: + if not tag_name: if self.tags: await LinePaginator.paginate( - self.accessible_tags(ctx.author), ctx, Embed(title="Available tags"), **self.PAGINATOR_DEFAULTS + self.accessible_tags(interaction.user), + interaction, Embed(title="Available tags"), + **self.PAGINATOR_DEFAULTS, ) else: - await ctx.send(embed=Embed(description="**There are no tags!**")) + await interaction.response.send_message(embed=Embed(description="**There are no tags!**")) return True - identifier = TagIdentifier.from_string(argument_string) + identifier = TagIdentifier.from_string(tag_name) if identifier.group is None: # Try to find accessible tags from a group matching the identifier's name. - if group_tags := self.accessible_tags_in_group(identifier.name, ctx.author): + if group_tags := self.accessible_tags_in_group(identifier.name, interaction.user): await LinePaginator.paginate( - group_tags, ctx, Embed(title=f"Tags under *{identifier.name}*"), **self.PAGINATOR_DEFAULTS + group_tags, interaction, Embed(title=f"Tags under *{identifier.name}*"), **self.PAGINATOR_DEFAULTS ) return True - embed = await self.get_tag_embed(ctx, identifier) + embed = await self.get_tag_embed(interaction, identifier) if embed is None: return False if embed is not COOLDOWN.obj: await wait_for_deletion( - await ctx.send(embed=embed), - (ctx.author.id,) + await interaction.response.send_message(embed=embed), + (interaction.user.id,) ) # A valid tag was found and was either sent, or is on cooldown return True + @get_command.autocomplete("tag_name") + async def tag_name_autocomplete( + self, + interaction: discord.Interaction, + current: str + ) -> list[app_commands.Choice[str]]: + """Autocompleter for `/tag get` command.""" + tag_names = [tag.name for tag in self.tags.keys()] + return [ + app_commands.Choice(name=tag, value=tag) + for tag in tag_names if current.lower() in tag + ] + + @tag_group.command(name="list") + async def list_command(self, interaction: discord.Interaction) -> bool: + """Lists all accessible tags.""" + if self.tags: + await LinePaginator.paginate( + self.accessible_tags(interaction.user), + interaction, + Embed(title="Available tags"), + **self.PAGINATOR_DEFAULTS, + ) + else: + await interaction.response.send_message(embed=Embed(description="**There are no tags!**")) + return True + async def setup(bot: Bot) -> None: """Load the Tags cog.""" await bot.add_cog(Tags(bot)) + await bot.tree.sync(guild=discord.Object(id=GUILD_ID)) diff --git a/bot/pagination.py b/bot/pagination.py index 0ef5808cc..1c63a4768 100644 --- a/bot/pagination.py +++ b/bot/pagination.py @@ -191,7 +191,7 @@ class LinePaginator(Paginator): async def paginate( cls, lines: t.List[str], - ctx: Context, + ctx: t.Union[Context, discord.Interaction], embed: discord.Embed, prefix: str = "", suffix: str = "", @@ -228,7 +228,10 @@ class LinePaginator(Paginator): current_page = 0 if not restrict_to_user: - restrict_to_user = ctx.author + if isinstance(ctx, discord.Interaction): + restrict_to_user = ctx.user + else: + restrict_to_user = ctx.author if not lines: if exception_on_empty_embed: @@ -261,6 +264,8 @@ class LinePaginator(Paginator): log.trace(f"Setting embed url to '{url}'") log.debug("There's less than two pages, so we won't paginate - sending single page on its own") + if isinstance(ctx, discord.Interaction): + return await ctx.response.send_message(embed=embed) return await ctx.send(embed=embed) else: if footer_text: @@ -274,7 +279,11 @@ class LinePaginator(Paginator): log.trace(f"Setting embed url to '{url}'") log.debug("Sending first page to channel...") - message = await ctx.send(embed=embed) + if isinstance(ctx, discord.Interaction): + await ctx.response.send_message(embed=embed) + message = await ctx.original_response() + else: + message = await ctx.send(embed=embed) log.debug("Adding emoji reactions to message...") @@ -292,7 +301,10 @@ class LinePaginator(Paginator): while True: try: - reaction, user = await ctx.bot.wait_for("reaction_add", timeout=timeout, check=check) + if isinstance(ctx, discord.Interaction): + reaction, user = await ctx.client.wait_for("reaction_add", timeout=timeout, check=check) + else: + reaction, user = await ctx.bot.wait_for("reaction_add", timeout=timeout, check=check) log.trace(f"Got reaction: {reaction}") except asyncio.TimeoutError: log.debug("Timed out waiting for a reaction") diff --git a/tests/bot/exts/backend/test_error_handler.py b/tests/bot/exts/backend/test_error_handler.py index adb0252a5..83bc3c4a1 100644 --- a/tests/bot/exts/backend/test_error_handler.py +++ b/tests/bot/exts/backend/test_error_handler.py @@ -9,7 +9,7 @@ from bot.exts.backend import error_handler from bot.exts.info.tags import Tags from bot.exts.moderation.silence import Silence from bot.utils.checks import InWhitelistCheckFailure -from tests.helpers import MockBot, MockContext, MockGuild, MockRole, MockTextChannel, MockVoiceChannel +from tests.helpers import MockBot, MockContext, MockGuild, MockInteraction, MockRole, MockTextChannel, MockVoiceChannel class ErrorHandlerTests(unittest.IsolatedAsyncioTestCase): @@ -331,7 +331,7 @@ class TryGetTagTests(unittest.IsolatedAsyncioTestCase): def setUp(self): self.bot = MockBot() - self.ctx = MockContext() + self.interaction = MockInteraction() self.tag = Tags(self.bot) self.cog = error_handler.ErrorHandler(self.bot) self.bot.get_command.return_value = self.tag.get_command @@ -339,57 +339,57 @@ class TryGetTagTests(unittest.IsolatedAsyncioTestCase): async def test_try_get_tag_get_command(self): """Should call `Bot.get_command` with `tags get` argument.""" self.bot.get_command.reset_mock() - await self.cog.try_get_tag(self.ctx) + await self.cog.try_get_tag(self.interaction) self.bot.get_command.assert_called_once_with("tags get") async def test_try_get_tag_invoked_from_error_handler(self): - """`self.ctx` should have `invoked_from_error_handler` `True`.""" - self.ctx.invoked_from_error_handler = False - await self.cog.try_get_tag(self.ctx) - self.assertTrue(self.ctx.invoked_from_error_handler) + """`self.interaction` should have `invoked_from_error_handler` `True`.""" + self.interaction.invoked_from_error_handler = False + await self.cog.try_get_tag(self.interaction) + self.assertTrue(self.interaction.invoked_from_error_handler) async def test_try_get_tag_no_permissions(self): """Test how to handle checks failing.""" self.tag.get_command.can_run = AsyncMock(return_value=False) - self.ctx.invoked_with = "foo" - self.assertIsNone(await self.cog.try_get_tag(self.ctx)) + self.interaction.invoked_with = "foo" + self.assertIsNone(await self.cog.try_get_tag(self.interaction, AsyncMock(return_value=False))) async def test_try_get_tag_command_error(self): """Should call `on_command_error` when `CommandError` raised.""" err = errors.CommandError() self.tag.get_command.can_run = AsyncMock(side_effect=err) self.cog.on_command_error = AsyncMock() - self.assertIsNone(await self.cog.try_get_tag(self.ctx)) - self.cog.on_command_error.assert_awaited_once_with(self.ctx, err) + self.assertIsNone(await self.cog.try_get_tag(self.interaction, AsyncMock(side_effect=err))) + self.cog.on_command_error.assert_awaited_once_with(self.interaction, err) async def test_dont_call_suggestion_tag_sent(self): """Should never call command suggestion if tag is already sent.""" - self.ctx.message = MagicMock(content="foo") - self.ctx.invoke = AsyncMock(return_value=True) + self.interaction.message = MagicMock(content="foo") + self.interaction.invoke = AsyncMock(return_value=True) self.cog.send_command_suggestion = AsyncMock() - await self.cog.try_get_tag(self.ctx) + await self.cog.try_get_tag(self.interaction, AsyncMock()) self.cog.send_command_suggestion.assert_not_awaited() @patch("bot.exts.backend.error_handler.MODERATION_ROLES", new=[1234]) async def test_dont_call_suggestion_if_user_mod(self): """Should not call command suggestion if user is a mod.""" - self.ctx.invoked_with = "foo" - self.ctx.invoke = AsyncMock(return_value=False) - self.ctx.author.roles = [MockRole(id=1234)] + self.interaction.invoked_with = "foo" + self.interaction.invoke = AsyncMock(return_value=False) + self.interaction.user.roles = [MockRole(id=1234)] self.cog.send_command_suggestion = AsyncMock() - await self.cog.try_get_tag(self.ctx) + await self.cog.try_get_tag(self.interaction, AsyncMock()) self.cog.send_command_suggestion.assert_not_awaited() async def test_call_suggestion(self): """Should call command suggestion if user is not a mod.""" - self.ctx.invoked_with = "foo" - self.ctx.invoke = AsyncMock(return_value=False) + self.interaction.invoked_with = "foo" + self.interaction.invoke = AsyncMock(return_value=False) self.cog.send_command_suggestion = AsyncMock() - await self.cog.try_get_tag(self.ctx) - self.cog.send_command_suggestion.assert_awaited_once_with(self.ctx, "foo") + await self.cog.try_get_tag(self.interaction, AsyncMock()) + self.cog.send_command_suggestion.assert_awaited_once_with(self.interaction, "foo") class IndividualErrorHandlerTests(unittest.IsolatedAsyncioTestCase): diff --git a/tests/helpers.py b/tests/helpers.py index 4b980ac21..2d20b4d07 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -479,6 +479,26 @@ class MockContext(CustomMockMixin, unittest.mock.MagicMock): self.invoked_from_error_handler = kwargs.get('invoked_from_error_handler', False) +class MockInteraction(CustomMockMixin, unittest.mock.MagicMock): + """ + A MagicMock subclass to mock Interaction objects. + + Instances of this class will follow the specifications of `discord.Interaction` + instances. For more information, see the `MockGuild` docstring. + """ + # spec_set = context_instance + + def __init__(self, **kwargs) -> None: + super().__init__(**kwargs) + self.me = kwargs.get('me', MockMember()) + self.client = kwargs.get('client', MockBot()) + self.guild = kwargs.get('guild', MockGuild()) + self.user = kwargs.get('user', MockMember()) + self.channel = kwargs.get('channel', MockTextChannel()) + self.message = kwargs.get('message', MockMessage()) + self.invoked_from_error_handler = kwargs.get('invoked_from_error_handler', False) + + attachment_instance = discord.Attachment(data=unittest.mock.MagicMock(id=1), state=unittest.mock.MagicMock()) |