matrix-alertbot/tests/test_command.py

import unittest
from typing import List
from unittest.mock import MagicMock, Mock, call, patch

import nio
from diskcache import Cache

import matrix_alertbot.callback
from matrix_alertbot.alertmanager import AlertmanagerClient
from matrix_alertbot.command import Command
from matrix_alertbot.errors import AlertmanagerError
from matrix_alertbot.matcher import AlertMatcher, AlertRegexMatcher

from tests.utils import make_awaitable


async def create_silence_raise_alertmanager_error(
    fingerprint: str, duration: str, user: str, matchers: List[AlertMatcher]
) -> str:
    if fingerprint == "fingerprint1":
        raise AlertmanagerError
    return "silence1"


async def delete_silence_raise_alertmanager_error(
    fingerprint: str, matchers: List[AlertMatcher]
) -> List[str]:
    if fingerprint == "fingerprint1":
        raise AlertmanagerError
    return ["silence1"]


class CommandTestCase(unittest.IsolatedAsyncioTestCase):
    def setUp(self) -> None:
        # Create a Command 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"
        # Pretend that attempting to send a message is always successful
        self.fake_client.room_send.return_value = make_awaitable(None)

        self.fake_fingerprints = ["fingerprint1", "fingerprint2"]
        self.fake_silences = ["silence1", "silence2"]

        self.fake_cache = MagicMock(spec=Cache)
        self.fake_cache.__getitem__.return_value = self.fake_fingerprints
        self.fake_cache.__contains__.return_value = True

        self.fake_alertmanager = Mock(spec=AlertmanagerClient)
        self.fake_alertmanager.delete_silences.return_value = self.fake_silences

        # 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"
        self.fake_room.user_name.side_effect = lambda x: x

        self.fake_event_id = "some event id"
        self.fake_sender = "@some_other_fake_user:example.com"

        # 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 "

    @patch.object(matrix_alertbot.command.Command, "_ack")
    async def test_process_ack_command(self, fake_ack: Mock) -> None:
        """Tests the callback for InviteMemberEvents"""
        # Tests that the bot attempts to join a room after being invited to it

        command = Command(
            self.fake_client,
            self.fake_cache,
            self.fake_alertmanager,
            self.fake_config,
            "ack",
            self.fake_room,
            self.fake_sender,
            self.fake_event_id,
        )
        await command.process()

        # Check that we attempted to process the command
        fake_ack.assert_called_once()

    @patch.object(matrix_alertbot.command.Command, "_unack")
    async def test_process_unack_command(self, fake_unack: Mock) -> None:
        """Tests the callback for InviteMemberEvents"""
        # Tests that the bot attempts to join a room after being invited to it

        for command_word in ("unack", "nack"):
            command = Command(
                self.fake_client,
                self.fake_cache,
                self.fake_alertmanager,
                self.fake_config,
                command_word,
                self.fake_room,
                self.fake_sender,
                self.fake_event_id,
            )
            await command.process()

        # Check that we attempted to process the command
        fake_unack.assert_has_calls([call(), call()])

    @patch.object(matrix_alertbot.command.Command, "_show_help")
    async def test_process_help_command(self, fake_help: Mock) -> None:
        """Tests the callback for InviteMemberEvents"""
        # Tests that the bot attempts to join a room after being invited to it

        command = Command(
            self.fake_client,
            self.fake_cache,
            self.fake_alertmanager,
            self.fake_config,
            "help",
            self.fake_room,
            self.fake_sender,
            self.fake_event_id,
        )
        await command.process()

        # Check that we attempted to process the command
        fake_help.assert_called_once()

    @patch.object(matrix_alertbot.command.Command, "_unknown_command")
    async def test_process_unknown_command(self, fake_unknown: Mock) -> None:
        """Tests the callback for InviteMemberEvents"""
        # Tests that the bot attempts to join a room after being invited to it

        command = Command(
            self.fake_client,
            self.fake_cache,
            self.fake_alertmanager,
            self.fake_config,
            "",
            self.fake_room,
            self.fake_sender,
            self.fake_event_id,
        )
        await command.process()

        # Check that we attempted to process the command
        fake_unknown.assert_called_once()

    @patch.object(matrix_alertbot.command, "send_text_to_room")
    async def test_ack_in_reply_without_duration_nor_matchers(
        self, fake_send_text_to_room: Mock
    ) -> None:
        """Tests the callback for InviteMemberEvents"""
        # Tests that the bot attempts to join a room after being invited to it

        command = Command(
            self.fake_client,
            self.fake_cache,
            self.fake_alertmanager,
            self.fake_config,
            "ack",
            self.fake_room,
            self.fake_sender,
            self.fake_event_id,
        )
        await command._ack()

        # Check that we attempted to create silences
        self.fake_alertmanager.create_silence.assert_has_calls(
            [
                call(fingerprint, "1d", self.fake_sender, [])
                for fingerprint in self.fake_fingerprints
            ]
        )
        fake_send_text_to_room.assert_called_once_with(
            self.fake_client,
            self.fake_room.room_id,
            "Created 2 silences with a duration of 1d.",
        )

    @patch.object(matrix_alertbot.command, "send_text_to_room")
    async def test_ack_in_reply_without_duration_and_with_matchers(
        self, fake_send_text_to_room: Mock
    ) -> None:
        """Tests the callback for InviteMemberEvents"""
        # Tests that the bot attempts to join a room after being invited to it
        matchers = [
            AlertMatcher(label="alertname", value="alert1"),
            AlertRegexMatcher(label="severity", regex="critical"),
        ]

        command = Command(
            self.fake_client,
            self.fake_cache,
            self.fake_alertmanager,
            self.fake_config,
            "ack alertname=alert1 severity=~critical",
            self.fake_room,
            self.fake_sender,
            self.fake_event_id,
        )
        await command._ack()

        # Check that we attempted to create silences
        self.fake_alertmanager.create_silence.assert_has_calls(
            [
                call(
                    fingerprint,
                    "1d",
                    self.fake_sender,
                    matchers,
                )
                for fingerprint in self.fake_fingerprints
            ]
        )
        fake_send_text_to_room.assert_called_once_with(
            self.fake_client,
            self.fake_room.room_id,
            "Created 2 silences with a duration of 1d.",
        )

    @patch.object(matrix_alertbot.command, "send_text_to_room")
    async def test_ack_in_reply_with_duration_and_without_matchers(
        self, fake_send_text_to_room: Mock
    ) -> None:
        """Tests the callback for InviteMemberEvents"""
        # Tests that the bot attempts to join a room after being invited to it

        command = Command(
            self.fake_client,
            self.fake_cache,
            self.fake_alertmanager,
            self.fake_config,
            "ack 1w 2d",
            self.fake_room,
            self.fake_sender,
            self.fake_event_id,
        )
        await command._ack()

        # Check that we attempted to create silences
        self.fake_alertmanager.create_silence.assert_has_calls(
            [
                call(fingerprint, "1w 2d", self.fake_sender, [])
                for fingerprint in self.fake_fingerprints
            ]
        )
        fake_send_text_to_room.assert_called_once_with(
            self.fake_client,
            self.fake_room.room_id,
            "Created 2 silences with a duration of 1w 2d.",
        )

    @patch.object(matrix_alertbot.command, "send_text_to_room")
    async def test_ack_in_reply_with_duration_and_matchers(
        self, fake_send_text_to_room: Mock
    ) -> None:
        """Tests the callback for InviteMemberEvents"""
        # Tests that the bot attempts to join a room after being invited to it
        matchers = [
            AlertMatcher(label="alertname", value="alert1"),
            AlertMatcher(label="severity", value="critical"),
        ]

        command = Command(
            self.fake_client,
            self.fake_cache,
            self.fake_alertmanager,
            self.fake_config,
            "ack 1w 2d alertname=alert1 severity=critical",
            self.fake_room,
            self.fake_sender,
            self.fake_event_id,
        )
        await command._ack()

        # Check that we attempted to create silences
        self.fake_alertmanager.create_silence.assert_has_calls(
            [
                call(
                    fingerprint,
                    "1w 2d",
                    self.fake_sender,
                    matchers,
                )
                for fingerprint in self.fake_fingerprints
            ]
        )
        fake_send_text_to_room.assert_called_once_with(
            self.fake_client,
            self.fake_room.room_id,
            "Created 2 silences with a duration of 1w 2d.",
        )

    @patch.object(matrix_alertbot.command, "send_text_to_room")
    async def test_ack_raise_alertmanager_error(
        self, fake_send_text_to_room: Mock
    ) -> None:
        """Tests the callback for InviteMemberEvents"""
        # Tests that the bot attempts to join a room after being invited to it

        command = Command(
            self.fake_client,
            self.fake_cache,
            self.fake_alertmanager,
            self.fake_config,
            "ack",
            self.fake_room,
            self.fake_sender,
            self.fake_event_id,
        )

        self.fake_alertmanager.create_silence.side_effect = (
            create_silence_raise_alertmanager_error
        )
        await command._ack()

        # Check that we attempted to create silences
        self.fake_alertmanager.create_silence.assert_has_calls(
            [
                call(fingerprint, "1d", self.fake_sender, [])
                for fingerprint in self.fake_fingerprints
            ]
        )
        fake_send_text_to_room.assert_called_once_with(
            self.fake_client,
            self.fake_room.room_id,
            "Created 1 silences with a duration of 1d.",
        )

    @patch.object(matrix_alertbot.command, "send_text_to_room")
    async def test_unack_in_reply_without_matchers(
        self, fake_send_text_to_room: Mock
    ) -> None:
        """Tests the callback for InviteMemberEvents"""
        # Tests that the bot attempts to join a room after being invited to it

        command = Command(
            self.fake_client,
            self.fake_cache,
            self.fake_alertmanager,
            self.fake_config,
            "unack",
            self.fake_room,
            self.fake_sender,
            self.fake_event_id,
        )
        await command._unack()

        # Check that we attempted to create silences
        self.fake_alertmanager.delete_silences.assert_has_calls(
            [call(fingerprint, []) for fingerprint in self.fake_fingerprints]
        )
        fake_send_text_to_room.assert_called_with(
            self.fake_client, self.fake_room.room_id, "Removed 4 silences."
        )

    @patch.object(matrix_alertbot.command, "send_text_to_room")
    async def test_unack_in_reply_with_matchers(
        self, fake_send_text_to_room: Mock
    ) -> None:
        """Tests the callback for InviteMemberEvents"""
        # Tests that the bot attempts to join a room after being invited to it

        matchers = [
            AlertMatcher(label="alertname", value="alert1"),
            AlertRegexMatcher(label="severity", regex="critical"),
        ]

        command = Command(
            self.fake_client,
            self.fake_cache,
            self.fake_alertmanager,
            self.fake_config,
            "unack alertname=alert1 severity=~critical",
            self.fake_room,
            self.fake_sender,
            self.fake_event_id,
        )
        await command._unack()

        # Check that we attempted to create silences
        self.fake_alertmanager.delete_silences.assert_has_calls(
            [call(fingerprint, matchers) for fingerprint in self.fake_fingerprints]
        )
        fake_send_text_to_room.assert_called_with(
            self.fake_client, self.fake_room.room_id, "Removed 4 silences."
        )

    @patch.object(matrix_alertbot.command, "send_text_to_room")
    async def test_unack_silence_raise_alertmanager_error(
        self, fake_send_text_to_room: Mock
    ) -> None:
        """Tests the callback for InviteMemberEvents"""
        # Tests that the bot attempts to join a room after being invited to it

        command = Command(
            self.fake_client,
            self.fake_cache,
            self.fake_alertmanager,
            self.fake_config,
            "unack",
            self.fake_room,
            self.fake_sender,
            self.fake_event_id,
        )

        self.fake_alertmanager.delete_silences.side_effect = (
            delete_silence_raise_alertmanager_error
        )
        await command._unack()

        # Check that we attempted to create silences
        self.fake_alertmanager.delete_silences.assert_has_calls(
            [call(fingerprint, []) for fingerprint in self.fake_fingerprints]
        )
        fake_send_text_to_room.assert_called_with(
            self.fake_client, self.fake_room.room_id, "Removed 1 silences."
        )


if __name__ == "__main__":
    unittest.main()