import unittest from unittest.mock import Mock, patch import nio from diskcache import Cache import matrix_alertbot.callback from matrix_alertbot.alertmanager import AlertmanagerClient from matrix_alertbot.callback import Callbacks from matrix_alertbot.command import BaseCommand from tests.utils import make_awaitable class CallbacksTestCase(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: # Create a Callbacks object and give it some Mock'd objects to use self.fake_client = Mock(spec=nio.AsyncClient) self.fake_client.user = "@fake_user:example.com" self.fake_cache = Mock(spec=Cache) self.fake_alertmanager = Mock(spec=AlertmanagerClient) # Create a fake room to play with self.fake_room = Mock(spec=nio.MatrixRoom) self.fake_room.room_id = "!abcdefg:example.com" self.fake_room.display_name = "Fake Room" # We don't spec config, as it doesn't currently have well defined attributes self.fake_config = Mock() self.fake_config.room_id = self.fake_room.room_id self.fake_config.command_prefix = "!alert " self.callbacks = Callbacks( self.fake_client, self.fake_cache, self.fake_alertmanager, self.fake_config ) async def test_invite(self) -> None: """Tests the callback for InviteMemberEvents""" # Tests that the bot attempts to join a room after being invited to it fake_invite_event = Mock(spec=nio.InviteMemberEvent) fake_invite_event.sender = "@some_other_fake_user:example.com" # Pretend that attempting to join a room is always successful self.fake_client.join.return_value = make_awaitable(None) # Pretend that we received an invite event await self.callbacks.invite(self.fake_room, fake_invite_event) # Check that we attempted to join the room self.fake_client.join.assert_called_once_with(self.fake_room.room_id) @patch.object(matrix_alertbot.callback.CommandFactory, "create", autospec=True) async def test_message_without_prefix(self, fake_command_create: Mock) -> None: """Tests the callback for RoomMessageText without any command prefix""" # Tests that the bot process messages in the room fake_message_event = Mock(spec=nio.RoomMessageText) fake_message_event.sender = "@some_other_fake_user:example.com" fake_message_event.body = "Hello world!" # Pretend that we received a text message event await self.callbacks.message(self.fake_room, fake_message_event) # Check that the command was not executed fake_command_create.assert_not_called() @patch.object(matrix_alertbot.callback.CommandFactory, "create", autospec=True) async def test_message_not_in_reply_with_prefix( self, fake_command_create: Mock ) -> None: """Tests the callback for RoomMessageText with the command prefix""" # Tests that the bot process messages in the room that contain a command fake_command = Mock(spec=BaseCommand) fake_command_create.return_value = fake_command fake_message_event = Mock(spec=nio.RoomMessageText) fake_message_event.event_id = "some event id" fake_message_event.sender = "@some_other_fake_user:example.com" fake_message_event.body = "!alert help" fake_message_event.source = {"content": {}} # Pretend that we received a text message event await self.callbacks.message(self.fake_room, fake_message_event) # Check that the command was not executed fake_command_create.assert_called_with( "help", self.fake_client, self.fake_alertmanager, self.fake_cache, self.fake_config, self.fake_room, fake_message_event.sender, fake_message_event.event_id, None, ) fake_command.process.assert_called_once() @patch.object(matrix_alertbot.callback.CommandFactory, "create", autospec=True) async def test_message_in_reply_with_prefix( self, fake_command_create: Mock ) -> None: """Tests the callback for RoomMessageText with the command prefix""" # Tests that the bot process messages in the room that contain a command fake_command = Mock(spec=BaseCommand) fake_command_create.return_value = fake_command fake_message_event = Mock(spec=nio.RoomMessageText) fake_message_event.event_id = "some event id" fake_message_event.sender = "@some_other_fake_user:example.com" fake_message_event.body = "!alert help" fake_message_event.source = { "content": { "m.relates_to": {"m.in_reply_to": {"event_id": "some alert event id"}} } } # Pretend that we received a text message event await self.callbacks.message(self.fake_room, fake_message_event) # Check that we attempted to execute the command fake_command_create.assert_called_once_with( "help", self.fake_client, self.fake_alertmanager, self.fake_cache, self.fake_config, self.fake_room, fake_message_event.sender, fake_message_event.event_id, "some alert event id", ) fake_command.process.assert_called_once() if __name__ == "__main__": unittest.main()