matrix-alertbot/tests/test_callback.py

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()