diff options
| author | 2021-01-02 20:23:17 -0800 | |
|---|---|---|
| committer | 2021-01-02 20:23:17 -0800 | |
| commit | 01406f86ad9b83d5b378c3e264c5d8b3e767ac4c (patch) | |
| tree | 046895a131aaba337f1489485478f6678eec696e /tests | |
| parent | Merge branch 'feat/F4zi/CommandSuggestion' of https://github.com/python-disco... (diff) | |
| parent | Merge pull request #1334 from python-discord/bug/precommit-pycharm (diff) | |
Rebased after a long time of being abandon
Since the cogs folder has been removed, the error_handler and tag cogs had to be removed and transfer into their respective places in the exts folder.
Diffstat (limited to 'tests')
59 files changed, 5004 insertions, 1624 deletions
diff --git a/tests/README.md b/tests/README.md index be78821bf..4f62edd68 100644 --- a/tests/README.md +++ b/tests/README.md @@ -83,7 +83,7 @@ TagContentConverter should return correct values for valid input. As we are trying to test our "units" of code independently, we want to make sure that we do not rely objects and data generated by "external" code. If we we did, then we wouldn't know if the failure we're observing was caused by the code we are actually trying to test or something external to it. -However, the features that we are trying to test often depend on those objects generated by external pieces of code. It would be difficult to test a bot command without having access to a `Context` instance. Fortunately, there's a solution for that: we use fake objects that act like the true object. We call these fake objects "mocks". +However, the features that we are trying to test often depend on those objects generated by external pieces of code. It would be difficult to test a bot command without having access to a `Context` instance. Fortunately, there's a solution for that: we use fake objects that act like the true object. We call these fake objects "mocks". To create these mock object, we mainly use the [`unittest.mock`](https://docs.python.org/3/library/unittest.mock.html) module. In addition, we have also defined a couple of specialized mock objects that mock specific `discord.py` types (see the section on the below.). @@ -114,13 +114,13 @@ class BotCogTests(unittest.TestCase): ### Mocking coroutines -By default, the `unittest.mock.Mock` and `unittest.mock.MagicMock` classes cannot mock coroutines, since the `__call__` method they provide is synchronous. In anticipation of the `AsyncMock` that will be [introduced in Python 3.8](https://docs.python.org/3.9/whatsnew/3.8.html#unittest), we have added an `AsyncMock` helper to [`helpers.py`](/tests/helpers.py). Do note that this drop-in replacement only implements an asynchronous `__call__` method, not the additional assertions that will come with the new `AsyncMock` type in Python 3.8. +By default, the `unittest.mock.Mock` and `unittest.mock.MagicMock` classes cannot mock coroutines, since the `__call__` method they provide is synchronous. In anticipation of the `AsyncMock` that will be [introduced in Python 3.8](https://docs.python.org/3.9/whatsnew/3.8.html#unittest), we have added an `AsyncMock` helper to [`helpers.py`](/tests/helpers.py). Do note that this drop-in replacement only implements an asynchronous `__call__` method, not the additional assertions that will come with the new `AsyncMock` type in Python 3.8. ### Special mocks for some `discord.py` types To quote Ned Batchelder, Mock objects are "automatic chameleons". This means that they will happily allow the access to any attribute or method and provide a mocked value in return. One downside to this is that if the code you are testing gets the name of the attribute wrong, your mock object will not complain and the test may still pass. -In order to avoid that, we have defined a number of Mock types in [`helpers.py`](/tests/helpers.py) that follow the specifications of the actual Discord types they are mocking. This means that trying to access an attribute or method on a mocked object that does not exist on the equivalent `discord.py` object will result in an `AttributeError`. In addition, these mocks have some sensible defaults and **pass `isinstance` checks for the types they are mocking**. +In order to avoid that, we have defined a number of Mock types in [`helpers.py`](/tests/helpers.py) that follow the specifications of the actual Discord types they are mocking. This means that trying to access an attribute or method on a mocked object that does not exist on the equivalent `discord.py` object will result in an `AttributeError`. In addition, these mocks have some sensible defaults and **pass `isinstance` checks for the types they are mocking**. These special mocks are added when they are needed, so if you think it would be sensible to add another one, feel free to propose one in your PR. @@ -144,7 +144,7 @@ Finally, there are some considerations to make when writing tests, both for writ ### Test coverage is a starting point -Having test coverage is a good starting point for unit testing: If a part of your code was not covered by a test, we know that we have not tested it properly. The reverse is unfortunately not true: Even if the code we are testing has 100% branch coverage, it does not mean it's fully tested or guaranteed to work. +Having test coverage is a good starting point for unit testing: If a part of your code was not covered by a test, we know that we have not tested it properly. The reverse is unfortunately not true: Even if the code we are testing has 100% branch coverage, it does not mean it's fully tested or guaranteed to work. One problem is that 100% branch coverage may be misleading if we haven't tested our code against all the realistic input it may get in production. For instance, take a look at the following `member_information` function and the test we've written for it: @@ -169,7 +169,7 @@ class FunctionsTests(unittest.TestCase): If you were to run this test, not only would the function pass the test, `coverage.py` will also tell us that the test provides 100% branch coverage for the function. Can you spot the bug the test suite did not catch? -The problem here is that we have only tested our function with a member object that had `None` for the `member.joined` attribute. This means that `member.joined.stfptime("%d-%m-%Y")` was never executed during our test, leading to us missing the spelling mistake in `stfptime` (it should be `strftime`). +The problem here is that we have only tested our function with a member object that had `None` for the `member.joined` attribute. This means that `member.joined.stfptime("%d-%m-%Y")` was never executed during our test, leading to us missing the spelling mistake in `stfptime` (it should be `strftime`). Adding another test would not increase the test coverage we have, but it does ensure that we'll notice that this function can fail with realistic data: diff --git a/tests/_autospec.py b/tests/_autospec.py new file mode 100644 index 000000000..ee2fc1973 --- /dev/null +++ b/tests/_autospec.py @@ -0,0 +1,64 @@ +import contextlib +import functools +import unittest.mock +from typing import Callable + + [email protected](unittest.mock._patch.decoration_helper) +def _decoration_helper(self, patched, args, keywargs): + """Skips adding patchings as args if their `dont_pass` attribute is True.""" + # Don't ask what this does. It's just a copy from stdlib, but with the dont_pass check added. + extra_args = [] + with contextlib.ExitStack() as exit_stack: + for patching in patched.patchings: + arg = exit_stack.enter_context(patching) + if not getattr(patching, "dont_pass", False): + # Only add the patching as an arg if dont_pass is False. + if patching.attribute_name is not None: + keywargs.update(arg) + elif patching.new is unittest.mock.DEFAULT: + extra_args.append(arg) + + args += tuple(extra_args) + yield args, keywargs + + [email protected](unittest.mock._patch.copy) +def _copy(self): + """Copy the `dont_pass` attribute along with the standard copy operation.""" + patcher_copy = _copy.original(self) + patcher_copy.dont_pass = getattr(self, "dont_pass", False) + return patcher_copy + + +# Monkey-patch the patcher class :) +_copy.original = unittest.mock._patch.copy +unittest.mock._patch.copy = _copy +unittest.mock._patch.decoration_helper = _decoration_helper + + +def autospec(target, *attributes: str, pass_mocks: bool = True, **patch_kwargs) -> Callable: + """ + Patch multiple `attributes` of a `target` with autospecced mocks and `spec_set` as True. + + If `pass_mocks` is True, pass the autospecced mocks as arguments to the decorated object. + """ + # Caller's kwargs should take priority and overwrite the defaults. + kwargs = dict(spec_set=True, autospec=True) + kwargs.update(patch_kwargs) + + # Import the target if it's a string. + # This is to support both object and string targets like patch.multiple. + if type(target) is str: + target = unittest.mock._importer(target) + + def decorator(func): + for attribute in attributes: + patcher = unittest.mock.patch.object(target, attribute, **kwargs) + if not pass_mocks: + # A custom attribute to keep track of which patchings should be skipped. + patcher.dont_pass = True + func = patcher(func) + return func + return decorator diff --git a/tests/base.py b/tests/base.py index 029a249ed..d99b9ac31 100644 --- a/tests/base.py +++ b/tests/base.py @@ -1,6 +1,12 @@ import logging import unittest from contextlib import contextmanager +from typing import Dict + +import discord +from discord.ext import commands + +from tests import helpers class _CaptureLogHandler(logging.Handler): @@ -16,11 +22,16 @@ class _CaptureLogHandler(logging.Handler): self.records.append(record) -class LoggingTestCase(unittest.TestCase): - """TestCase subclass that adds more logging assertion tools.""" +class LoggingTestsMixin: + """ + A mixin that defines additional test methods for logging behavior. + + This mixin relies on the availability of the `fail` attribute defined by the + test classes included in Python's unittest method to signal test failure. + """ @contextmanager - def assertNotLogs(self, logger=None, level=None, msg=None): + def assertNotLogs(self, logger=None, level=None, msg=None): # noqa: N802 """ Asserts that no logs of `level` and higher were emitted by `logger`. @@ -65,3 +76,30 @@ class LoggingTestCase(unittest.TestCase): standard_message = self._truncateMessage(base_message, record_message) msg = self._formatMessage(msg, standard_message) self.fail(msg) + + +class CommandTestCase(unittest.IsolatedAsyncioTestCase): + """TestCase with additional assertions that are useful for testing Discord commands.""" + + async def assertHasPermissionsCheck( # noqa: N802 + self, + cmd: commands.Command, + permissions: Dict[str, bool], + ) -> None: + """ + Test that `cmd` raises a `MissingPermissions` exception if author lacks `permissions`. + + Every permission in `permissions` is expected to be reported as missing. In other words, do + not include permissions which should not raise an exception along with those which should. + """ + # Invert permission values because it's more intuitive to pass to this assertion the same + # permissions as those given to the check decorator. + permissions = {k: not v for k, v in permissions.items()} + + ctx = helpers.MockContext() + ctx.channel.permissions_for.return_value = discord.Permissions(**permissions) + + with self.assertRaises(commands.MissingPermissions) as cm: + await cmd.can_run(ctx) + + self.assertCountEqual(permissions.keys(), cm.exception.missing_perms) diff --git a/tests/bot/cogs/sync/test_roles.py b/tests/bot/cogs/sync/test_roles.py deleted file mode 100644 index 27ae27639..000000000 --- a/tests/bot/cogs/sync/test_roles.py +++ /dev/null @@ -1,126 +0,0 @@ -import unittest - -from bot.cogs.sync.syncers import Role, get_roles_for_sync - - -class GetRolesForSyncTests(unittest.TestCase): - """Tests constructing the roles to synchronize with the site.""" - - def test_get_roles_for_sync_empty_return_for_equal_roles(self): - """No roles should be synced when no diff is found.""" - api_roles = {Role(id=41, name='name', colour=33, permissions=0x8, position=1)} - guild_roles = {Role(id=41, name='name', colour=33, permissions=0x8, position=1)} - - self.assertEqual( - get_roles_for_sync(guild_roles, api_roles), - (set(), set(), set()) - ) - - def test_get_roles_for_sync_returns_roles_to_update_with_non_id_diff(self): - """Roles to be synced are returned when non-ID attributes differ.""" - api_roles = {Role(id=41, name='old name', colour=35, permissions=0x8, position=1)} - guild_roles = {Role(id=41, name='new name', colour=33, permissions=0x8, position=2)} - - self.assertEqual( - get_roles_for_sync(guild_roles, api_roles), - (set(), guild_roles, set()) - ) - - def test_get_roles_only_returns_roles_that_require_update(self): - """Roles that require an update should be returned as the second tuple element.""" - api_roles = { - Role(id=41, name='old name', colour=33, permissions=0x8, position=1), - Role(id=53, name='other role', colour=55, permissions=0, position=3) - } - guild_roles = { - Role(id=41, name='new name', colour=35, permissions=0x8, position=2), - Role(id=53, name='other role', colour=55, permissions=0, position=3) - } - - self.assertEqual( - get_roles_for_sync(guild_roles, api_roles), - ( - set(), - {Role(id=41, name='new name', colour=35, permissions=0x8, position=2)}, - set(), - ) - ) - - def test_get_roles_returns_new_roles_in_first_tuple_element(self): - """Newly created roles are returned as the first tuple element.""" - api_roles = { - Role(id=41, name='name', colour=35, permissions=0x8, position=1), - } - guild_roles = { - Role(id=41, name='name', colour=35, permissions=0x8, position=1), - Role(id=53, name='other role', colour=55, permissions=0, position=2) - } - - self.assertEqual( - get_roles_for_sync(guild_roles, api_roles), - ( - {Role(id=53, name='other role', colour=55, permissions=0, position=2)}, - set(), - set(), - ) - ) - - def test_get_roles_returns_roles_to_update_and_new_roles(self): - """Newly created and updated roles should be returned together.""" - api_roles = { - Role(id=41, name='old name', colour=35, permissions=0x8, position=1), - } - guild_roles = { - Role(id=41, name='new name', colour=40, permissions=0x16, position=2), - Role(id=53, name='other role', colour=55, permissions=0, position=3) - } - - self.assertEqual( - get_roles_for_sync(guild_roles, api_roles), - ( - {Role(id=53, name='other role', colour=55, permissions=0, position=3)}, - {Role(id=41, name='new name', colour=40, permissions=0x16, position=2)}, - set(), - ) - ) - - def test_get_roles_returns_roles_to_delete(self): - """Roles to be deleted should be returned as the third tuple element.""" - api_roles = { - Role(id=41, name='name', colour=35, permissions=0x8, position=1), - Role(id=61, name='to delete', colour=99, permissions=0x9, position=2), - } - guild_roles = { - Role(id=41, name='name', colour=35, permissions=0x8, position=1), - } - - self.assertEqual( - get_roles_for_sync(guild_roles, api_roles), - ( - set(), - set(), - {Role(id=61, name='to delete', colour=99, permissions=0x9, position=2)}, - ) - ) - - def test_get_roles_returns_roles_to_delete_update_and_new_roles(self): - """When roles were added, updated, and removed, all of them are returned properly.""" - api_roles = { - Role(id=41, name='not changed', colour=35, permissions=0x8, position=1), - Role(id=61, name='to delete', colour=99, permissions=0x9, position=2), - Role(id=71, name='to update', colour=99, permissions=0x9, position=3), - } - guild_roles = { - Role(id=41, name='not changed', colour=35, permissions=0x8, position=1), - Role(id=81, name='to create', colour=99, permissions=0x9, position=4), - Role(id=71, name='updated', colour=101, permissions=0x5, position=3), - } - - self.assertEqual( - get_roles_for_sync(guild_roles, api_roles), - ( - {Role(id=81, name='to create', colour=99, permissions=0x9, position=4)}, - {Role(id=71, name='updated', colour=101, permissions=0x5, position=3)}, - {Role(id=61, name='to delete', colour=99, permissions=0x9, position=2)}, - ) - ) diff --git a/tests/bot/cogs/sync/test_users.py b/tests/bot/cogs/sync/test_users.py deleted file mode 100644 index ccaf67490..000000000 --- a/tests/bot/cogs/sync/test_users.py +++ /dev/null @@ -1,84 +0,0 @@ -import unittest - -from bot.cogs.sync.syncers import User, get_users_for_sync - - -def fake_user(**kwargs): - kwargs.setdefault('id', 43) - kwargs.setdefault('name', 'bob the test man') - kwargs.setdefault('discriminator', 1337) - kwargs.setdefault('avatar_hash', None) - kwargs.setdefault('roles', (666,)) - kwargs.setdefault('in_guild', True) - return User(**kwargs) - - -class GetUsersForSyncTests(unittest.TestCase): - """Tests constructing the users to synchronize with the site.""" - - def test_get_users_for_sync_returns_nothing_for_empty_params(self): - """When no users are given, none are returned.""" - self.assertEqual( - get_users_for_sync({}, {}), - (set(), set()) - ) - - def test_get_users_for_sync_returns_nothing_for_equal_users(self): - """When no users are updated, none are returned.""" - api_users = {43: fake_user()} - guild_users = {43: fake_user()} - - self.assertEqual( - get_users_for_sync(guild_users, api_users), - (set(), set()) - ) - - def test_get_users_for_sync_returns_users_to_update_on_non_id_field_diff(self): - """When a non-ID-field differs, the user to update is returned.""" - api_users = {43: fake_user()} - guild_users = {43: fake_user(name='new fancy name')} - - self.assertEqual( - get_users_for_sync(guild_users, api_users), - (set(), {fake_user(name='new fancy name')}) - ) - - def test_get_users_for_sync_returns_users_to_create_with_new_ids_on_guild(self): - """When new users join the guild, they are returned as the first tuple element.""" - api_users = {43: fake_user()} - guild_users = {43: fake_user(), 63: fake_user(id=63)} - - self.assertEqual( - get_users_for_sync(guild_users, api_users), - ({fake_user(id=63)}, set()) - ) - - def test_get_users_for_sync_updates_in_guild_field_on_user_leave(self): - """When a user leaves the guild, the `in_guild` flag is updated to `False`.""" - api_users = {43: fake_user(), 63: fake_user(id=63)} - guild_users = {43: fake_user()} - - self.assertEqual( - get_users_for_sync(guild_users, api_users), - (set(), {fake_user(id=63, in_guild=False)}) - ) - - def test_get_users_for_sync_updates_and_creates_users_as_needed(self): - """When one user left and another one was updated, both are returned.""" - api_users = {43: fake_user()} - guild_users = {63: fake_user(id=63)} - - self.assertEqual( - get_users_for_sync(guild_users, api_users), - ({fake_user(id=63)}, {fake_user(in_guild=False)}) - ) - - def test_get_users_for_sync_does_not_duplicate_update_users(self): - """When the API knows a user the guild doesn't, nothing is performed.""" - api_users = {43: fake_user(in_guild=False)} - guild_users = {} - - self.assertEqual( - get_users_for_sync(guild_users, api_users), - (set(), set()) - ) diff --git a/tests/bot/cogs/test_duck_pond.py b/tests/bot/cogs/test_duck_pond.py deleted file mode 100644 index d07b2bce1..000000000 --- a/tests/bot/cogs/test_duck_pond.py +++ /dev/null @@ -1,584 +0,0 @@ -import asyncio -import logging -import typing -import unittest -from unittest.mock import MagicMock, patch - -import discord - -from bot import constants -from bot.cogs import duck_pond -from tests import base -from tests import helpers - -MODULE_PATH = "bot.cogs.duck_pond" - - -class DuckPondTests(base.LoggingTestCase): - """Tests for DuckPond functionality.""" - - @classmethod - def setUpClass(cls): - """Sets up the objects that only have to be initialized once.""" - cls.nonstaff_member = helpers.MockMember(name="Non-staffer") - - cls.staff_role = helpers.MockRole(name="Staff role", id=constants.STAFF_ROLES[0]) - cls.staff_member = helpers.MockMember(name="staffer", roles=[cls.staff_role]) - - cls.checkmark_emoji = "\N{White Heavy Check Mark}" - cls.thumbs_up_emoji = "\N{Thumbs Up Sign}" - cls.unicode_duck_emoji = "\N{Duck}" - cls.duck_pond_emoji = helpers.MockPartialEmoji(id=constants.DuckPond.custom_emojis[0]) - cls.non_duck_custom_emoji = helpers.MockPartialEmoji(id=123) - - def setUp(self): - """Sets up the objects that need to be refreshed before each test.""" - self.bot = helpers.MockBot(user=helpers.MockMember(id=46692)) - self.cog = duck_pond.DuckPond(bot=self.bot) - - def test_duck_pond_correctly_initializes(self): - """`__init__ should set `bot` and `webhook_id` attributes and schedule `fetch_webhook`.""" - bot = helpers.MockBot() - cog = MagicMock() - - duck_pond.DuckPond.__init__(cog, bot) - - self.assertEqual(cog.bot, bot) - self.assertEqual(cog.webhook_id, constants.Webhooks.duck_pond) - bot.loop.create_loop.called_once_with(cog.fetch_webhook()) - - def test_fetch_webhook_succeeds_without_connectivity_issues(self): - """The `fetch_webhook` method waits until `READY` event and sets the `webhook` attribute.""" - self.bot.fetch_webhook.return_value = "dummy webhook" - self.cog.webhook_id = 1 - - asyncio.run(self.cog.fetch_webhook()) - - self.bot.wait_until_ready.assert_called_once() - self.bot.fetch_webhook.assert_called_once_with(1) - self.assertEqual(self.cog.webhook, "dummy webhook") - - def test_fetch_webhook_logs_when_unable_to_fetch_webhook(self): - """The `fetch_webhook` method should log an exception when it fails to fetch the webhook.""" - self.bot.fetch_webhook.side_effect = discord.HTTPException(response=MagicMock(), message="Not found.") - self.cog.webhook_id = 1 - - log = logging.getLogger('bot.cogs.duck_pond') - with self.assertLogs(logger=log, level=logging.ERROR) as log_watcher: - asyncio.run(self.cog.fetch_webhook()) - - self.bot.wait_until_ready.assert_called_once() - self.bot.fetch_webhook.assert_called_once_with(1) - - self.assertEqual(len(log_watcher.records), 1) - - record = log_watcher.records[0] - self.assertEqual(record.levelno, logging.ERROR) - - def test_is_staff_returns_correct_values_based_on_instance_passed(self): - """The `is_staff` method should return correct values based on the instance passed.""" - test_cases = ( - (helpers.MockUser(name="User instance"), False), - (helpers.MockMember(name="Member instance without staff role"), False), - (helpers.MockMember(name="Member instance with staff role", roles=[self.staff_role]), True) - ) - - for user, expected_return in test_cases: - actual_return = self.cog.is_staff(user) - with self.subTest(user_type=user.name, expected_return=expected_return, actual_return=actual_return): - self.assertEqual(expected_return, actual_return) - - @helpers.async_test - async def test_has_green_checkmark_correctly_detects_presence_of_green_checkmark_emoji(self): - """The `has_green_checkmark` method should only return `True` if one is present.""" - test_cases = ( - ( - "No reactions", helpers.MockMessage(), False - ), - ( - "No green check mark reactions", - helpers.MockMessage(reactions=[ - helpers.MockReaction(emoji=self.unicode_duck_emoji, users=[self.bot.user]), - helpers.MockReaction(emoji=self.thumbs_up_emoji, users=[self.bot.user]) - ]), - False - ), - ( - "Green check mark reaction, but not from our bot", - helpers.MockMessage(reactions=[ - helpers.MockReaction(emoji=self.unicode_duck_emoji, users=[self.bot.user]), - helpers.MockReaction(emoji=self.checkmark_emoji, users=[self.staff_member]) - ]), - False - ), - ( - "Green check mark reaction, with one from the bot", - helpers.MockMessage(reactions=[ - helpers.MockReaction(emoji=self.unicode_duck_emoji, users=[self.bot.user]), - helpers.MockReaction(emoji=self.checkmark_emoji, users=[self.staff_member, self.bot.user]) - ]), - True - ) - ) - - for description, message, expected_return in test_cases: - actual_return = await self.cog.has_green_checkmark(message) - with self.subTest( - test_case=description, - expected_return=expected_return, - actual_return=actual_return - ): - self.assertEqual(expected_return, actual_return) - - def test_send_webhook_correctly_passes_on_arguments(self): - """The `send_webhook` method should pass the arguments to the webhook correctly.""" - self.cog.webhook = helpers.MockAsyncWebhook() - - content = "fake content" - username = "fake username" - avatar_url = "fake avatar_url" - embed = "fake embed" - - asyncio.run(self.cog.send_webhook(content, username, avatar_url, embed)) - - self.cog.webhook.send.assert_called_once_with( - content=content, - username=username, - avatar_url=avatar_url, - embed=embed - ) - - def test_send_webhook_logs_when_sending_message_fails(self): - """The `send_webhook` method should catch a `discord.HTTPException` and log accordingly.""" - self.cog.webhook = helpers.MockAsyncWebhook() - self.cog.webhook.send.side_effect = discord.HTTPException(response=MagicMock(), message="Something failed.") - - log = logging.getLogger('bot.cogs.duck_pond') - with self.assertLogs(logger=log, level=logging.ERROR) as log_watcher: - asyncio.run(self.cog.send_webhook()) - - self.assertEqual(len(log_watcher.records), 1) - - record = log_watcher.records[0] - self.assertEqual(record.levelno, logging.ERROR) - - def _get_reaction( - self, - emoji: typing.Union[str, helpers.MockEmoji], - staff: int = 0, - nonstaff: int = 0 - ) -> helpers.MockReaction: - staffers = [helpers.MockMember(roles=[self.staff_role]) for _ in range(staff)] - nonstaffers = [helpers.MockMember() for _ in range(nonstaff)] - return helpers.MockReaction(emoji=emoji, users=staffers + nonstaffers) - - @helpers.async_test - async def test_count_ducks_correctly_counts_the_number_of_eligible_duck_emojis(self): - """The `count_ducks` method should return the number of unique staffers who gave a duck.""" - test_cases = ( - # Simple test cases - # A message without reactions should return 0 - ( - "No reactions", - helpers.MockMessage(), - 0 - ), - # A message with a non-duck reaction from a non-staffer should return 0 - ( - "Non-duck reaction from non-staffer", - helpers.MockMessage(reactions=[self._get_reaction(emoji=self.thumbs_up_emoji, nonstaff=1)]), - 0 - ), - # A message with a non-duck reaction from a staffer should return 0 - ( - "Non-duck reaction from staffer", - helpers.MockMessage(reactions=[self._get_reaction(emoji=self.non_duck_custom_emoji, staff=1)]), - 0 - ), - # A message with a non-duck reaction from a non-staffer and staffer should return 0 - ( - "Non-duck reaction from staffer + non-staffer", - helpers.MockMessage(reactions=[self._get_reaction(emoji=self.thumbs_up_emoji, staff=1, nonstaff=1)]), - 0 - ), - # A message with a unicode duck reaction from a non-staffer should return 0 - ( - "Unicode Duck Reaction from non-staffer", - helpers.MockMessage(reactions=[self._get_reaction(emoji=self.unicode_duck_emoji, nonstaff=1)]), - 0 - ), - # A message with a unicode duck reaction from a staffer should return 1 - ( - "Unicode Duck Reaction from staffer", - helpers.MockMessage(reactions=[self._get_reaction(emoji=self.unicode_duck_emoji, staff=1)]), - 1 - ), - # A message with a unicode duck reaction from a non-staffer and staffer should return 1 - ( - "Unicode Duck Reaction from staffer + non-staffer", - helpers.MockMessage(reactions=[self._get_reaction(emoji=self.unicode_duck_emoji, staff=1, nonstaff=1)]), - 1 - ), - # A message with a duckpond duck reaction from a non-staffer should return 0 - ( - "Duckpond Duck Reaction from non-staffer", - helpers.MockMessage(reactions=[self._get_reaction(emoji=self.duck_pond_emoji, nonstaff=1)]), - 0 - ), - # A message with a duckpond duck reaction from a staffer should return 1 - ( - "Duckpond Duck Reaction from staffer", - helpers.MockMessage(reactions=[self._get_reaction(emoji=self.duck_pond_emoji, staff=1)]), - 1 - ), - # A message with a duckpond duck reaction from a non-staffer and staffer should return 1 - ( - "Duckpond Duck Reaction from staffer + non-staffer", - helpers.MockMessage(reactions=[self._get_reaction(emoji=self.duck_pond_emoji, staff=1, nonstaff=1)]), - 1 - ), - - # Complex test cases - # A message with duckpond duck reactions from 3 staffers and 2 non-staffers returns 3 - ( - "Duckpond Duck Reaction from 3 staffers + 2 non-staffers", - helpers.MockMessage(reactions=[self._get_reaction(emoji=self.duck_pond_emoji, staff=3, nonstaff=2)]), - 3 - ), - # A staffer with multiple duck reactions only counts once - ( - "Two different duck reactions from the same staffer", - helpers.MockMessage( - reactions=[ - helpers.MockReaction(emoji=self.duck_pond_emoji, users=[self.staff_member]), - helpers.MockReaction(emoji=self.unicode_duck_emoji, users=[self.staff_member]), - ] - ), - 1 - ), - # A non-string emoji does not count (to test the `isinstance(reaction.emoji, str)` elif) - ( - "Reaction with non-Emoji/str emoij from 3 staffers + 2 non-staffers", - helpers.MockMessage(reactions=[self._get_reaction(emoji=100, staff=3, nonstaff=2)]), - 0 - ), - # We correctly sum when multiple reactions are provided. - ( - "Duckpond Duck Reaction from 3 staffers + 2 non-staffers", - helpers.MockMessage( - reactions=[ - self._get_reaction(emoji=self.duck_pond_emoji, staff=3, nonstaff=2), - self._get_reaction(emoji=self.unicode_duck_emoji, staff=4, nonstaff=9), - ] - ), - 3 + 4 - ), - ) - - for description, message, expected_count in test_cases: - actual_count = await self.cog.count_ducks(message) - with self.subTest(test_case=description, expected_count=expected_count, actual_count=actual_count): - self.assertEqual(expected_count, actual_count) - - @helpers.async_test - async def test_relay_message_correctly_relays_content_and_attachments(self): - """The `relay_message` method should correctly relay message content and attachments.""" - send_webhook_path = f"{MODULE_PATH}.DuckPond.send_webhook" - send_attachments_path = f"{MODULE_PATH}.send_attachments" - - self.cog.webhook = helpers.MockAsyncWebhook() - - test_values = ( - (helpers.MockMessage(clean_content="", attachments=[]), False, False), - (helpers.MockMessage(clean_content="message", attachments=[]), True, False), - (helpers.MockMessage(clean_content="", attachments=["attachment"]), False, True), - (helpers.MockMessage(clean_content="message", attachments=["attachment"]), True, True), - ) - - for message, expect_webhook_call, expect_attachment_call in test_values: - with patch(send_webhook_path, new_callable=helpers.AsyncMock) as send_webhook: - with patch(send_attachments_path, new_callable=helpers.AsyncMock) as send_attachments: - with self.subTest(clean_content=message.clean_content, attachments=message.attachments): - await self.cog.relay_message(message) - - self.assertEqual(expect_webhook_call, send_webhook.called) - self.assertEqual(expect_attachment_call, send_attachments.called) - - message.add_reaction.assert_called_once_with(self.checkmark_emoji) - - @patch(f"{MODULE_PATH}.send_attachments", new_callable=helpers.AsyncMock) - @helpers.async_test - async def test_relay_message_handles_irretrievable_attachment_exceptions(self, send_attachments): - """The `relay_message` method should handle irretrievable attachments.""" - message = helpers.MockMessage(clean_content="message", attachments=["attachment"]) - side_effects = (discord.errors.Forbidden(MagicMock(), ""), discord.errors.NotFound(MagicMock(), "")) - - self.cog.webhook = helpers.MockAsyncWebhook() - log = logging.getLogger("bot.cogs.duck_pond") - - for side_effect in side_effects: - send_attachments.side_effect = side_effect - with patch(f"{MODULE_PATH}.DuckPond.send_webhook", new_callable=helpers.AsyncMock) as send_webhook: - with self.subTest(side_effect=type(side_effect).__name__): - with self.assertNotLogs(logger=log, level=logging.ERROR): - await self.cog.relay_message(message) - - self.assertEqual(send_webhook.call_count, 2) - - @patch(f"{MODULE_PATH}.DuckPond.send_webhook", new_callable=helpers.AsyncMock) - @patch(f"{MODULE_PATH}.send_attachments", new_callable=helpers.AsyncMock) - @helpers.async_test - async def test_relay_message_handles_attachment_http_error(self, send_attachments, send_webhook): - """The `relay_message` method should handle irretrievable attachments.""" - message = helpers.MockMessage(clean_content="message", attachments=["attachment"]) - - self.cog.webhook = helpers.MockAsyncWebhook() - log = logging.getLogger("bot.cogs.duck_pond") - - side_effect = discord.HTTPException(MagicMock(), "") - send_attachments.side_effect = side_effect - with self.subTest(side_effect=type(side_effect).__name__): - with self.assertLogs(logger=log, level=logging.ERROR) as log_watcher: - await self.cog.relay_message(message) - - send_webhook.assert_called_once_with( - content=message.clean_content, - username=message.author.display_name, - avatar_url=message.author.avatar_url - ) - - self.assertEqual(len(log_watcher.records), 1) - - record = log_watcher.records[0] - self.assertEqual(record.levelno, logging.ERROR) - - def _mock_payload(self, label: str, is_custom_emoji: bool, id_: int, emoji_name: str): - """Creates a mock `on_raw_reaction_add` payload with the specified emoji data.""" - payload = MagicMock(name=label) - payload.emoji.is_custom_emoji.return_value = is_custom_emoji - payload.emoji.id = id_ - payload.emoji.name = emoji_name - return payload - - @helpers.async_test - async def test_payload_has_duckpond_emoji_correctly_detects_relevant_emojis(self): - """The `on_raw_reaction_add` event handler should ignore irrelevant emojis.""" - test_values = ( - # Custom Emojis - ( - self._mock_payload( - label="Custom Duckpond Emoji", - is_custom_emoji=True, - id_=constants.DuckPond.custom_emojis[0], - emoji_name="" - ), - True - ), - ( - self._mock_payload( - label="Custom Non-Duckpond Emoji", - is_custom_emoji=True, - id_=123, - emoji_name="" - ), - False - ), - # Unicode Emojis - ( - self._mock_payload( - label="Unicode Duck Emoji", - is_custom_emoji=False, - id_=1, - emoji_name=self.unicode_duck_emoji - ), - True - ), - ( - self._mock_payload( - label="Unicode Non-Duck Emoji", - is_custom_emoji=False, - id_=1, - emoji_name=self.thumbs_up_emoji - ), - False - ), - ) - - for payload, expected_return in test_values: - actual_return = self.cog._payload_has_duckpond_emoji(payload) - with self.subTest(case=payload._mock_name, expected_return=expected_return, actual_return=actual_return): - self.assertEqual(expected_return, actual_return) - - @patch(f"{MODULE_PATH}.discord.utils.get") - @patch(f"{MODULE_PATH}.DuckPond._payload_has_duckpond_emoji", new=MagicMock(return_value=False)) - def test_on_raw_reaction_add_returns_early_with_payload_without_duck_emoji(self, utils_get): - """The `on_raw_reaction_add` method should return early if the payload does not contain a duck emoji.""" - self.assertIsNone(asyncio.run(self.cog.on_raw_reaction_add(payload=MagicMock()))) - - # Ensure we've returned before making an unnecessary API call in the lines of code after the emoji check - utils_get.assert_not_called() - - def _raw_reaction_mocks(self, channel_id, message_id, user_id): - """Sets up mocks for tests of the `on_raw_reaction_add` event listener.""" - channel = helpers.MockTextChannel(id=channel_id) - self.bot.get_all_channels.return_value = (channel,) - - message = helpers.MockMessage(id=message_id) - - channel.fetch_message.return_value = message - - member = helpers.MockMember(id=user_id, roles=[self.staff_role]) - message.guild.members = (member,) - - payload = MagicMock(channel_id=channel_id, message_id=message_id, user_id=user_id) - - return channel, message, member, payload - - @helpers.async_test - async def test_on_raw_reaction_add_returns_for_bot_and_non_staff_members(self): - """The `on_raw_reaction_add` event handler should return for bot users or non-staff members.""" - channel_id = 1234 - message_id = 2345 - user_id = 3456 - - channel, message, _, payload = self._raw_reaction_mocks(channel_id, message_id, user_id) - - test_cases = ( - ("non-staff member", helpers.MockMember(id=user_id)), - ("bot staff member", helpers.MockMember(id=user_id, roles=[self.staff_role], bot=True)), - ) - - payload.emoji = self.duck_pond_emoji - - for description, member in test_cases: - message.guild.members = (member, ) - with self.subTest(test_case=description), patch(f"{MODULE_PATH}.DuckPond.has_green_checkmark") as checkmark: - checkmark.side_effect = AssertionError( - "Expected method to return before calling `self.has_green_checkmark`." - ) - self.assertIsNone(await self.cog.on_raw_reaction_add(payload)) - - # Check that we did make it past the payload checks - channel.fetch_message.assert_called_once() - channel.fetch_message.reset_mock() - - @patch(f"{MODULE_PATH}.DuckPond.is_staff") - @patch(f"{MODULE_PATH}.DuckPond.count_ducks", new_callable=helpers.AsyncMock) - def test_on_raw_reaction_add_returns_on_message_with_green_checkmark_placed_by_bot(self, count_ducks, is_staff): - """The `on_raw_reaction_add` event should return when the message has a green check mark placed by the bot.""" - channel_id = 31415926535 - message_id = 27182818284 - user_id = 16180339887 - - channel, message, member, payload = self._raw_reaction_mocks(channel_id, message_id, user_id) - - payload.emoji = helpers.MockPartialEmoji(name=self.unicode_duck_emoji) - payload.emoji.is_custom_emoji.return_value = False - - message.reactions = [helpers.MockReaction(emoji=self.checkmark_emoji, users=[self.bot.user])] - - is_staff.return_value = True - count_ducks.side_effect = AssertionError("Expected method to return before calling `self.count_ducks`") - - self.assertIsNone(asyncio.run(self.cog.on_raw_reaction_add(payload))) - - # Assert that we've made it past `self.is_staff` - is_staff.assert_called_once() - - @helpers.async_test - async def test_on_raw_reaction_add_does_not_relay_below_duck_threshold(self): - """The `on_raw_reaction_add` listener should not relay messages or attachments below the duck threshold.""" - test_cases = ( - (constants.DuckPond.threshold - 1, False), - (constants.DuckPond.threshold, True), - (constants.DuckPond.threshold + 1, True), - ) - - channel, message, member, payload = self._raw_reaction_mocks(channel_id=3, message_id=4, user_id=5) - - payload.emoji = self.duck_pond_emoji - - for duck_count, should_relay in test_cases: - with patch(f"{MODULE_PATH}.DuckPond.relay_message", new_callable=helpers.AsyncMock) as relay_message: - with patch(f"{MODULE_PATH}.DuckPond.count_ducks", new_callable=helpers.AsyncMock) as count_ducks: - count_ducks.return_value = duck_count - with self.subTest(duck_count=duck_count, should_relay=should_relay): - await self.cog.on_raw_reaction_add(payload) - - # Confirm that we've made it past counting - count_ducks.assert_called_once() - - # Did we relay a message? - has_relayed = relay_message.called - self.assertEqual(has_relayed, should_relay) - - if should_relay: - relay_message.assert_called_once_with(message) - - @helpers.async_test - async def test_on_raw_reaction_remove_prevents_removal_of_green_checkmark_depending_on_the_duck_count(self): - """The `on_raw_reaction_remove` listener prevents removal of the check mark on messages with enough ducks.""" - checkmark = helpers.MockPartialEmoji(name=self.checkmark_emoji) - - message = helpers.MockMessage(id=1234) - - channel = helpers.MockTextChannel(id=98765) - channel.fetch_message.return_value = message - - self.bot.get_all_channels.return_value = (channel, ) - - payload = MagicMock(channel_id=channel.id, message_id=message.id, emoji=checkmark) - - test_cases = ( - (constants.DuckPond.threshold - 1, False), - (constants.DuckPond.threshold, True), - (constants.DuckPond.threshold + 1, True), - ) - for duck_count, should_re_add_checkmark in test_cases: - with patch(f"{MODULE_PATH}.DuckPond.count_ducks", new_callable=helpers.AsyncMock) as count_ducks: - count_ducks.return_value = duck_count - with self.subTest(duck_count=duck_count, should_re_add_checkmark=should_re_add_checkmark): - await self.cog.on_raw_reaction_remove(payload) - - # Check if we fetched the message - channel.fetch_message.assert_called_once_with(message.id) - - # Check if we actually counted the number of ducks - count_ducks.assert_called_once_with(message) - - has_re_added_checkmark = message.add_reaction.called - self.assertEqual(should_re_add_checkmark, has_re_added_checkmark) - - if should_re_add_checkmark: - message.add_reaction.assert_called_once_with(self.checkmark_emoji) - message.add_reaction.reset_mock() - - # reset mocks - channel.fetch_message.reset_mock() - message.reset_mock() - - def test_on_raw_reaction_remove_ignores_removal_of_non_checkmark_reactions(self): - """The `on_raw_reaction_remove` listener should ignore the removal of non-check mark emojis.""" - channel = helpers.MockTextChannel(id=98765) - - channel.fetch_message.side_effect = AssertionError( - "Expected method to return before calling `channel.fetch_message`" - ) - - self.bot.get_all_channels.return_value = (channel, ) - - payload = MagicMock(emoji=helpers.MockPartialEmoji(name=self.thumbs_up_emoji), channel_id=channel.id) - - self.assertIsNone(asyncio.run(self.cog.on_raw_reaction_remove(payload))) - - channel.fetch_message.assert_not_called() - - -class DuckPondSetupTests(unittest.TestCase): - """Tests setup of the `DuckPond` cog.""" - - def test_setup(self): - """Setup of the extension should call add_cog.""" - bot = helpers.MockBot() - duck_pond.setup(bot) - bot.add_cog.assert_called_once() diff --git a/tests/bot/cogs/test_token_remover.py b/tests/bot/cogs/test_token_remover.py deleted file mode 100644 index a54b839d7..000000000 --- a/tests/bot/cogs/test_token_remover.py +++ /dev/null @@ -1,131 +0,0 @@ -import asyncio -import logging -import unittest -from unittest.mock import MagicMock - -from discord import Colour - -from bot.cogs.token_remover import ( - DELETION_MESSAGE_TEMPLATE, - TokenRemover, - setup as setup_cog, -) -from bot.constants import Channels, Colours, Event, Icons -from tests.helpers import AsyncMock, MockBot, MockMessage - - -class TokenRemoverTests(unittest.TestCase): - """Tests the `TokenRemover` cog.""" - - def setUp(self): - """Adds the cog, a bot, and a message to the instance for usage in tests.""" - self.bot = MockBot() - self.bot.get_cog.return_value = MagicMock() - self.bot.get_cog.return_value.send_log_message = AsyncMock() - self.cog = TokenRemover(bot=self.bot) - - self.msg = MockMessage(id=555, content='') - self.msg.author.__str__ = MagicMock() - self.msg.author.__str__.return_value = 'lemon' - self.msg.author.bot = False - self.msg.author.avatar_url_as.return_value = 'picture-lemon.png' - self.msg.author.id = 42 - self.msg.author.mention = '@lemon' - self.msg.channel.mention = "#lemonade-stand" - - def test_is_valid_user_id_is_true_for_numeric_content(self): - """A string decoding to numeric characters is a valid user ID.""" - # MTIz = base64(123) - self.assertTrue(TokenRemover.is_valid_user_id('MTIz')) - - def test_is_valid_user_id_is_false_for_alphabetic_content(self): - """A string decoding to alphabetic characters is not a valid user ID.""" - # YWJj = base64(abc) - self.assertFalse(TokenRemover.is_valid_user_id('YWJj')) - - def test_is_valid_timestamp_is_true_for_valid_timestamps(self): - """A string decoding to a valid timestamp should be recognized as such.""" - self.assertTrue(TokenRemover.is_valid_timestamp('DN9r_A')) - - def test_is_valid_timestamp_is_false_for_invalid_values(self): - """A string not decoding to a valid timestamp should not be recognized as such.""" - # MTIz = base64(123) - self.assertFalse(TokenRemover.is_valid_timestamp('MTIz')) - - def test_mod_log_property(self): - """The `mod_log` property should ask the bot to return the `ModLog` cog.""" - self.bot.get_cog.return_value = 'lemon' - self.assertEqual(self.cog.mod_log, self.bot.get_cog.return_value) - self.bot.get_cog.assert_called_once_with('ModLog') - - def test_ignores_bot_messages(self): - """When the message event handler is called with a bot message, nothing is done.""" - self.msg.author.bot = True - coroutine = self.cog.on_message(self.msg) - self.assertIsNone(asyncio.run(coroutine)) - - def test_ignores_messages_without_tokens(self): - """Messages without anything looking like a token are ignored.""" - for content in ('', 'lemon wins'): - with self.subTest(content=content): - self.msg.content = content - coroutine = self.cog.on_message(self.msg) - self.assertIsNone(asyncio.run(coroutine)) - - def test_ignores_messages_with_invalid_tokens(self): - """Messages with values that are invalid tokens are ignored.""" - for content in ('foo.bar.baz', 'x.y.'): - with self.subTest(content=content): - self.msg.content = content - coroutine = self.cog.on_message(self.msg) - self.assertIsNone(asyncio.run(coroutine)) - - def test_censors_valid_tokens(self): - """Valid tokens are censored.""" - cases = ( - # (content, censored_token) - ('MTIz.DN9R_A.xyz', 'MTIz.DN9R_A.xxx'), - ) - - for content, censored_token in cases: - with self.subTest(content=content, censored_token=censored_token): - self.msg.content = content - coroutine = self.cog.on_message(self.msg) - with self.assertLogs(logger='bot.cogs.token_remover', level=logging.DEBUG) as cm: - self.assertIsNone(asyncio.run(coroutine)) # no return value - - [line] = cm.output - log_message = ( - "Censored a seemingly valid token sent by " - "lemon (`42`) in #lemonade-stand, " - f"token was `{censored_token}`" - ) - self.assertIn(log_message, line) - - self.msg.delete.assert_called_once_with() - self.msg.channel.send.assert_called_once_with( - DELETION_MESSAGE_TEMPLATE.format(mention='@lemon') - ) - self.bot.get_cog.assert_called_with('ModLog') - self.msg.author.avatar_url_as.assert_called_once_with(static_format='png') - - mod_log = self.bot.get_cog.return_value - mod_log.ignore.assert_called_once_with(Event.message_delete, self.msg.id) - mod_log.send_log_message.assert_called_once_with( - icon_url=Icons.token_removed, - colour=Colour(Colours.soft_red), - title="Token removed!", - text=log_message, - thumbnail='picture-lemon.png', - channel_id=Channels.mod_alerts - ) - - -class TokenRemoverSetupTests(unittest.TestCase): - """Tests setup of the `TokenRemover` cog.""" - - def test_setup(self): - """Setup of the extension should call add_cog.""" - bot = MockBot() - setup_cog(bot) - bot.add_cog.assert_called_once() diff --git a/tests/bot/cogs/__init__.py b/tests/bot/exts/__init__.py index e69de29bb..e69de29bb 100644 --- a/tests/bot/cogs/__init__.py +++ b/tests/bot/exts/__init__.py diff --git a/tests/bot/cogs/sync/__init__.py b/tests/bot/exts/backend/__init__.py index e69de29bb..e69de29bb 100644 --- a/tests/bot/cogs/sync/__init__.py +++ b/tests/bot/exts/backend/__init__.py diff --git a/tests/bot/patches/__init__.py b/tests/bot/exts/backend/sync/__init__.py index e69de29bb..e69de29bb 100644 --- a/tests/bot/patches/__init__.py +++ b/tests/bot/exts/backend/sync/__init__.py diff --git a/tests/bot/exts/backend/sync/test_base.py b/tests/bot/exts/backend/sync/test_base.py new file mode 100644 index 000000000..3ad9db9c3 --- /dev/null +++ b/tests/bot/exts/backend/sync/test_base.py @@ -0,0 +1,66 @@ +import unittest +from unittest import mock + + +from bot.api import ResponseCodeError +from bot.exts.backend.sync._syncers import Syncer +from tests import helpers + + +class TestSyncer(Syncer): + """Syncer subclass with mocks for abstract methods for testing purposes.""" + + name = "test" + _get_diff = mock.AsyncMock() + _sync = mock.AsyncMock() + + +class SyncerSyncTests(unittest.IsolatedAsyncioTestCase): + """Tests for main function orchestrating the sync.""" + + def setUp(self): + patcher = mock.patch("bot.instance", new=helpers.MockBot(user=helpers.MockMember(bot=True))) + self.bot = patcher.start() + self.addCleanup(patcher.stop) + + self.guild = helpers.MockGuild() + + TestSyncer._get_diff.reset_mock(return_value=True, side_effect=True) + TestSyncer._sync.reset_mock(return_value=True, side_effect=True) + + # Make sure `_get_diff` returns a MagicMock, not an AsyncMock + TestSyncer._get_diff.return_value = mock.MagicMock() + + async def test_sync_message_edited(self): + """The message should be edited if one was sent, even if the sync has an API error.""" + subtests = ( + (None, None, False), + (helpers.MockMessage(), None, True), + (helpers.MockMessage(), ResponseCodeError(mock.MagicMock()), True), + ) + + for message, side_effect, should_edit in subtests: + with self.subTest(message=message, side_effect=side_effect, should_edit=should_edit): + TestSyncer._sync.side_effect = side_effect + ctx = helpers.MockContext() + ctx.send.return_value = message + + await TestSyncer.sync(self.guild, ctx) + + if should_edit: + message.edit.assert_called_once() + self.assertIn("content", message.edit.call_args[1]) + + async def test_sync_message_sent(self): + """If ctx is given, a new message should be sent.""" + subtests = ( + (None, None), + (helpers.MockContext(), helpers.MockMessage()), + ) + + for ctx, message in subtests: + with self.subTest(ctx=ctx, message=message): + await TestSyncer.sync(self.guild, ctx) + + if ctx is not None: + ctx.send.assert_called_once() diff --git a/tests/bot/exts/backend/sync/test_cog.py b/tests/bot/exts/backend/sync/test_cog.py new file mode 100644 index 000000000..22a07313e --- /dev/null +++ b/tests/bot/exts/backend/sync/test_cog.py @@ -0,0 +1,414 @@ +import unittest +from unittest import mock + +import discord + +from bot import constants +from bot.api import ResponseCodeError +from bot.exts.backend import sync +from bot.exts.backend.sync._cog import Sync +from bot.exts.backend.sync._syncers import Syncer +from tests import helpers +from tests.base import CommandTestCase + + +class SyncExtensionTests(unittest.IsolatedAsyncioTestCase): + """Tests for the sync extension.""" + + @staticmethod + def test_extension_setup(): + """The Sync cog should be added.""" + bot = helpers.MockBot() + sync.setup(bot) + bot.add_cog.assert_called_once() + + +class SyncCogTestCase(unittest.IsolatedAsyncioTestCase): + """Base class for Sync cog tests. Sets up patches for syncers.""" + + def setUp(self): + self.bot = helpers.MockBot() + + role_syncer_patcher = mock.patch( + "bot.exts.backend.sync._syncers.RoleSyncer", + autospec=Syncer, + spec_set=True + ) + user_syncer_patcher = mock.patch( + "bot.exts.backend.sync._syncers.UserSyncer", + autospec=Syncer, + spec_set=True + ) + + self.RoleSyncer = role_syncer_patcher.start() + self.UserSyncer = user_syncer_patcher.start() + + self.addCleanup(role_syncer_patcher.stop) + self.addCleanup(user_syncer_patcher.stop) + + self.cog = Sync(self.bot) + + @staticmethod + def response_error(status: int) -> ResponseCodeError: + """Fixture to return a ResponseCodeError with the given status code.""" + response = mock.MagicMock() + response.status = status + + return ResponseCodeError(response) + + +class SyncCogTests(SyncCogTestCase): + """Tests for the Sync cog.""" + + @mock.patch.object(Sync, "sync_guild", new_callable=mock.MagicMock) + def test_sync_cog_init(self, sync_guild): + """Should instantiate syncers and run a sync for the guild.""" + # Reset because a Sync cog was already instantiated in setUp. + self.RoleSyncer.reset_mock() + self.UserSyncer.reset_mock() + self.bot.loop.create_task = mock.MagicMock() + + mock_sync_guild_coro = mock.MagicMock() + sync_guild.return_value = mock_sync_guild_coro + + Sync(self.bot) + + sync_guild.assert_called_once_with() + self.bot.loop.create_task.assert_called_once_with(mock_sync_guild_coro) + + async def test_sync_cog_sync_guild(self): + """Roles and users should be synced only if a guild is successfully retrieved.""" + for guild in (helpers.MockGuild(), None): + with self.subTest(guild=guild): + self.bot.reset_mock() + self.RoleSyncer.reset_mock() + self.UserSyncer.reset_mock() + + self.bot.get_guild = mock.MagicMock(return_value=guild) + + await self.cog.sync_guild() + + self.bot.wait_until_guild_available.assert_called_once() + self.bot.get_guild.assert_called_once_with(constants.Guild.id) + + if guild is None: + self.RoleSyncer.sync.assert_not_called() + self.UserSyncer.sync.assert_not_called() + else: + self.RoleSyncer.sync.assert_called_once_with(guild) + self.UserSyncer.sync.assert_called_once_with(guild) + + async def patch_user_helper(self, side_effect: BaseException) -> None: + """Helper to set a side effect for bot.api_client.patch and then assert it is called.""" + self.bot.api_client.patch.reset_mock(side_effect=True) + self.bot.api_client.patch.side_effect = side_effect + + user_id, updated_information = 5, {"key": 123} + await self.cog.patch_user(user_id, updated_information) + + self.bot.api_client.patch.assert_called_once_with( + f"bot/users/{user_id}", + json=updated_information, + ) + + async def test_sync_cog_patch_user(self): + """A PATCH request should be sent and 404 errors ignored.""" + for side_effect in (None, self.response_error(404)): + with self.subTest(side_effect=side_effect): + await self.patch_user_helper(side_effect) + + async def test_sync_cog_patch_user_non_404(self): + """A PATCH request should be sent and the error raised if it's not a 404.""" + with self.assertRaises(ResponseCodeError): + await self.patch_user_helper(self.response_error(500)) + + +class SyncCogListenerTests(SyncCogTestCase): + """Tests for the listeners of the Sync cog.""" + + def setUp(self): + super().setUp() + self.cog.patch_user = mock.AsyncMock(spec_set=self.cog.patch_user) + + self.guild_id_patcher = mock.patch("bot.exts.backend.sync._cog.constants.Guild.id", 5) + self.guild_id = self.guild_id_patcher.start() + + self.guild = helpers.MockGuild(id=self.guild_id) + self.other_guild = helpers.MockGuild(id=0) + + def tearDown(self): + self.guild_id_patcher.stop() + + async def test_sync_cog_on_guild_role_create(self): + """A POST request should be sent with the new role's data.""" + self.assertTrue(self.cog.on_guild_role_create.__cog_listener__) + + role_data = { + "colour": 49, + "id": 777, + "name": "rolename", + "permissions": 8, + "position": 23, + } + role = helpers.MockRole(**role_data, guild=self.guild) + await self.cog.on_guild_role_create(role) + + self.bot.api_client.post.assert_called_once_with("bot/roles", json=role_data) + + async def test_sync_cog_on_guild_role_create_ignores_guilds(self): + """Events from other guilds should be ignored.""" + role = helpers.MockRole(guild=self.other_guild) + await self.cog.on_guild_role_create(role) + self.bot.api_client.post.assert_not_awaited() + + async def test_sync_cog_on_guild_role_delete(self): + """A DELETE request should be sent.""" + self.assertTrue(self.cog.on_guild_role_delete.__cog_listener__) + + role = helpers.MockRole(id=99, guild=self.guild) + await self.cog.on_guild_role_delete(role) + + self.bot.api_client.delete.assert_called_once_with("bot/roles/99") + + async def test_sync_cog_on_guild_role_delete_ignores_guilds(self): + """Events from other guilds should be ignored.""" + role = helpers.MockRole(guild=self.other_guild) + await self.cog.on_guild_role_delete(role) + self.bot.api_client.delete.assert_not_awaited() + + async def test_sync_cog_on_guild_role_update(self): + """A PUT request should be sent if the colour, name, permissions, or position changes.""" + self.assertTrue(self.cog.on_guild_role_update.__cog_listener__) + + role_data = { + "colour": 49, + "id": 777, + "name": "rolename", + "permissions": 8, + "position": 23, + } + subtests = ( + (True, ("colour", "name", "permissions", "position")), + (False, ("hoist", "mentionable")), + ) + + for should_put, attributes in subtests: + for attribute in attributes: + with self.subTest(should_put=should_put, changed_attribute=attribute): + self.bot.api_client.put.reset_mock() + + after_role_data = role_data.copy() + after_role_data[attribute] = 876 + + before_role = helpers.MockRole(**role_data, guild=self.guild) + after_role = helpers.MockRole(**after_role_data, guild=self.guild) + + await self.cog.on_guild_role_update(before_role, after_role) + + if should_put: + self.bot.api_client.put.assert_called_once_with( + f"bot/roles/{after_role.id}", + json=after_role_data + ) + else: + self.bot.api_client.put.assert_not_called() + + async def test_sync_cog_on_guild_role_update_ignores_guilds(self): + """Events from other guilds should be ignored.""" + role = helpers.MockRole(guild=self.other_guild) + await self.cog.on_guild_role_update(role, role) + self.bot.api_client.put.assert_not_awaited() + + async def test_sync_cog_on_member_remove(self): + """Member should be patched to set in_guild as False.""" + self.assertTrue(self.cog.on_member_remove.__cog_listener__) + + member = helpers.MockMember(guild=self.guild) + await self.cog.on_member_remove(member) + + self.cog.patch_user.assert_called_once_with( + member.id, + json={"in_guild": False} + ) + + async def test_sync_cog_on_member_remove_ignores_guilds(self): + """Events from other guilds should be ignored.""" + member = helpers.MockMember(guild=self.other_guild) + await self.cog.on_member_remove(member) + self.cog.patch_user.assert_not_awaited() + + async def test_sync_cog_on_member_update_roles(self): + """Members should be patched if their roles have changed.""" + self.assertTrue(self.cog.on_member_update.__cog_listener__) + + # Roles are intentionally unsorted. + before_roles = [helpers.MockRole(id=12), helpers.MockRole(id=30), helpers.MockRole(id=20)] + before_member = helpers.MockMember(roles=before_roles, guild=self.guild) + after_member = helpers.MockMember(roles=before_roles[1:], guild=self.guild) + + await self.cog.on_member_update(before_member, after_member) + + data = {"roles": sorted(role.id for role in after_member.roles)} + self.cog.patch_user.assert_called_once_with(after_member.id, json=data) + + async def test_sync_cog_on_member_update_other(self): + """Members should not be patched if other attributes have changed.""" + self.assertTrue(self.cog.on_member_update.__cog_listener__) + + subtests = ( + ("activities", discord.Game("Pong"), discord.Game("Frogger")), + ("nick", "old nick", "new nick"), + ("status", discord.Status.online, discord.Status.offline), + ) + + for attribute, old_value, new_value in subtests: + with self.subTest(attribute=attribute): + self.cog.patch_user.reset_mock() + + before_member = helpers.MockMember(**{attribute: old_value}, guild=self.guild) + after_member = helpers.MockMember(**{attribute: new_value}, guild=self.guild) + + await self.cog.on_member_update(before_member, after_member) + + self.cog.patch_user.assert_not_called() + + async def test_sync_cog_on_member_update_ignores_guilds(self): + """Events from other guilds should be ignored.""" + member = helpers.MockMember(guild=self.other_guild) + await self.cog.on_member_update(member, member) + self.cog.patch_user.assert_not_awaited() + + async def test_sync_cog_on_user_update(self): + """A user should be patched only if the name, discriminator, or avatar changes.""" + self.assertTrue(self.cog.on_user_update.__cog_listener__) + + before_data = { + "name": "old name", + "discriminator": "1234", + "bot": False, + } + + subtests = ( + (True, "name", "name", "new name", "new name"), + (True, "discriminator", "discriminator", "8765", 8765), + (False, "bot", "bot", True, True), + ) + + for should_patch, attribute, api_field, value, api_value in subtests: + with self.subTest(attribute=attribute): + self.cog.patch_user.reset_mock() + + after_data = before_data.copy() + after_data[attribute] = value + before_user = helpers.MockUser(**before_data) + after_user = helpers.MockUser(**after_data) + + await self.cog.on_user_update(before_user, after_user) + + if should_patch: + self.cog.patch_user.assert_called_once() + + # Don't care if *all* keys are present; only the changed one is required + call_args = self.cog.patch_user.call_args + self.assertEqual(call_args.args[0], after_user.id) + self.assertIn("json", call_args.kwargs) + + self.assertIn("ignore_404", call_args.kwargs) + self.assertTrue(call_args.kwargs["ignore_404"]) + + json = call_args.kwargs["json"] + self.assertIn(api_field, json) + self.assertEqual(json[api_field], api_value) + else: + self.cog.patch_user.assert_not_called() + + async def on_member_join_helper(self, side_effect: Exception) -> dict: + """ + Helper to set `side_effect` for on_member_join and assert a PUT request was sent. + + The request data for the mock member is returned. All exceptions will be re-raised. + """ + member = helpers.MockMember( + discriminator="1234", + roles=[helpers.MockRole(id=22), helpers.MockRole(id=12)], + guild=self.guild, + ) + + data = { + "discriminator": int(member.discriminator), + "id": member.id, + "in_guild": True, + "name": member.name, + "roles": sorted(role.id for role in member.roles) + } + + self.bot.api_client.put.reset_mock(side_effect=True) + self.bot.api_client.put.side_effect = side_effect + + try: + await self.cog.on_member_join(member) + except Exception: + raise + finally: + self.bot.api_client.put.assert_called_once_with( + f"bot/users/{member.id}", + json=data + ) + + return data + + async def test_sync_cog_on_member_join(self): + """Should PUT user's data or POST it if the user doesn't exist.""" + for side_effect in (None, self.response_error(404)): + with self.subTest(side_effect=side_effect): + self.bot.api_client.post.reset_mock() + data = await self.on_member_join_helper(side_effect) + + if side_effect: + self.bot.api_client.post.assert_called_once_with("bot/users", json=data) + else: + self.bot.api_client.post.assert_not_called() + + async def test_sync_cog_on_member_join_non_404(self): + """ResponseCodeError should be re-raised if status code isn't a 404.""" + with self.assertRaises(ResponseCodeError): + await self.on_member_join_helper(self.response_error(500)) + + self.bot.api_client.post.assert_not_called() + + async def test_sync_cog_on_member_join_ignores_guilds(self): + """Events from other guilds should be ignored.""" + member = helpers.MockMember(guild=self.other_guild) + await self.cog.on_member_join(member) + self.bot.api_client.post.assert_not_awaited() + self.bot.api_client.put.assert_not_awaited() + + +class SyncCogCommandTests(SyncCogTestCase, CommandTestCase): + """Tests for the commands in the Sync cog.""" + + async def test_sync_roles_command(self): + """sync() should be called on the RoleSyncer.""" + ctx = helpers.MockContext() + await self.cog.sync_roles_command(self.cog, ctx) + + self.RoleSyncer.sync.assert_called_once_with(ctx.guild, ctx) + + async def test_sync_users_command(self): + """sync() should be called on the UserSyncer.""" + ctx = helpers.MockContext() + await self.cog.sync_users_command(self.cog, ctx) + + self.UserSyncer.sync.assert_called_once_with(ctx.guild, ctx) + + async def test_commands_require_admin(self): + """The sync commands should only run if the author has the administrator permission.""" + cmds = ( + self.cog.sync_group, + self.cog.sync_roles_command, + self.cog.sync_users_command, + ) + + for cmd in cmds: + with self.subTest(cmd=cmd): + await self.assertHasPermissionsCheck(cmd, {"administrator": True}) diff --git a/tests/bot/exts/backend/sync/test_roles.py b/tests/bot/exts/backend/sync/test_roles.py new file mode 100644 index 000000000..541074336 --- /dev/null +++ b/tests/bot/exts/backend/sync/test_roles.py @@ -0,0 +1,159 @@ +import unittest +from unittest import mock + +import discord + +from bot.exts.backend.sync._syncers import RoleSyncer, _Diff, _Role +from tests import helpers + + +def fake_role(**kwargs): + """Fixture to return a dictionary representing a role with default values set.""" + kwargs.setdefault("id", 9) + kwargs.setdefault("name", "fake role") + kwargs.setdefault("colour", 7) + kwargs.setdefault("permissions", 0) + kwargs.setdefault("position", 55) + + return kwargs + + +class RoleSyncerDiffTests(unittest.IsolatedAsyncioTestCase): + """Tests for determining differences between roles in the DB and roles in the Guild cache.""" + + def setUp(self): + patcher = mock.patch("bot.instance", new=helpers.MockBot()) + self.bot = patcher.start() + self.addCleanup(patcher.stop) + + @staticmethod + def get_guild(*roles): + """Fixture to return a guild object with the given roles.""" + guild = helpers.MockGuild() + guild.roles = [] + + for role in roles: + mock_role = helpers.MockRole(**role) + mock_role.colour = discord.Colour(role["colour"]) + mock_role.permissions = discord.Permissions(role["permissions"]) + guild.roles.append(mock_role) + + return guild + + async def test_empty_diff_for_identical_roles(self): + """No differences should be found if the roles in the guild and DB are identical.""" + self.bot.api_client.get.return_value = [fake_role()] + guild = self.get_guild(fake_role()) + + actual_diff = await RoleSyncer._get_diff(guild) + expected_diff = (set(), set(), set()) + + self.assertEqual(actual_diff, expected_diff) + + async def test_diff_for_updated_roles(self): + """Only updated roles should be added to the 'updated' set of the diff.""" + updated_role = fake_role(id=41, name="new") + + self.bot.api_client.get.return_value = [fake_role(id=41, name="old"), fake_role()] + guild = self.get_guild(updated_role, fake_role()) + + actual_diff = await RoleSyncer._get_diff(guild) + expected_diff = (set(), {_Role(**updated_role)}, set()) + + self.assertEqual(actual_diff, expected_diff) + + async def test_diff_for_new_roles(self): + """Only new roles should be added to the 'created' set of the diff.""" + new_role = fake_role(id=41, name="new") + + self.bot.api_client.get.return_value = [fake_role()] + guild = self.get_guild(fake_role(), new_role) + + actual_diff = await RoleSyncer._get_diff(guild) + expected_diff = ({_Role(**new_role)}, set(), set()) + + self.assertEqual(actual_diff, expected_diff) + + async def test_diff_for_deleted_roles(self): + """Only deleted roles should be added to the 'deleted' set of the diff.""" + deleted_role = fake_role(id=61, name="deleted") + + self.bot.api_client.get.return_value = [fake_role(), deleted_role] + guild = self.get_guild(fake_role()) + + actual_diff = await RoleSyncer._get_diff(guild) + expected_diff = (set(), set(), {_Role(**deleted_role)}) + + self.assertEqual(actual_diff, expected_diff) + + async def test_diff_for_new_updated_and_deleted_roles(self): + """When roles are added, updated, and removed, all of them are returned properly.""" + new = fake_role(id=41, name="new") + updated = fake_role(id=71, name="updated") + deleted = fake_role(id=61, name="deleted") + + self.bot.api_client.get.return_value = [ + fake_role(), + fake_role(id=71, name="updated name"), + deleted, + ] + guild = self.get_guild(fake_role(), new, updated) + + actual_diff = await RoleSyncer._get_diff(guild) + expected_diff = ({_Role(**new)}, {_Role(**updated)}, {_Role(**deleted)}) + + self.assertEqual(actual_diff, expected_diff) + + +class RoleSyncerSyncTests(unittest.IsolatedAsyncioTestCase): + """Tests for the API requests that sync roles.""" + + def setUp(self): + patcher = mock.patch("bot.instance", new=helpers.MockBot()) + self.bot = patcher.start() + self.addCleanup(patcher.stop) + + async def test_sync_created_roles(self): + """Only POST requests should be made with the correct payload.""" + roles = [fake_role(id=111), fake_role(id=222)] + + role_tuples = {_Role(**role) for role in roles} + diff = _Diff(role_tuples, set(), set()) + await RoleSyncer._sync(diff) + + calls = [mock.call("bot/roles", json=role) for role in roles] + self.bot.api_client.post.assert_has_calls(calls, any_order=True) + self.assertEqual(self.bot.api_client.post.call_count, len(roles)) + + self.bot.api_client.put.assert_not_called() + self.bot.api_client.delete.assert_not_called() + + async def test_sync_updated_roles(self): + """Only PUT requests should be made with the correct payload.""" + roles = [fake_role(id=111), fake_role(id=222)] + + role_tuples = {_Role(**role) for role in roles} + diff = _Diff(set(), role_tuples, set()) + await RoleSyncer._sync(diff) + + calls = [mock.call(f"bot/roles/{role['id']}", json=role) for role in roles] + self.bot.api_client.put.assert_has_calls(calls, any_order=True) + self.assertEqual(self.bot.api_client.put.call_count, len(roles)) + + self.bot.api_client.post.assert_not_called() + self.bot.api_client.delete.assert_not_called() + + async def test_sync_deleted_roles(self): + """Only DELETE requests should be made with the correct payload.""" + roles = [fake_role(id=111), fake_role(id=222)] + + role_tuples = {_Role(**role) for role in roles} + diff = _Diff(set(), set(), role_tuples) + await RoleSyncer._sync(diff) + + calls = [mock.call(f"bot/roles/{role['id']}") for role in roles] + self.bot.api_client.delete.assert_has_calls(calls, any_order=True) + self.assertEqual(self.bot.api_client.delete.call_count, len(roles)) + + self.bot.api_client.post.assert_not_called() + self.bot.api_client.put.assert_not_called() diff --git a/tests/bot/exts/backend/sync/test_users.py b/tests/bot/exts/backend/sync/test_users.py new file mode 100644 index 000000000..61673e1bb --- /dev/null +++ b/tests/bot/exts/backend/sync/test_users.py @@ -0,0 +1,217 @@ +import unittest +from unittest import mock + +from bot.exts.backend.sync._syncers import UserSyncer, _Diff +from tests import helpers + + +def fake_user(**kwargs): + """Fixture to return a dictionary representing a user with default values set.""" + kwargs.setdefault("id", 43) + kwargs.setdefault("name", "bob the test man") + kwargs.setdefault("discriminator", 1337) + kwargs.setdefault("roles", [666]) + kwargs.setdefault("in_guild", True) + + return kwargs + + +class UserSyncerDiffTests(unittest.IsolatedAsyncioTestCase): + """Tests for determining differences between users in the DB and users in the Guild cache.""" + + def setUp(self): + patcher = mock.patch("bot.instance", new=helpers.MockBot()) + self.bot = patcher.start() + self.addCleanup(patcher.stop) + + @staticmethod + def get_guild(*members): + """Fixture to return a guild object with the given members.""" + guild = helpers.MockGuild() + guild.members = [] + + for member in members: + member = member.copy() + del member["in_guild"] + + mock_member = helpers.MockMember(**member) + mock_member.roles = [helpers.MockRole(id=role_id) for role_id in member["roles"]] + + guild.members.append(mock_member) + + return guild + + @staticmethod + def get_mock_member(member: dict): + member = member.copy() + del member["in_guild"] + mock_member = helpers.MockMember(**member) + mock_member.roles = [helpers.MockRole(id=role_id) for role_id in member["roles"]] + return mock_member + + async def test_empty_diff_for_no_users(self): + """When no users are given, an empty diff should be returned.""" + self.bot.api_client.get.return_value = { + "count": 3, + "next_page_no": None, + "previous_page_no": None, + "results": [] + } + guild = self.get_guild() + + actual_diff = await UserSyncer._get_diff(guild) + expected_diff = ([], [], None) + + self.assertEqual(actual_diff, expected_diff) + + async def test_empty_diff_for_identical_users(self): + """No differences should be found if the users in the guild and DB are identical.""" + self.bot.api_client.get.return_value = { + "count": 3, + "next_page_no": None, + "previous_page_no": None, + "results": [fake_user()] + } + guild = self.get_guild(fake_user()) + + guild.get_member.return_value = self.get_mock_member(fake_user()) + actual_diff = await UserSyncer._get_diff(guild) + expected_diff = ([], [], None) + + self.assertEqual(actual_diff, expected_diff) + + async def test_diff_for_updated_users(self): + """Only updated users should be added to the 'updated' set of the diff.""" + updated_user = fake_user(id=99, name="new") + + self.bot.api_client.get.return_value = { + "count": 3, + "next_page_no": None, + "previous_page_no": None, + "results": [fake_user(id=99, name="old"), fake_user()] + } + guild = self.get_guild(updated_user, fake_user()) + guild.get_member.side_effect = [ + self.get_mock_member(updated_user), + self.get_mock_member(fake_user()) + ] + + actual_diff = await UserSyncer._get_diff(guild) + expected_diff = ([], [{"id": 99, "name": "new"}], None) + + self.assertEqual(actual_diff, expected_diff) + + async def test_diff_for_new_users(self): + """Only new users should be added to the 'created' list of the diff.""" + new_user = fake_user(id=99, name="new") + + self.bot.api_client.get.return_value = { + "count": 3, + "next_page_no": None, + "previous_page_no": None, + "results": [fake_user()] + } + guild = self.get_guild(fake_user(), new_user) + guild.get_member.side_effect = [ + self.get_mock_member(fake_user()), + self.get_mock_member(new_user) + ] + actual_diff = await UserSyncer._get_diff(guild) + expected_diff = ([new_user], [], None) + + self.assertEqual(actual_diff, expected_diff) + + async def test_diff_sets_in_guild_false_for_leaving_users(self): + """When a user leaves the guild, the `in_guild` flag is updated to `False`.""" + self.bot.api_client.get.return_value = { + "count": 3, + "next_page_no": None, + "previous_page_no": None, + "results": [fake_user(), fake_user(id=63)] + } + guild = self.get_guild(fake_user()) + guild.get_member.side_effect = [ + self.get_mock_member(fake_user()), + None + ] + + actual_diff = await UserSyncer._get_diff(guild) + expected_diff = ([], [{"id": 63, "in_guild": False}], None) + + self.assertEqual(actual_diff, expected_diff) + + async def test_diff_for_new_updated_and_leaving_users(self): + """When users are added, updated, and removed, all of them are returned properly.""" + new_user = fake_user(id=99, name="new") + + updated_user = fake_user(id=55, name="updated") + + self.bot.api_client.get.return_value = { + "count": 3, + "next_page_no": None, + "previous_page_no": None, + "results": [fake_user(), fake_user(id=55), fake_user(id=63)] + } + guild = self.get_guild(fake_user(), new_user, updated_user) + guild.get_member.side_effect = [ + self.get_mock_member(fake_user()), + self.get_mock_member(updated_user), + None + ] + + actual_diff = await UserSyncer._get_diff(guild) + expected_diff = ([new_user], [{"id": 55, "name": "updated"}, {"id": 63, "in_guild": False}], None) + + self.assertEqual(actual_diff, expected_diff) + + async def test_empty_diff_for_db_users_not_in_guild(self): + """When the DB knows a user, but the guild doesn't, no difference is found.""" + self.bot.api_client.get.return_value = { + "count": 3, + "next_page_no": None, + "previous_page_no": None, + "results": [fake_user(), fake_user(id=63, in_guild=False)] + } + guild = self.get_guild(fake_user()) + guild.get_member.side_effect = [ + self.get_mock_member(fake_user()), + None + ] + + actual_diff = await UserSyncer._get_diff(guild) + expected_diff = ([], [], None) + + self.assertEqual(actual_diff, expected_diff) + + +class UserSyncerSyncTests(unittest.IsolatedAsyncioTestCase): + """Tests for the API requests that sync users.""" + + def setUp(self): + patcher = mock.patch("bot.instance", new=helpers.MockBot()) + self.bot = patcher.start() + self.addCleanup(patcher.stop) + + async def test_sync_created_users(self): + """Only POST requests should be made with the correct payload.""" + users = [fake_user(id=111), fake_user(id=222)] + + diff = _Diff(users, [], None) + await UserSyncer._sync(diff) + + self.bot.api_client.post.assert_called_once_with("bot/users", json=diff.created) + + self.bot.api_client.put.assert_not_called() + self.bot.api_client.delete.assert_not_called() + + async def test_sync_updated_users(self): + """Only PUT requests should be made with the correct payload.""" + users = [fake_user(id=111), fake_user(id=222)] + + diff = _Diff([], users, None) + await UserSyncer._sync(diff) + + self.bot.api_client.patch.assert_called_once_with("bot/users/bulk_patch", json=diff.updated) + + self.bot.api_client.post.assert_not_called() + self.bot.api_client.delete.assert_not_called() diff --git a/tests/bot/exts/backend/test_logging.py b/tests/bot/exts/backend/test_logging.py new file mode 100644 index 000000000..466f207d9 --- /dev/null +++ b/tests/bot/exts/backend/test_logging.py @@ -0,0 +1,32 @@ +import unittest +from unittest.mock import patch + +from bot import constants +from bot.exts.backend.logging import Logging +from tests.helpers import MockBot, MockTextChannel + + +class LoggingTests(unittest.IsolatedAsyncioTestCase): + """Test cases for connected login.""" + + def setUp(self): + self.bot = MockBot() + self.cog = Logging(self.bot) + self.dev_log = MockTextChannel(id=1234, name="dev-log") + + @patch("bot.exts.backend.logging.DEBUG_MODE", False) + async def test_debug_mode_false(self): + """Should send connected message to dev-log.""" + self.bot.get_channel.return_value = self.dev_log + + await self.cog.startup_greeting() + self.bot.wait_until_guild_available.assert_awaited_once_with() + self.bot.get_channel.assert_called_once_with(constants.Channels.dev_log) + self.dev_log.send.assert_awaited_once() + + @patch("bot.exts.backend.logging.DEBUG_MODE", True) + async def test_debug_mode_true(self): + """Should not send anything to dev-log.""" + await self.cog.startup_greeting() + self.bot.wait_until_guild_available.assert_awaited_once_with() + self.bot.get_channel.assert_not_called() diff --git a/tests/bot/exts/filters/__init__.py b/tests/bot/exts/filters/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/tests/bot/exts/filters/__init__.py diff --git a/tests/bot/exts/filters/test_antimalware.py b/tests/bot/exts/filters/test_antimalware.py new file mode 100644 index 000000000..3393c6cdc --- /dev/null +++ b/tests/bot/exts/filters/test_antimalware.py @@ -0,0 +1,187 @@ +import unittest +from unittest.mock import AsyncMock, Mock + +from discord import NotFound + +from bot.constants import Channels, STAFF_ROLES +from bot.exts.filters import antimalware +from tests.helpers import MockAttachment, MockBot, MockMessage, MockRole + + +class AntiMalwareCogTests(unittest.IsolatedAsyncioTestCase): + """Test the AntiMalware cog.""" + + def setUp(self): + """Sets up fresh objects for each test.""" + self.bot = MockBot() + self.bot.filter_list_cache = { + "FILE_FORMAT.True": { + ".first": {}, + ".second": {}, + ".third": {}, + } + } + self.cog = antimalware.AntiMalware(self.bot) + self.message = MockMessage() + self.message.webhook_id = None + self.message.author.bot = None + self.whitelist = [".first", ".second", ".third"] + + async def test_message_with_allowed_attachment(self): + """Messages with allowed extensions should not be deleted""" + attachment = MockAttachment(filename="python.first") + self.message.attachments = [attachment] + + await self.cog.on_message(self.message) + self.message.delete.assert_not_called() + + async def test_message_without_attachment(self): + """Messages without attachments should result in no action.""" + await self.cog.on_message(self.message) + self.message.delete.assert_not_called() + + async def test_direct_message_with_attachment(self): + """Direct messages should have no action taken.""" + attachment = MockAttachment(filename="python.disallowed") + self.message.attachments = [attachment] + self.message.guild = None + + await self.cog.on_message(self.message) + + self.message.delete.assert_not_called() + + async def test_webhook_message_with_illegal_extension(self): + """A webhook message containing an illegal extension should be ignored.""" + attachment = MockAttachment(filename="python.disallowed") + self.message.webhook_id = 697140105563078727 + self.message.attachments = [attachment] + + await self.cog.on_message(self.message) + + self.message.delete.assert_not_called() + + async def test_bot_message_with_illegal_extension(self): + """A bot message containing an illegal extension should be ignored.""" + attachment = MockAttachment(filename="python.disallowed") + self.message.author.bot = 409107086526644234 + self.message.attachments = [attachment] + + await self.cog.on_message(self.message) + + self.message.delete.assert_not_called() + + async def test_message_with_illegal_extension_gets_deleted(self): + """A message containing an illegal extension should send an embed.""" + attachment = MockAttachment(filename="python.disallowed") + self.message.attachments = [attachment] + + await self.cog.on_message(self.message) + + self.message.delete.assert_called_once() + + async def test_message_send_by_staff(self): + """A message send by a member of staff should be ignored.""" + staff_role = MockRole(id=STAFF_ROLES[0]) + self.message.author.roles.append(staff_role) + attachment = MockAttachment(filename="python.disallowed") + self.message.attachments = [attachment] + + await self.cog.on_message(self.message) + + self.message.delete.assert_not_called() + + async def test_python_file_redirect_embed_description(self): + """A message containing a .py file should result in an embed redirecting the user to our paste site""" + attachment = MockAttachment(filename="python.py") + self.message.attachments = [attachment] + self.message.channel.send = AsyncMock() + + await self.cog.on_message(self.message) + self.message.channel.send.assert_called_once() + args, kwargs = self.message.channel.send.call_args + embed = kwargs.pop("embed") + + self.assertEqual(embed.description, antimalware.PY_EMBED_DESCRIPTION) + + async def test_txt_file_redirect_embed_description(self): + """A message containing a .txt file should result in the correct embed.""" + attachment = MockAttachment(filename="python.txt") + self.message.attachments = [attachment] + self.message.channel.send = AsyncMock() + antimalware.TXT_EMBED_DESCRIPTION = Mock() + antimalware.TXT_EMBED_DESCRIPTION.format.return_value = "test" + + await self.cog.on_message(self.message) + self.message.channel.send.assert_called_once() + args, kwargs = self.message.channel.send.call_args + embed = kwargs.pop("embed") + cmd_channel = self.bot.get_channel(Channels.bot_commands) + + self.assertEqual(embed.description, antimalware.TXT_EMBED_DESCRIPTION.format.return_value) + antimalware.TXT_EMBED_DESCRIPTION.format.assert_called_with(cmd_channel_mention=cmd_channel.mention) + + async def test_other_disallowed_extension_embed_description(self): + """Test the description for a non .py/.txt disallowed extension.""" + attachment = MockAttachment(filename="python.disallowed") + self.message.attachments = [attachment] + self.message.channel.send = AsyncMock() + antimalware.DISALLOWED_EMBED_DESCRIPTION = Mock() + antimalware.DISALLOWED_EMBED_DESCRIPTION.format.return_value = "test" + + await self.cog.on_message(self.message) + self.message.channel.send.assert_called_once() + args, kwargs = self.message.channel.send.call_args + embed = kwargs.pop("embed") + meta_channel = self.bot.get_channel(Channels.meta) + + self.assertEqual(embed.description, antimalware.DISALLOWED_EMBED_DESCRIPTION.format.return_value) + antimalware.DISALLOWED_EMBED_DESCRIPTION.format.assert_called_with( + joined_whitelist=", ".join(self.whitelist), + blocked_extensions_str=".disallowed", + meta_channel_mention=meta_channel.mention + ) + + async def test_removing_deleted_message_logs(self): + """Removing an already deleted message logs the correct message""" + attachment = MockAttachment(filename="python.disallowed") + self.message.attachments = [attachment] + self.message.delete = AsyncMock(side_effect=NotFound(response=Mock(status=""), message="")) + + with self.assertLogs(logger=antimalware.log, level="INFO"): + await self.cog.on_message(self.message) + self.message.delete.assert_called_once() + + async def test_message_with_illegal_attachment_logs(self): + """Deleting a message with an illegal attachment should result in a log.""" + attachment = MockAttachment(filename="python.disallowed") + self.message.attachments = [attachment] + + with self.assertLogs(logger=antimalware.log, level="INFO"): + await self.cog.on_message(self.message) + + async def test_get_disallowed_extensions(self): + """The return value should include all non-whitelisted extensions.""" + test_values = ( + ([], []), + (self.whitelist, []), + ([".first"], []), + ([".first", ".disallowed"], [".disallowed"]), + ([".disallowed"], [".disallowed"]), + ([".disallowed", ".illegal"], [".disallowed", ".illegal"]), + ) + + for extensions, expected_disallowed_extensions in test_values: + with self.subTest(extensions=extensions, expected_disallowed_extensions=expected_disallowed_extensions): + self.message.attachments = [MockAttachment(filename=f"filename{extension}") for extension in extensions] + disallowed_extensions = self.cog._get_disallowed_extensions(self.message) + self.assertCountEqual(disallowed_extensions, expected_disallowed_extensions) + + +class AntiMalwareSetupTests(unittest.TestCase): + """Tests setup of the `AntiMalware` cog.""" + + def test_setup(self): + """Setup of the extension should call add_cog.""" + bot = MockBot() + antimalware.setup(bot) + bot.add_cog.assert_called_once() diff --git a/tests/bot/cogs/test_antispam.py b/tests/bot/exts/filters/test_antispam.py index ce5472c71..6a0e4fded 100644 --- a/tests/bot/cogs/test_antispam.py +++ b/tests/bot/exts/filters/test_antispam.py @@ -1,6 +1,6 @@ import unittest -from bot.cogs import antispam +from bot.exts.filters import antispam class AntispamConfigurationValidationTests(unittest.TestCase): diff --git a/tests/bot/cogs/test_security.py b/tests/bot/exts/filters/test_security.py index 9d1a62f7e..c0c3baa42 100644 --- a/tests/bot/cogs/test_security.py +++ b/tests/bot/exts/filters/test_security.py @@ -3,7 +3,7 @@ from unittest.mock import MagicMock from discord.ext.commands import NoPrivateMessage -from bot.cogs import security +from bot.exts.filters import security from tests.helpers import MockBot, MockContext diff --git a/tests/bot/exts/filters/test_token_remover.py b/tests/bot/exts/filters/test_token_remover.py new file mode 100644 index 000000000..f99cc3370 --- /dev/null +++ b/tests/bot/exts/filters/test_token_remover.py @@ -0,0 +1,408 @@ +import unittest +from re import Match +from unittest import mock +from unittest.mock import MagicMock + +from discord import Colour, NotFound + +from bot import constants +from bot.exts.filters import token_remover +from bot.exts.filters.token_remover import Token, TokenRemover +from bot.exts.moderation.modlog import ModLog +from bot.utils.messages import format_user +from tests.helpers import MockBot, MockMessage, autospec + + +class TokenRemoverTests(unittest.IsolatedAsyncioTestCase): + """Tests the `TokenRemover` cog.""" + + def setUp(self): + """Adds the cog, a bot, and a message to the instance for usage in tests.""" + self.bot = MockBot() + self.cog = TokenRemover(bot=self.bot) + + self.msg = MockMessage(id=555, content="hello world") + self.msg.channel.mention = "#lemonade-stand" + self.msg.guild.get_member.return_value.bot = False + self.msg.guild.get_member.return_value.__str__.return_value = "Woody" + self.msg.author.__str__ = MagicMock(return_value=self.msg.author.name) + self.msg.author.avatar_url_as.return_value = "picture-lemon.png" + + def test_extract_user_id_valid(self): + """Should consider user IDs valid if they decode into an integer ID.""" + id_pairs = ( + ("NDcyMjY1OTQzMDYyNDEzMzMy", 472265943062413332), + ("NDc1MDczNjI5Mzk5NTQ3OTA0", 475073629399547904), + ("NDY3MjIzMjMwNjUwNzc3NjQx", 467223230650777641), + ) + + for token_id, user_id in id_pairs: + with self.subTest(token_id=token_id): + result = TokenRemover.extract_user_id(token_id) + self.assertEqual(result, user_id) + + def test_extract_user_id_invalid(self): + """Should consider non-digit and non-ASCII IDs invalid.""" + ids = ( + ("SGVsbG8gd29ybGQ", "non-digit ASCII"), + ("0J_RgNC40LLQtdGCINC80LjRgA", "cyrillic text"), + ("4pO14p6L4p6C4pG34p264pGl8J-EiOKSj-KCieKBsA", "Unicode digits"), + ("4oaA4oaB4oWh4oWi4Lyz4Lyq4Lyr4LG9", "Unicode numerals"), + ("8J2fjvCdn5nwnZ-k8J2fr_Cdn7rgravvvJngr6c", "Unicode decimals"), + ("{hello}[world]&(bye!)", "ASCII invalid Base64"), + ("Þíß-ï§-ňøẗ-våłìÐ", "Unicode invalid Base64"), + ) + + for user_id, msg in ids: + with self.subTest(msg=msg): + result = TokenRemover.extract_user_id(user_id) + self.assertIsNone(result) + + def test_is_valid_timestamp_valid(self): + """Should consider timestamps valid if they're greater than the Discord epoch.""" + timestamps = ( + "XsyRkw", + "Xrim9Q", + "XsyR-w", + "XsySD_", + "Dn9r_A", + ) + + for timestamp in timestamps: + with self.subTest(timestamp=timestamp): + result = TokenRemover.is_valid_timestamp(timestamp) + self.assertTrue(result) + + def test_is_valid_timestamp_invalid(self): + """Should consider timestamps invalid if they're before Discord epoch or can't be parsed.""" + timestamps = ( + ("B4Yffw", "DISCORD_EPOCH - TOKEN_EPOCH - 1"), + ("ew", "123"), + ("AoIKgA", "42076800"), + ("{hello}[world]&(bye!)", "ASCII invalid Base64"), + ("Þíß-ï§-ňøẗ-våłìÐ", "Unicode invalid Base64"), + ) + + for timestamp, msg in timestamps: + with self.subTest(msg=msg): + result = TokenRemover.is_valid_timestamp(timestamp) + self.assertFalse(result) + + def test_is_valid_hmac_valid(self): + """Should consider an HMAC valid if it has at least 3 unique characters.""" + valid_hmacs = ( + "VXmErH7j511turNpfURmb0rVNm8", + "Ysnu2wacjaKs7qnoo46S8Dm2us8", + "sJf6omBPORBPju3WJEIAcwW9Zds", + "s45jqDV_Iisn-symw0yDRrk_jf4", + ) + + for hmac in valid_hmacs: + with self.subTest(msg=hmac): + result = TokenRemover.is_maybe_valid_hmac(hmac) + self.assertTrue(result) + + def test_is_invalid_hmac_invalid(self): + """Should consider an HMAC invalid if has fewer than 3 unique characters.""" + invalid_hmacs = ( + ("xxxxxxxxxxxxxxxxxx", "Single character"), + ("XxXxXxXxXxXxXxXxXx", "Single character alternating case"), + ("ASFasfASFasfASFASsf", "Three characters alternating-case"), + ("asdasdasdasdasdasdasd", "Three characters one case"), + ) + + for hmac, msg in invalid_hmacs: + with self.subTest(msg=msg): + result = TokenRemover.is_maybe_valid_hmac(hmac) + self.assertFalse(result) + + def test_mod_log_property(self): + """The `mod_log` property should ask the bot to return the `ModLog` cog.""" + self.bot.get_cog.return_value = 'lemon' + self.assertEqual(self.cog.mod_log, self.bot.get_cog.return_value) + self.bot.get_cog.assert_called_once_with('ModLog') + + async def test_on_message_edit_uses_on_message(self): + """The edit listener should delegate handling of the message to the normal listener.""" + self.cog.on_message = mock.create_autospec(self.cog.on_message, spec_set=True) + + await self.cog.on_message_edit(MockMessage(), self.msg) + self.cog.on_message.assert_awaited_once_with(self.msg) + + @autospec(TokenRemover, "find_token_in_message", "take_action") + async def test_on_message_takes_action(self, find_token_in_message, take_action): + """Should take action if a valid token is found when a message is sent.""" + cog = TokenRemover(self.bot) + found_token = "foobar" + find_token_in_message.return_value = found_token + + await cog.on_message(self.msg) + + find_token_in_message.assert_called_once_with(self.msg) + take_action.assert_awaited_once_with(cog, self.msg, found_token) + + @autospec(TokenRemover, "find_token_in_message", "take_action") + async def test_on_message_skips_missing_token(self, find_token_in_message, take_action): + """Shouldn't take action if a valid token isn't found when a message is sent.""" + cog = TokenRemover(self.bot) + find_token_in_message.return_value = False + + await cog.on_message(self.msg) + + find_token_in_message.assert_called_once_with(self.msg) + take_action.assert_not_awaited() + + @autospec(TokenRemover, "find_token_in_message") + async def test_on_message_ignores_dms_bots(self, find_token_in_message): + """Shouldn't parse a message if it is a DM or authored by a bot.""" + cog = TokenRemover(self.bot) + dm_msg = MockMessage(guild=None) + bot_msg = MockMessage(author=MagicMock(bot=True)) + + for msg in (dm_msg, bot_msg): + await cog.on_message(msg) + find_token_in_message.assert_not_called() + + @autospec("bot.exts.filters.token_remover", "TOKEN_RE") + def test_find_token_no_matches(self, token_re): + """None should be returned if the regex matches no tokens in a message.""" + token_re.finditer.return_value = () + + return_value = TokenRemover.find_token_in_message(self.msg) + + self.assertIsNone(return_value) + token_re.finditer.assert_called_once_with(self.msg.content) + + @autospec(TokenRemover, "extract_user_id", "is_valid_timestamp", "is_maybe_valid_hmac") + @autospec("bot.exts.filters.token_remover", "Token") + @autospec("bot.exts.filters.token_remover", "TOKEN_RE") + def test_find_token_valid_match( + self, + token_re, + token_cls, + extract_user_id, + is_valid_timestamp, + is_maybe_valid_hmac, + ): + """The first match with a valid user ID, timestamp, and HMAC should be returned as a `Token`.""" + matches = [ + mock.create_autospec(Match, spec_set=True, instance=True), + mock.create_autospec(Match, spec_set=True, instance=True), + ] + tokens = [ + mock.create_autospec(Token, spec_set=True, instance=True), + mock.create_autospec(Token, spec_set=True, instance=True), + ] + + token_re.finditer.return_value = matches + token_cls.side_effect = tokens + extract_user_id.side_effect = (None, True) # The 1st match will be invalid, 2nd one valid. + is_valid_timestamp.return_value = True + is_maybe_valid_hmac.return_value = True + + return_value = TokenRemover.find_token_in_message(self.msg) + + self.assertEqual(tokens[1], return_value) + token_re.finditer.assert_called_once_with(self.msg.content) + + @autospec(TokenRemover, "extract_user_id", "is_valid_timestamp", "is_maybe_valid_hmac") + @autospec("bot.exts.filters.token_remover", "Token") + @autospec("bot.exts.filters.token_remover", "TOKEN_RE") + def test_find_token_invalid_matches( + self, + token_re, + token_cls, + extract_user_id, + is_valid_timestamp, + is_maybe_valid_hmac, + ): + """None should be returned if no matches have valid user IDs, HMACs, and timestamps.""" + token_re.finditer.return_value = [mock.create_autospec(Match, spec_set=True, instance=True)] + token_cls.return_value = mock.create_autospec(Token, spec_set=True, instance=True) + extract_user_id.return_value = None + is_valid_timestamp.return_value = False + is_maybe_valid_hmac.return_value = False + + return_value = TokenRemover.find_token_in_message(self.msg) + + self.assertIsNone(return_value) + token_re.finditer.assert_called_once_with(self.msg.content) + + def test_regex_invalid_tokens(self): + """Messages without anything looking like a token are not matched.""" + tokens = ( + "", + "lemon wins", + "..", + "x.y", + "x.y.", + ".y.z", + ".y.", + "..z", + "x..z", + " . . ", + "\n.\n.\n", + "hellö.world.bye", + "base64.nötbåse64.morebase64", + "19jd3J.dfkm3d.€víł§tüff", + ) + + for token in tokens: + with self.subTest(token=token): + results = token_remover.TOKEN_RE.findall(token) + self.assertEqual(len(results), 0) + + def test_regex_valid_tokens(self): + """Messages that look like tokens should be matched.""" + # Don't worry, these tokens have been invalidated. + tokens = ( + "NDcyMjY1OTQzMDYy_DEzMz-y.XsyRkw.VXmErH7j511turNpfURmb0rVNm8", + "NDcyMjY1OTQzMDYyNDEzMzMy.Xrim9Q.Ysnu2wacjaKs7qnoo46S8Dm2us8", + "NDc1MDczNjI5Mzk5NTQ3OTA0.XsyR-w.sJf6omBPORBPju3WJEIAcwW9Zds", + "NDY3MjIzMjMwNjUwNzc3NjQx.XsySD_.s45jqDV_Iisn-symw0yDRrk_jf4", + ) + + for token in tokens: + with self.subTest(token=token): + results = token_remover.TOKEN_RE.fullmatch(token) + self.assertIsNotNone(results, f"{token} was not matched by the regex") + + def test_regex_matches_multiple_valid(self): + """Should support multiple matches in the middle of a string.""" + token_1 = "NDY3MjIzMjMwNjUwNzc3NjQx.XsyWGg.uFNEQPCc4ePwGh7egG8UicQssz8" + token_2 = "NDcyMjY1OTQzMDYyNDEzMzMy.XsyWMw.l8XPnDqb0lp-EiQ2g_0xVFT1pyc" + message = f"garbage {token_1} hello {token_2} world" + + results = token_remover.TOKEN_RE.finditer(message) + results = [match[0] for match in results] + self.assertCountEqual((token_1, token_2), results) + + @autospec("bot.exts.filters.token_remover", "LOG_MESSAGE") + def test_format_log_message(self, log_message): + """Should correctly format the log message with info from the message and token.""" + token = Token("NDcyMjY1OTQzMDYyNDEzMzMy", "XsySD_", "s45jqDV_Iisn-symw0yDRrk_jf4") + log_message.format.return_value = "Howdy" + + return_value = TokenRemover.format_log_message(self.msg, token) + + self.assertEqual(return_value, log_message.format.return_value) + log_message.format.assert_called_once_with( + author=format_user(self.msg.author), + channel=self.msg.channel.mention, + user_id=token.user_id, + timestamp=token.timestamp, + hmac="x" * len(token.hmac), + ) + + @autospec("bot.exts.filters.token_remover", "UNKNOWN_USER_LOG_MESSAGE") + def test_format_userid_log_message_unknown(self, unknown_user_log_message): + """Should correctly format the user ID portion when the actual user it belongs to is unknown.""" + token = Token("NDcyMjY1OTQzMDYyNDEzMzMy", "XsySD_", "s45jqDV_Iisn-symw0yDRrk_jf4") + unknown_user_log_message.format.return_value = " Partner" + msg = MockMessage(id=555, content="hello world") + msg.guild.get_member.return_value = None + + return_value = TokenRemover.format_userid_log_message(msg, token) + + self.assertEqual(return_value, (unknown_user_log_message.format.return_value, False)) + unknown_user_log_message.format.assert_called_once_with(user_id=472265943062413332) + + @autospec("bot.exts.filters.token_remover", "KNOWN_USER_LOG_MESSAGE") + def test_format_userid_log_message_bot(self, known_user_log_message): + """Should correctly format the user ID portion when the ID belongs to a known bot.""" + token = Token("NDcyMjY1OTQzMDYyNDEzMzMy", "XsySD_", "s45jqDV_Iisn-symw0yDRrk_jf4") + known_user_log_message.format.return_value = " Partner" + msg = MockMessage(id=555, content="hello world") + msg.guild.get_member.return_value.__str__.return_value = "Sam" + msg.guild.get_member.return_value.bot = True + + return_value = TokenRemover.format_userid_log_message(msg, token) + + self.assertEqual(return_value, (known_user_log_message.format.return_value, False)) + + known_user_log_message.format.assert_called_once_with( + user_id=472265943062413332, + user_name="Sam", + kind="BOT", + ) + + @autospec("bot.exts.filters.token_remover", "KNOWN_USER_LOG_MESSAGE") + def test_format_log_message_user_token_user(self, user_token_message): + """Should correctly format the user ID portion when the ID belongs to a known user.""" + token = Token("NDY3MjIzMjMwNjUwNzc3NjQx", "XsySD_", "s45jqDV_Iisn-symw0yDRrk_jf4") + user_token_message.format.return_value = "Partner" + + return_value = TokenRemover.format_userid_log_message(self.msg, token) + + self.assertEqual(return_value, (user_token_message.format.return_value, True)) + user_token_message.format.assert_called_once_with( + user_id=467223230650777641, + user_name="Woody", + kind="USER", + ) + + @mock.patch.object(TokenRemover, "mod_log", new_callable=mock.PropertyMock) + @autospec("bot.exts.filters.token_remover", "log") + @autospec(TokenRemover, "format_log_message", "format_userid_log_message") + async def test_take_action(self, format_log_message, format_userid_log_message, logger, mod_log_property): + """Should delete the message and send a mod log.""" + cog = TokenRemover(self.bot) + mod_log = mock.create_autospec(ModLog, spec_set=True, instance=True) + token = mock.create_autospec(Token, spec_set=True, instance=True) + token.user_id = "no-id" + log_msg = "testing123" + userid_log_message = "userid-log-message" + + mod_log_property.return_value = mod_log + format_log_message.return_value = log_msg + format_userid_log_message.return_value = (userid_log_message, True) + + await cog.take_action(self.msg, token) + + self.msg.delete.assert_called_once_with() + self.msg.channel.send.assert_called_once_with( + token_remover.DELETION_MESSAGE_TEMPLATE.format(mention=self.msg.author.mention) + ) + + format_log_message.assert_called_once_with(self.msg, token) + format_userid_log_message.assert_called_once_with(self.msg, token) + logger.debug.assert_called_with(log_msg) + self.bot.stats.incr.assert_called_once_with("tokens.removed_tokens") + + mod_log.ignore.assert_called_once_with(constants.Event.message_delete, self.msg.id) + mod_log.send_log_message.assert_called_once_with( + icon_url=constants.Icons.token_removed, + colour=Colour(constants.Colours.soft_red), + title="Token removed!", + text=log_msg + "\n" + userid_log_message, + thumbnail=self.msg.author.avatar_url_as.return_value, + channel_id=constants.Channels.mod_alerts, + ping_everyone=True, + ) + + @mock.patch.object(TokenRemover, "mod_log", new_callable=mock.PropertyMock) + async def test_take_action_delete_failure(self, mod_log_property): + """Shouldn't send any messages if the token message can't be deleted.""" + cog = TokenRemover(self.bot) + mod_log_property.return_value = mock.create_autospec(ModLog, spec_set=True, instance=True) + self.msg.delete.side_effect = NotFound(MagicMock(), MagicMock()) + + token = mock.create_autospec(Token, spec_set=True, instance=True) + await cog.take_action(self.msg, token) + + self.msg.delete.assert_called_once_with() + self.msg.channel.send.assert_not_awaited() + + +class TokenRemoverExtensionTests(unittest.TestCase): + """Tests for the token_remover extension.""" + + @autospec("bot.exts.filters.token_remover", "TokenRemover") + def test_extension_setup(self, cog): + """The TokenRemover cog should be added.""" + bot = MockBot() + token_remover.setup(bot) + + cog.assert_called_once_with(bot) + bot.add_cog.assert_called_once() + self.assertTrue(isinstance(bot.add_cog.call_args.args[0], TokenRemover)) diff --git a/tests/bot/exts/info/__init__.py b/tests/bot/exts/info/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/tests/bot/exts/info/__init__.py diff --git a/tests/bot/cogs/test_information.py b/tests/bot/exts/info/test_information.py index 4496a2ae0..d077be960 100644 --- a/tests/bot/cogs/test_information.py +++ b/tests/bot/exts/info/test_information.py @@ -1,4 +1,3 @@ -import asyncio import textwrap import unittest import unittest.mock @@ -6,20 +5,19 @@ import unittest.mock import discord from bot import constants -from bot.cogs import information -from bot.decorators import InChannelCheckFailure +from bot.exts.info import information +from bot.utils.checks import InWhitelistCheckFailure from tests import helpers +COG_PATH = "bot.exts.info.information.Information" -COG_PATH = "bot.cogs.information.Information" - -class InformationCogTests(unittest.TestCase): +class InformationCogTests(unittest.IsolatedAsyncioTestCase): """Tests the Information cog.""" @classmethod def setUpClass(cls): - cls.moderator_role = helpers.MockRole(name="Moderator", id=constants.Roles.moderator) + cls.moderator_role = helpers.MockRole(name="Moderator", id=constants.Roles.moderators) def setUp(self): """Sets up fresh objects for each test.""" @@ -30,27 +28,24 @@ class InformationCogTests(unittest.TestCase): self.ctx = helpers.MockContext() self.ctx.author.roles.append(self.moderator_role) - def test_roles_command_command(self): + async def test_roles_command_command(self): """Test if the `role_info` command correctly returns the `moderator_role`.""" self.ctx.guild.roles.append(self.moderator_role) - self.cog.roles_info.can_run = helpers.AsyncMock() + self.cog.roles_info.can_run = unittest.mock.AsyncMock() self.cog.roles_info.can_run.return_value = True - coroutine = self.cog.roles_info.callback(self.cog, self.ctx) - - self.assertIsNone(asyncio.run(coroutine)) + self.assertIsNone(await self.cog.roles_info(self.cog, self.ctx)) self.ctx.send.assert_called_once() _, kwargs = self.ctx.send.call_args embed = kwargs.pop('embed') - self.assertEqual(embed.title, "Role information") + self.assertEqual(embed.title, "Role information (Total 1 role)") self.assertEqual(embed.colour, discord.Colour.blurple()) - self.assertEqual(embed.description, f"`{self.moderator_role.id}` - {self.moderator_role.mention}\n") - self.assertEqual(embed.footer.text, "Total roles: 1") + self.assertEqual(embed.description, f"\n`{self.moderator_role.id}` - {self.moderator_role.mention}\n") - def test_role_info_command(self): + async def test_role_info_command(self): """Tests the `role info` command.""" dummy_role = helpers.MockRole( name="Dummy", @@ -72,12 +67,10 @@ class InformationCogTests(unittest.TestCase): self.ctx.guild.roles.append([dummy_role, admin_role]) - self.cog.role_info.can_run = helpers.AsyncMock() + self.cog.role_info.can_run = unittest.mock.AsyncMock() self.cog.role_info.can_run.return_value = True - coroutine = self.cog.role_info.callback(self.cog, self.ctx, dummy_role, admin_role) - - self.assertIsNone(asyncio.run(coroutine)) + self.assertIsNone(await self.cog.role_info(self.cog, self.ctx, dummy_role, admin_role)) self.assertEqual(self.ctx.send.call_count, 2) @@ -99,86 +92,18 @@ class InformationCogTests(unittest.TestCase): self.assertEqual(admin_embed.title, "Admins info") self.assertEqual(admin_embed.colour, discord.Colour.red()) - @unittest.mock.patch('bot.cogs.information.time_since') - def test_server_info_command(self, time_since_patch): - time_since_patch.return_value = '2 days ago' - - self.ctx.guild = helpers.MockGuild( - features=('lemons', 'apples'), - region="The Moon", - roles=[self.moderator_role], - channels=[ - discord.TextChannel( - state={}, - guild=self.ctx.guild, - data={'id': 42, 'name': 'lemons-offering', 'position': 22, 'type': 'text'} - ), - discord.CategoryChannel( - state={}, - guild=self.ctx.guild, - data={'id': 5125, 'name': 'the-lemon-collection', 'position': 22, 'type': 'category'} - ), - discord.VoiceChannel( - state={}, - guild=self.ctx.guild, - data={'id': 15290, 'name': 'listen-to-lemon', 'position': 22, 'type': 'voice'} - ) - ], - members=[ - *(helpers.MockMember(status='online') for _ in range(2)), - *(helpers.MockMember(status='idle') for _ in range(1)), - *(helpers.MockMember(status='dnd') for _ in range(4)), - *(helpers.MockMember(status='offline') for _ in range(3)), - ], - member_count=1_234, - icon_url='a-lemon.jpg', - ) - - coroutine = self.cog.server_info.callback(self.cog, self.ctx) - self.assertIsNone(asyncio.run(coroutine)) - time_since_patch.assert_called_once_with(self.ctx.guild.created_at, precision='days') - _, kwargs = self.ctx.send.call_args - embed = kwargs.pop('embed') - self.assertEqual(embed.colour, discord.Colour.blurple()) - self.assertEqual( - embed.description, - textwrap.dedent( - f""" - **Server information** - Created: {time_since_patch.return_value} - Voice region: {self.ctx.guild.region} - Features: {', '.join(self.ctx.guild.features)} - - **Counts** - Members: {self.ctx.guild.member_count:,} - Roles: {len(self.ctx.guild.roles)} - Text: 1 - Voice: 1 - Channel categories: 1 - - **Members** - {constants.Emojis.status_online} 2 - {constants.Emojis.status_idle} 1 - {constants.Emojis.status_dnd} 4 - {constants.Emojis.status_offline} 3 - """ - ) - ) - self.assertEqual(embed.thumbnail.url, 'a-lemon.jpg') - - -class UserInfractionHelperMethodTests(unittest.TestCase): +class UserInfractionHelperMethodTests(unittest.IsolatedAsyncioTestCase): """Tests for the helper methods of the `!user` command.""" def setUp(self): """Common set-up steps done before for each test.""" self.bot = helpers.MockBot() - self.bot.api_client.get = helpers.AsyncMock() + self.bot.api_client.get = unittest.mock.AsyncMock() self.cog = information.Information(self.bot) self.member = helpers.MockMember(id=1234) - def test_user_command_helper_method_get_requests(self): + async def test_user_command_helper_method_get_requests(self): """The helper methods should form the correct get requests.""" test_values = ( { @@ -200,11 +125,11 @@ class UserInfractionHelperMethodTests(unittest.TestCase): endpoint, params = test_value["expected_args"] with self.subTest(method=helper_method, endpoint=endpoint, params=params): - asyncio.run(helper_method(self.member)) + await helper_method(self.member) self.bot.api_client.get.assert_called_once_with(endpoint, params=params) self.bot.api_client.get.reset_mock() - def _method_subtests(self, method, test_values, default_header): + async def _method_subtests(self, method, test_values, default_header): """Helper method that runs the subtests for the different helper methods.""" for test_value in test_values: api_response = test_value["api response"] @@ -213,12 +138,12 @@ class UserInfractionHelperMethodTests(unittest.TestCase): with self.subTest(method=method, api_response=api_response, expected_lines=expected_lines): self.bot.api_client.get.return_value = api_response - expected_output = "\n".join(default_header + expected_lines) - actual_output = asyncio.run(method(self.member)) + expected_output = "\n".join(expected_lines) + actual_output = await method(self.member) - self.assertEqual(expected_output, actual_output) + self.assertEqual((default_header, expected_output), actual_output) - def test_basic_user_infraction_counts_returns_correct_strings(self): + async def test_basic_user_infraction_counts_returns_correct_strings(self): """The method should correctly list both the total and active number of non-hidden infractions.""" test_values = ( # No infractions means zero counts @@ -247,16 +172,16 @@ class UserInfractionHelperMethodTests(unittest.TestCase): }, ) - header = ["**Infractions**"] + header = "Infractions" - self._method_subtests(self.cog.basic_user_infraction_counts, test_values, header) + await self._method_subtests(self.cog.basic_user_infraction_counts, test_values, header) - def test_expanded_user_infraction_counts_returns_correct_strings(self): + async def test_expanded_user_infraction_counts_returns_correct_strings(self): """The method should correctly list the total and active number of all infractions split by infraction type.""" test_values = ( { "api response": [], - "expected_lines": ["This user has never received an infraction."], + "expected_lines": ["No infractions"], }, # Shows non-hidden inactive infraction as expected { @@ -302,24 +227,24 @@ class UserInfractionHelperMethodTests(unittest.TestCase): }, ) - header = ["**Infractions**"] + header = "Infractions" - self._method_subtests(self.cog.expanded_user_infraction_counts, test_values, header) + await self._method_subtests(self.cog.expanded_user_infraction_counts, test_values, header) - def test_user_nomination_counts_returns_correct_strings(self): + async def test_user_nomination_counts_returns_correct_strings(self): """The method should list the number of active and historical nominations for the user.""" test_values = ( { "api response": [], - "expected_lines": ["This user has never been nominated."], + "expected_lines": ["No nominations"], }, { "api response": [{'active': True}], - "expected_lines": ["This user is **currently** nominated (1 nomination in total)."], + "expected_lines": ["This user is **currently** nominated", "(1 nomination in total)"], }, { "api response": [{'active': True}, {'active': False}], - "expected_lines": ["This user is **currently** nominated (2 nominations in total)."], + "expected_lines": ["This user is **currently** nominated", "(2 nominations in total)"], }, { "api response": [{'active': False}], @@ -332,48 +257,57 @@ class UserInfractionHelperMethodTests(unittest.TestCase): ) - header = ["**Nominations**"] + header = "Nominations" - self._method_subtests(self.cog.user_nomination_counts, test_values, header) + await self._method_subtests(self.cog.user_nomination_counts, test_values, header) [email protected]("bot.cogs.information.time_since", new=unittest.mock.MagicMock(return_value="1 year ago")) [email protected]("bot.cogs.information.constants.MODERATION_CHANNELS", new=[50]) -class UserEmbedTests(unittest.TestCase): [email protected]("bot.exts.info.information.time_since", new=unittest.mock.MagicMock(return_value="1 year ago")) [email protected]("bot.exts.info.information.constants.MODERATION_CHANNELS", new=[50]) +class UserEmbedTests(unittest.IsolatedAsyncioTestCase): """Tests for the creation of the `!user` embed.""" def setUp(self): """Common set-up steps done before for each test.""" self.bot = helpers.MockBot() - self.bot.api_client.get = helpers.AsyncMock() + self.bot.api_client.get = unittest.mock.AsyncMock() self.cog = information.Information(self.bot) - @unittest.mock.patch(f"{COG_PATH}.basic_user_infraction_counts", new=helpers.AsyncMock(return_value="")) - def test_create_user_embed_uses_string_representation_of_user_in_title_if_nick_is_not_available(self): + @unittest.mock.patch( + f"{COG_PATH}.basic_user_infraction_counts", + new=unittest.mock.AsyncMock(return_value=("Infractions", "basic infractions")) + ) + async def test_create_user_embed_uses_string_representation_of_user_in_title_if_nick_is_not_available(self): """The embed should use the string representation of the user if they don't have a nick.""" ctx = helpers.MockContext(channel=helpers.MockTextChannel(id=1)) user = helpers.MockMember() user.nick = None user.__str__ = unittest.mock.Mock(return_value="Mr. Hemlock") - embed = asyncio.run(self.cog.create_user_embed(ctx, user)) + embed = await self.cog.create_user_embed(ctx, user) self.assertEqual(embed.title, "Mr. Hemlock") - @unittest.mock.patch(f"{COG_PATH}.basic_user_infraction_counts", new=helpers.AsyncMock(return_value="")) - def test_create_user_embed_uses_nick_in_title_if_available(self): + @unittest.mock.patch( + f"{COG_PATH}.basic_user_infraction_counts", + new=unittest.mock.AsyncMock(return_value=("Infractions", "basic infractions")) + ) + async def test_create_user_embed_uses_nick_in_title_if_available(self): """The embed should use the nick if it's available.""" ctx = helpers.MockContext(channel=helpers.MockTextChannel(id=1)) user = helpers.MockMember() user.nick = "Cat lover" user.__str__ = unittest.mock.Mock(return_value="Mr. Hemlock") - embed = asyncio.run(self.cog.create_user_embed(ctx, user)) + embed = await self.cog.create_user_embed(ctx, user) self.assertEqual(embed.title, "Cat lover (Mr. Hemlock)") - @unittest.mock.patch(f"{COG_PATH}.basic_user_infraction_counts", new=helpers.AsyncMock(return_value="")) - def test_create_user_embed_ignores_everyone_role(self): + @unittest.mock.patch( + f"{COG_PATH}.basic_user_infraction_counts", + new=unittest.mock.AsyncMock(return_value=("Infractions", "basic infractions")) + ) + async def test_create_user_embed_ignores_everyone_role(self): """Created `!user` embeds should not contain mention of the @everyone-role.""" ctx = helpers.MockContext(channel=helpers.MockTextChannel(id=1)) admins_role = helpers.MockRole(name='Admins') @@ -382,80 +316,93 @@ class UserEmbedTests(unittest.TestCase): # A `MockMember` has the @Everyone role by default; we add the Admins to that. user = helpers.MockMember(roles=[admins_role], top_role=admins_role) - embed = asyncio.run(self.cog.create_user_embed(ctx, user)) + embed = await self.cog.create_user_embed(ctx, user) - self.assertIn("&Admins", embed.description) - self.assertNotIn("&Everyone", embed.description) + self.assertIn("&Admins", embed.fields[1].value) + self.assertNotIn("&Everyone", embed.fields[1].value) - @unittest.mock.patch(f"{COG_PATH}.expanded_user_infraction_counts", new_callable=helpers.AsyncMock) - @unittest.mock.patch(f"{COG_PATH}.user_nomination_counts", new_callable=helpers.AsyncMock) - def test_create_user_embed_expanded_information_in_moderation_channels(self, nomination_counts, infraction_counts): + @unittest.mock.patch(f"{COG_PATH}.expanded_user_infraction_counts", new_callable=unittest.mock.AsyncMock) + @unittest.mock.patch(f"{COG_PATH}.user_nomination_counts", new_callable=unittest.mock.AsyncMock) + async def test_create_user_embed_expanded_information_in_moderation_channels( + self, + nomination_counts, + infraction_counts + ): """The embed should contain expanded infractions and nomination info in mod channels.""" ctx = helpers.MockContext(channel=helpers.MockTextChannel(id=50)) moderators_role = helpers.MockRole(name='Moderators') moderators_role.colour = 100 - infraction_counts.return_value = "expanded infractions info" - nomination_counts.return_value = "nomination info" + infraction_counts.return_value = ("Infractions", "expanded infractions info") + nomination_counts.return_value = ("Nominations", "nomination info") user = helpers.MockMember(id=314, roles=[moderators_role], top_role=moderators_role) - embed = asyncio.run(self.cog.create_user_embed(ctx, user)) + embed = await self.cog.create_user_embed(ctx, user) infraction_counts.assert_called_once_with(user) nomination_counts.assert_called_once_with(user) self.assertEqual( textwrap.dedent(f""" - **User Information** Created: {"1 year ago"} Profile: {user.mention} ID: {user.id} + """).strip(), + embed.fields[0].value + ) - **Member Information** + self.assertEqual( + textwrap.dedent(f""" Joined: {"1 year ago"} + Verified: {"True"} Roles: &Moderators - - expanded infractions info - - nomination info """).strip(), - embed.description + embed.fields[1].value ) - @unittest.mock.patch(f"{COG_PATH}.basic_user_infraction_counts", new_callable=helpers.AsyncMock) - def test_create_user_embed_basic_information_outside_of_moderation_channels(self, infraction_counts): + @unittest.mock.patch(f"{COG_PATH}.basic_user_infraction_counts", new_callable=unittest.mock.AsyncMock) + async def test_create_user_embed_basic_information_outside_of_moderation_channels(self, infraction_counts): """The embed should contain only basic infraction data outside of mod channels.""" ctx = helpers.MockContext(channel=helpers.MockTextChannel(id=100)) moderators_role = helpers.MockRole(name='Moderators') moderators_role.colour = 100 - infraction_counts.return_value = "basic infractions info" + infraction_counts.return_value = ("Infractions", "basic infractions info") user = helpers.MockMember(id=314, roles=[moderators_role], top_role=moderators_role) - embed = asyncio.run(self.cog.create_user_embed(ctx, user)) + embed = await self.cog.create_user_embed(ctx, user) infraction_counts.assert_called_once_with(user) self.assertEqual( textwrap.dedent(f""" - **User Information** Created: {"1 year ago"} Profile: {user.mention} ID: {user.id} + """).strip(), + embed.fields[0].value + ) - **Member Information** + self.assertEqual( + textwrap.dedent(f""" Joined: {"1 year ago"} Roles: &Moderators - - basic infractions info """).strip(), - embed.description + embed.fields[1].value + ) + + self.assertEqual( + "basic infractions info", + embed.fields[2].value ) - @unittest.mock.patch(f"{COG_PATH}.basic_user_infraction_counts", new=helpers.AsyncMock(return_value="")) - def test_create_user_embed_uses_top_role_colour_when_user_has_roles(self): + @unittest.mock.patch( + f"{COG_PATH}.basic_user_infraction_counts", + new=unittest.mock.AsyncMock(return_value=("Infractions", "basic infractions")) + ) + async def test_create_user_embed_uses_top_role_colour_when_user_has_roles(self): """The embed should be created with the colour of the top role, if a top role is available.""" ctx = helpers.MockContext() @@ -463,35 +410,41 @@ class UserEmbedTests(unittest.TestCase): moderators_role.colour = 100 user = helpers.MockMember(id=314, roles=[moderators_role], top_role=moderators_role) - embed = asyncio.run(self.cog.create_user_embed(ctx, user)) + embed = await self.cog.create_user_embed(ctx, user) self.assertEqual(embed.colour, discord.Colour(moderators_role.colour)) - @unittest.mock.patch(f"{COG_PATH}.basic_user_infraction_counts", new=helpers.AsyncMock(return_value="")) - def test_create_user_embed_uses_blurple_colour_when_user_has_no_roles(self): + @unittest.mock.patch( + f"{COG_PATH}.basic_user_infraction_counts", + new=unittest.mock.AsyncMock(return_value=("Infractions", "basic infractions")) + ) + async def test_create_user_embed_uses_blurple_colour_when_user_has_no_roles(self): """The embed should be created with a blurple colour if the user has no assigned roles.""" ctx = helpers.MockContext() user = helpers.MockMember(id=217) - embed = asyncio.run(self.cog.create_user_embed(ctx, user)) + embed = await self.cog.create_user_embed(ctx, user) self.assertEqual(embed.colour, discord.Colour.blurple()) - @unittest.mock.patch(f"{COG_PATH}.basic_user_infraction_counts", new=helpers.AsyncMock(return_value="")) - def test_create_user_embed_uses_png_format_of_user_avatar_as_thumbnail(self): + @unittest.mock.patch( + f"{COG_PATH}.basic_user_infraction_counts", + new=unittest.mock.AsyncMock(return_value=("Infractions", "basic infractions")) + ) + async def test_create_user_embed_uses_png_format_of_user_avatar_as_thumbnail(self): """The embed thumbnail should be set to the user's avatar in `png` format.""" ctx = helpers.MockContext() user = helpers.MockMember(id=217) user.avatar_url_as.return_value = "avatar url" - embed = asyncio.run(self.cog.create_user_embed(ctx, user)) + embed = await 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") [email protected]("bot.cogs.information.constants") -class UserCommandTests(unittest.TestCase): [email protected]("bot.exts.info.information.constants") +class UserCommandTests(unittest.IsolatedAsyncioTestCase): """Tests for the `!user` command.""" def setUp(self): @@ -507,76 +460,70 @@ class UserCommandTests(unittest.TestCase): self.moderator = helpers.MockMember(id=2, name="riffautae", roles=[self.moderator_role]) self.target = helpers.MockMember(id=3, name="__fluzz__") - def test_regular_member_cannot_target_another_member(self, constants): + # There's no way to mock the channel constant without deferring imports. The constant is + # used as a default value for a parameter, which gets defined upon import. + self.bot_command_channel = helpers.MockTextChannel(id=constants.Channels.bot_commands) + + async def test_regular_member_cannot_target_another_member(self, constants): """A regular user should not be able to use `!user` targeting another user.""" constants.MODERATION_ROLES = [self.moderator_role.id] - ctx = helpers.MockContext(author=self.author) - asyncio.run(self.cog.user_info.callback(self.cog, ctx, self.target)) + await self.cog.user_info(self.cog, ctx, self.target) ctx.send.assert_called_once_with("You may not use this command on users other than yourself.") - def test_regular_member_cannot_use_command_outside_of_bot_commands(self, constants): + async def test_regular_member_cannot_use_command_outside_of_bot_commands(self, constants): """A regular user should not be able to use this command outside of bot-commands.""" constants.MODERATION_ROLES = [self.moderator_role.id] constants.STAFF_ROLES = [self.moderator_role.id] - constants.Channels.bot = 50 - 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): - asyncio.run(self.cog.user_info.callback(self.cog, ctx)) + with self.assertRaises(InWhitelistCheckFailure, msg=msg): + await self.cog.user_info(self.cog, ctx) - @unittest.mock.patch("bot.cogs.information.Information.create_user_embed", new_callable=helpers.AsyncMock) - def test_regular_user_may_use_command_in_bot_commands_channel(self, create_embed, constants): + @unittest.mock.patch("bot.exts.info.information.Information.create_user_embed") + async def test_regular_user_may_use_command_in_bot_commands_channel(self, create_embed, constants): """A regular user should be allowed to use `!user` targeting themselves in bot-commands.""" constants.STAFF_ROLES = [self.moderator_role.id] - constants.Channels.bot = 50 + ctx = helpers.MockContext(author=self.author, channel=self.bot_command_channel) - ctx = helpers.MockContext(author=self.author, channel=helpers.MockTextChannel(id=50)) - - asyncio.run(self.cog.user_info.callback(self.cog, ctx)) + await self.cog.user_info(self.cog, ctx) create_embed.assert_called_once_with(ctx, self.author) ctx.send.assert_called_once() - @unittest.mock.patch("bot.cogs.information.Information.create_user_embed", new_callable=helpers.AsyncMock) - def test_regular_user_can_explicitly_target_themselves(self, create_embed, constants): + @unittest.mock.patch("bot.exts.info.information.Information.create_user_embed") + async def test_regular_user_can_explicitly_target_themselves(self, create_embed, _): """A user should target itself with `!user` when a `user` argument was not provided.""" constants.STAFF_ROLES = [self.moderator_role.id] - constants.Channels.bot = 50 - - ctx = helpers.MockContext(author=self.author, channel=helpers.MockTextChannel(id=50)) + ctx = helpers.MockContext(author=self.author, channel=self.bot_command_channel) - asyncio.run(self.cog.user_info.callback(self.cog, ctx, self.author)) + await self.cog.user_info(self.cog, ctx, self.author) create_embed.assert_called_once_with(ctx, self.author) ctx.send.assert_called_once() - @unittest.mock.patch("bot.cogs.information.Information.create_user_embed", new_callable=helpers.AsyncMock) - def test_staff_members_can_bypass_channel_restriction(self, create_embed, constants): + @unittest.mock.patch("bot.exts.info.information.Information.create_user_embed") + async def test_staff_members_can_bypass_channel_restriction(self, create_embed, constants): """Staff members should be able to bypass the bot-commands channel restriction.""" constants.STAFF_ROLES = [self.moderator_role.id] - constants.Channels.bot = 50 - ctx = helpers.MockContext(author=self.moderator, channel=helpers.MockTextChannel(id=200)) - asyncio.run(self.cog.user_info.callback(self.cog, ctx)) + await self.cog.user_info(self.cog, ctx) create_embed.assert_called_once_with(ctx, self.moderator) ctx.send.assert_called_once() - @unittest.mock.patch("bot.cogs.information.Information.create_user_embed", new_callable=helpers.AsyncMock) - def test_moderators_can_target_another_member(self, create_embed, constants): + @unittest.mock.patch("bot.exts.info.information.Information.create_user_embed") + async def test_moderators_can_target_another_member(self, create_embed, constants): """A moderator should be able to use `!user` targeting another user.""" constants.MODERATION_ROLES = [self.moderator_role.id] constants.STAFF_ROLES = [self.moderator_role.id] - ctx = helpers.MockContext(author=self.moderator, channel=helpers.MockTextChannel(id=50)) - asyncio.run(self.cog.user_info.callback(self.cog, ctx, self.target)) + await self.cog.user_info(self.cog, ctx, self.target) create_embed.assert_called_once_with(ctx, self.target) ctx.send.assert_called_once() diff --git a/tests/bot/exts/moderation/__init__.py b/tests/bot/exts/moderation/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/tests/bot/exts/moderation/__init__.py diff --git a/tests/bot/exts/moderation/infraction/__init__.py b/tests/bot/exts/moderation/infraction/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/tests/bot/exts/moderation/infraction/__init__.py diff --git a/tests/bot/exts/moderation/infraction/test_infractions.py b/tests/bot/exts/moderation/infraction/test_infractions.py new file mode 100644 index 000000000..bf557a484 --- /dev/null +++ b/tests/bot/exts/moderation/infraction/test_infractions.py @@ -0,0 +1,201 @@ +import textwrap +import unittest +from unittest.mock import AsyncMock, MagicMock, Mock, patch + +from bot.constants import Event +from bot.exts.moderation.infraction.infractions import Infractions +from tests.helpers import MockBot, MockContext, MockGuild, MockMember, MockRole + + +class TruncationTests(unittest.IsolatedAsyncioTestCase): + """Tests for ban and kick command reason truncation.""" + + def setUp(self): + self.bot = MockBot() + self.cog = Infractions(self.bot) + self.user = MockMember(id=1234, top_role=MockRole(id=3577, position=10)) + self.target = MockMember(id=1265, top_role=MockRole(id=9876, position=0)) + self.guild = MockGuild(id=4567) + self.ctx = MockContext(bot=self.bot, author=self.user, guild=self.guild) + + @patch("bot.exts.moderation.infraction._utils.get_active_infraction") + @patch("bot.exts.moderation.infraction._utils.post_infraction") + async def test_apply_ban_reason_truncation(self, post_infraction_mock, get_active_mock): + """Should truncate reason for `ctx.guild.ban`.""" + get_active_mock.return_value = None + post_infraction_mock.return_value = {"foo": "bar"} + + self.cog.apply_infraction = AsyncMock() + self.bot.get_cog.return_value = AsyncMock() + self.cog.mod_log.ignore = Mock() + self.ctx.guild.ban = Mock() + + await self.cog.apply_ban(self.ctx, self.target, "foo bar" * 3000) + self.ctx.guild.ban.assert_called_once_with( + self.target, + reason=textwrap.shorten("foo bar" * 3000, 512, placeholder="..."), + delete_message_days=0 + ) + self.cog.apply_infraction.assert_awaited_once_with( + self.ctx, {"foo": "bar"}, self.target, self.ctx.guild.ban.return_value + ) + + @patch("bot.exts.moderation.infraction._utils.post_infraction") + async def test_apply_kick_reason_truncation(self, post_infraction_mock): + """Should truncate reason for `Member.kick`.""" + post_infraction_mock.return_value = {"foo": "bar"} + + self.cog.apply_infraction = AsyncMock() + self.cog.mod_log.ignore = Mock() + self.target.kick = Mock() + + await self.cog.apply_kick(self.ctx, self.target, "foo bar" * 3000) + self.target.kick.assert_called_once_with(reason=textwrap.shorten("foo bar" * 3000, 512, placeholder="...")) + self.cog.apply_infraction.assert_awaited_once_with( + self.ctx, {"foo": "bar"}, self.target, self.target.kick.return_value + ) + + +@patch("bot.exts.moderation.infraction.infractions.constants.Roles.voice_verified", new=123456) +class VoiceBanTests(unittest.IsolatedAsyncioTestCase): + """Tests for voice ban related functions and commands.""" + + def setUp(self): + self.bot = MockBot() + self.mod = MockMember(top_role=10) + self.user = MockMember(top_role=1, roles=[MockRole(id=123456)]) + self.guild = MockGuild() + self.ctx = MockContext(bot=self.bot, author=self.mod) + self.cog = Infractions(self.bot) + + async def test_permanent_voice_ban(self): + """Should call voice ban applying function without expiry.""" + self.cog.apply_voice_ban = AsyncMock() + self.assertIsNone(await self.cog.voiceban(self.cog, self.ctx, self.user, reason="foobar")) + self.cog.apply_voice_ban.assert_awaited_once_with(self.ctx, self.user, "foobar") + + async def test_temporary_voice_ban(self): + """Should call voice ban applying function with expiry.""" + self.cog.apply_voice_ban = AsyncMock() + self.assertIsNone(await self.cog.tempvoiceban(self.cog, self.ctx, self.user, "baz", reason="foobar")) + self.cog.apply_voice_ban.assert_awaited_once_with(self.ctx, self.user, "foobar", expires_at="baz") + + async def test_voice_unban(self): + """Should call infraction pardoning function.""" + self.cog.pardon_infraction = AsyncMock() + self.assertIsNone(await self.cog.unvoiceban(self.cog, self.ctx, self.user)) + self.cog.pardon_infraction.assert_awaited_once_with(self.ctx, "voice_ban", self.user) + + @patch("bot.exts.moderation.infraction.infractions._utils.post_infraction") + @patch("bot.exts.moderation.infraction.infractions._utils.get_active_infraction") + async def test_voice_ban_user_have_active_infraction(self, get_active_infraction, post_infraction_mock): + """Should return early when user already have Voice Ban infraction.""" + get_active_infraction.return_value = {"foo": "bar"} + self.assertIsNone(await self.cog.apply_voice_ban(self.ctx, self.user, "foobar")) + get_active_infraction.assert_awaited_once_with(self.ctx, self.user, "voice_ban") + post_infraction_mock.assert_not_awaited() + + @patch("bot.exts.moderation.infraction.infractions._utils.post_infraction") + @patch("bot.exts.moderation.infraction.infractions._utils.get_active_infraction") + async def test_voice_ban_infraction_post_failed(self, get_active_infraction, post_infraction_mock): + """Should return early when posting infraction fails.""" + self.cog.mod_log.ignore = MagicMock() + get_active_infraction.return_value = None + post_infraction_mock.return_value = None + self.assertIsNone(await self.cog.apply_voice_ban(self.ctx, self.user, "foobar")) + post_infraction_mock.assert_awaited_once() + self.cog.mod_log.ignore.assert_not_called() + + @patch("bot.exts.moderation.infraction.infractions._utils.post_infraction") + @patch("bot.exts.moderation.infraction.infractions._utils.get_active_infraction") + async def test_voice_ban_infraction_post_add_kwargs(self, get_active_infraction, post_infraction_mock): + """Should pass all kwargs passed to apply_voice_ban to post_infraction.""" + get_active_infraction.return_value = None + # We don't want that this continue yet + post_infraction_mock.return_value = None + self.assertIsNone(await self.cog.apply_voice_ban(self.ctx, self.user, "foobar", my_kwarg=23)) + post_infraction_mock.assert_awaited_once_with( + self.ctx, self.user, "voice_ban", "foobar", active=True, my_kwarg=23 + ) + + @patch("bot.exts.moderation.infraction.infractions._utils.post_infraction") + @patch("bot.exts.moderation.infraction.infractions._utils.get_active_infraction") + async def test_voice_ban_mod_log_ignore(self, get_active_infraction, post_infraction_mock): + """Should ignore Voice Verified role removing.""" + self.cog.mod_log.ignore = MagicMock() + self.cog.apply_infraction = AsyncMock() + self.user.remove_roles = MagicMock(return_value="my_return_value") + + get_active_infraction.return_value = None + post_infraction_mock.return_value = {"foo": "bar"} + + self.assertIsNone(await self.cog.apply_voice_ban(self.ctx, self.user, "foobar")) + self.cog.mod_log.ignore.assert_called_once_with(Event.member_update, self.user.id) + + @patch("bot.exts.moderation.infraction.infractions._utils.post_infraction") + @patch("bot.exts.moderation.infraction.infractions._utils.get_active_infraction") + async def test_voice_ban_apply_infraction(self, get_active_infraction, post_infraction_mock): + """Should ignore Voice Verified role removing.""" + self.cog.mod_log.ignore = MagicMock() + self.cog.apply_infraction = AsyncMock() + self.user.remove_roles = MagicMock(return_value="my_return_value") + + get_active_infraction.return_value = None + post_infraction_mock.return_value = {"foo": "bar"} + + self.assertIsNone(await self.cog.apply_voice_ban(self.ctx, self.user, "foobar")) + self.user.remove_roles.assert_called_once_with(self.cog._voice_verified_role, reason="foobar") + self.cog.apply_infraction.assert_awaited_once_with(self.ctx, {"foo": "bar"}, self.user, "my_return_value") + + @patch("bot.exts.moderation.infraction.infractions._utils.post_infraction") + @patch("bot.exts.moderation.infraction.infractions._utils.get_active_infraction") + async def test_voice_ban_truncate_reason(self, get_active_infraction, post_infraction_mock): + """Should truncate reason for voice ban.""" + self.cog.mod_log.ignore = MagicMock() + self.cog.apply_infraction = AsyncMock() + self.user.remove_roles = MagicMock(return_value="my_return_value") + + get_active_infraction.return_value = None + post_infraction_mock.return_value = {"foo": "bar"} + + self.assertIsNone(await self.cog.apply_voice_ban(self.ctx, self.user, "foobar" * 3000)) + self.user.remove_roles.assert_called_once_with( + self.cog._voice_verified_role, reason=textwrap.shorten("foobar" * 3000, 512, placeholder="...") + ) + self.cog.apply_infraction.assert_awaited_once_with(self.ctx, {"foo": "bar"}, self.user, "my_return_value") + + async def test_voice_unban_user_not_found(self): + """Should include info to return dict when user was not found from guild.""" + self.guild.get_member.return_value = None + result = await self.cog.pardon_voice_ban(self.user.id, self.guild, "foobar") + self.assertEqual(result, {"Info": "User was not found in the guild."}) + + @patch("bot.exts.moderation.infraction.infractions._utils.notify_pardon") + @patch("bot.exts.moderation.infraction.infractions.format_user") + async def test_voice_unban_user_found(self, format_user_mock, notify_pardon_mock): + """Should add role back with ignoring, notify user and return log dictionary..""" + self.guild.get_member.return_value = self.user + notify_pardon_mock.return_value = True + format_user_mock.return_value = "my-user" + + result = await self.cog.pardon_voice_ban(self.user.id, self.guild, "foobar") + self.assertEqual(result, { + "Member": "my-user", + "DM": "Sent" + }) + notify_pardon_mock.assert_awaited_once() + + @patch("bot.exts.moderation.infraction.infractions._utils.notify_pardon") + @patch("bot.exts.moderation.infraction.infractions.format_user") + async def test_voice_unban_dm_fail(self, format_user_mock, notify_pardon_mock): + """Should add role back with ignoring, notify user and return log dictionary..""" + self.guild.get_member.return_value = self.user + notify_pardon_mock.return_value = False + format_user_mock.return_value = "my-user" + + result = await self.cog.pardon_voice_ban(self.user.id, self.guild, "foobar") + self.assertEqual(result, { + "Member": "my-user", + "DM": "**Failed**" + }) + notify_pardon_mock.assert_awaited_once() diff --git a/tests/bot/exts/moderation/infraction/test_utils.py b/tests/bot/exts/moderation/infraction/test_utils.py new file mode 100644 index 000000000..5b62463e0 --- /dev/null +++ b/tests/bot/exts/moderation/infraction/test_utils.py @@ -0,0 +1,359 @@ +import unittest +from collections import namedtuple +from datetime import datetime +from unittest.mock import AsyncMock, MagicMock, call, patch + +from discord import Embed, Forbidden, HTTPException, NotFound + +from bot.api import ResponseCodeError +from bot.constants import Colours, Icons +from bot.exts.moderation.infraction import _utils as utils +from tests.helpers import MockBot, MockContext, MockMember, MockUser + + +class ModerationUtilsTests(unittest.IsolatedAsyncioTestCase): + """Tests Moderation utils.""" + + def setUp(self): + self.bot = MockBot() + self.member = MockMember(id=1234) + self.user = MockUser(id=1234) + self.ctx = MockContext(bot=self.bot, author=self.member) + + async def test_post_user(self): + """Should POST a new user and return the response if successful or otherwise send an error message.""" + user = MockUser(discriminator=5678, id=1234, name="Test user") + not_user = MagicMock(discriminator=3333, id=5678, name="Wrong user") + test_cases = [ + { + "user": user, + "post_result": "bar", + "raise_error": None, + "payload": { + "discriminator": 5678, + "id": self.user.id, + "in_guild": False, + "name": "Test user", + "roles": [] + } + }, + { + "user": self.member, + "post_result": "foo", + "raise_error": ResponseCodeError(MagicMock(status=400), "foo"), + "payload": { + "discriminator": 0, + "id": self.member.id, + "in_guild": False, + "name": "Name unknown", + "roles": [] + } + }, + { + "user": not_user, + "post_result": "bar", + "raise_error": None, + "payload": { + "discriminator": not_user.discriminator, + "id": not_user.id, + "in_guild": False, + "name": not_user.name, + "roles": [] + } + } + ] + + for case in test_cases: + user = case["user"] + post_result = case["post_result"] + raise_error = case["raise_error"] + payload = case["payload"] + + with self.subTest(user=user, post_result=post_result, raise_error=raise_error, payload=payload): + self.bot.api_client.post.reset_mock(side_effect=True) + self.ctx.bot.api_client.post.return_value = post_result + + self.ctx.bot.api_client.post.side_effect = raise_error + + result = await utils.post_user(self.ctx, user) + + if raise_error: + self.assertIsNone(result) + self.ctx.send.assert_awaited_once() + self.assertIn(str(raise_error.status), self.ctx.send.call_args[0][0]) + else: + self.assertEqual(result, post_result) + self.bot.api_client.post.assert_awaited_once_with("bot/users", json=payload) + + async def test_get_active_infraction(self): + """ + Should request the API for active infractions and return infraction if the user has one or `None` otherwise. + + A message should be sent to the context indicating a user already has an infraction, if that's the case. + """ + test_case = namedtuple("test_case", ["get_return_value", "expected_output", "infraction_nr", "send_msg"]) + test_cases = [ + test_case([], None, None, True), + test_case([{"id": 123987}], {"id": 123987}, "123987", False), + test_case([{"id": 123987}], {"id": 123987}, "123987", True) + ] + + for case in test_cases: + with self.subTest(return_value=case.get_return_value, expected=case.expected_output): + self.bot.api_client.get.reset_mock() + self.ctx.send.reset_mock() + + params = { + "active": "true", + "type": "ban", + "user__id": str(self.member.id) + } + + self.bot.api_client.get.return_value = case.get_return_value + + result = await utils.get_active_infraction(self.ctx, self.member, "ban", send_msg=case.send_msg) + self.assertEqual(result, case.expected_output) + self.bot.api_client.get.assert_awaited_once_with("bot/infractions", params=params) + + if case.send_msg and case.get_return_value: + self.ctx.send.assert_awaited_once() + sent_message = self.ctx.send.call_args[0][0] + self.assertIn(case.infraction_nr, sent_message) + self.assertIn("ban", sent_message) + else: + self.ctx.send.assert_not_awaited() + + @patch("bot.exts.moderation.infraction._utils.send_private_embed") + async def test_notify_infraction(self, send_private_embed_mock): + """ + Should send an embed of a certain format as a DM and return `True` if DM successful. + + Appealable infractions should have the appeal message in the embed's footer. + """ + test_cases = [ + { + "args": (self.user, "ban", "2020-02-26 09:20 (23 hours and 59 minutes)"), + "expected_output": Embed( + title=utils.INFRACTION_TITLE, + description=utils.INFRACTION_DESCRIPTION_TEMPLATE.format( + type="Ban", + expires="2020-02-26 09:20 (23 hours and 59 minutes)", + reason="No reason provided." + ), + colour=Colours.soft_red, + url=utils.RULES_URL + ).set_author( + name=utils.INFRACTION_AUTHOR_NAME, + url=utils.RULES_URL, + icon_url=Icons.token_removed + ).set_footer(text=utils.INFRACTION_APPEAL_FOOTER), + "send_result": True + }, + { + "args": (self.user, "warning", None, "Test reason."), + "expected_output": Embed( + title=utils.INFRACTION_TITLE, + description=utils.INFRACTION_DESCRIPTION_TEMPLATE.format( + type="Warning", + expires="N/A", + reason="Test reason." + ), + colour=Colours.soft_red, + url=utils.RULES_URL + ).set_author( + name=utils.INFRACTION_AUTHOR_NAME, + url=utils.RULES_URL, + icon_url=Icons.token_removed + ), + "send_result": False + }, + { + "args": (self.user, "note", None, None, Icons.defcon_denied), + "expected_output": Embed( + title=utils.INFRACTION_TITLE, + description=utils.INFRACTION_DESCRIPTION_TEMPLATE.format( + type="Note", + expires="N/A", + reason="No reason provided." + ), + colour=Colours.soft_red, + url=utils.RULES_URL + ).set_author( + name=utils.INFRACTION_AUTHOR_NAME, + url=utils.RULES_URL, + icon_url=Icons.defcon_denied + ), + "send_result": False + }, + { + "args": (self.user, "mute", "2020-02-26 09:20 (23 hours and 59 minutes)", "Test", Icons.defcon_denied), + "expected_output": Embed( + title=utils.INFRACTION_TITLE, + description=utils.INFRACTION_DESCRIPTION_TEMPLATE.format( + type="Mute", + expires="2020-02-26 09:20 (23 hours and 59 minutes)", + reason="Test" + ), + colour=Colours.soft_red, + url=utils.RULES_URL + ).set_author( + name=utils.INFRACTION_AUTHOR_NAME, + url=utils.RULES_URL, + icon_url=Icons.defcon_denied + ).set_footer(text=utils.INFRACTION_APPEAL_FOOTER), + "send_result": False + }, + { + "args": (self.user, "mute", None, "foo bar" * 4000, Icons.defcon_denied), + "expected_output": Embed( + title=utils.INFRACTION_TITLE, + description=utils.INFRACTION_DESCRIPTION_TEMPLATE.format( + type="Mute", + expires="N/A", + reason="foo bar" * 4000 + )[:2045] + "...", + colour=Colours.soft_red, + url=utils.RULES_URL + ).set_author( + name=utils.INFRACTION_AUTHOR_NAME, + url=utils.RULES_URL, + icon_url=Icons.defcon_denied + ).set_footer(text=utils.INFRACTION_APPEAL_FOOTER), + "send_result": True + } + ] + + for case in test_cases: + with self.subTest(args=case["args"], expected=case["expected_output"], send=case["send_result"]): + send_private_embed_mock.reset_mock() + + send_private_embed_mock.return_value = case["send_result"] + result = await utils.notify_infraction(*case["args"]) + + self.assertEqual(case["send_result"], result) + + embed = send_private_embed_mock.call_args[0][1] + + self.assertEqual(embed.to_dict(), case["expected_output"].to_dict()) + + send_private_embed_mock.assert_awaited_once_with(case["args"][0], embed) + + @patch("bot.exts.moderation.infraction._utils.send_private_embed") + async def test_notify_pardon(self, send_private_embed_mock): + """Should send an embed of a certain format as a DM and return `True` if DM successful.""" + test_case = namedtuple("test_case", ["args", "icon", "send_result"]) + test_cases = [ + test_case((self.user, "Test title", "Example content"), Icons.user_verified, True), + test_case((self.user, "Test title", "Example content", Icons.user_update), Icons.user_update, False) + ] + + for case in test_cases: + expected = Embed( + description="Example content", + colour=Colours.soft_green + ).set_author( + name="Test title", + icon_url=case.icon + ) + + with self.subTest(args=case.args, expected=expected): + send_private_embed_mock.reset_mock() + + send_private_embed_mock.return_value = case.send_result + + result = await utils.notify_pardon(*case.args) + self.assertEqual(case.send_result, result) + + embed = send_private_embed_mock.call_args[0][1] + self.assertEqual(embed.to_dict(), expected.to_dict()) + + send_private_embed_mock.assert_awaited_once_with(case.args[0], embed) + + async def test_send_private_embed(self): + """Should DM the user and return `True` on success or `False` on failure.""" + embed = Embed(title="Test", description="Test val") + + test_case = namedtuple("test_case", ["expected_output", "raised_exception"]) + test_cases = [ + test_case(True, None), + test_case(False, HTTPException(AsyncMock(), AsyncMock())), + test_case(False, Forbidden(AsyncMock(), AsyncMock())), + test_case(False, NotFound(AsyncMock(), AsyncMock())) + ] + + for case in test_cases: + with self.subTest(expected=case.expected_output, raised=case.raised_exception): + self.user.send.reset_mock(side_effect=True) + self.user.send.side_effect = case.raised_exception + + result = await utils.send_private_embed(self.user, embed) + + self.assertEqual(result, case.expected_output) + self.user.send.assert_awaited_once_with(embed=embed) + + +class TestPostInfraction(unittest.IsolatedAsyncioTestCase): + """Tests for the `post_infraction` function.""" + + def setUp(self): + self.bot = MockBot() + self.member = MockMember(id=1234) + self.user = MockUser(id=1234) + self.ctx = MockContext(bot=self.bot, author=self.member) + + async def test_normal_post_infraction(self): + """Should return response from POST request if there are no errors.""" + now = datetime.now() + payload = { + "actor": self.ctx.author.id, + "hidden": True, + "reason": "Test reason", + "type": "ban", + "user": self.member.id, + "active": False, + "expires_at": now.isoformat() + } + + self.ctx.bot.api_client.post.return_value = "foo" + actual = await utils.post_infraction(self.ctx, self.member, "ban", "Test reason", now, True, False) + + self.assertEqual(actual, "foo") + self.ctx.bot.api_client.post.assert_awaited_once_with("bot/infractions", json=payload) + + async def test_unknown_error_post_infraction(self): + """Should send an error message to chat when a non-400 error occurs.""" + self.ctx.bot.api_client.post.side_effect = ResponseCodeError(AsyncMock(), AsyncMock()) + self.ctx.bot.api_client.post.side_effect.status = 500 + + actual = await utils.post_infraction(self.ctx, self.user, "ban", "Test reason") + self.assertIsNone(actual) + + self.assertTrue("500" in self.ctx.send.call_args[0][0]) + + @patch("bot.exts.moderation.infraction._utils.post_user", return_value=None) + async def test_user_not_found_none_post_infraction(self, post_user_mock): + """Should abort and return `None` when a new user fails to be posted.""" + self.bot.api_client.post.side_effect = ResponseCodeError(MagicMock(status=400), {"user": "foo"}) + + actual = await utils.post_infraction(self.ctx, self.user, "mute", "Test reason") + self.assertIsNone(actual) + post_user_mock.assert_awaited_once_with(self.ctx, self.user) + + @patch("bot.exts.moderation.infraction._utils.post_user", return_value="bar") + async def test_first_fail_second_success_user_post_infraction(self, post_user_mock): + """Should post the user if they don't exist, POST infraction again, and return the response if successful.""" + payload = { + "actor": self.ctx.author.id, + "hidden": False, + "reason": "Test reason", + "type": "mute", + "user": self.user.id, + "active": True + } + + self.bot.api_client.post.side_effect = [ResponseCodeError(MagicMock(status=400), {"user": "foo"}), "foo"] + + actual = await utils.post_infraction(self.ctx, self.user, "mute", "Test reason") + self.assertEqual(actual, "foo") + self.bot.api_client.post.assert_has_awaits([call("bot/infractions", json=payload)] * 2) + post_user_mock.assert_awaited_once_with(self.ctx, self.user) diff --git a/tests/bot/exts/moderation/test_incidents.py b/tests/bot/exts/moderation/test_incidents.py new file mode 100644 index 000000000..cbf7f7bcf --- /dev/null +++ b/tests/bot/exts/moderation/test_incidents.py @@ -0,0 +1,770 @@ +import asyncio +import enum +import logging +import typing as t +import unittest +from unittest.mock import AsyncMock, MagicMock, call, patch + +import aiohttp +import discord + +from bot.constants import Colours +from bot.exts.moderation import incidents +from tests.helpers import ( + MockAsyncWebhook, + MockAttachment, + MockBot, + MockMember, + MockMessage, + MockReaction, + MockRole, + MockTextChannel, + MockUser, +) + + +class MockAsyncIterable: + """ + Helper for mocking asynchronous for loops. + + It does not appear that the `unittest` library currently provides anything that would + allow us to simply mock an async iterator, such as `discord.TextChannel.history`. + + We therefore write our own helper to wrap a regular synchronous iterable, and feed + its values via `__anext__` rather than `__next__`. + + This class was written for the purposes of testing the `Incidents` cog - it may not + be generic enough to be placed in the `tests.helpers` module. + """ + + def __init__(self, messages: t.Iterable): + """Take a sync iterable to be wrapped.""" + self.iter_messages = iter(messages) + + def __aiter__(self): + """Return `self` as we provide the `__anext__` method.""" + return self + + async def __anext__(self): + """ + Feed the next item, or raise `StopAsyncIteration`. + + Since we're wrapping a sync iterator, it will communicate that it has been depleted + by raising a `StopIteration`. The `async for` construct does not expect it, and we + therefore need to substitute it for the appropriate exception type. + """ + try: + return next(self.iter_messages) + except StopIteration: + raise StopAsyncIteration + + +class MockSignal(enum.Enum): + A = "A" + B = "B" + + +mock_404 = discord.NotFound( + response=MagicMock(aiohttp.ClientResponse), # Mock the erroneous response + message="Not found", +) + + +class TestDownloadFile(unittest.IsolatedAsyncioTestCase): + """Collection of tests for the `download_file` helper function.""" + + async def test_download_file_success(self): + """If `to_file` succeeds, function returns the acquired `discord.File`.""" + file = MagicMock(discord.File, filename="bigbadlemon.jpg") + attachment = MockAttachment(to_file=AsyncMock(return_value=file)) + + acquired_file = await incidents.download_file(attachment) + self.assertIs(file, acquired_file) + + async def test_download_file_404(self): + """If `to_file` encounters a 404, function handles the exception & returns None.""" + attachment = MockAttachment(to_file=AsyncMock(side_effect=mock_404)) + + acquired_file = await incidents.download_file(attachment) + self.assertIsNone(acquired_file) + + async def test_download_file_fail(self): + """If `to_file` fails on a non-404 error, function logs the exception & returns None.""" + arbitrary_error = discord.HTTPException(MagicMock(aiohttp.ClientResponse), "Arbitrary API error") + attachment = MockAttachment(to_file=AsyncMock(side_effect=arbitrary_error)) + + with self.assertLogs(logger=incidents.log, level=logging.ERROR): + acquired_file = await incidents.download_file(attachment) + + self.assertIsNone(acquired_file) + + +class TestMakeEmbed(unittest.IsolatedAsyncioTestCase): + """Collection of tests for the `make_embed` helper function.""" + + async def test_make_embed_actioned(self): + """Embed is coloured green and footer contains 'Actioned' when `outcome=Signal.ACTIONED`.""" + embed, file = await incidents.make_embed(MockMessage(), incidents.Signal.ACTIONED, MockMember()) + + self.assertEqual(embed.colour.value, Colours.soft_green) + self.assertIn("Actioned", embed.footer.text) + + async def test_make_embed_not_actioned(self): + """Embed is coloured red and footer contains 'Rejected' when `outcome=Signal.NOT_ACTIONED`.""" + embed, file = await incidents.make_embed(MockMessage(), incidents.Signal.NOT_ACTIONED, MockMember()) + + self.assertEqual(embed.colour.value, Colours.soft_red) + self.assertIn("Rejected", embed.footer.text) + + async def test_make_embed_content(self): + """Incident content appears as embed description.""" + incident = MockMessage(content="this is an incident") + embed, file = await incidents.make_embed(incident, incidents.Signal.ACTIONED, MockMember()) + + self.assertEqual(incident.content, embed.description) + + async def test_make_embed_with_attachment_succeeds(self): + """Incident's attachment is downloaded and displayed in the embed's image field.""" + file = MagicMock(discord.File, filename="bigbadjoe.jpg") + attachment = MockAttachment(filename="bigbadjoe.jpg") + incident = MockMessage(content="this is an incident", attachments=[attachment]) + + # Patch `download_file` to return our `file` + with patch("bot.exts.moderation.incidents.download_file", AsyncMock(return_value=file)): + embed, returned_file = await incidents.make_embed(incident, incidents.Signal.ACTIONED, MockMember()) + + self.assertIs(file, returned_file) + self.assertEqual("attachment://bigbadjoe.jpg", embed.image.url) + + async def test_make_embed_with_attachment_fails(self): + """Incident's attachment fails to download, proxy url is linked instead.""" + attachment = MockAttachment(proxy_url="discord.com/bigbadjoe.jpg") + incident = MockMessage(content="this is an incident", attachments=[attachment]) + + # Patch `download_file` to return None as if the download failed + with patch("bot.exts.moderation.incidents.download_file", AsyncMock(return_value=None)): + embed, returned_file = await incidents.make_embed(incident, incidents.Signal.ACTIONED, MockMember()) + + self.assertIsNone(returned_file) + + # The author name field is simply expected to have something in it, we do not assert the message + self.assertGreater(len(embed.author.name), 0) + self.assertEqual(embed.author.url, "discord.com/bigbadjoe.jpg") # However, it should link the exact url + + +@patch("bot.constants.Channels.incidents", 123) +class TestIsIncident(unittest.TestCase): + """ + Collection of tests for the `is_incident` helper function. + + In `setUp`, we will create a mock message which should qualify as an incident. Each + test case will then mutate this instance to make it **not** qualify, in various ways. + + Notice that we patch the #incidents channel id globally for this class. + """ + + def setUp(self) -> None: + """Prepare a mock message which should qualify as an incident.""" + self.incident = MockMessage( + channel=MockTextChannel(id=123), + content="this is an incident", + author=MockUser(bot=False), + pinned=False, + ) + + def test_is_incident_true(self): + """Message qualifies as an incident if unchanged.""" + self.assertTrue(incidents.is_incident(self.incident)) + + def check_false(self): + """Assert that `self.incident` does **not** qualify as an incident.""" + self.assertFalse(incidents.is_incident(self.incident)) + + def test_is_incident_false_channel(self): + """Message doesn't qualify if sent outside of #incidents.""" + self.incident.channel = MockTextChannel(id=456) + self.check_false() + + def test_is_incident_false_content(self): + """Message doesn't qualify if content begins with hash symbol.""" + self.incident.content = "# this is a comment message" + self.check_false() + + def test_is_incident_false_author(self): + """Message doesn't qualify if author is a bot.""" + self.incident.author = MockUser(bot=True) + self.check_false() + + def test_is_incident_false_pinned(self): + """Message doesn't qualify if it is pinned.""" + self.incident.pinned = True + self.check_false() + + +class TestOwnReactions(unittest.TestCase): + """Assertions for the `own_reactions` function.""" + + def test_own_reactions(self): + """Only bot's own emoji are extracted from the input incident.""" + reactions = ( + MockReaction(emoji="A", me=True), + MockReaction(emoji="B", me=True), + MockReaction(emoji="C", me=False), + ) + message = MockMessage(reactions=reactions) + self.assertSetEqual(incidents.own_reactions(message), {"A", "B"}) + + +@patch("bot.exts.moderation.incidents.ALL_SIGNALS", {"A", "B"}) +class TestHasSignals(unittest.TestCase): + """ + Assertions for the `has_signals` function. + + We patch `ALL_SIGNALS` globally. Each test function then patches `own_reactions` + as appropriate. + """ + + def test_has_signals_true(self): + """True when `own_reactions` returns all emoji in `ALL_SIGNALS`.""" + message = MockMessage() + own_reactions = MagicMock(return_value={"A", "B"}) + + with patch("bot.exts.moderation.incidents.own_reactions", own_reactions): + self.assertTrue(incidents.has_signals(message)) + + def test_has_signals_false(self): + """False when `own_reactions` does not return all emoji in `ALL_SIGNALS`.""" + message = MockMessage() + own_reactions = MagicMock(return_value={"A", "C"}) + + with patch("bot.exts.moderation.incidents.own_reactions", own_reactions): + self.assertFalse(incidents.has_signals(message)) + + +@patch("bot.exts.moderation.incidents.Signal", MockSignal) +class TestAddSignals(unittest.IsolatedAsyncioTestCase): + """ + Assertions for the `add_signals` coroutine. + + These are all fairly similar and could go into a single test function, but I found the + patching & sub-testing fairly awkward in that case and decided to split them up + to avoid unnecessary syntax noise. + """ + + def setUp(self): + """Prepare a mock incident message for tests to use.""" + self.incident = MockMessage() + + @patch("bot.exts.moderation.incidents.own_reactions", MagicMock(return_value=set())) + async def test_add_signals_missing(self): + """All emoji are added when none are present.""" + await incidents.add_signals(self.incident) + self.incident.add_reaction.assert_has_calls([call("A"), call("B")]) + + @patch("bot.exts.moderation.incidents.own_reactions", MagicMock(return_value={"A"})) + async def test_add_signals_partial(self): + """Only missing emoji are added when some are present.""" + await incidents.add_signals(self.incident) + self.incident.add_reaction.assert_has_calls([call("B")]) + + @patch("bot.exts.moderation.incidents.own_reactions", MagicMock(return_value={"A", "B"})) + async def test_add_signals_present(self): + """No emoji are added when all are present.""" + await incidents.add_signals(self.incident) + self.incident.add_reaction.assert_not_called() + + +class TestIncidents(unittest.IsolatedAsyncioTestCase): + """ + Tests for bound methods of the `Incidents` cog. + + Use this as a base class for `Incidents` tests - it will prepare a fresh instance + for each test function, but not make any assertions on its own. Tests can mutate + the instance as they wish. + """ + + def setUp(self): + """ + Prepare a fresh `Incidents` instance for each test. + + Note that this will not schedule `crawl_incidents` in the background, as everything + is being mocked. The `crawl_task` attribute will end up being None. + """ + self.cog_instance = incidents.Incidents(MockBot()) + + +@patch("asyncio.sleep", AsyncMock()) # Prevent the coro from sleeping to speed up the test +class TestCrawlIncidents(TestIncidents): + """ + Tests for the `Incidents.crawl_incidents` coroutine. + + Apart from `test_crawl_incidents_waits_until_cache_ready`, all tests in this class + will patch the return values of `is_incident` and `has_signal` and then observe + whether the `AsyncMock` for `add_signals` was awaited or not. + + The `add_signals` mock is added by each test separately to ensure it is clean (has not + been awaited by another test yet). The mock can be reset, but this appears to be the + cleaner way. + + For each test, we inject a mock channel with a history of 1 message only (see: `setUp`). + """ + + def setUp(self): + """For each test, ensure `bot.get_channel` returns a channel with 1 arbitrary message.""" + super().setUp() # First ensure we get `cog_instance` from parent + + incidents_history = MagicMock(return_value=MockAsyncIterable([MockMessage()])) + self.cog_instance.bot.get_channel = MagicMock(return_value=MockTextChannel(history=incidents_history)) + + async def test_crawl_incidents_waits_until_cache_ready(self): + """ + The coroutine will await the `wait_until_guild_available` event. + + Since this task is schedule in the `__init__`, it is critical that it waits for the + cache to be ready, so that it can safely get the #incidents channel. + """ + await self.cog_instance.crawl_incidents() + self.cog_instance.bot.wait_until_guild_available.assert_awaited() + + @patch("bot.exts.moderation.incidents.add_signals", AsyncMock()) + @patch("bot.exts.moderation.incidents.is_incident", MagicMock(return_value=False)) # Message doesn't qualify + @patch("bot.exts.moderation.incidents.has_signals", MagicMock(return_value=False)) + async def test_crawl_incidents_noop_if_is_not_incident(self): + """Signals are not added for a non-incident message.""" + await self.cog_instance.crawl_incidents() + incidents.add_signals.assert_not_awaited() + + @patch("bot.exts.moderation.incidents.add_signals", AsyncMock()) + @patch("bot.exts.moderation.incidents.is_incident", MagicMock(return_value=True)) # Message qualifies + @patch("bot.exts.moderation.incidents.has_signals", MagicMock(return_value=True)) # But already has signals + async def test_crawl_incidents_noop_if_message_already_has_signals(self): + """Signals are not added for messages which already have them.""" + await self.cog_instance.crawl_incidents() + incidents.add_signals.assert_not_awaited() + + @patch("bot.exts.moderation.incidents.add_signals", AsyncMock()) + @patch("bot.exts.moderation.incidents.is_incident", MagicMock(return_value=True)) # Message qualifies + @patch("bot.exts.moderation.incidents.has_signals", MagicMock(return_value=False)) # And doesn't have signals + async def test_crawl_incidents_add_signals_called(self): + """Message has signals added as it does not have them yet and qualifies as an incident.""" + await self.cog_instance.crawl_incidents() + incidents.add_signals.assert_awaited_once() + + +class TestArchive(TestIncidents): + """Tests for the `Incidents.archive` coroutine.""" + + async def test_archive_webhook_not_found(self): + """ + Method recovers and returns False when the webhook is not found. + + Implicitly, this also tests that the error is handled internally and doesn't + propagate out of the method, which is just as important. + """ + self.cog_instance.bot.fetch_webhook = AsyncMock(side_effect=mock_404) + self.assertFalse( + await self.cog_instance.archive(incident=MockMessage(), outcome=MagicMock(), actioned_by=MockMember()) + ) + + async def test_archive_relays_incident(self): + """ + If webhook is found, method relays `incident` properly. + + This test will assert that the fetched webhook's `send` method is fed the correct arguments, + and that the `archive` method returns True. + """ + webhook = MockAsyncWebhook() + self.cog_instance.bot.fetch_webhook = AsyncMock(return_value=webhook) # Patch in our webhook + + # Define our own `incident` to be archived + incident = MockMessage( + content="this is an incident", + author=MockUser(name="author_name", avatar_url="author_avatar"), + id=123, + ) + built_embed = MagicMock(discord.Embed, id=123) # We patch `make_embed` to return this + + with patch("bot.exts.moderation.incidents.make_embed", AsyncMock(return_value=(built_embed, None))): + archive_return = await self.cog_instance.archive(incident, MagicMock(value="A"), MockMember()) + + # Now we check that the webhook was given the correct args, and that `archive` returned True + webhook.send.assert_called_once_with( + embed=built_embed, + username="author_name", + avatar_url="author_avatar", + file=None, + ) + self.assertTrue(archive_return) + + async def test_archive_clyde_username(self): + """ + The archive webhook username is cleansed using `sub_clyde`. + + Discord will reject any webhook with "clyde" in the username field, as it impersonates + the official Clyde bot. Since we do not control what the username will be (the incident + author name is used), we must ensure the name is cleansed, otherwise the relay may fail. + + This test assumes the username is passed as a kwarg. If this test fails, please review + whether the passed argument is being retrieved correctly. + """ + webhook = MockAsyncWebhook() + self.cog_instance.bot.fetch_webhook = AsyncMock(return_value=webhook) + + message_from_clyde = MockMessage(author=MockUser(name="clyde the great")) + await self.cog_instance.archive(message_from_clyde, MagicMock(incidents.Signal), MockMember()) + + self.assertNotIn("clyde", webhook.send.call_args.kwargs["username"]) + + +class TestMakeConfirmationTask(TestIncidents): + """ + Tests for the `Incidents.make_confirmation_task` method. + + Writing tests for this method is difficult, as it mostly just delegates the provided + information elsewhere. There is very little internal logic. Whether our approach + works conceptually is difficult to prove using unit tests. + """ + + def test_make_confirmation_task_check(self): + """ + The internal check will recognize the passed incident. + + This is a little tricky - we first pass a message with a specific `id` in, and then + retrieve the built check from the `call_args` of the `wait_for` method. This relies + on the check being passed as a kwarg. + + Once the check is retrieved, we assert that it gives True for our incident's `id`, + and False for any other. + + If this function begins to fail, first check that `created_check` is being retrieved + correctly. It should be the function that is built locally in the tested method. + """ + self.cog_instance.make_confirmation_task(MockMessage(id=123)) + + self.cog_instance.bot.wait_for.assert_called_once() + created_check = self.cog_instance.bot.wait_for.call_args.kwargs["check"] + + # The `message_id` matches the `id` of our incident + self.assertTrue(created_check(payload=MagicMock(message_id=123))) + + # This `message_id` does not match + self.assertFalse(created_check(payload=MagicMock(message_id=0))) + + +@patch("bot.exts.moderation.incidents.ALLOWED_ROLES", {1, 2}) +@patch("bot.exts.moderation.incidents.Incidents.make_confirmation_task", AsyncMock()) # Generic awaitable +class TestProcessEvent(TestIncidents): + """Tests for the `Incidents.process_event` coroutine.""" + + async def test_process_event_bad_role(self): + """The reaction is removed when the author lacks all allowed roles.""" + incident = MockMessage() + member = MockMember(roles=[MockRole(id=0)]) # Must have role 1 or 2 + + await self.cog_instance.process_event("reaction", incident, member) + incident.remove_reaction.assert_called_once_with("reaction", member) + + async def test_process_event_bad_emoji(self): + """ + The reaction is removed when an invalid emoji is used. + + This requires that we pass in a `member` with valid roles, as we need the role check + to succeed. + """ + incident = MockMessage() + member = MockMember(roles=[MockRole(id=1)]) # Member has allowed role + + await self.cog_instance.process_event("invalid_signal", incident, member) + incident.remove_reaction.assert_called_once_with("invalid_signal", member) + + async def test_process_event_no_archive_on_investigating(self): + """Message is not archived on `Signal.INVESTIGATING`.""" + with patch("bot.exts.moderation.incidents.Incidents.archive", AsyncMock()) as mocked_archive: + await self.cog_instance.process_event( + reaction=incidents.Signal.INVESTIGATING.value, + incident=MockMessage(), + member=MockMember(roles=[MockRole(id=1)]), + ) + + mocked_archive.assert_not_called() + + async def test_process_event_no_delete_if_archive_fails(self): + """ + Original message is not deleted when `Incidents.archive` returns False. + + This is the way of signaling that the relay failed, and we should not remove the original, + as that would result in losing the incident record. + """ + incident = MockMessage() + + with patch("bot.exts.moderation.incidents.Incidents.archive", AsyncMock(return_value=False)): + await self.cog_instance.process_event( + reaction=incidents.Signal.ACTIONED.value, + incident=incident, + member=MockMember(roles=[MockRole(id=1)]) + ) + + incident.delete.assert_not_called() + + async def test_process_event_confirmation_task_is_awaited(self): + """Task given by `Incidents.make_confirmation_task` is awaited before method exits.""" + mock_task = AsyncMock() + + with patch("bot.exts.moderation.incidents.Incidents.make_confirmation_task", mock_task): + await self.cog_instance.process_event( + reaction=incidents.Signal.ACTIONED.value, + incident=MockMessage(), + member=MockMember(roles=[MockRole(id=1)]) + ) + + mock_task.assert_awaited() + + async def test_process_event_confirmation_task_timeout_is_handled(self): + """ + Confirmation task `asyncio.TimeoutError` is handled gracefully. + + We have `make_confirmation_task` return a mock with a side effect, and then catch the + exception should it propagate out of `process_event`. This is so that we can then manually + fail the test with a more informative message than just the plain traceback. + """ + mock_task = AsyncMock(side_effect=asyncio.TimeoutError()) + + try: + with patch("bot.exts.moderation.incidents.Incidents.make_confirmation_task", mock_task): + await self.cog_instance.process_event( + reaction=incidents.Signal.ACTIONED.value, + incident=MockMessage(), + member=MockMember(roles=[MockRole(id=1)]) + ) + except asyncio.TimeoutError: + self.fail("TimeoutError was not handled gracefully, and propagated out of `process_event`!") + + +class TestResolveMessage(TestIncidents): + """Tests for the `Incidents.resolve_message` coroutine.""" + + async def test_resolve_message_pass_message_id(self): + """Method will call `_get_message` with the passed `message_id`.""" + await self.cog_instance.resolve_message(123) + self.cog_instance.bot._connection._get_message.assert_called_once_with(123) + + async def test_resolve_message_in_cache(self): + """ + No API call is made if the queried message exists in the cache. + + We mock the `_get_message` return value regardless of input. Whether it finds the message + internally is considered d.py's responsibility, not ours. + """ + cached_message = MockMessage(id=123) + self.cog_instance.bot._connection._get_message = MagicMock(return_value=cached_message) + + return_value = await self.cog_instance.resolve_message(123) + + self.assertIs(return_value, cached_message) + self.cog_instance.bot.get_channel.assert_not_called() # The `fetch_message` line was never hit + + async def test_resolve_message_not_in_cache(self): + """ + The message is retrieved from the API if it isn't cached. + + This is desired behaviour for messages which exist, but were sent before the bot's + current session. + """ + self.cog_instance.bot._connection._get_message = MagicMock(return_value=None) # Cache returns None + + # API returns our message + uncached_message = MockMessage() + fetch_message = AsyncMock(return_value=uncached_message) + self.cog_instance.bot.get_channel = MagicMock(return_value=MockTextChannel(fetch_message=fetch_message)) + + retrieved_message = await self.cog_instance.resolve_message(123) + self.assertIs(retrieved_message, uncached_message) + + async def test_resolve_message_doesnt_exist(self): + """ + If the API returns a 404, the function handles it gracefully and returns None. + + This is an edge-case happening with racing events - event A will relay the message + to the archive and delete the original. Once event B acquires the `event_lock`, + it will not find the message in the cache, and will ask the API. + """ + self.cog_instance.bot._connection._get_message = MagicMock(return_value=None) # Cache returns None + + fetch_message = AsyncMock(side_effect=mock_404) + self.cog_instance.bot.get_channel = MagicMock(return_value=MockTextChannel(fetch_message=fetch_message)) + + self.assertIsNone(await self.cog_instance.resolve_message(123)) + + async def test_resolve_message_fetch_fails(self): + """ + Non-404 errors are handled, logged & None is returned. + + In contrast with a 404, this should make an error-level log. We assert that at least + one such log was made - we do not make any assertions about the log's message. + """ + self.cog_instance.bot._connection._get_message = MagicMock(return_value=None) # Cache returns None + + arbitrary_error = discord.HTTPException( + response=MagicMock(aiohttp.ClientResponse), + message="Arbitrary error", + ) + fetch_message = AsyncMock(side_effect=arbitrary_error) + self.cog_instance.bot.get_channel = MagicMock(return_value=MockTextChannel(fetch_message=fetch_message)) + + with self.assertLogs(logger=incidents.log, level=logging.ERROR): + self.assertIsNone(await self.cog_instance.resolve_message(123)) + + +@patch("bot.constants.Channels.incidents", 123) +class TestOnRawReactionAdd(TestIncidents): + """ + Tests for the `Incidents.on_raw_reaction_add` listener. + + Writing tests for this listener comes with additional complexity due to the listener + awaiting the `crawl_task` task. See `asyncSetUp` for further details, which attempts + to make unit testing this function possible. + """ + + def setUp(self): + """ + Prepare & assign `payload` attribute. + + This attribute represents an *ideal* payload which will not be rejected by the + listener. As each test will receive a fresh instance, it can be mutated to + observe how the listener's behaviour changes with different attributes on + the passed payload. + """ + super().setUp() # Ensure `cog_instance` is assigned + + self.payload = MagicMock( + discord.RawReactionActionEvent, + channel_id=123, # Patched at class level + message_id=456, + member=MockMember(bot=False), + emoji="reaction", + ) + + async def asyncSetUp(self): # noqa: N802 + """ + Prepare an empty task and assign it as `crawl_task`. + + It appears that the `unittest` framework does not provide anything for mocking + asyncio tasks. An `AsyncMock` instance can be called and then awaited, however, + it does not provide the `done` method or any other parts of the `asyncio.Task` + interface. + + Although we do not need to make any assertions about the task itself while + testing the listener, the code will still await it and call the `done` method, + and so we must inject something that will not fail on either action. + + Note that this is done in an `asyncSetUp`, which runs after `setUp`. + The justification is that creating an actual task requires the event + loop to be ready, which is not the case in the `setUp`. + """ + mock_task = asyncio.create_task(AsyncMock()()) # Mock async func, then a coro + self.cog_instance.crawl_task = mock_task + + async def test_on_raw_reaction_add_wrong_channel(self): + """ + Events outside of #incidents will be ignored. + + We check this by asserting that `resolve_message` was never queried. + """ + self.payload.channel_id = 0 + self.cog_instance.resolve_message = AsyncMock() + + await self.cog_instance.on_raw_reaction_add(self.payload) + self.cog_instance.resolve_message.assert_not_called() + + async def test_on_raw_reaction_add_user_is_bot(self): + """ + Events dispatched by bot accounts will be ignored. + + We check this by asserting that `resolve_message` was never queried. + """ + self.payload.member = MockMember(bot=True) + self.cog_instance.resolve_message = AsyncMock() + + await self.cog_instance.on_raw_reaction_add(self.payload) + self.cog_instance.resolve_message.assert_not_called() + + async def test_on_raw_reaction_add_message_doesnt_exist(self): + """ + Listener gracefully handles the case where `resolve_message` gives None. + + We check this by asserting that `process_event` was never called. + """ + self.cog_instance.process_event = AsyncMock() + self.cog_instance.resolve_message = AsyncMock(return_value=None) + + await self.cog_instance.on_raw_reaction_add(self.payload) + self.cog_instance.process_event.assert_not_called() + + async def test_on_raw_reaction_add_message_is_not_an_incident(self): + """ + The event won't be processed if the related message is not an incident. + + This is an edge-case that can happen if someone manually leaves a reaction + on a pinned message, or a comment. + + We check this by asserting that `process_event` was never called. + """ + self.cog_instance.process_event = AsyncMock() + self.cog_instance.resolve_message = AsyncMock(return_value=MockMessage()) + + with patch("bot.exts.moderation.incidents.is_incident", MagicMock(return_value=False)): + await self.cog_instance.on_raw_reaction_add(self.payload) + + self.cog_instance.process_event.assert_not_called() + + async def test_on_raw_reaction_add_valid_event_is_processed(self): + """ + If the reaction event is valid, it is passed to `process_event`. + + This is the case when everything goes right: + * The reaction was placed in #incidents, and not by a bot + * The message was found successfully + * The message qualifies as an incident + + Additionally, we check that all arguments were passed as expected. + """ + incident = MockMessage(id=1) + + self.cog_instance.process_event = AsyncMock() + self.cog_instance.resolve_message = AsyncMock(return_value=incident) + + with patch("bot.exts.moderation.incidents.is_incident", MagicMock(return_value=True)): + await self.cog_instance.on_raw_reaction_add(self.payload) + + self.cog_instance.process_event.assert_called_with( + "reaction", # Defined in `self.payload` + incident, + self.payload.member, + ) + + +class TestOnMessage(TestIncidents): + """ + Tests for the `Incidents.on_message` listener. + + Notice the decorators mocking the `is_incident` return value. The `is_incidents` + function is tested in `TestIsIncident` - here we do not worry about it. + """ + + @patch("bot.exts.moderation.incidents.is_incident", MagicMock(return_value=True)) + async def test_on_message_incident(self): + """Messages qualifying as incidents are passed to `add_signals`.""" + incident = MockMessage() + + with patch("bot.exts.moderation.incidents.add_signals", AsyncMock()) as mock_add_signals: + await self.cog_instance.on_message(incident) + + mock_add_signals.assert_called_once_with(incident) + + @patch("bot.exts.moderation.incidents.is_incident", MagicMock(return_value=False)) + async def test_on_message_non_incident(self): + """Messages not qualifying as incidents are ignored.""" + with patch("bot.exts.moderation.incidents.add_signals", AsyncMock()) as mock_add_signals: + await self.cog_instance.on_message(MockMessage()) + + mock_add_signals.assert_not_called() diff --git a/tests/bot/exts/moderation/test_modlog.py b/tests/bot/exts/moderation/test_modlog.py new file mode 100644 index 000000000..f8f142484 --- /dev/null +++ b/tests/bot/exts/moderation/test_modlog.py @@ -0,0 +1,29 @@ +import unittest + +import discord + +from bot.exts.moderation.modlog import ModLog +from tests.helpers import MockBot, MockTextChannel + + +class ModLogTests(unittest.IsolatedAsyncioTestCase): + """Tests for moderation logs.""" + + def setUp(self): + self.bot = MockBot() + self.cog = ModLog(self.bot) + self.channel = MockTextChannel() + + async def test_log_entry_description_truncation(self): + """Test that embed description for ModLog entry is truncated.""" + self.bot.get_channel.return_value = self.channel + await self.cog.send_log_message( + icon_url="foo", + colour=discord.Colour.blue(), + title="bar", + text="foo bar" * 3000 + ) + embed = self.channel.send.call_args[1]["embed"] + self.assertEqual( + embed.description, ("foo bar" * 3000)[:2045] + "..." + ) diff --git a/tests/bot/exts/moderation/test_silence.py b/tests/bot/exts/moderation/test_silence.py new file mode 100644 index 000000000..fa5fc9e81 --- /dev/null +++ b/tests/bot/exts/moderation/test_silence.py @@ -0,0 +1,493 @@ +import asyncio +import unittest +from datetime import datetime, timezone +from unittest import mock +from unittest.mock import Mock + +from async_rediscache import RedisSession +from discord import PermissionOverwrite + +from bot.constants import Channels, Guild, Roles +from bot.exts.moderation import silence +from tests.helpers import MockBot, MockContext, MockTextChannel, autospec + +redis_session = None +redis_loop = asyncio.get_event_loop() + + +def setUpModule(): # noqa: N802 + """Create and connect to the fakeredis session.""" + global redis_session + redis_session = RedisSession(use_fakeredis=True) + redis_loop.run_until_complete(redis_session.connect()) + + +def tearDownModule(): # noqa: N802 + """Close the fakeredis session.""" + if redis_session: + redis_loop.run_until_complete(redis_session.close()) + + +# Have to subclass it because builtins can't be patched. +class PatchedDatetime(datetime): + """A datetime object with a mocked now() function.""" + + now = mock.create_autospec(datetime, "now") + + +class SilenceNotifierTests(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.alert_channel = MockTextChannel() + self.notifier = silence.SilenceNotifier(self.alert_channel) + self.notifier.stop = self.notifier_stop_mock = Mock() + self.notifier.start = self.notifier_start_mock = Mock() + + def test_add_channel_adds_channel(self): + """Channel is added to `_silenced_channels` with the current loop.""" + channel = Mock() + with mock.patch.object(self.notifier, "_silenced_channels") as silenced_channels: + self.notifier.add_channel(channel) + silenced_channels.__setitem__.assert_called_with(channel, self.notifier._current_loop) + + def test_add_channel_starts_loop(self): + """Loop is started if `_silenced_channels` was empty.""" + self.notifier.add_channel(Mock()) + self.notifier_start_mock.assert_called_once() + + def test_add_channel_skips_start_with_channels(self): + """Loop start is not called when `_silenced_channels` is not empty.""" + with mock.patch.object(self.notifier, "_silenced_channels"): + self.notifier.add_channel(Mock()) + self.notifier_start_mock.assert_not_called() + + def test_remove_channel_removes_channel(self): + """Channel is removed from `_silenced_channels`.""" + channel = Mock() + with mock.patch.object(self.notifier, "_silenced_channels") as silenced_channels: + self.notifier.remove_channel(channel) + silenced_channels.__delitem__.assert_called_with(channel) + + def test_remove_channel_stops_loop(self): + """Notifier loop is stopped if `_silenced_channels` is empty after remove.""" + with mock.patch.object(self.notifier, "_silenced_channels", __bool__=lambda _: False): + self.notifier.remove_channel(Mock()) + self.notifier_stop_mock.assert_called_once() + + def test_remove_channel_skips_stop_with_channels(self): + """Notifier loop is not stopped if `_silenced_channels` is not empty after remove.""" + self.notifier.remove_channel(Mock()) + self.notifier_stop_mock.assert_not_called() + + async def test_notifier_private_sends_alert(self): + """Alert is sent on 15 min intervals.""" + test_cases = (900, 1800, 2700) + for current_loop in test_cases: + with self.subTest(current_loop=current_loop): + with mock.patch.object(self.notifier, "_current_loop", new=current_loop): + await self.notifier._notifier() + self.alert_channel.send.assert_called_once_with( + f"<@&{Roles.moderators}> currently silenced channels: " + ) + self.alert_channel.send.reset_mock() + + async def test_notifier_skips_alert(self): + """Alert is skipped on first loop or not an increment of 900.""" + test_cases = (0, 15, 5000) + for current_loop in test_cases: + with self.subTest(current_loop=current_loop): + with mock.patch.object(self.notifier, "_current_loop", new=current_loop): + await self.notifier._notifier() + self.alert_channel.send.assert_not_called() + + +@autospec(silence.Silence, "previous_overwrites", "unsilence_timestamps", pass_mocks=False) +class SilenceCogTests(unittest.IsolatedAsyncioTestCase): + """Tests for the general functionality of the Silence cog.""" + + @autospec(silence, "Scheduler", pass_mocks=False) + def setUp(self) -> None: + self.bot = MockBot() + self.cog = silence.Silence(self.bot) + + @autospec(silence, "SilenceNotifier", pass_mocks=False) + async def test_async_init_got_guild(self): + """Bot got guild after it became available.""" + await self.cog._async_init() + self.bot.wait_until_guild_available.assert_awaited_once() + self.bot.get_guild.assert_called_once_with(Guild.id) + + @autospec(silence, "SilenceNotifier", pass_mocks=False) + async def test_async_init_got_channels(self): + """Got channels from bot.""" + self.bot.get_channel.side_effect = lambda id_: MockTextChannel(id=id_) + + await self.cog._async_init() + self.assertEqual(self.cog._mod_alerts_channel.id, Channels.mod_alerts) + + @autospec(silence, "SilenceNotifier") + async def test_async_init_got_notifier(self, notifier): + """Notifier was started with channel.""" + self.bot.get_channel.side_effect = lambda id_: MockTextChannel(id=id_) + + await self.cog._async_init() + notifier.assert_called_once_with(MockTextChannel(id=Channels.mod_log)) + self.assertEqual(self.cog.notifier, notifier.return_value) + + @autospec(silence, "SilenceNotifier", pass_mocks=False) + async def test_async_init_rescheduled(self): + """`_reschedule_` coroutine was awaited.""" + self.cog._reschedule = mock.create_autospec(self.cog._reschedule) + await self.cog._async_init() + self.cog._reschedule.assert_awaited_once_with() + + def test_cog_unload_cancelled_tasks(self): + """The init task was cancelled.""" + self.cog._init_task = asyncio.Future() + self.cog.cog_unload() + + # It's too annoying to test cancel_all since it's a done callback and wrapped in a lambda. + self.assertTrue(self.cog._init_task.cancelled()) + + @autospec("discord.ext.commands", "has_any_role") + @mock.patch.object(silence, "MODERATION_ROLES", new=(1, 2, 3)) + async def test_cog_check(self, role_check): + """Role check was called with `MODERATION_ROLES`""" + ctx = MockContext() + role_check.return_value.predicate = mock.AsyncMock() + + await self.cog.cog_check(ctx) + role_check.assert_called_once_with(*(1, 2, 3)) + role_check.return_value.predicate.assert_awaited_once_with(ctx) + + +@autospec(silence.Silence, "previous_overwrites", "unsilence_timestamps", pass_mocks=False) +class RescheduleTests(unittest.IsolatedAsyncioTestCase): + """Tests for the rescheduling of cached unsilences.""" + + @autospec(silence, "Scheduler", "SilenceNotifier", pass_mocks=False) + def setUp(self): + self.bot = MockBot() + self.cog = silence.Silence(self.bot) + self.cog._unsilence_wrapper = mock.create_autospec(self.cog._unsilence_wrapper) + + with mock.patch.object(self.cog, "_reschedule", autospec=True): + asyncio.run(self.cog._async_init()) # Populate instance attributes. + + async def test_skipped_missing_channel(self): + """Did nothing because the channel couldn't be retrieved.""" + self.cog.unsilence_timestamps.items.return_value = [(123, -1), (123, 1), (123, 10000000000)] + self.bot.get_channel.return_value = None + + await self.cog._reschedule() + + self.cog.notifier.add_channel.assert_not_called() + self.cog._unsilence_wrapper.assert_not_called() + self.cog.scheduler.schedule_later.assert_not_called() + + async def test_added_permanent_to_notifier(self): + """Permanently silenced channels were added to the notifier.""" + channels = [MockTextChannel(id=123), MockTextChannel(id=456)] + self.bot.get_channel.side_effect = channels + self.cog.unsilence_timestamps.items.return_value = [(123, -1), (456, -1)] + + await self.cog._reschedule() + + self.cog.notifier.add_channel.assert_any_call(channels[0]) + self.cog.notifier.add_channel.assert_any_call(channels[1]) + + self.cog._unsilence_wrapper.assert_not_called() + self.cog.scheduler.schedule_later.assert_not_called() + + async def test_unsilenced_expired(self): + """Unsilenced expired silences.""" + channels = [MockTextChannel(id=123), MockTextChannel(id=456)] + self.bot.get_channel.side_effect = channels + self.cog.unsilence_timestamps.items.return_value = [(123, 100), (456, 200)] + + await self.cog._reschedule() + + self.cog._unsilence_wrapper.assert_any_call(channels[0]) + self.cog._unsilence_wrapper.assert_any_call(channels[1]) + + self.cog.notifier.add_channel.assert_not_called() + self.cog.scheduler.schedule_later.assert_not_called() + + @mock.patch.object(silence, "datetime", new=PatchedDatetime) + async def test_rescheduled_active(self): + """Rescheduled active silences.""" + channels = [MockTextChannel(id=123), MockTextChannel(id=456)] + self.bot.get_channel.side_effect = channels + self.cog.unsilence_timestamps.items.return_value = [(123, 2000), (456, 3000)] + silence.datetime.now.return_value = datetime.fromtimestamp(1000, tz=timezone.utc) + + self.cog._unsilence_wrapper = mock.MagicMock() + unsilence_return = self.cog._unsilence_wrapper.return_value + + await self.cog._reschedule() + + # Yuck. + calls = [mock.call(1000, 123, unsilence_return), mock.call(2000, 456, unsilence_return)] + self.cog.scheduler.schedule_later.assert_has_calls(calls) + + unsilence_calls = [mock.call(channel) for channel in channels] + self.cog._unsilence_wrapper.assert_has_calls(unsilence_calls) + + self.cog.notifier.add_channel.assert_not_called() + + +@autospec(silence.Silence, "previous_overwrites", "unsilence_timestamps", pass_mocks=False) +class SilenceTests(unittest.IsolatedAsyncioTestCase): + """Tests for the silence command and its related helper methods.""" + + @autospec(silence.Silence, "_reschedule", pass_mocks=False) + @autospec(silence, "Scheduler", "SilenceNotifier", pass_mocks=False) + def setUp(self) -> None: + self.bot = MockBot() + self.cog = silence.Silence(self.bot) + self.cog._init_task = asyncio.Future() + self.cog._init_task.set_result(None) + + # Avoid unawaited coroutine warnings. + self.cog.scheduler.schedule_later.side_effect = lambda delay, task_id, coro: coro.close() + + asyncio.run(self.cog._async_init()) # Populate instance attributes. + + self.channel = MockTextChannel() + self.overwrite = PermissionOverwrite(stream=True, send_messages=True, add_reactions=False) + self.channel.overwrites_for.return_value = self.overwrite + + async def test_sent_correct_message(self): + """Appropriate failure/success message was sent by the command.""" + test_cases = ( + (0.0001, silence.MSG_SILENCE_SUCCESS.format(duration=0.0001), True,), + (None, silence.MSG_SILENCE_PERMANENT, True,), + (5, silence.MSG_SILENCE_FAIL, False,), + ) + for duration, message, was_silenced in test_cases: + ctx = MockContext() + with mock.patch.object(self.cog, "_set_silence_overwrites", return_value=was_silenced): + with self.subTest(was_silenced=was_silenced, message=message, duration=duration): + await self.cog.silence.callback(self.cog, ctx, duration) + ctx.send.assert_called_once_with(message) + + async def test_skipped_already_silenced(self): + """Permissions were not set and `False` was returned for an already silenced channel.""" + subtests = ( + (False, PermissionOverwrite(send_messages=False, add_reactions=False)), + (True, PermissionOverwrite(send_messages=True, add_reactions=True)), + (True, PermissionOverwrite(send_messages=False, add_reactions=False)), + ) + + for contains, overwrite in subtests: + with self.subTest(contains=contains, overwrite=overwrite): + self.cog.scheduler.__contains__.return_value = contains + channel = MockTextChannel() + channel.overwrites_for.return_value = overwrite + + self.assertFalse(await self.cog._set_silence_overwrites(channel)) + channel.set_permissions.assert_not_called() + + async def test_silenced_channel(self): + """Channel had `send_message` and `add_reactions` permissions revoked for verified role.""" + self.assertTrue(await self.cog._set_silence_overwrites(self.channel)) + self.assertFalse(self.overwrite.send_messages) + self.assertFalse(self.overwrite.add_reactions) + self.channel.set_permissions.assert_awaited_once_with( + self.cog._everyone_role, + overwrite=self.overwrite + ) + + async def test_preserved_other_overwrites(self): + """Channel's other unrelated overwrites were not changed.""" + prev_overwrite_dict = dict(self.overwrite) + await self.cog._set_silence_overwrites(self.channel) + new_overwrite_dict = dict(self.overwrite) + + # Remove 'send_messages' & 'add_reactions' keys because they were changed by the method. + del prev_overwrite_dict['send_messages'] + del prev_overwrite_dict['add_reactions'] + del new_overwrite_dict['send_messages'] + del new_overwrite_dict['add_reactions'] + + self.assertDictEqual(prev_overwrite_dict, new_overwrite_dict) + + async def test_temp_not_added_to_notifier(self): + """Channel was not added to notifier if a duration was set for the silence.""" + with mock.patch.object(self.cog, "_set_silence_overwrites", return_value=True): + await self.cog.silence.callback(self.cog, MockContext(), 15) + self.cog.notifier.add_channel.assert_not_called() + + async def test_indefinite_added_to_notifier(self): + """Channel was added to notifier if a duration was not set for the silence.""" + with mock.patch.object(self.cog, "_set_silence_overwrites", return_value=True): + await self.cog.silence.callback(self.cog, MockContext(), None) + self.cog.notifier.add_channel.assert_called_once() + + async def test_silenced_not_added_to_notifier(self): + """Channel was not added to the notifier if it was already silenced.""" + with mock.patch.object(self.cog, "_set_silence_overwrites", return_value=False): + await self.cog.silence.callback(self.cog, MockContext(), 15) + self.cog.notifier.add_channel.assert_not_called() + + async def test_cached_previous_overwrites(self): + """Channel's previous overwrites were cached.""" + overwrite_json = '{"send_messages": true, "add_reactions": false}' + await self.cog._set_silence_overwrites(self.channel) + self.cog.previous_overwrites.set.assert_called_once_with(self.channel.id, overwrite_json) + + @autospec(silence, "datetime") + async def test_cached_unsilence_time(self, datetime_mock): + """The UTC POSIX timestamp for the unsilence was cached.""" + now_timestamp = 100 + duration = 15 + timestamp = now_timestamp + duration * 60 + datetime_mock.now.return_value = datetime.fromtimestamp(now_timestamp, tz=timezone.utc) + + ctx = MockContext(channel=self.channel) + await self.cog.silence.callback(self.cog, ctx, duration) + + self.cog.unsilence_timestamps.set.assert_awaited_once_with(ctx.channel.id, timestamp) + datetime_mock.now.assert_called_once_with(tz=timezone.utc) # Ensure it's using an aware dt. + + async def test_cached_indefinite_time(self): + """A value of -1 was cached for a permanent silence.""" + ctx = MockContext(channel=self.channel) + await self.cog.silence.callback(self.cog, ctx, None) + self.cog.unsilence_timestamps.set.assert_awaited_once_with(ctx.channel.id, -1) + + async def test_scheduled_task(self): + """An unsilence task was scheduled.""" + ctx = MockContext(channel=self.channel, invoke=mock.MagicMock()) + + await self.cog.silence.callback(self.cog, ctx, 5) + + args = (300, ctx.channel.id, ctx.invoke.return_value) + self.cog.scheduler.schedule_later.assert_called_once_with(*args) + ctx.invoke.assert_called_once_with(self.cog.unsilence) + + async def test_permanent_not_scheduled(self): + """A task was not scheduled for a permanent silence.""" + ctx = MockContext(channel=self.channel) + await self.cog.silence.callback(self.cog, ctx, None) + self.cog.scheduler.schedule_later.assert_not_called() + + +@autospec(silence.Silence, "unsilence_timestamps", pass_mocks=False) +class UnsilenceTests(unittest.IsolatedAsyncioTestCase): + """Tests for the unsilence command and its related helper methods.""" + + @autospec(silence.Silence, "_reschedule", pass_mocks=False) + @autospec(silence, "Scheduler", "SilenceNotifier", pass_mocks=False) + def setUp(self) -> None: + self.bot = MockBot(get_channel=lambda _: MockTextChannel()) + self.cog = silence.Silence(self.bot) + self.cog._init_task = asyncio.Future() + self.cog._init_task.set_result(None) + + overwrites_cache = mock.create_autospec(self.cog.previous_overwrites, spec_set=True) + self.cog.previous_overwrites = overwrites_cache + + asyncio.run(self.cog._async_init()) # Populate instance attributes. + + self.cog.scheduler.__contains__.return_value = True + overwrites_cache.get.return_value = '{"send_messages": true, "add_reactions": false}' + self.channel = MockTextChannel() + self.overwrite = PermissionOverwrite(stream=True, send_messages=False, add_reactions=False) + self.channel.overwrites_for.return_value = self.overwrite + + async def test_sent_correct_message(self): + """Appropriate failure/success message was sent by the command.""" + unsilenced_overwrite = PermissionOverwrite(send_messages=True, add_reactions=True) + test_cases = ( + (True, silence.MSG_UNSILENCE_SUCCESS, unsilenced_overwrite), + (False, silence.MSG_UNSILENCE_FAIL, unsilenced_overwrite), + (False, silence.MSG_UNSILENCE_MANUAL, self.overwrite), + (False, silence.MSG_UNSILENCE_MANUAL, PermissionOverwrite(send_messages=False)), + (False, silence.MSG_UNSILENCE_MANUAL, PermissionOverwrite(add_reactions=False)), + ) + for was_unsilenced, message, overwrite in test_cases: + ctx = MockContext() + with self.subTest(was_unsilenced=was_unsilenced, message=message, overwrite=overwrite): + with mock.patch.object(self.cog, "_unsilence", return_value=was_unsilenced): + ctx.channel.overwrites_for.return_value = overwrite + await self.cog.unsilence.callback(self.cog, ctx) + ctx.channel.send.assert_called_once_with(message) + + async def test_skipped_already_unsilenced(self): + """Permissions were not set and `False` was returned for an already unsilenced channel.""" + self.cog.scheduler.__contains__.return_value = False + self.cog.previous_overwrites.get.return_value = None + channel = MockTextChannel() + + self.assertFalse(await self.cog._unsilence(channel)) + channel.set_permissions.assert_not_called() + + async def test_restored_overwrites(self): + """Channel's `send_message` and `add_reactions` overwrites were restored.""" + await self.cog._unsilence(self.channel) + self.channel.set_permissions.assert_awaited_once_with( + self.cog._everyone_role, + overwrite=self.overwrite, + ) + + # Recall that these values are determined by the fixture. + self.assertTrue(self.overwrite.send_messages) + self.assertFalse(self.overwrite.add_reactions) + + async def test_cache_miss_used_default_overwrites(self): + """Both overwrites were set to None due previous values not being found in the cache.""" + self.cog.previous_overwrites.get.return_value = None + + await self.cog._unsilence(self.channel) + self.channel.set_permissions.assert_awaited_once_with( + self.cog._everyone_role, + overwrite=self.overwrite, + ) + + self.assertIsNone(self.overwrite.send_messages) + self.assertIsNone(self.overwrite.add_reactions) + + async def test_cache_miss_sent_mod_alert(self): + """A message was sent to the mod alerts channel.""" + self.cog.previous_overwrites.get.return_value = None + + await self.cog._unsilence(self.channel) + self.cog._mod_alerts_channel.send.assert_awaited_once() + + async def test_removed_notifier(self): + """Channel was removed from `notifier`.""" + await self.cog._unsilence(self.channel) + self.cog.notifier.remove_channel.assert_called_once_with(self.channel) + + async def test_deleted_cached_overwrite(self): + """Channel was deleted from the overwrites cache.""" + await self.cog._unsilence(self.channel) + self.cog.previous_overwrites.delete.assert_awaited_once_with(self.channel.id) + + async def test_deleted_cached_time(self): + """Channel was deleted from the timestamp cache.""" + await self.cog._unsilence(self.channel) + self.cog.unsilence_timestamps.delete.assert_awaited_once_with(self.channel.id) + + async def test_cancelled_task(self): + """The scheduled unsilence task should be cancelled.""" + await self.cog._unsilence(self.channel) + self.cog.scheduler.cancel.assert_called_once_with(self.channel.id) + + async def test_preserved_other_overwrites(self): + """Channel's other unrelated overwrites were not changed, including cache misses.""" + for overwrite_json in ('{"send_messages": true, "add_reactions": null}', None): + with self.subTest(overwrite_json=overwrite_json): + self.cog.previous_overwrites.get.return_value = overwrite_json + + prev_overwrite_dict = dict(self.overwrite) + await self.cog._unsilence(self.channel) + new_overwrite_dict = dict(self.overwrite) + + # Remove these keys because they were modified by the unsilence. + del prev_overwrite_dict['send_messages'] + del prev_overwrite_dict['add_reactions'] + del new_overwrite_dict['send_messages'] + del new_overwrite_dict['add_reactions'] + + self.assertDictEqual(prev_overwrite_dict, new_overwrite_dict) diff --git a/tests/bot/exts/moderation/test_slowmode.py b/tests/bot/exts/moderation/test_slowmode.py new file mode 100644 index 000000000..dad751e0d --- /dev/null +++ b/tests/bot/exts/moderation/test_slowmode.py @@ -0,0 +1,113 @@ +import unittest +from unittest import mock + +from dateutil.relativedelta import relativedelta + +from bot.constants import Emojis +from bot.exts.moderation.slowmode import Slowmode +from tests.helpers import MockBot, MockContext, MockTextChannel + + +class SlowmodeTests(unittest.IsolatedAsyncioTestCase): + + def setUp(self) -> None: + self.bot = MockBot() + self.cog = Slowmode(self.bot) + self.ctx = MockContext() + + async def test_get_slowmode_no_channel(self) -> None: + """Get slowmode without a given channel.""" + self.ctx.channel = MockTextChannel(name='python-general', slowmode_delay=5) + + await self.cog.get_slowmode(self.cog, self.ctx, None) + self.ctx.send.assert_called_once_with("The slowmode delay for #python-general is 5 seconds.") + + async def test_get_slowmode_with_channel(self) -> None: + """Get slowmode with a given channel.""" + text_channel = MockTextChannel(name='python-language', slowmode_delay=2) + + await self.cog.get_slowmode(self.cog, self.ctx, text_channel) + self.ctx.send.assert_called_once_with('The slowmode delay for #python-language is 2 seconds.') + + async def test_set_slowmode_no_channel(self) -> None: + """Set slowmode without a given channel.""" + test_cases = ( + ('helpers', 23, True, f'{Emojis.check_mark} The slowmode delay for #helpers is now 23 seconds.'), + ('mods', 76526, False, f'{Emojis.cross_mark} The slowmode delay must be between 0 and 6 hours.'), + ('admins', 97, True, f'{Emojis.check_mark} The slowmode delay for #admins is now 1 minute and 37 seconds.') + ) + + for channel_name, seconds, edited, result_msg in test_cases: + with self.subTest( + channel_mention=channel_name, + seconds=seconds, + edited=edited, + result_msg=result_msg + ): + self.ctx.channel = MockTextChannel(name=channel_name) + + await self.cog.set_slowmode(self.cog, self.ctx, None, relativedelta(seconds=seconds)) + + if edited: + self.ctx.channel.edit.assert_awaited_once_with(slowmode_delay=float(seconds)) + else: + self.ctx.channel.edit.assert_not_called() + + self.ctx.send.assert_called_once_with(result_msg) + + self.ctx.reset_mock() + + async def test_set_slowmode_with_channel(self) -> None: + """Set slowmode with a given channel.""" + test_cases = ( + ('bot-commands', 12, True, f'{Emojis.check_mark} The slowmode delay for #bot-commands is now 12 seconds.'), + ('mod-spam', 21, True, f'{Emojis.check_mark} The slowmode delay for #mod-spam is now 21 seconds.'), + ('admin-spam', 4323598, False, f'{Emojis.cross_mark} The slowmode delay must be between 0 and 6 hours.') + ) + + for channel_name, seconds, edited, result_msg in test_cases: + with self.subTest( + channel_mention=channel_name, + seconds=seconds, + edited=edited, + result_msg=result_msg + ): + text_channel = MockTextChannel(name=channel_name) + + await self.cog.set_slowmode(self.cog, self.ctx, text_channel, relativedelta(seconds=seconds)) + + if edited: + text_channel.edit.assert_awaited_once_with(slowmode_delay=float(seconds)) + else: + text_channel.edit.assert_not_called() + + self.ctx.send.assert_called_once_with(result_msg) + + self.ctx.reset_mock() + + async def test_reset_slowmode_no_channel(self) -> None: + """Reset slowmode without a given channel.""" + self.ctx.channel = MockTextChannel(name='careers', slowmode_delay=6) + + await self.cog.reset_slowmode(self.cog, self.ctx, None) + self.ctx.send.assert_called_once_with( + f'{Emojis.check_mark} The slowmode delay for #careers has been reset to 0 seconds.' + ) + + async def test_reset_slowmode_with_channel(self) -> None: + """Reset slowmode with a given channel.""" + text_channel = MockTextChannel(name='meta', slowmode_delay=1) + + await self.cog.reset_slowmode(self.cog, self.ctx, text_channel) + self.ctx.send.assert_called_once_with( + f'{Emojis.check_mark} The slowmode delay for #meta has been reset to 0 seconds.' + ) + + @mock.patch("bot.exts.moderation.slowmode.has_any_role") + @mock.patch("bot.exts.moderation.slowmode.MODERATION_ROLES", new=(1, 2, 3)) + async def test_cog_check(self, role_check): + """Role check is called with `MODERATION_ROLES`""" + role_check.return_value.predicate = mock.AsyncMock() + await self.cog.cog_check(self.ctx) + role_check.assert_called_once_with(*(1, 2, 3)) + role_check.return_value.predicate.assert_awaited_once_with(self.ctx) diff --git a/tests/bot/exts/test_cogs.py b/tests/bot/exts/test_cogs.py new file mode 100644 index 000000000..f8e120262 --- /dev/null +++ b/tests/bot/exts/test_cogs.py @@ -0,0 +1,82 @@ +"""Test suite for general tests which apply to all cogs.""" + +import importlib +import pkgutil +import typing as t +import unittest +from collections import defaultdict +from types import ModuleType +from unittest import mock + +from discord.ext import commands + +from bot import exts + + +class CommandNameTests(unittest.TestCase): + """Tests for shadowing command names and aliases.""" + + @staticmethod + def walk_commands(cog: commands.Cog) -> t.Iterator[commands.Command]: + """An iterator that recursively walks through `cog`'s commands and subcommands.""" + # Can't use Bot.walk_commands() or Cog.get_commands() cause those are instance methods. + for command in cog.__cog_commands__: + if command.parent is None: + yield command + if isinstance(command, commands.GroupMixin): + # Annoyingly it returns duplicates for each alias so use a set to fix that + yield from set(command.walk_commands()) + + @staticmethod + def walk_modules() -> t.Iterator[ModuleType]: + """Yield imported modules from the bot.exts subpackage.""" + def on_error(name: str) -> t.NoReturn: + raise ImportError(name=name) # pragma: no cover + + # The mock prevents asyncio.get_event_loop() from being called. + with mock.patch("discord.ext.tasks.loop"): + prefix = f"{exts.__name__}." + for module in pkgutil.walk_packages(exts.__path__, prefix, onerror=on_error): + if not module.ispkg: + yield importlib.import_module(module.name) + + @staticmethod + def walk_cogs(module: ModuleType) -> t.Iterator[commands.Cog]: + """Yield all cogs defined in an extension.""" + for obj in module.__dict__.values(): + # Check if it's a class type cause otherwise issubclass() may raise a TypeError. + is_cog = isinstance(obj, type) and issubclass(obj, commands.Cog) + if is_cog and obj.__module__ == module.__name__: + yield obj + + @staticmethod + def get_qualified_names(command: commands.Command) -> t.List[str]: + """Return a list of all qualified names, including aliases, for the `command`.""" + names = [f"{command.full_parent_name} {alias}".strip() for alias in command.aliases] + names.append(command.qualified_name) + names += getattr(command, "root_aliases", []) + + return names + + def get_all_commands(self) -> t.Iterator[commands.Command]: + """Yield all commands for all cogs in all extensions.""" + for module in self.walk_modules(): + for cog in self.walk_cogs(module): + for cmd in self.walk_commands(cog): + yield cmd + + def test_names_dont_shadow(self): + """Names and aliases of commands should be unique.""" + all_names = defaultdict(list) + for cmd in self.get_all_commands(): + func_name = f"{cmd.module}.{cmd.callback.__qualname__}" + + for name in self.get_qualified_names(cmd): + with self.subTest(cmd=func_name, name=name): + 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}." + ) + + all_names[name].append(func_name) diff --git a/tests/bot/exts/utils/__init__.py b/tests/bot/exts/utils/__init__.py new file mode 100644 index 000000000..e69de29bb --- /dev/null +++ b/tests/bot/exts/utils/__init__.py diff --git a/tests/bot/exts/utils/test_jams.py b/tests/bot/exts/utils/test_jams.py new file mode 100644 index 000000000..45e7b5b51 --- /dev/null +++ b/tests/bot/exts/utils/test_jams.py @@ -0,0 +1,173 @@ +import unittest +from unittest.mock import AsyncMock, MagicMock, create_autospec + +from discord import CategoryChannel + +from bot.constants import Roles +from bot.exts.utils import jams +from tests.helpers import MockBot, MockContext, MockGuild, MockMember, MockRole, MockTextChannel + + +def get_mock_category(channel_count: int, name: str) -> CategoryChannel: + """Return a mocked code jam category.""" + category = create_autospec(CategoryChannel, spec_set=True, instance=True) + category.name = name + category.channels = [MockTextChannel() for _ in range(channel_count)] + + return category + + +class JamCreateTeamTests(unittest.IsolatedAsyncioTestCase): + """Tests for `createteam` command.""" + + def setUp(self): + self.bot = MockBot() + self.admin_role = MockRole(name="Admins", id=Roles.admins) + self.command_user = MockMember([self.admin_role]) + self.guild = MockGuild([self.admin_role]) + self.ctx = MockContext(bot=self.bot, author=self.command_user, guild=self.guild) + self.cog = jams.CodeJams(self.bot) + + async def test_too_small_amount_of_team_members_passed(self): + """Should `ctx.send` and exit early when too small amount of members.""" + for case in (1, 2): + with self.subTest(amount_of_members=case): + self.cog.create_channels = AsyncMock() + self.cog.add_roles = AsyncMock() + + self.ctx.reset_mock() + members = (MockMember() for _ in range(case)) + await self.cog.createteam(self.cog, self.ctx, "foo", members) + + self.ctx.send.assert_awaited_once() + self.cog.create_channels.assert_not_awaited() + self.cog.add_roles.assert_not_awaited() + + async def test_duplicate_members_provided(self): + """Should `ctx.send` and exit early because duplicate members provided and total there is only 1 member.""" + self.cog.create_channels = AsyncMock() + self.cog.add_roles = AsyncMock() + + member = MockMember() + await self.cog.createteam(self.cog, self.ctx, "foo", (member for _ in range(5))) + + self.ctx.send.assert_awaited_once() + self.cog.create_channels.assert_not_awaited() + self.cog.add_roles.assert_not_awaited() + + async def test_result_sending(self): + """Should call `ctx.send` when everything goes right.""" + self.cog.create_channels = AsyncMock() + self.cog.add_roles = AsyncMock() + + members = [MockMember() for _ in range(5)] + await self.cog.createteam(self.cog, self.ctx, "foo", members) + + self.cog.create_channels.assert_awaited_once() + self.cog.add_roles.assert_awaited_once() + self.ctx.send.assert_awaited_once() + + async def test_category_doesnt_exist(self): + """Should create a new code jam category.""" + subtests = ( + [], + [get_mock_category(jams.MAX_CHANNELS - 1, jams.CATEGORY_NAME)], + [get_mock_category(jams.MAX_CHANNELS - 2, "other")], + ) + + for categories in subtests: + self.guild.reset_mock() + self.guild.categories = categories + + with self.subTest(categories=categories): + actual_category = await self.cog.get_category(self.guild) + + self.guild.create_category_channel.assert_awaited_once() + category_overwrites = self.guild.create_category_channel.call_args[1]["overwrites"] + + self.assertFalse(category_overwrites[self.guild.default_role].read_messages) + self.assertTrue(category_overwrites[self.guild.me].read_messages) + self.assertEqual(self.guild.create_category_channel.return_value, actual_category) + + async def test_category_channel_exist(self): + """Should not try to create category channel.""" + expected_category = get_mock_category(jams.MAX_CHANNELS - 2, jams.CATEGORY_NAME) + self.guild.categories = [ + get_mock_category(jams.MAX_CHANNELS - 2, "other"), + expected_category, + get_mock_category(0, jams.CATEGORY_NAME), + ] + + actual_category = await self.cog.get_category(self.guild) + self.assertEqual(expected_category, actual_category) + + async def test_channel_overwrites(self): + """Should have correct permission overwrites for users and roles.""" + leader = MockMember() + members = [leader] + [MockMember() for _ in range(4)] + overwrites = self.cog.get_overwrites(members, self.guild) + + # Leader permission overwrites + self.assertTrue(overwrites[leader].manage_messages) + self.assertTrue(overwrites[leader].read_messages) + self.assertTrue(overwrites[leader].manage_webhooks) + self.assertTrue(overwrites[leader].connect) + + # Other members permission overwrites + for member in members[1:]: + self.assertTrue(overwrites[member].read_messages) + self.assertTrue(overwrites[member].connect) + + # Everyone and verified role overwrite + self.assertFalse(overwrites[self.guild.default_role].read_messages) + self.assertFalse(overwrites[self.guild.default_role].connect) + self.assertFalse(overwrites[self.guild.get_role(Roles.verified)].read_messages) + self.assertFalse(overwrites[self.guild.get_role(Roles.verified)].connect) + + async def test_team_channels_creation(self): + """Should create new voice and text channel for team.""" + members = [MockMember() for _ in range(5)] + + self.cog.get_overwrites = MagicMock() + self.cog.get_category = AsyncMock() + self.ctx.guild.create_text_channel.return_value = MockTextChannel(mention="foobar-channel") + actual = await self.cog.create_channels(self.guild, "my-team", members) + + self.assertEqual("foobar-channel", actual) + self.cog.get_overwrites.assert_called_once_with(members, self.guild) + self.cog.get_category.assert_awaited_once_with(self.guild) + + self.guild.create_text_channel.assert_awaited_once_with( + "my-team", + overwrites=self.cog.get_overwrites.return_value, + category=self.cog.get_category.return_value + ) + self.guild.create_voice_channel.assert_awaited_once_with( + "My Team", + overwrites=self.cog.get_overwrites.return_value, + category=self.cog.get_category.return_value + ) + + async def test_jam_roles_adding(self): + """Should add team leader role to leader and jam role to every team member.""" + leader_role = MockRole(name="Team Leader") + jam_role = MockRole(name="Jammer") + self.guild.get_role.side_effect = [leader_role, jam_role] + + leader = MockMember() + members = [leader] + [MockMember() for _ in range(4)] + await self.cog.add_roles(self.guild, members) + + leader.add_roles.assert_any_await(leader_role) + for member in members: + member.add_roles.assert_any_await(jam_role) + + +class CodeJamSetup(unittest.TestCase): + """Test for `setup` function of `CodeJam` cog.""" + + def test_setup(self): + """Should call `bot.add_cog`.""" + bot = MockBot() + jams.setup(bot) + bot.add_cog.assert_called_once() diff --git a/tests/bot/exts/utils/test_snekbox.py b/tests/bot/exts/utils/test_snekbox.py new file mode 100644 index 000000000..321a92445 --- /dev/null +++ b/tests/bot/exts/utils/test_snekbox.py @@ -0,0 +1,384 @@ +import asyncio +import unittest +from unittest.mock import AsyncMock, MagicMock, Mock, call, create_autospec, patch + +from discord.ext import commands + +from bot import constants +from bot.exts.utils import snekbox +from bot.exts.utils.snekbox import Snekbox +from tests.helpers import MockBot, MockContext, MockMessage, MockReaction, MockUser + + +class SnekboxTests(unittest.IsolatedAsyncioTestCase): + def setUp(self): + """Add mocked bot and cog to the instance.""" + self.bot = MockBot() + self.cog = Snekbox(bot=self.bot) + + async def test_post_eval(self): + """Post the eval code to the URLs.snekbox_eval_api endpoint.""" + resp = MagicMock() + resp.json = AsyncMock(return_value="return") + + context_manager = MagicMock() + context_manager.__aenter__.return_value = resp + self.bot.http_session.post.return_value = context_manager + + self.assertEqual(await self.cog.post_eval("import random"), "return") + self.bot.http_session.post.assert_called_with( + constants.URLs.snekbox_eval_api, + json={"input": "import random"}, + raise_for_status=True + ) + resp.json.assert_awaited_once() + + async def test_upload_output_reject_too_long(self): + """Reject output longer than MAX_PASTE_LEN.""" + result = await self.cog.upload_output("-" * (snekbox.MAX_PASTE_LEN + 1)) + self.assertEqual(result, "too long to upload") + + @patch("bot.exts.utils.snekbox.send_to_paste_service") + async def test_upload_output(self, mock_paste_util): + """Upload the eval output to the URLs.paste_service.format(key="documents") endpoint.""" + await self.cog.upload_output("Test output.") + mock_paste_util.assert_called_once_with("Test output.", extension="txt") + + def test_prepare_input(self): + cases = ( + ('print("Hello world!")', 'print("Hello world!")', 'non-formatted'), + ('`print("Hello world!")`', 'print("Hello world!")', 'one line code block'), + ('```\nprint("Hello world!")```', 'print("Hello world!")', 'multiline code block'), + ('```py\nprint("Hello world!")```', 'print("Hello world!")', 'multiline python code block'), + ('text```print("Hello world!")```text', 'print("Hello world!")', 'code block surrounded by text'), + ('```print("Hello world!")```\ntext\n```py\nprint("Hello world!")```', + 'print("Hello world!")\nprint("Hello world!")', 'two code blocks with text in-between'), + ('`print("Hello world!")`\ntext\n```print("How\'s it going?")```', + 'print("How\'s it going?")', 'code block preceded by inline code'), + ('`print("Hello world!")`\ntext\n`print("Hello world!")`', + 'print("Hello world!")', 'one inline code block of two') + ) + for case, expected, testname in cases: + with self.subTest(msg=f'Extract code from {testname}.'): + self.assertEqual(self.cog.prepare_input(case), expected) + + def test_get_results_message(self): + """Return error and message according to the eval result.""" + cases = ( + ('ERROR', None, ('Your eval job has failed', 'ERROR')), + ('', 128 + snekbox.SIGKILL, ('Your eval job timed out or ran out of memory', '')), + ('', 255, ('Your eval job has failed', 'A fatal NsJail error occurred')) + ) + for stdout, returncode, expected in cases: + with self.subTest(stdout=stdout, returncode=returncode, expected=expected): + actual = self.cog.get_results_message({'stdout': stdout, 'returncode': returncode}) + self.assertEqual(actual, expected) + + @patch('bot.exts.utils.snekbox.Signals', side_effect=ValueError) + def test_get_results_message_invalid_signal(self, mock_signals: Mock): + self.assertEqual( + self.cog.get_results_message({'stdout': '', 'returncode': 127}), + ('Your eval job has completed with return code 127', '') + ) + + @patch('bot.exts.utils.snekbox.Signals') + def test_get_results_message_valid_signal(self, mock_signals: Mock): + mock_signals.return_value.name = 'SIGTEST' + self.assertEqual( + self.cog.get_results_message({'stdout': '', 'returncode': 127}), + ('Your eval job has completed with return code 127 (SIGTEST)', '') + ) + + def test_get_status_emoji(self): + """Return emoji according to the eval result.""" + cases = ( + (' ', -1, ':warning:'), + ('Hello world!', 0, ':white_check_mark:'), + ('Invalid beard size', -1, ':x:') + ) + for stdout, returncode, expected in cases: + with self.subTest(stdout=stdout, returncode=returncode, expected=expected): + actual = self.cog.get_status_emoji({'stdout': stdout, 'returncode': returncode}) + self.assertEqual(actual, expected) + + async def test_format_output(self): + """Test output formatting.""" + self.cog.upload_output = AsyncMock(return_value='https://testificate.com/') + + too_many_lines = ( + '001 | v\n002 | e\n003 | r\n004 | y\n005 | l\n006 | o\n' + '007 | n\n008 | g\n009 | b\n010 | e\n011 | a\n... (truncated - too many lines)' + ) + too_long_too_many_lines = ( + "\n".join( + f"{i:03d} | {line}" for i, line in enumerate(['verylongbeard' * 10] * 15, 1) + )[:1000] + "\n... (truncated - too long, too many lines)" + ) + + cases = ( + ('', ('[No output]', None), 'No output'), + ('My awesome output', ('My awesome output', None), 'One line output'), + ('<@', ("<@\u200B", None), r'Convert <@ to <@\u200B'), + ('<!@', ("<!@\u200B", None), r'Convert <!@ to <!@\u200B'), + ( + '\u202E\u202E\u202E', + ('Code block escape attempt detected; will not output result', 'https://testificate.com/'), + 'Detect RIGHT-TO-LEFT OVERRIDE' + ), + ( + '\u200B\u200B\u200B', + ('Code block escape attempt detected; will not output result', 'https://testificate.com/'), + 'Detect ZERO WIDTH SPACE' + ), + ('long\nbeard', ('001 | long\n002 | beard', None), 'Two line output'), + ( + 'v\ne\nr\ny\nl\no\nn\ng\nb\ne\na\nr\nd', + (too_many_lines, 'https://testificate.com/'), + '12 lines output' + ), + ( + 'verylongbeard' * 100, + ('verylongbeard' * 76 + 'verylongbear\n... (truncated - too long)', 'https://testificate.com/'), + '1300 characters output' + ), + ( + ('verylongbeard' * 10 + '\n') * 15, + (too_long_too_many_lines, 'https://testificate.com/'), + '15 lines, 1965 characters output' + ), + ) + for case, expected, testname in cases: + with self.subTest(msg=testname, case=case, expected=expected): + self.assertEqual(await self.cog.format_output(case), expected) + + async def test_eval_command_evaluate_once(self): + """Test the eval command procedure.""" + ctx = MockContext() + response = MockMessage() + self.cog.prepare_input = MagicMock(return_value='MyAwesomeFormattedCode') + self.cog.send_eval = AsyncMock(return_value=response) + self.cog.continue_eval = AsyncMock(return_value=None) + + await self.cog.eval_command(self.cog, ctx=ctx, code='MyAwesomeCode') + self.cog.prepare_input.assert_called_once_with('MyAwesomeCode') + self.cog.send_eval.assert_called_once_with(ctx, 'MyAwesomeFormattedCode') + self.cog.continue_eval.assert_called_once_with(ctx, response) + + async def test_eval_command_evaluate_twice(self): + """Test the eval and re-eval command procedure.""" + ctx = MockContext() + response = MockMessage() + self.cog.prepare_input = MagicMock(return_value='MyAwesomeFormattedCode') + self.cog.send_eval = AsyncMock(return_value=response) + self.cog.continue_eval = AsyncMock() + self.cog.continue_eval.side_effect = ('MyAwesomeCode-2', None) + + await self.cog.eval_command(self.cog, ctx=ctx, code='MyAwesomeCode') + self.cog.prepare_input.has_calls(call('MyAwesomeCode'), call('MyAwesomeCode-2')) + self.cog.send_eval.assert_called_with(ctx, 'MyAwesomeFormattedCode') + self.cog.continue_eval.assert_called_with(ctx, response) + + async def test_eval_command_reject_two_eval_at_the_same_time(self): + """Test if the eval command rejects an eval if the author already have a running eval.""" + ctx = MockContext() + ctx.author.id = 42 + ctx.author.mention = '@LemonLemonishBeard#0042' + ctx.send = AsyncMock() + self.cog.jobs = (42,) + await self.cog.eval_command(self.cog, ctx=ctx, code='MyAwesomeCode') + ctx.send.assert_called_once_with( + "@LemonLemonishBeard#0042 You've already got a job running - please wait for it to finish!" + ) + + async def test_eval_command_call_help(self): + """Test if the eval command call the help command if no code is provided.""" + ctx = MockContext(command="sentinel") + await self.cog.eval_command(self.cog, ctx=ctx, code='') + ctx.send_help.assert_called_once_with(ctx.command) + + async def test_send_eval(self): + """Test the send_eval function.""" + ctx = MockContext() + ctx.message = MockMessage() + ctx.send = AsyncMock() + ctx.author.mention = '@LemonLemonishBeard#0042' + + self.cog.post_eval = AsyncMock(return_value={'stdout': '', 'returncode': 0}) + self.cog.get_results_message = MagicMock(return_value=('Return code 0', '')) + self.cog.get_status_emoji = MagicMock(return_value=':yay!:') + self.cog.format_output = AsyncMock(return_value=('[No output]', None)) + + mocked_filter_cog = MagicMock() + mocked_filter_cog.filter_eval = AsyncMock(return_value=False) + self.bot.get_cog.return_value = mocked_filter_cog + + await self.cog.send_eval(ctx, 'MyAwesomeCode') + ctx.send.assert_called_once_with( + '@LemonLemonishBeard#0042 :yay!: Return code 0.\n\n```\n[No output]\n```' + ) + self.cog.post_eval.assert_called_once_with('MyAwesomeCode') + self.cog.get_status_emoji.assert_called_once_with({'stdout': '', 'returncode': 0}) + self.cog.get_results_message.assert_called_once_with({'stdout': '', 'returncode': 0}) + self.cog.format_output.assert_called_once_with('') + + async def test_send_eval_with_paste_link(self): + """Test the send_eval function with a too long output that generate a paste link.""" + ctx = MockContext() + ctx.message = MockMessage() + ctx.send = AsyncMock() + ctx.author.mention = '@LemonLemonishBeard#0042' + + self.cog.post_eval = AsyncMock(return_value={'stdout': 'Way too long beard', 'returncode': 0}) + self.cog.get_results_message = MagicMock(return_value=('Return code 0', '')) + self.cog.get_status_emoji = MagicMock(return_value=':yay!:') + self.cog.format_output = AsyncMock(return_value=('Way too long beard', 'lookatmybeard.com')) + + mocked_filter_cog = MagicMock() + mocked_filter_cog.filter_eval = AsyncMock(return_value=False) + self.bot.get_cog.return_value = mocked_filter_cog + + await self.cog.send_eval(ctx, 'MyAwesomeCode') + ctx.send.assert_called_once_with( + '@LemonLemonishBeard#0042 :yay!: Return code 0.' + '\n\n```\nWay too long beard\n```\nFull output: lookatmybeard.com' + ) + self.cog.post_eval.assert_called_once_with('MyAwesomeCode') + self.cog.get_status_emoji.assert_called_once_with({'stdout': 'Way too long beard', 'returncode': 0}) + self.cog.get_results_message.assert_called_once_with({'stdout': 'Way too long beard', 'returncode': 0}) + self.cog.format_output.assert_called_once_with('Way too long beard') + + async def test_send_eval_with_non_zero_eval(self): + """Test the send_eval function with a code returning a non-zero code.""" + ctx = MockContext() + ctx.message = MockMessage() + ctx.send = AsyncMock() + ctx.author.mention = '@LemonLemonishBeard#0042' + self.cog.post_eval = AsyncMock(return_value={'stdout': 'ERROR', 'returncode': 127}) + self.cog.get_results_message = MagicMock(return_value=('Return code 127', 'Beard got stuck in the eval')) + self.cog.get_status_emoji = MagicMock(return_value=':nope!:') + self.cog.format_output = AsyncMock() # This function isn't called + + mocked_filter_cog = MagicMock() + mocked_filter_cog.filter_eval = AsyncMock(return_value=False) + self.bot.get_cog.return_value = mocked_filter_cog + + await self.cog.send_eval(ctx, 'MyAwesomeCode') + ctx.send.assert_called_once_with( + '@LemonLemonishBeard#0042 :nope!: Return code 127.\n\n```\nBeard got stuck in the eval\n```' + ) + self.cog.post_eval.assert_called_once_with('MyAwesomeCode') + self.cog.get_status_emoji.assert_called_once_with({'stdout': 'ERROR', 'returncode': 127}) + self.cog.get_results_message.assert_called_once_with({'stdout': 'ERROR', 'returncode': 127}) + self.cog.format_output.assert_not_called() + + @patch("bot.exts.utils.snekbox.partial") + async def test_continue_eval_does_continue(self, partial_mock): + """Test that the continue_eval function does continue if required conditions are met.""" + ctx = MockContext(message=MockMessage(add_reaction=AsyncMock(), clear_reactions=AsyncMock())) + response = MockMessage(delete=AsyncMock()) + new_msg = MockMessage() + self.bot.wait_for.side_effect = ((None, new_msg), None) + expected = "NewCode" + self.cog.get_code = create_autospec(self.cog.get_code, spec_set=True, return_value=expected) + + actual = await self.cog.continue_eval(ctx, response) + self.cog.get_code.assert_awaited_once_with(new_msg) + self.assertEqual(actual, expected) + self.bot.wait_for.assert_has_awaits( + ( + call( + 'message_edit', + check=partial_mock(snekbox.predicate_eval_message_edit, ctx), + timeout=snekbox.REEVAL_TIMEOUT, + ), + call('reaction_add', check=partial_mock(snekbox.predicate_eval_emoji_reaction, ctx), timeout=10) + ) + ) + ctx.message.add_reaction.assert_called_once_with(snekbox.REEVAL_EMOJI) + ctx.message.clear_reaction.assert_called_once_with(snekbox.REEVAL_EMOJI) + response.delete.assert_called_once() + + async def test_continue_eval_does_not_continue(self): + ctx = MockContext(message=MockMessage(clear_reactions=AsyncMock())) + self.bot.wait_for.side_effect = asyncio.TimeoutError + + actual = await self.cog.continue_eval(ctx, MockMessage()) + self.assertEqual(actual, None) + ctx.message.clear_reaction.assert_called_once_with(snekbox.REEVAL_EMOJI) + + async def test_get_code(self): + """Should return 1st arg (or None) if eval cmd in message, otherwise return full content.""" + prefix = constants.Bot.prefix + subtests = ( + (self.cog.eval_command, f"{prefix}{self.cog.eval_command.name} print(1)", "print(1)"), + (self.cog.eval_command, f"{prefix}{self.cog.eval_command.name}", None), + (MagicMock(spec=commands.Command), f"{prefix}tags get foo"), + (None, "print(123)") + ) + + for command, content, *expected_code in subtests: + if not expected_code: + expected_code = content + else: + [expected_code] = expected_code + + with self.subTest(content=content, expected_code=expected_code): + self.bot.get_context.reset_mock() + self.bot.get_context.return_value = MockContext(command=command) + message = MockMessage(content=content) + + actual_code = await self.cog.get_code(message) + + self.bot.get_context.assert_awaited_once_with(message) + self.assertEqual(actual_code, expected_code) + + def test_predicate_eval_message_edit(self): + """Test the predicate_eval_message_edit function.""" + msg0 = MockMessage(id=1, content='abc') + msg1 = MockMessage(id=2, content='abcdef') + msg2 = MockMessage(id=1, content='abcdef') + + cases = ( + (msg0, msg0, False, 'same ID, same content'), + (msg0, msg1, False, 'different ID, different content'), + (msg0, msg2, True, 'same ID, different content') + ) + for ctx_msg, new_msg, expected, testname in cases: + with self.subTest(msg=f'Messages with {testname} return {expected}'): + ctx = MockContext(message=ctx_msg) + actual = snekbox.predicate_eval_message_edit(ctx, ctx_msg, new_msg) + self.assertEqual(actual, expected) + + def test_predicate_eval_emoji_reaction(self): + """Test the predicate_eval_emoji_reaction function.""" + valid_reaction = MockReaction(message=MockMessage(id=1)) + valid_reaction.__str__.return_value = snekbox.REEVAL_EMOJI + valid_ctx = MockContext(message=MockMessage(id=1), author=MockUser(id=2)) + valid_user = MockUser(id=2) + + invalid_reaction_id = MockReaction(message=MockMessage(id=42)) + invalid_reaction_id.__str__.return_value = snekbox.REEVAL_EMOJI + invalid_user_id = MockUser(id=42) + invalid_reaction_str = MockReaction(message=MockMessage(id=1)) + invalid_reaction_str.__str__.return_value = ':longbeard:' + + cases = ( + (invalid_reaction_id, valid_user, False, 'invalid reaction ID'), + (valid_reaction, invalid_user_id, False, 'invalid user ID'), + (invalid_reaction_str, valid_user, False, 'invalid reaction __str__'), + (valid_reaction, valid_user, True, 'matching attributes') + ) + for reaction, user, expected, testname in cases: + with self.subTest(msg=f'Test with {testname} and expected return {expected}'): + actual = snekbox.predicate_eval_emoji_reaction(valid_ctx, reaction, user) + self.assertEqual(actual, expected) + + +class SnekboxSetupTests(unittest.TestCase): + """Tests setup of the `Snekbox` cog.""" + + def test_setup(self): + """Setup of the extension should call add_cog.""" + bot = MockBot() + snekbox.setup(bot) + bot.add_cog.assert_called_once() diff --git a/tests/bot/rules/__init__.py b/tests/bot/rules/__init__.py index 36c986fe1..0d570f5a3 100644 --- a/tests/bot/rules/__init__.py +++ b/tests/bot/rules/__init__.py @@ -12,7 +12,7 @@ class DisallowedCase(NamedTuple): n_violations: int -class RuleTest(unittest.TestCase, metaclass=ABCMeta): +class RuleTest(unittest.IsolatedAsyncioTestCase, metaclass=ABCMeta): """ Abstract class for antispam rule test cases. @@ -68,9 +68,9 @@ class RuleTest(unittest.TestCase, metaclass=ABCMeta): @abstractmethod def relevant_messages(self, case: DisallowedCase) -> Iterable[MockMessage]: """Give expected relevant messages for `case`.""" - raise NotImplementedError + raise NotImplementedError # pragma: no cover @abstractmethod def get_report(self, case: DisallowedCase) -> str: """Give expected error report for `case`.""" - raise NotImplementedError + raise NotImplementedError # pragma: no cover diff --git a/tests/bot/rules/test_attachments.py b/tests/bot/rules/test_attachments.py index e54b4b5b8..d7e779221 100644 --- a/tests/bot/rules/test_attachments.py +++ b/tests/bot/rules/test_attachments.py @@ -2,7 +2,7 @@ from typing import Iterable from bot.rules import attachments from tests.bot.rules import DisallowedCase, RuleTest -from tests.helpers import MockMessage, async_test +from tests.helpers import MockMessage def make_msg(author: str, total_attachments: int) -> MockMessage: @@ -17,7 +17,6 @@ class AttachmentRuleTests(RuleTest): self.apply = attachments.apply self.config = {"max": 5, "interval": 10} - @async_test async def test_allows_messages_without_too_many_attachments(self): """Messages without too many attachments are allowed as-is.""" cases = ( @@ -28,7 +27,6 @@ class AttachmentRuleTests(RuleTest): await self.run_allowed(cases) - @async_test async def test_disallows_messages_with_too_many_attachments(self): """Messages with too many attachments trigger the rule.""" cases = ( diff --git a/tests/bot/rules/test_burst.py b/tests/bot/rules/test_burst.py index 72f0be0c7..03682966b 100644 --- a/tests/bot/rules/test_burst.py +++ b/tests/bot/rules/test_burst.py @@ -2,7 +2,7 @@ from typing import Iterable from bot.rules import burst from tests.bot.rules import DisallowedCase, RuleTest -from tests.helpers import MockMessage, async_test +from tests.helpers import MockMessage def make_msg(author: str) -> MockMessage: @@ -21,7 +21,6 @@ class BurstRuleTests(RuleTest): self.apply = burst.apply self.config = {"max": 2, "interval": 10} - @async_test async def test_allows_messages_within_limit(self): """Cases which do not violate the rule.""" cases = ( @@ -31,7 +30,6 @@ class BurstRuleTests(RuleTest): await self.run_allowed(cases) - @async_test async def test_disallows_messages_beyond_limit(self): """Cases where the amount of messages exceeds the limit, triggering the rule.""" cases = ( diff --git a/tests/bot/rules/test_burst_shared.py b/tests/bot/rules/test_burst_shared.py index 47367a5f8..3275143d5 100644 --- a/tests/bot/rules/test_burst_shared.py +++ b/tests/bot/rules/test_burst_shared.py @@ -2,7 +2,7 @@ from typing import Iterable from bot.rules import burst_shared from tests.bot.rules import DisallowedCase, RuleTest -from tests.helpers import MockMessage, async_test +from tests.helpers import MockMessage def make_msg(author: str) -> MockMessage: @@ -21,7 +21,6 @@ class BurstSharedRuleTests(RuleTest): self.apply = burst_shared.apply self.config = {"max": 2, "interval": 10} - @async_test async def test_allows_messages_within_limit(self): """ Cases that do not violate the rule. @@ -34,7 +33,6 @@ class BurstSharedRuleTests(RuleTest): await self.run_allowed(cases) - @async_test async def test_disallows_messages_beyond_limit(self): """Cases where the amount of messages exceeds the limit, triggering the rule.""" cases = ( diff --git a/tests/bot/rules/test_chars.py b/tests/bot/rules/test_chars.py index 7cc36f49e..f1e3c76a7 100644 --- a/tests/bot/rules/test_chars.py +++ b/tests/bot/rules/test_chars.py @@ -2,7 +2,7 @@ from typing import Iterable from bot.rules import chars from tests.bot.rules import DisallowedCase, RuleTest -from tests.helpers import MockMessage, async_test +from tests.helpers import MockMessage def make_msg(author: str, n_chars: int) -> MockMessage: @@ -20,7 +20,6 @@ class CharsRuleTests(RuleTest): "interval": 10, } - @async_test async def test_allows_messages_within_limit(self): """Cases with a total amount of chars within limit.""" cases = ( @@ -31,7 +30,6 @@ class CharsRuleTests(RuleTest): await self.run_allowed(cases) - @async_test async def test_disallows_messages_beyond_limit(self): """Cases where the total amount of chars exceeds the limit, triggering the rule.""" cases = ( diff --git a/tests/bot/rules/test_discord_emojis.py b/tests/bot/rules/test_discord_emojis.py index 0239b0b00..66c2d9f92 100644 --- a/tests/bot/rules/test_discord_emojis.py +++ b/tests/bot/rules/test_discord_emojis.py @@ -2,14 +2,15 @@ from typing import Iterable from bot.rules import discord_emojis from tests.bot.rules import DisallowedCase, RuleTest -from tests.helpers import MockMessage, async_test +from tests.helpers import MockMessage discord_emoji = "<:abcd:1234>" # Discord emojis follow the format <:name:id> +unicode_emoji = "🧪" -def make_msg(author: str, n_emojis: int) -> MockMessage: +def make_msg(author: str, n_emojis: int, emoji: str = discord_emoji) -> MockMessage: """Build a MockMessage instance with content containing `n_emojis` arbitrary emojis.""" - return MockMessage(author=author, content=discord_emoji * n_emojis) + return MockMessage(author=author, content=emoji * n_emojis) class DiscordEmojisRuleTests(RuleTest): @@ -19,19 +20,23 @@ class DiscordEmojisRuleTests(RuleTest): self.apply = discord_emojis.apply self.config = {"max": 2, "interval": 10} - @async_test async def test_allows_messages_within_limit(self): - """Cases with a total amount of discord emojis within limit.""" + """Cases with a total amount of discord and unicode emojis within limit.""" cases = ( [make_msg("bob", 2)], [make_msg("alice", 1), make_msg("bob", 2), make_msg("alice", 1)], + [make_msg("bob", 2, unicode_emoji)], + [ + make_msg("alice", 1, unicode_emoji), + make_msg("bob", 2, unicode_emoji), + make_msg("alice", 1, unicode_emoji) + ], ) await self.run_allowed(cases) - @async_test async def test_disallows_messages_beyond_limit(self): - """Cases with more than the allowed amount of discord emojis.""" + """Cases with more than the allowed amount of discord and unicode emojis.""" cases = ( DisallowedCase( [make_msg("bob", 3)], @@ -43,6 +48,20 @@ class DiscordEmojisRuleTests(RuleTest): ("alice",), 4, ), + DisallowedCase( + [make_msg("bob", 3, unicode_emoji)], + ("bob",), + 3, + ), + DisallowedCase( + [ + make_msg("alice", 2, unicode_emoji), + make_msg("bob", 2, unicode_emoji), + make_msg("alice", 2, unicode_emoji) + ], + ("alice",), + 4 + ) ) await self.run_disallowed(cases) diff --git a/tests/bot/rules/test_duplicates.py b/tests/bot/rules/test_duplicates.py index 59e0fb6ef..9bd886a77 100644 --- a/tests/bot/rules/test_duplicates.py +++ b/tests/bot/rules/test_duplicates.py @@ -2,7 +2,7 @@ from typing import Iterable from bot.rules import duplicates from tests.bot.rules import DisallowedCase, RuleTest -from tests.helpers import MockMessage, async_test +from tests.helpers import MockMessage def make_msg(author: str, content: str) -> MockMessage: @@ -17,7 +17,6 @@ class DuplicatesRuleTests(RuleTest): self.apply = duplicates.apply self.config = {"max": 2, "interval": 10} - @async_test async def test_allows_messages_within_limit(self): """Cases which do not violate the rule.""" cases = ( @@ -28,7 +27,6 @@ class DuplicatesRuleTests(RuleTest): await self.run_allowed(cases) - @async_test async def test_disallows_messages_beyond_limit(self): """Cases with too many duplicate messages from the same author.""" cases = ( diff --git a/tests/bot/rules/test_links.py b/tests/bot/rules/test_links.py index 3c3f90e5f..b091bd9d7 100644 --- a/tests/bot/rules/test_links.py +++ b/tests/bot/rules/test_links.py @@ -2,7 +2,7 @@ from typing import Iterable from bot.rules import links from tests.bot.rules import DisallowedCase, RuleTest -from tests.helpers import MockMessage, async_test +from tests.helpers import MockMessage def make_msg(author: str, total_links: int) -> MockMessage: @@ -21,7 +21,6 @@ class LinksTests(RuleTest): "interval": 10 } - @async_test async def test_links_within_limit(self): """Messages with an allowed amount of links.""" cases = ( @@ -34,7 +33,6 @@ class LinksTests(RuleTest): await self.run_allowed(cases) - @async_test async def test_links_exceeding_limit(self): """Messages with a a higher than allowed amount of links.""" cases = ( diff --git a/tests/bot/rules/test_mentions.py b/tests/bot/rules/test_mentions.py index ebcdabac6..6444532f2 100644 --- a/tests/bot/rules/test_mentions.py +++ b/tests/bot/rules/test_mentions.py @@ -2,7 +2,7 @@ from typing import Iterable from bot.rules import mentions from tests.bot.rules import DisallowedCase, RuleTest -from tests.helpers import MockMessage, async_test +from tests.helpers import MockMessage def make_msg(author: str, total_mentions: int) -> MockMessage: @@ -20,7 +20,6 @@ class TestMentions(RuleTest): "interval": 10, } - @async_test async def test_mentions_within_limit(self): """Messages with an allowed amount of mentions.""" cases = ( @@ -32,7 +31,6 @@ class TestMentions(RuleTest): await self.run_allowed(cases) - @async_test async def test_mentions_exceeding_limit(self): """Messages with a higher than allowed amount of mentions.""" cases = ( diff --git a/tests/bot/rules/test_newlines.py b/tests/bot/rules/test_newlines.py index d61c4609d..e35377773 100644 --- a/tests/bot/rules/test_newlines.py +++ b/tests/bot/rules/test_newlines.py @@ -2,7 +2,7 @@ from typing import Iterable, List from bot.rules import newlines from tests.bot.rules import DisallowedCase, RuleTest -from tests.helpers import MockMessage, async_test +from tests.helpers import MockMessage def make_msg(author: str, newline_groups: List[int]) -> MockMessage: @@ -29,7 +29,6 @@ class TotalNewlinesRuleTests(RuleTest): "interval": 10, } - @async_test async def test_allows_messages_within_limit(self): """Cases which do not violate the rule.""" cases = ( @@ -41,7 +40,6 @@ class TotalNewlinesRuleTests(RuleTest): await self.run_allowed(cases) - @async_test async def test_disallows_messages_total(self): """Cases which violate the rule by having too many newlines in total.""" cases = ( @@ -79,7 +77,6 @@ class GroupNewlinesRuleTests(RuleTest): self.apply = newlines.apply self.config = {"max": 5, "max_consecutive": 3, "interval": 10} - @async_test async def test_disallows_messages_consecutive(self): """Cases which violate the rule due to having too many consecutive newlines.""" cases = ( diff --git a/tests/bot/rules/test_role_mentions.py b/tests/bot/rules/test_role_mentions.py index b339cccf7..26c05d527 100644 --- a/tests/bot/rules/test_role_mentions.py +++ b/tests/bot/rules/test_role_mentions.py @@ -2,7 +2,7 @@ from typing import Iterable from bot.rules import role_mentions from tests.bot.rules import DisallowedCase, RuleTest -from tests.helpers import MockMessage, async_test +from tests.helpers import MockMessage def make_msg(author: str, n_mentions: int) -> MockMessage: @@ -17,7 +17,6 @@ class RoleMentionsRuleTests(RuleTest): self.apply = role_mentions.apply self.config = {"max": 2, "interval": 10} - @async_test async def test_allows_messages_within_limit(self): """Cases with a total amount of role mentions within limit.""" cases = ( @@ -27,7 +26,6 @@ class RoleMentionsRuleTests(RuleTest): await self.run_allowed(cases) - @async_test async def test_disallows_messages_beyond_limit(self): """Cases with more than the allowed amount of role mentions.""" cases = ( diff --git a/tests/bot/test_api.py b/tests/bot/test_api.py index bdfcc73e4..76bcb481d 100644 --- a/tests/bot/test_api.py +++ b/tests/bot/test_api.py @@ -2,10 +2,9 @@ import unittest from unittest.mock import MagicMock from bot import api -from tests.helpers import async_test -class APIClientTests(unittest.TestCase): +class APIClientTests(unittest.IsolatedAsyncioTestCase): """Tests for the bot's API client.""" @classmethod @@ -14,15 +13,6 @@ class APIClientTests(unittest.TestCase): cls.error_api_response = MagicMock() cls.error_api_response.status = 999 - def test_loop_is_not_running_by_default(self): - """The event loop should not be running by default.""" - self.assertFalse(api.loop_is_running()) - - @async_test - async def test_loop_is_running_in_async_context(self): - """The event loop should be running in an async context.""" - self.assertTrue(api.loop_is_running()) - def test_response_code_error_default_initialization(self): """Test the default initialization of `ResponseCodeError` without `text` or `json`""" error = api.ResponseCodeError(response=self.error_api_response) diff --git a/tests/bot/test_constants.py b/tests/bot/test_constants.py index dae7c066c..f10d6fbe8 100644 --- a/tests/bot/test_constants.py +++ b/tests/bot/test_constants.py @@ -1,14 +1,40 @@ import inspect +import typing import unittest from bot import constants +def is_annotation_instance(value: typing.Any, annotation: typing.Any) -> bool: + """ + Return True if `value` is an instance of the type represented by `annotation`. + + This doesn't account for things like Unions or checking for homogenous types in collections. + """ + origin = typing.get_origin(annotation) + + # This is done in case a bare e.g. `typing.List` is used. + # In such case, for the assertion to pass, the type needs to be normalised to e.g. `list`. + # `get_origin()` does this normalisation for us. + type_ = annotation if origin is None else origin + + return isinstance(value, type_) + + +def is_any_instance(value: typing.Any, types: typing.Collection) -> bool: + """Return True if `value` is an instance of any type in `types`.""" + for type_ in types: + if is_annotation_instance(value, type_): + return True + + return False + + class ConstantsTests(unittest.TestCase): """Tests for our constants.""" def test_section_configuration_matches_type_specification(self): - """The section annotations should match the actual types of the sections.""" + """"The section annotations should match the actual types of the sections.""" sections = ( cls @@ -17,10 +43,15 @@ class ConstantsTests(unittest.TestCase): ) for section in sections: for name, annotation in section.__annotations__.items(): - with self.subTest(section=section, name=name, annotation=annotation): + with self.subTest(section=section.__name__, name=name, annotation=annotation): value = getattr(section, name) + origin = typing.get_origin(annotation) + annotation_args = typing.get_args(annotation) + failure_msg = f"{value} is not an instance of {annotation}" - if getattr(annotation, '_name', None) in ('Dict', 'List'): - self.skipTest("Cannot validate containers yet.") - - self.assertIsInstance(value, annotation) + if origin is typing.Union: + is_instance = is_any_instance(value, annotation_args) + self.assertTrue(is_instance, failure_msg) + else: + is_instance = is_annotation_instance(value, annotation) + self.assertTrue(is_instance, failure_msg) diff --git a/tests/bot/test_converters.py b/tests/bot/test_converters.py index b2b78d9dd..c42111f3f 100644 --- a/tests/bot/test_converters.py +++ b/tests/bot/test_converters.py @@ -1,5 +1,5 @@ -import asyncio import datetime +import re import unittest from unittest.mock import MagicMock, patch @@ -8,6 +8,7 @@ from discord.ext.commands import BadArgument from bot.converters import ( Duration, + HushDurationConverter, ISODateTime, TagContentConverter, TagNameConverter, @@ -15,7 +16,7 @@ from bot.converters import ( ) -class ConverterTests(unittest.TestCase): +class ConverterTests(unittest.IsolatedAsyncioTestCase): """Tests our custom argument converters.""" @classmethod @@ -25,7 +26,7 @@ class ConverterTests(unittest.TestCase): cls.fixed_utc_now = datetime.datetime.fromisoformat('2019-01-01T00:00:00') - def test_tag_content_converter_for_valid(self): + async def test_tag_content_converter_for_valid(self): """TagContentConverter should return correct values for valid input.""" test_values = ( ('hello', 'hello'), @@ -34,10 +35,10 @@ class ConverterTests(unittest.TestCase): for content, expected_conversion in test_values: with self.subTest(content=content, expected_conversion=expected_conversion): - conversion = asyncio.run(TagContentConverter.convert(self.context, content)) + conversion = await TagContentConverter.convert(self.context, content) self.assertEqual(conversion, expected_conversion) - def test_tag_content_converter_for_invalid(self): + async def test_tag_content_converter_for_invalid(self): """TagContentConverter should raise the proper exception for invalid input.""" test_values = ( ('', "Tag contents should not be empty, or filled with whitespace."), @@ -46,10 +47,10 @@ class ConverterTests(unittest.TestCase): for value, exception_message in test_values: with self.subTest(tag_content=value, exception_message=exception_message): - with self.assertRaises(BadArgument, msg=exception_message): - asyncio.run(TagContentConverter.convert(self.context, value)) + with self.assertRaisesRegex(BadArgument, re.escape(exception_message)): + await TagContentConverter.convert(self.context, value) - def test_tag_name_converter_for_valid(self): + async def test_tag_name_converter_for_valid(self): """TagNameConverter should return the correct values for valid tag names.""" test_values = ( ('tracebacks', 'tracebacks'), @@ -59,44 +60,44 @@ class ConverterTests(unittest.TestCase): for name, expected_conversion in test_values: with self.subTest(name=name, expected_conversion=expected_conversion): - conversion = asyncio.run(TagNameConverter.convert(self.context, name)) + conversion = await TagNameConverter.convert(self.context, name) self.assertEqual(conversion, expected_conversion) - def test_tag_name_converter_for_invalid(self): + async def test_tag_name_converter_for_invalid(self): """TagNameConverter should raise the correct exception for invalid tag names.""" test_values = ( ('👋', "Don't be ridiculous, you can't use that character!"), ('', "Tag names should not be empty, or filled with whitespace."), (' ', "Tag names should not be empty, or filled with whitespace."), - ('42', "Tag names can't be numbers."), + ('42', "Tag names must contain at least one letter."), ('x' * 128, "Are you insane? That's way too long!"), ) for invalid_name, exception_message in test_values: with self.subTest(invalid_name=invalid_name, exception_message=exception_message): - with self.assertRaises(BadArgument, msg=exception_message): - asyncio.run(TagNameConverter.convert(self.context, invalid_name)) + with self.assertRaisesRegex(BadArgument, re.escape(exception_message)): + await TagNameConverter.convert(self.context, invalid_name) - def test_valid_python_identifier_for_valid(self): + async def test_valid_python_identifier_for_valid(self): """ValidPythonIdentifier returns valid identifiers unchanged.""" test_values = ('foo', 'lemon') for name in test_values: with self.subTest(identifier=name): - conversion = asyncio.run(ValidPythonIdentifier.convert(self.context, name)) + conversion = await ValidPythonIdentifier.convert(self.context, name) self.assertEqual(name, conversion) - def test_valid_python_identifier_for_invalid(self): + async def test_valid_python_identifier_for_invalid(self): """ValidPythonIdentifier raises the proper exception for invalid identifiers.""" test_values = ('nested.stuff', '#####') for name in test_values: with self.subTest(identifier=name): exception_message = f'`{name}` is not a valid Python identifier' - with self.assertRaises(BadArgument, msg=exception_message): - asyncio.run(ValidPythonIdentifier.convert(self.context, name)) + with self.assertRaisesRegex(BadArgument, re.escape(exception_message)): + await ValidPythonIdentifier.convert(self.context, name) - def test_duration_converter_for_valid(self): + async def test_duration_converter_for_valid(self): """Duration returns the correct `datetime` for valid duration strings.""" test_values = ( # Simple duration strings @@ -158,35 +159,35 @@ class ConverterTests(unittest.TestCase): mock_datetime.utcnow.return_value = self.fixed_utc_now with self.subTest(duration=duration, duration_dict=duration_dict): - converted_datetime = asyncio.run(converter.convert(self.context, duration)) + converted_datetime = await converter.convert(self.context, duration) self.assertEqual(converted_datetime, expected_datetime) - def test_duration_converter_for_invalid(self): + async def test_duration_converter_for_invalid(self): """Duration raises the right exception for invalid duration strings.""" test_values = ( # Units in wrong order - ('1d1w'), - ('1s1y'), + '1d1w', + '1s1y', # Duplicated units - ('1 year 2 years'), - ('1 M 10 minutes'), + '1 year 2 years', + '1 M 10 minutes', # Unknown substrings - ('1MVes'), - ('1y3breads'), + '1MVes', + '1y3breads', # Missing amount - ('ym'), + 'ym', # Incorrect whitespace - (" 1y"), - ("1S "), - ("1y 1m"), + " 1y", + "1S ", + "1y 1m", # Garbage - ('Guido van Rossum'), - ('lemon lemon lemon lemon lemon lemon lemon'), + 'Guido van Rossum', + 'lemon lemon lemon lemon lemon lemon lemon', ) converter = Duration() @@ -194,10 +195,21 @@ class ConverterTests(unittest.TestCase): for invalid_duration in test_values: with self.subTest(invalid_duration=invalid_duration): exception_message = f'`{invalid_duration}` is not a valid duration string.' - with self.assertRaises(BadArgument, msg=exception_message): - asyncio.run(converter.convert(self.context, invalid_duration)) + with self.assertRaisesRegex(BadArgument, re.escape(exception_message)): + await converter.convert(self.context, invalid_duration) - def test_isodatetime_converter_for_valid(self): + @patch("bot.converters.datetime") + async def test_duration_converter_out_of_range(self, mock_datetime): + """Duration converter should raise BadArgument if datetime raises a ValueError.""" + mock_datetime.__add__.side_effect = ValueError + mock_datetime.utcnow.return_value = mock_datetime + + duration = f"{datetime.MAXYEAR}y" + exception_message = f"`{duration}` results in a datetime outside the supported range." + with self.assertRaisesRegex(BadArgument, re.escape(exception_message)): + await Duration().convert(self.context, duration) + + async def test_isodatetime_converter_for_valid(self): """ISODateTime converter returns correct datetime for valid datetime string.""" test_values = ( # `YYYY-mm-ddTHH:MM:SSZ` | `YYYY-mm-dd HH:MM:SSZ` @@ -242,32 +254,61 @@ class ConverterTests(unittest.TestCase): for datetime_string, expected_dt in test_values: with self.subTest(datetime_string=datetime_string, expected_dt=expected_dt): - converted_dt = asyncio.run(converter.convert(self.context, datetime_string)) + converted_dt = await converter.convert(self.context, datetime_string) self.assertIsNone(converted_dt.tzinfo) self.assertEqual(converted_dt, expected_dt) - def test_isodatetime_converter_for_invalid(self): + async def test_isodatetime_converter_for_invalid(self): """ISODateTime converter raises the correct exception for invalid datetime strings.""" test_values = ( # Make sure it doesn't interfere with the Duration converter - ('1Y'), - ('1d'), - ('1H'), + '1Y', + '1d', + '1H', # Check if it fails when only providing the optional time part - ('10:10:10'), - ('10:00'), + '10:10:10', + '10:00', # Invalid date format - ('19-01-01'), + '19-01-01', # Other non-valid strings - ('fisk the tag master'), + 'fisk the tag master', ) converter = ISODateTime() for datetime_string in test_values: with self.subTest(datetime_string=datetime_string): exception_message = f"`{datetime_string}` is not a valid ISO-8601 datetime string" - with self.assertRaises(BadArgument, msg=exception_message): - asyncio.run(converter.convert(self.context, datetime_string)) + with self.assertRaisesRegex(BadArgument, re.escape(exception_message)): + await converter.convert(self.context, datetime_string) + + async def test_hush_duration_converter_for_valid(self): + """HushDurationConverter returns correct value for minutes duration or `"forever"` strings.""" + test_values = ( + ("0", 0), + ("15", 15), + ("10", 10), + ("5m", 5), + ("5M", 5), + ("forever", None), + ) + converter = HushDurationConverter() + for minutes_string, expected_minutes in test_values: + with self.subTest(minutes_string=minutes_string, expected_minutes=expected_minutes): + converted = await converter.convert(self.context, minutes_string) + self.assertEqual(expected_minutes, converted) + + async def test_hush_duration_converter_for_invalid(self): + """HushDurationConverter raises correct exception for invalid minutes duration strings.""" + test_values = ( + ("16", "Duration must be at most 15 minutes."), + ("10d", "10d is not a valid minutes duration."), + ("-1", "-1 is not a valid minutes duration."), + ) + converter = HushDurationConverter() + for invalid_minutes_string, exception_message in test_values: + with self.subTest(invalid_minutes_string=invalid_minutes_string, exception_message=exception_message): + with self.assertRaisesRegex(BadArgument, re.escape(exception_message)): + await converter.convert(self.context, invalid_minutes_string) diff --git a/tests/bot/test_decorators.py b/tests/bot/test_decorators.py new file mode 100644 index 000000000..3d450caa0 --- /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 in_whitelist +from bot.utils.checks import InWhitelistCheckFailure +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/bot/test_pagination.py b/tests/bot/test_pagination.py index 0a734b505..630f2516d 100644 --- a/tests/bot/test_pagination.py +++ b/tests/bot/test_pagination.py @@ -8,29 +8,39 @@ class LinePaginatorTests(TestCase): def setUp(self): """Create a paginator for the test method.""" - self.paginator = pagination.LinePaginator(prefix='', suffix='', max_size=30) - - def test_add_line_raises_on_too_long_lines(self): - """`add_line` should raise a `RuntimeError` for too long lines.""" - message = f"Line exceeds maximum page size {self.paginator.max_size - 2}" - with self.assertRaises(RuntimeError, msg=message): - self.paginator.add_line('x' * self.paginator.max_size) + self.paginator = pagination.LinePaginator(prefix='', suffix='', max_size=30, + scale_to_size=50) def test_add_line_works_on_small_lines(self): """`add_line` should allow small lines to be added.""" self.paginator.add_line('x' * (self.paginator.max_size - 3)) - - -class ImagePaginatorTests(TestCase): - """Tests functionality of the `ImagePaginator`.""" - - def setUp(self): - """Create a paginator for the test method.""" - self.paginator = pagination.ImagePaginator() - - def test_add_image_appends_image(self): - """`add_image` appends the image to the image list.""" - image = 'lemon' - self.paginator.add_image(image) - - assert self.paginator.images == [image] + # Note that the page isn't added to _pages until it's full. + self.assertEqual(len(self.paginator._pages), 0) + + def test_add_line_works_on_long_lines(self): + """After additional lines after `max_size` is exceeded should go on the next page.""" + self.paginator.add_line('x' * self.paginator.max_size) + self.assertEqual(len(self.paginator._pages), 0) + + # Any additional lines should start a new page after `max_size` is exceeded. + self.paginator.add_line('x') + self.assertEqual(len(self.paginator._pages), 1) + + def test_add_line_continuation(self): + """When `scale_to_size` is exceeded, remaining words should be split onto the next page.""" + self.paginator.add_line('zyz ' * (self.paginator.scale_to_size//4 + 1)) + self.assertEqual(len(self.paginator._pages), 1) + + def test_add_line_no_continuation(self): + """If adding a new line to an existing page would exceed `max_size`, it should start a new + page rather than using continuation. + """ + self.paginator.add_line('z' * (self.paginator.max_size - 3)) + self.paginator.add_line('z') + self.assertEqual(len(self.paginator._pages), 1) + + def test_add_line_truncates_very_long_words(self): + """`add_line` should truncate if a single long word exceeds `scale_to_size`.""" + self.paginator.add_line('x' * (self.paginator.scale_to_size + 1)) + # Note: item at index 1 is the truncated line, index 0 is prefix + self.assertEqual(self.paginator._current_page[1], 'x' * self.paginator.scale_to_size) diff --git a/tests/bot/test_utils.py b/tests/bot/test_utils.py deleted file mode 100644 index 58ae2a81a..000000000 --- a/tests/bot/test_utils.py +++ /dev/null @@ -1,52 +0,0 @@ -import unittest - -from bot import utils - - -class CaseInsensitiveDictTests(unittest.TestCase): - """Tests for the `CaseInsensitiveDict` container.""" - - def test_case_insensitive_key_access(self): - """Tests case insensitive key access and storage.""" - instance = utils.CaseInsensitiveDict() - - key = 'LEMON' - value = 'trees' - - instance[key] = value - self.assertIn(key, instance) - self.assertEqual(instance.get(key), value) - self.assertEqual(instance.get(key.casefold()), value) - self.assertEqual(instance.pop(key.casefold()), value) - self.assertNotIn(key, instance) - self.assertNotIn(key.casefold(), instance) - - instance.setdefault(key, value) - del instance[key] - self.assertNotIn(key, instance) - - def test_initialization_from_kwargs(self): - """Tests creating the dictionary from keyword arguments.""" - instance = utils.CaseInsensitiveDict({'FOO': 'bar'}) - self.assertEqual(instance['foo'], 'bar') - - def test_update_from_other_mapping(self): - """Tests updating the dictionary from another mapping.""" - instance = utils.CaseInsensitiveDict() - instance.update({'FOO': 'bar'}) - self.assertEqual(instance['foo'], 'bar') - - -class ChunkTests(unittest.TestCase): - """Tests the `chunk` method.""" - - def test_empty_chunking(self): - """Tests chunking on an empty iterable.""" - generator = utils.chunks(iterable=[], size=5) - self.assertEqual(list(generator), []) - - def test_list_chunking(self): - """Tests chunking a non-empty list.""" - iterable = [1, 2, 3, 4, 5] - generator = utils.chunks(iterable=iterable, size=2) - self.assertEqual(list(generator), [[1, 2], [3, 4], [5]]) diff --git a/tests/bot/utils/test_checks.py b/tests/bot/utils/test_checks.py index 9610771e5..883465e0b 100644 --- a/tests/bot/utils/test_checks.py +++ b/tests/bot/utils/test_checks.py @@ -1,51 +1,93 @@ import unittest +from unittest.mock import MagicMock + +from discord import DMChannel from bot.utils import checks +from bot.utils.checks import InWhitelistCheckFailure from tests.helpers import MockContext, MockRole -class ChecksTests(unittest.TestCase): +class ChecksTests(unittest.IsolatedAsyncioTestCase): """Tests the check functions defined in `bot.checks`.""" def setUp(self): self.ctx = MockContext() - def test_with_role_check_without_guild(self): - """`with_role_check` returns `False` if `Context.guild` is None.""" - self.ctx.guild = None - self.assertFalse(checks.with_role_check(self.ctx)) + async def test_has_any_role_check_without_guild(self): + """`has_any_role_check` returns `False` for non-guild channels.""" + self.ctx.channel = MagicMock(DMChannel) + self.assertFalse(await checks.has_any_role_check(self.ctx)) - def test_with_role_check_without_required_roles(self): - """`with_role_check` returns `False` if `Context.author` lacks the required role.""" + async def test_has_any_role_check_without_required_roles(self): + """`has_any_role_check` returns `False` if `Context.author` lacks the required role.""" self.ctx.author.roles = [] - self.assertFalse(checks.with_role_check(self.ctx)) + self.assertFalse(await checks.has_any_role_check(self.ctx)) - def test_with_role_check_with_guild_and_required_role(self): - """`with_role_check` returns `True` if `Context.author` has the required role.""" + async def test_has_any_role_check_with_guild_and_required_role(self): + """`has_any_role_check` returns `True` if `Context.author` has the required role.""" self.ctx.author.roles.append(MockRole(id=10)) - self.assertTrue(checks.with_role_check(self.ctx, 10)) + self.assertTrue(await checks.has_any_role_check(self.ctx, 10)) - def test_without_role_check_without_guild(self): - """`without_role_check` should return `False` when `Context.guild` is None.""" - self.ctx.guild = None - self.assertFalse(checks.without_role_check(self.ctx)) + async def test_has_no_roles_check_without_guild(self): + """`has_no_roles_check` should return `False` when `Context.guild` is None.""" + self.ctx.channel = MagicMock(DMChannel) + self.assertFalse(await checks.has_no_roles_check(self.ctx)) - def test_without_role_check_returns_false_with_unwanted_role(self): - """`without_role_check` returns `False` if `Context.author` has unwanted role.""" + async def test_has_no_roles_check_returns_false_with_unwanted_role(self): + """`has_no_roles_check` returns `False` if `Context.author` has unwanted role.""" role_id = 42 self.ctx.author.roles.append(MockRole(id=role_id)) - self.assertFalse(checks.without_role_check(self.ctx, role_id)) + self.assertFalse(await checks.has_no_roles_check(self.ctx, role_id)) - def test_without_role_check_returns_true_without_unwanted_role(self): - """`without_role_check` returns `True` if `Context.author` does not have unwanted role.""" + async def test_has_no_roles_check_returns_true_without_unwanted_role(self): + """`has_no_roles_check` returns `True` if `Context.author` does not have unwanted role.""" role_id = 42 self.ctx.author.roles.append(MockRole(id=role_id)) - self.assertTrue(checks.without_role_check(self.ctx, role_id + 10)) + self.assertTrue(await checks.has_no_roles_check(self.ctx, role_id + 10)) + + def test_in_whitelist_check_correct_channel(self): + """`in_whitelist_check` returns `True` if `Context.channel.id` is in the channel list.""" + channel_id = 3 + self.ctx.channel.id = channel_id + self.assertTrue(checks.in_whitelist_check(self.ctx, [channel_id])) + + def test_in_whitelist_check_incorrect_channel(self): + """`in_whitelist_check` raises InWhitelistCheckFailure if there's no channel match.""" + self.ctx.channel.id = 3 + with self.assertRaises(InWhitelistCheckFailure): + checks.in_whitelist_check(self.ctx, [4]) + + def test_in_whitelist_check_correct_category(self): + """`in_whitelist_check` returns `True` if `Context.channel.category_id` is in the category list.""" + category_id = 3 + self.ctx.channel.category_id = category_id + self.assertTrue(checks.in_whitelist_check(self.ctx, categories=[category_id])) + + def test_in_whitelist_check_incorrect_category(self): + """`in_whitelist_check` raises InWhitelistCheckFailure if there's no category match.""" + self.ctx.channel.category_id = 3 + with self.assertRaises(InWhitelistCheckFailure): + checks.in_whitelist_check(self.ctx, categories=[4]) + + def test_in_whitelist_check_correct_role(self): + """`in_whitelist_check` returns `True` if any of the `Context.author.roles` are in the roles list.""" + self.ctx.author.roles = (MagicMock(id=1), MagicMock(id=2)) + self.assertTrue(checks.in_whitelist_check(self.ctx, roles=[2, 6])) + + def test_in_whitelist_check_incorrect_role(self): + """`in_whitelist_check` raises InWhitelistCheckFailure if there's no role match.""" + self.ctx.author.roles = (MagicMock(id=1), MagicMock(id=2)) + with self.assertRaises(InWhitelistCheckFailure): + checks.in_whitelist_check(self.ctx, roles=[4]) - def test_in_channel_check_for_correct_channel(self): - self.ctx.channel.id = 42 - self.assertTrue(checks.in_channel_check(self.ctx, *[42])) + def test_in_whitelist_check_fail_silently(self): + """`in_whitelist_check` test no exception raised if `fail_silently` is `True`""" + self.assertFalse(checks.in_whitelist_check(self.ctx, roles=[2, 6], fail_silently=True)) - def test_in_channel_check_for_incorrect_channel(self): - self.ctx.channel.id = 42 + 10 - self.assertFalse(checks.in_channel_check(self.ctx, *[42])) + def test_in_whitelist_check_complex(self): + """`in_whitelist_check` test with multiple parameters""" + self.ctx.author.roles = (MagicMock(id=1), MagicMock(id=2)) + self.ctx.channel.category_id = 3 + self.ctx.channel.id = 5 + self.assertTrue(checks.in_whitelist_check(self.ctx, channels=[1], categories=[8], roles=[2])) diff --git a/tests/bot/utils/test_messages.py b/tests/bot/utils/test_messages.py new file mode 100644 index 000000000..9c22c9751 --- /dev/null +++ b/tests/bot/utils/test_messages.py @@ -0,0 +1,27 @@ +import unittest + +from bot.utils import messages + + +class TestMessages(unittest.TestCase): + """Tests for functions in the `bot.utils.messages` module.""" + + def test_sub_clyde(self): + """Uppercase E's and lowercase e's are substituted with their cyrillic counterparts.""" + sub_e = "\u0435" + sub_E = "\u0415" # noqa: N806: Uppercase E in variable name + + test_cases = ( + (None, None), + ("", ""), + ("clyde", f"clyd{sub_e}"), + ("CLYDE", f"CLYD{sub_E}"), + ("cLyDe", f"cLyD{sub_e}"), + ("BIGclyde", f"BIGclyd{sub_e}"), + ("small clydeus the unholy", f"small clyd{sub_e}us the unholy"), + ("BIGCLYDE, babyclyde", f"BIGCLYD{sub_E}, babyclyd{sub_e}"), + ) + + for username_in, username_out in test_cases: + with self.subTest(input=username_in, expected_output=username_out): + self.assertEqual(messages.sub_clyde(username_in), username_out) diff --git a/tests/bot/utils/test_services.py b/tests/bot/utils/test_services.py new file mode 100644 index 000000000..1b48f6560 --- /dev/null +++ b/tests/bot/utils/test_services.py @@ -0,0 +1,77 @@ +import logging +import unittest +from unittest.mock import AsyncMock, MagicMock, Mock, patch + +from aiohttp import ClientConnectorError + +from bot.utils.services import FAILED_REQUEST_ATTEMPTS, send_to_paste_service +from tests.helpers import MockBot + + +class PasteTests(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + patcher = patch("bot.instance", new=MockBot()) + self.bot = patcher.start() + self.addCleanup(patcher.stop) + + @patch("bot.utils.services.URLs.paste_service", "https://paste_service.com/{key}") + async def test_url_and_sent_contents(self): + """Correct url was used and post was called with expected data.""" + response = MagicMock( + json=AsyncMock(return_value={"key": ""}) + ) + self.bot.http_session.post.return_value.__aenter__.return_value = response + self.bot.http_session.post.reset_mock() + await send_to_paste_service("Content") + self.bot.http_session.post.assert_called_once_with("https://paste_service.com/documents", data="Content") + + @patch("bot.utils.services.URLs.paste_service", "https://paste_service.com/{key}") + async def test_paste_returns_correct_url_on_success(self): + """Url with specified extension is returned on successful requests.""" + key = "paste_key" + test_cases = ( + (f"https://paste_service.com/{key}.txt", "txt"), + (f"https://paste_service.com/{key}.py", "py"), + (f"https://paste_service.com/{key}", ""), + ) + response = MagicMock( + json=AsyncMock(return_value={"key": key}) + ) + self.bot.http_session.post.return_value.__aenter__.return_value = response + + for expected_output, extension in test_cases: + with self.subTest(msg=f"Send contents with extension {repr(extension)}"): + self.assertEqual( + await send_to_paste_service("", extension=extension), + expected_output + ) + + async def test_request_repeated_on_json_errors(self): + """Json with error message and invalid json are handled as errors and requests repeated.""" + test_cases = ({"message": "error"}, {"unexpected_key": None}, {}) + self.bot.http_session.post.return_value.__aenter__.return_value = response = MagicMock() + self.bot.http_session.post.reset_mock() + + for error_json in test_cases: + with self.subTest(error_json=error_json): + response.json = AsyncMock(return_value=error_json) + result = await send_to_paste_service("") + self.assertEqual(self.bot.http_session.post.call_count, FAILED_REQUEST_ATTEMPTS) + self.assertIsNone(result) + + self.bot.http_session.post.reset_mock() + + async def test_request_repeated_on_connection_errors(self): + """Requests are repeated in the case of connection errors.""" + self.bot.http_session.post = MagicMock(side_effect=ClientConnectorError(Mock(), Mock())) + result = await send_to_paste_service("") + self.assertEqual(self.bot.http_session.post.call_count, FAILED_REQUEST_ATTEMPTS) + self.assertIsNone(result) + + async def test_general_error_handled_and_request_repeated(self): + """All `Exception`s are handled, logged and request repeated.""" + self.bot.http_session.post = MagicMock(side_effect=Exception) + result = await send_to_paste_service("") + self.assertEqual(self.bot.http_session.post.call_count, FAILED_REQUEST_ATTEMPTS) + self.assertLogs("bot.utils", logging.ERROR) + self.assertIsNone(result) diff --git a/tests/bot/utils/test_time.py b/tests/bot/utils/test_time.py index 69f35f2f5..694d3a40f 100644 --- a/tests/bot/utils/test_time.py +++ b/tests/bot/utils/test_time.py @@ -1,12 +1,11 @@ import asyncio import unittest from datetime import datetime, timezone -from unittest.mock import patch +from unittest.mock import AsyncMock, patch from dateutil.relativedelta import relativedelta from bot.utils import time -from tests.helpers import AsyncMock class TimeTests(unittest.TestCase): @@ -44,7 +43,7 @@ class TimeTests(unittest.TestCase): for max_units in test_cases: with self.subTest(max_units=max_units), self.assertRaises(ValueError) as error: time.humanize_delta(relativedelta(days=2, hours=2), 'hours', max_units) - self.assertEqual(str(error), 'max_units must be positive') + self.assertEqual(str(error.exception), 'max_units must be positive') def test_parse_rfc1123(self): """Testing parse_rfc1123.""" diff --git a/tests/helpers.py b/tests/helpers.py index 5df796c23..496363ae3 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -1,18 +1,20 @@ from __future__ import annotations -import asyncio import collections -import functools -import inspect import itertools import logging import unittest.mock -from typing import Any, Iterable, Optional +from asyncio import AbstractEventLoop +from typing import Iterable, Optional import discord +from aiohttp import ClientSession from discord.ext.commands import Context +from bot.api import APIClient +from bot.async_stats import AsyncStatsClient from bot.bot import Bot +from tests._autospec import autospec # noqa: F401 other modules import it via this module for logger in logging.Logger.manager.loggerDict.values(): @@ -25,21 +27,6 @@ for logger in logging.Logger.manager.loggerDict.values(): logger.setLevel(logging.CRITICAL) -def async_test(wrapped): - """ - Run a test case via asyncio. - Example: - >>> @async_test - ... async def lemon_wins(): - ... assert True - """ - - @functools.wraps(wrapped) - def wrapper(*args, **kwargs): - return asyncio.run(wrapped(*args, **kwargs)) - return wrapper - - class HashableMixin(discord.mixins.EqualityComparable): """ Mixin that provides similar hashing and equality functionality as discord.py's `Hashable` mixin. @@ -68,24 +55,31 @@ class CustomMockMixin: """ Provides common functionality for our custom Mock types. - The cooperative `__init__` automatically creates `AsyncMock` attributes for every coroutine - function `inspect` detects in the `spec` instance we provide. In addition, this mixin takes care - of making sure child mocks are instantiated with the correct class. By default, the mock of the - children will be `unittest.mock.MagicMock`, but this can be overwritten by setting the attribute - `child_mock_type` on the custom mock inheriting from this mixin. + The `_get_child_mock` method automatically returns an AsyncMock for coroutine methods of the mock + object. As discord.py also uses synchronous methods that nonetheless return coroutine objects, the + class attribute `additional_spec_asyncs` can be overwritten with an iterable containing additional + attribute names that should also mocked with an AsyncMock instead of a regular MagicMock/Mock. The + class method `spec_set` can be overwritten with the object that should be uses as the specification + for the mock. + + Mock/MagicMock subclasses that use this mixin only need to define `__init__` method if they need to + implement custom behavior. """ child_mock_type = unittest.mock.MagicMock discord_id = itertools.count(0) + spec_set = None + additional_spec_asyncs = None - def __init__(self, spec_set: Any = None, **kwargs): + def __init__(self, **kwargs): name = kwargs.pop('name', None) # `name` has special meaning for Mock classes, so we need to set it manually. - super().__init__(spec_set=spec_set, **kwargs) + super().__init__(spec_set=self.spec_set, **kwargs) + + if self.additional_spec_asyncs: + self._spec_asyncs.extend(self.additional_spec_asyncs) if name: self.name = name - if spec_set: - self._extract_coroutine_methods_from_spec_instance(spec_set) def _get_child_mock(self, **kw): """ @@ -99,7 +93,16 @@ class CustomMockMixin: This override will look for an attribute called `child_mock_type` and use that as the type of the child mock. """ - klass = self.child_mock_type + _new_name = kw.get("_new_name") + if _new_name in self.__dict__['_spec_asyncs']: + return unittest.mock.AsyncMock(**kw) + + _type = type(self) + if issubclass(_type, unittest.mock.MagicMock) and _new_name in unittest.mock._async_method_magics: + # Any asynchronous magic becomes an AsyncMock + klass = unittest.mock.AsyncMock + else: + klass = self.child_mock_type if self._mock_sealed: attribute = "." + kw["name"] if "name" in kw else "()" @@ -108,95 +111,6 @@ class CustomMockMixin: return klass(**kw) - def _extract_coroutine_methods_from_spec_instance(self, source: Any) -> None: - """Automatically detect coroutine functions in `source` and set them as AsyncMock attributes.""" - for name, _method in inspect.getmembers(source, inspect.iscoroutinefunction): - setattr(self, name, AsyncMock()) - - -# TODO: Remove me in Python 3.8 -class AsyncMock(CustomMockMixin, unittest.mock.MagicMock): - """ - A MagicMock subclass to mock async callables. - - Python 3.8 will introduce an AsyncMock class in the standard library that will have some more - features; this stand-in only overwrites the `__call__` method to an async version. - """ - - async def __call__(self, *args, **kwargs): - return super().__call__(*args, **kwargs) - - -class AsyncIteratorMock: - """ - A class to mock asynchronous iterators. - - This allows async for, which is used in certain Discord.py objects. For example, - an async iterator is returned by the Reaction.users() method. - """ - - def __init__(self, iterable: Iterable = None): - if iterable is None: - iterable = [] - - self.iter = iter(iterable) - self.iterable = iterable - - self.call_count = 0 - - def __aiter__(self): - return self - - async def __anext__(self): - try: - return next(self.iter) - except StopIteration: - raise StopAsyncIteration - - def __call__(self): - """ - Keeps track of the number of times an instance has been called. - - This is useful, since it typically shows that the iterator has actually been used somewhere after we have - instantiated the mock for an attribute that normally returns an iterator when called. - """ - self.call_count += 1 - return self - - @property - def return_value(self): - """Makes `self.iterable` accessible as self.return_value.""" - return self.iterable - - @return_value.setter - def return_value(self, iterable): - """Stores the `return_value` as `self.iterable` and its iterator as `self.iter`.""" - self.iter = iter(iterable) - self.iterable = iterable - - def assert_called(self): - """Asserts if the AsyncIteratorMock instance has been called at least once.""" - if self.call_count == 0: - raise AssertionError("Expected AsyncIteratorMock to have been called.") - - def assert_called_once(self): - """Asserts if the AsyncIteratorMock instance has been called exactly once.""" - if self.call_count != 1: - raise AssertionError( - f"Expected AsyncIteratorMock to have been called once. Called {self.call_count} times." - ) - - def assert_not_called(self): - """Asserts if the AsyncIteratorMock instance has not been called.""" - if self.call_count != 0: - raise AssertionError( - f"Expected AsyncIteratorMock to not have been called once. Called {self.call_count} times." - ) - - def reset_mock(self): - """Resets the call count, but not the return value or iterator.""" - self.call_count = 0 - # Create a guild instance to get a realistic Mock of `discord.Guild` guild_data = { @@ -247,9 +161,11 @@ class MockGuild(CustomMockMixin, unittest.mock.Mock, HashableMixin): For more info, see the `Mocking` section in `tests/README.md`. """ + spec_set = guild_instance + def __init__(self, roles: Optional[Iterable[MockRole]] = None, **kwargs) -> None: default_kwargs = {'id': next(self.discord_id), 'members': []} - super().__init__(spec_set=guild_instance, **collections.ChainMap(kwargs, default_kwargs)) + super().__init__(**collections.ChainMap(kwargs, default_kwargs)) self.roles = [MockRole(name="@everyone", position=1, id=0)] if roles: @@ -268,9 +184,23 @@ class MockRole(CustomMockMixin, unittest.mock.Mock, ColourMixin, HashableMixin): Instances of this class will follow the specifications of `discord.Role` instances. For more information, see the `MockGuild` docstring. """ + spec_set = role_instance + def __init__(self, **kwargs) -> None: - default_kwargs = {'id': next(self.discord_id), 'name': 'role', 'position': 1} - super().__init__(spec_set=role_instance, **collections.ChainMap(kwargs, default_kwargs)) + default_kwargs = { + 'id': next(self.discord_id), + 'name': 'role', + 'position': 1, + 'colour': discord.Colour(0xdeadbf), + 'permissions': discord.Permissions(), + } + super().__init__(**collections.ChainMap(kwargs, default_kwargs)) + + if isinstance(self.colour, int): + self.colour = discord.Colour(self.colour) + + if isinstance(self.permissions, int): + self.permissions = discord.Permissions(self.permissions) if 'mention' not in kwargs: self.mention = f'&{self.name}' @@ -279,6 +209,10 @@ class MockRole(CustomMockMixin, unittest.mock.Mock, ColourMixin, HashableMixin): """Simplified position-based comparisons similar to those of `discord.Role`.""" return self.position < other.position + def __ge__(self, other): + """Simplified position-based comparisons similar to those of `discord.Role`.""" + return self.position >= other.position + # Create a Member instance to get a realistic Mock of `discord.Member` member_data = {'user': 'lemon', 'roles': [1]} @@ -293,9 +227,11 @@ class MockMember(CustomMockMixin, unittest.mock.Mock, ColourMixin, HashableMixin Instances of this class will follow the specifications of `discord.Member` instances. For more information, see the `MockGuild` docstring. """ + spec_set = member_instance + def __init__(self, roles: Optional[Iterable[MockRole]] = None, **kwargs) -> None: - default_kwargs = {'name': 'member', 'id': next(self.discord_id), 'bot': False} - super().__init__(spec_set=member_instance, **collections.ChainMap(kwargs, default_kwargs)) + default_kwargs = {'name': 'member', 'id': next(self.discord_id), 'bot': False, "pending": False} + super().__init__(**collections.ChainMap(kwargs, default_kwargs)) self.roles = [MockRole(name="@everyone", position=1, id=0)] if roles: @@ -316,18 +252,36 @@ class MockUser(CustomMockMixin, unittest.mock.Mock, ColourMixin, HashableMixin): Instances of this class will follow the specifications of `discord.User` instances. For more information, see the `MockGuild` docstring. """ + spec_set = user_instance + def __init__(self, **kwargs) -> None: default_kwargs = {'name': 'user', 'id': next(self.discord_id), 'bot': False} - super().__init__(spec_set=user_instance, **collections.ChainMap(kwargs, default_kwargs)) + super().__init__(**collections.ChainMap(kwargs, default_kwargs)) if 'mention' not in kwargs: self.mention = f"@{self.name}" -# Create a Bot instance to get a realistic MagicMock of `discord.ext.commands.Bot` -bot_instance = Bot(command_prefix=unittest.mock.MagicMock()) -bot_instance.http_session = None -bot_instance.api_client = None +class MockAPIClient(CustomMockMixin, unittest.mock.MagicMock): + """ + A MagicMock subclass to mock APIClient objects. + + Instances of this class will follow the specifications of `bot.api.APIClient` instances. + For more information, see the `MockGuild` docstring. + """ + spec_set = APIClient + + +def _get_mock_loop() -> unittest.mock.Mock: + """Return a mocked asyncio.AbstractEventLoop.""" + loop = unittest.mock.create_autospec(spec=AbstractEventLoop, spec_set=True) + + # Since calling `create_task` on our MockBot does not actually schedule the coroutine object + # as a task in the asyncio loop, this `side_effect` calls `close()` on the coroutine object + # to prevent "has not been awaited"-warnings. + loop.create_task.side_effect = lambda coroutine: coroutine.close() + + return loop class MockBot(CustomMockMixin, unittest.mock.MagicMock): @@ -337,19 +291,20 @@ class MockBot(CustomMockMixin, unittest.mock.MagicMock): Instances of this class will follow the specifications of `discord.ext.commands.Bot` instances. For more information, see the `MockGuild` docstring. """ + spec_set = Bot( + command_prefix=unittest.mock.MagicMock(), + loop=_get_mock_loop(), + redis_session=unittest.mock.MagicMock(), + ) + additional_spec_asyncs = ("wait_for", "redis_ready") def __init__(self, **kwargs) -> None: - super().__init__(spec_set=bot_instance, **kwargs) - - # self.wait_for is *not* a coroutine function, but returns a coroutine nonetheless and - # and should therefore be awaited. (The documentation calls it a coroutine as well, which - # is technically incorrect, since it's a regular def.) - self.wait_for = AsyncMock() + super().__init__(**kwargs) - # Since calling `create_task` on our MockBot does not actually schedule the coroutine object - # as a task in the asyncio loop, this `side_effect` calls `close()` on the coroutine object - # to prevent "has not been awaited"-warnings. - self.loop.create_task.side_effect = lambda coroutine: coroutine.close() + self.loop = _get_mock_loop() + self.api_client = MockAPIClient(loop=self.loop) + self.http_session = unittest.mock.create_autospec(spec=ClientSession, spec_set=True) + self.stats = unittest.mock.create_autospec(spec=AsyncStatsClient, spec_set=True) # Create a TextChannel instance to get a realistic MagicMock of `discord.TextChannel` @@ -375,15 +330,37 @@ class MockTextChannel(CustomMockMixin, unittest.mock.Mock, HashableMixin): Instances of this class will follow the specifications of `discord.TextChannel` instances. For more information, see the `MockGuild` docstring. """ + 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__(spec_set=channel_instance, **collections.ChainMap(kwargs, default_kwargs)) + super().__init__(**collections.ChainMap(kwargs, default_kwargs)) if 'mention' not in kwargs: 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, @@ -417,9 +394,10 @@ class MockContext(CustomMockMixin, unittest.mock.MagicMock): Instances of this class will follow the specifications of `discord.ext.commands.Context` instances. For more information, see the `MockGuild` docstring. """ + spec_set = context_instance def __init__(self, **kwargs) -> None: - super().__init__(spec_set=context_instance, **kwargs) + super().__init__(**kwargs) self.bot = kwargs.get('bot', MockBot()) self.guild = kwargs.get('guild', MockGuild()) self.author = kwargs.get('author', MockMember()) @@ -436,8 +414,7 @@ class MockAttachment(CustomMockMixin, unittest.mock.MagicMock): Instances of this class will follow the specifications of `discord.Attachment` instances. For more information, see the `MockGuild` docstring. """ - def __init__(self, **kwargs) -> None: - super().__init__(spec_set=attachment_instance, **kwargs) + spec_set = attachment_instance class MockMessage(CustomMockMixin, unittest.mock.MagicMock): @@ -447,10 +424,11 @@ class MockMessage(CustomMockMixin, unittest.mock.MagicMock): Instances of this class will follow the specifications of `discord.Message` instances. For more information, see the `MockGuild` docstring. """ + spec_set = message_instance def __init__(self, **kwargs) -> None: default_kwargs = {'attachments': []} - super().__init__(spec_set=message_instance, **collections.ChainMap(kwargs, default_kwargs)) + super().__init__(**collections.ChainMap(kwargs, default_kwargs)) self.author = kwargs.get('author', MockMember()) self.channel = kwargs.get('channel', MockTextChannel()) @@ -466,9 +444,10 @@ class MockEmoji(CustomMockMixin, unittest.mock.MagicMock): Instances of this class will follow the specifications of `discord.Emoji` instances. For more information, see the `MockGuild` docstring. """ + spec_set = emoji_instance def __init__(self, **kwargs) -> None: - super().__init__(spec_set=emoji_instance, **kwargs) + super().__init__(**kwargs) self.guild = kwargs.get('guild', MockGuild()) @@ -482,9 +461,7 @@ class MockPartialEmoji(CustomMockMixin, unittest.mock.MagicMock): Instances of this class will follow the specifications of `discord.PartialEmoji` instances. For more information, see the `MockGuild` docstring. """ - - def __init__(self, **kwargs) -> None: - super().__init__(spec_set=partial_emoji_instance, **kwargs) + spec_set = partial_emoji_instance reaction_instance = discord.Reaction(message=MockMessage(), data={'me': True}, emoji=MockEmoji()) @@ -497,12 +474,19 @@ class MockReaction(CustomMockMixin, unittest.mock.MagicMock): Instances of this class will follow the specifications of `discord.Reaction` instances. For more information, see the `MockGuild` docstring. """ + spec_set = reaction_instance def __init__(self, **kwargs) -> None: - super().__init__(spec_set=reaction_instance, **kwargs) + _users = kwargs.pop("users", []) + super().__init__(**kwargs) self.emoji = kwargs.get('emoji', MockEmoji()) self.message = kwargs.get('message', MockMessage()) - self.users = AsyncIteratorMock(kwargs.get('users', [])) + + user_iterator = unittest.mock.AsyncMock() + user_iterator.__aiter__.return_value = _users + self.users.return_value = user_iterator + + self.__str__.return_value = str(self.emoji) webhook_instance = discord.Webhook(data=unittest.mock.MagicMock(), adapter=unittest.mock.MagicMock()) @@ -515,13 +499,5 @@ class MockAsyncWebhook(CustomMockMixin, unittest.mock.MagicMock): Instances of this class will follow the specifications of `discord.Webhook` instances. For more information, see the `MockGuild` docstring. """ - - def __init__(self, **kwargs) -> None: - super().__init__(spec_set=webhook_instance, **kwargs) - - # Because Webhooks can also use a synchronous "WebhookAdapter", the methods are not defined - # as coroutines. That's why we need to set the methods manually. - self.send = AsyncMock() - self.edit = AsyncMock() - self.delete = AsyncMock() - self.execute = AsyncMock() + spec_set = webhook_instance + additional_spec_asyncs = ("send", "edit", "delete", "execute") diff --git a/tests/test_base.py b/tests/test_base.py index a16e2af8f..a7db4bf3e 100644 --- a/tests/test_base.py +++ b/tests/test_base.py @@ -3,7 +3,11 @@ import unittest import unittest.mock -from tests.base import LoggingTestCase, _CaptureLogHandler +from tests.base import LoggingTestsMixin, _CaptureLogHandler + + +class LoggingTestCase(LoggingTestsMixin, unittest.TestCase): + pass class LoggingTestCaseTests(unittest.TestCase): @@ -18,24 +22,14 @@ class LoggingTestCaseTests(unittest.TestCase): try: with LoggingTestCase.assertNotLogs(self, level=logging.DEBUG): pass - except AssertionError: + except AssertionError: # pragma: no cover self.fail("`self.assertNotLogs` raised an AssertionError when it should not!") - @unittest.mock.patch("tests.base.LoggingTestCase.assertNotLogs") - def test_the_test_function_assert_not_logs_does_not_raise_with_no_logs(self, assertNotLogs): - """Test if test_assert_not_logs_does_not_raise_with_no_logs captures exception correctly.""" - assertNotLogs.return_value = iter([None]) - assertNotLogs.side_effect = AssertionError - - message = "`self.assertNotLogs` raised an AssertionError when it should not!" - with self.assertRaises(AssertionError, msg=message): - self.test_assert_not_logs_does_not_raise_with_no_logs() - def test_assert_not_logs_raises_correct_assertion_error_when_logs_are_emitted(self): """Test if LoggingTestCase.assertNotLogs raises AssertionError when logs were emitted.""" msg_regex = ( r"1 logs of DEBUG or higher were triggered on root:\n" - r'<LogRecord: tests\.test_base, [\d]+, .+/tests/test_base\.py, [\d]+, "Log!">' + r'<LogRecord: tests\.test_base, [\d]+, .+[/\\]tests[/\\]test_base\.py, [\d]+, "Log!">' ) with self.assertRaisesRegex(AssertionError, msg_regex): with LoggingTestCase.assertNotLogs(self, level=logging.DEBUG): diff --git a/tests/test_helpers.py b/tests/test_helpers.py index 7894e104a..81285e009 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -1,5 +1,4 @@ import asyncio -import inspect import unittest import unittest.mock @@ -214,6 +213,11 @@ class DiscordMocksTests(unittest.TestCase): with self.assertRaises(RuntimeError, msg="cannot reuse already awaited coroutine"): asyncio.run(coroutine_object) + def test_user_mock_uses_explicitly_passed_mention_attribute(self): + """MockUser should use an explicitly passed value for user.mention.""" + user = helpers.MockUser(mention="hello") + self.assertEqual(user.mention, "hello") + class MockObjectTests(unittest.TestCase): """Tests the mock objects and mixins we've defined.""" @@ -341,65 +345,10 @@ class MockObjectTests(unittest.TestCase): attribute = getattr(mock, valid_attribute) self.assertTrue(isinstance(attribute, mock_type.child_mock_type)) - def test_extract_coroutine_methods_from_spec_instance_should_extract_all_and_only_coroutines(self): - """Test if all coroutine functions are extracted, but not regular methods or attributes.""" - class CoroutineDonor: - def __init__(self): - self.some_attribute = 'alpha' - - async def first_coroutine(): - """This coroutine function should be extracted.""" - - async def second_coroutine(): - """This coroutine function should be extracted.""" - - def regular_method(): - """This regular function should not be extracted.""" - - class Receiver: + def test_custom_mock_mixin_mocks_async_magic_methods_with_async_mock(self): + """The CustomMockMixin should mock async magic methods with an AsyncMock.""" + class MyMock(helpers.CustomMockMixin, unittest.mock.MagicMock): pass - donor = CoroutineDonor() - receiver = Receiver() - - helpers.CustomMockMixin._extract_coroutine_methods_from_spec_instance(receiver, donor) - - self.assertIsInstance(receiver.first_coroutine, helpers.AsyncMock) - self.assertIsInstance(receiver.second_coroutine, helpers.AsyncMock) - self.assertFalse(hasattr(receiver, 'regular_method')) - self.assertFalse(hasattr(receiver, 'some_attribute')) - - @unittest.mock.patch("builtins.super", new=unittest.mock.MagicMock()) - @unittest.mock.patch("tests.helpers.CustomMockMixin._extract_coroutine_methods_from_spec_instance") - def test_custom_mock_mixin_init_with_spec(self, extract_method_mock): - """Test if CustomMockMixin correctly passes on spec/kwargs and calls the extraction method.""" - spec_set = "pydis" - - helpers.CustomMockMixin(spec_set=spec_set) - - extract_method_mock.assert_called_once_with(spec_set) - - @unittest.mock.patch("builtins.super", new=unittest.mock.MagicMock()) - @unittest.mock.patch("tests.helpers.CustomMockMixin._extract_coroutine_methods_from_spec_instance") - def test_custom_mock_mixin_init_without_spec(self, extract_method_mock): - """Test if CustomMockMixin correctly passes on spec/kwargs and calls the extraction method.""" - helpers.CustomMockMixin() - - extract_method_mock.assert_not_called() - - def test_async_mock_provides_coroutine_for_dunder_call(self): - """Test if AsyncMock objects have a coroutine for their __call__ method.""" - async_mock = helpers.AsyncMock() - self.assertTrue(inspect.iscoroutinefunction(async_mock.__call__)) - - coroutine = async_mock() - self.assertTrue(inspect.iscoroutine(coroutine)) - self.assertIsNotNone(asyncio.run(coroutine)) - - def test_async_test_decorator_allows_synchronous_call_to_async_def(self): - """Test if the `async_test` decorator allows an `async def` to be called synchronously.""" - @helpers.async_test - async def kosayoda(): - return "return value" - - self.assertEqual(kosayoda(), "return value") + mock = MyMock() + self.assertIsInstance(mock.__aenter__, unittest.mock.AsyncMock) diff --git a/tests/utils/test_time.py b/tests/utils/test_time.py deleted file mode 100644 index 4baa6395c..000000000 --- a/tests/utils/test_time.py +++ /dev/null @@ -1,62 +0,0 @@ -import asyncio -from datetime import datetime, timezone -from unittest.mock import patch - -import pytest -from dateutil.relativedelta import relativedelta - -from bot.utils import time -from tests.helpers import AsyncMock - - - ('delta', 'precision', 'max_units', 'expected'), - ( - (relativedelta(days=2), 'seconds', 1, '2 days'), - (relativedelta(days=2, hours=2), 'seconds', 2, '2 days and 2 hours'), - (relativedelta(days=2, hours=2), 'seconds', 1, '2 days'), - (relativedelta(days=2, hours=2), 'days', 2, '2 days'), - - # Does not abort for unknown units, as the unit name is checked - # against the attribute of the relativedelta instance. - (relativedelta(days=2, hours=2), 'elephants', 2, '2 days and 2 hours'), - - # Very high maximum units, but it only ever iterates over - # each value the relativedelta might have. - (relativedelta(days=2, hours=2), 'hours', 20, '2 days and 2 hours'), - ) -) -def test_humanize_delta( - delta: relativedelta, - precision: str, - max_units: int, - expected: str -): - assert time.humanize_delta(delta, precision, max_units) == expected - - [email protected]('max_units', (-1, 0)) -def test_humanize_delta_raises_for_invalid_max_units(max_units: int): - with pytest.raises(ValueError, match='max_units must be positive'): - time.humanize_delta(relativedelta(days=2, hours=2), 'hours', max_units) - - - ('stamp', 'expected'), - ( - ('Sun, 15 Sep 2019 12:00:00 GMT', datetime(2019, 9, 15, 12, 0, 0, tzinfo=timezone.utc)), - ) -) -def test_parse_rfc1123(stamp: str, expected: str): - assert time.parse_rfc1123(stamp) == expected - - -@patch('asyncio.sleep', new_callable=AsyncMock) -def test_wait_until(sleep_patch): - start = datetime(2019, 1, 1, 0, 0) - then = datetime(2019, 1, 1, 0, 10) - - # No return value - assert asyncio.run(time.wait_until(then, start)) is None - - sleep_patch.assert_called_once_with(10 * 60) |