332 lines
12 KiB
Python
332 lines
12 KiB
Python
|
from typing import Dict
|
||
|
|
import unittest
|
||
|
|
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.callback import Callbacks
|
||
|
|
from matrix_alertbot.command import Command
|
||
|
|
|
||
|
|
from tests.utils import make_awaitable
|
||
|
|
|
||
|
|
|
||
|
|
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 = Mock(return_value=["fingerprint1", "fingerprint2"])
|
||
|
|
self.fake_cache = MagicMock(spec=Cache)
|
||
|
|
self.fake_cache.__getitem__ = self.fake_fingerprints
|
||
|
|
|
||
|
|
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"
|
||
|
|
self.fake_room.user_name.side_effect = lambda x: x
|
||
|
|
|
||
|
|
self.fake_source_not_in_reply = {"content": {}}
|
||
|
|
self.fake_source_in_reply = {
|
||
|
|
"content": {
|
||
|
|
"m.relates_to": {"m.in_reply_to": {"event_id": "some event id"}}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
# 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
|
||
|
|
|
||
|
|
fake_message_event = Mock(spec=nio.RoomMessageText)
|
||
|
|
|
||
|
|
command = Command(
|
||
|
|
self.fake_client,
|
||
|
|
self.fake_cache,
|
||
|
|
self.fake_alertmanager,
|
||
|
|
self.fake_config,
|
||
|
|
"ack",
|
||
|
|
self.fake_room,
|
||
|
|
fake_message_event,
|
||
|
|
)
|
||
|
|
await command.process()
|
||
|
|
|
||
|
|
@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
|
||
|
|
|
||
|
|
fake_message_event = Mock(spec=nio.RoomMessageText)
|
||
|
|
|
||
|
|
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,
|
||
|
|
fake_message_event,
|
||
|
|
)
|
||
|
|
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
|
||
|
|
|
||
|
|
fake_message_event = Mock(spec=nio.RoomMessageText)
|
||
|
|
|
||
|
|
command = Command(
|
||
|
|
self.fake_client,
|
||
|
|
self.fake_cache,
|
||
|
|
self.fake_alertmanager,
|
||
|
|
self.fake_config,
|
||
|
|
"help",
|
||
|
|
self.fake_room,
|
||
|
|
fake_message_event,
|
||
|
|
)
|
||
|
|
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
|
||
|
|
|
||
|
|
fake_message_event = Mock(spec=nio.RoomMessageText)
|
||
|
|
|
||
|
|
command = Command(
|
||
|
|
self.fake_client,
|
||
|
|
self.fake_cache,
|
||
|
|
self.fake_alertmanager,
|
||
|
|
self.fake_config,
|
||
|
|
"",
|
||
|
|
self.fake_room,
|
||
|
|
fake_message_event,
|
||
|
|
)
|
||
|
|
await command.process()
|
||
|
|
|
||
|
|
# Check that we attempted to process the command
|
||
|
|
fake_unknown.assert_called_once()
|
||
|
|
|
||
|
|
async def test_ack_not_in_reply_without_duration(self) -> None:
|
||
|
|
"""Tests the callback for InviteMemberEvents"""
|
||
|
|
# Tests that the bot attempts to join a room after being invited to it
|
||
|
|
|
||
|
|
fake_message_event = Mock(spec=nio.RoomMessageText)
|
||
|
|
fake_message_event.sender = "@some_other_fake_user:example.com"
|
||
|
|
fake_message_event.body = ""
|
||
|
|
fake_message_event.source = self.fake_source_not_in_reply
|
||
|
|
|
||
|
|
command = Command(
|
||
|
|
self.fake_client,
|
||
|
|
self.fake_cache,
|
||
|
|
self.fake_alertmanager,
|
||
|
|
self.fake_config,
|
||
|
|
"ack",
|
||
|
|
self.fake_room,
|
||
|
|
fake_message_event,
|
||
|
|
)
|
||
|
|
await command._ack()
|
||
|
|
|
||
|
|
# Check that we didn't attempt to create silences
|
||
|
|
self.fake_alertmanager.create_silence.assert_not_called()
|
||
|
|
self.fake_client.room_send.assert_not_called()
|
||
|
|
|
||
|
|
async def test_ack_not_in_reply_with_duration(self) -> None:
|
||
|
|
"""Tests the callback for InviteMemberEvents"""
|
||
|
|
# Tests that the bot attempts to join a room after being invited to it
|
||
|
|
|
||
|
|
fake_message_event = Mock(spec=nio.RoomMessageText)
|
||
|
|
fake_message_event.sender = "@some_other_fake_user:example.com"
|
||
|
|
fake_message_event.body = ""
|
||
|
|
fake_message_event.source = self.fake_source_not_in_reply
|
||
|
|
|
||
|
|
command = Command(
|
||
|
|
self.fake_client,
|
||
|
|
self.fake_cache,
|
||
|
|
self.fake_alertmanager,
|
||
|
|
self.fake_config,
|
||
|
|
"ack 2d",
|
||
|
|
self.fake_room,
|
||
|
|
fake_message_event,
|
||
|
|
)
|
||
|
|
await command._ack()
|
||
|
|
|
||
|
|
# Check that we didn't attempt to create silences
|
||
|
|
self.fake_alertmanager.create_silence.assert_not_called()
|
||
|
|
self.fake_client.room_send.assert_not_called()
|
||
|
|
|
||
|
|
@patch.object(matrix_alertbot.command, "send_text_to_room")
|
||
|
|
async def test_ack_in_reply_without_duration(
|
||
|
|
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
|
||
|
|
|
||
|
|
fake_message_event = Mock(spec=nio.RoomMessageText)
|
||
|
|
fake_message_event.sender = "@some_other_fake_user:example.com"
|
||
|
|
fake_message_event.body = ""
|
||
|
|
fake_message_event.source = self.fake_source_in_reply
|
||
|
|
|
||
|
|
command = Command(
|
||
|
|
self.fake_client,
|
||
|
|
self.fake_cache,
|
||
|
|
self.fake_alertmanager,
|
||
|
|
self.fake_config,
|
||
|
|
"ack",
|
||
|
|
self.fake_room,
|
||
|
|
fake_message_event,
|
||
|
|
)
|
||
|
|
await command._ack()
|
||
|
|
|
||
|
|
# Check that we attempted to create silences
|
||
|
|
self.fake_alertmanager.create_silence.assert_has_calls(
|
||
|
|
list(
|
||
|
|
call(
|
||
|
|
fingerprint,
|
||
|
|
"1d",
|
||
|
|
fake_message_event.sender,
|
||
|
|
)
|
||
|
|
for fingerprint in self.fake_fingerprints.return_value
|
||
|
|
)
|
||
|
|
)
|
||
|
|
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(
|
||
|
|
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
|
||
|
|
|
||
|
|
fake_message_event = Mock(spec=nio.RoomMessageText)
|
||
|
|
fake_message_event.sender = "@some_other_fake_user:example.com"
|
||
|
|
fake_message_event.body = ""
|
||
|
|
fake_message_event.source = self.fake_source_in_reply
|
||
|
|
|
||
|
|
command = Command(
|
||
|
|
self.fake_client,
|
||
|
|
self.fake_cache,
|
||
|
|
self.fake_alertmanager,
|
||
|
|
self.fake_config,
|
||
|
|
"ack 2d",
|
||
|
|
self.fake_room,
|
||
|
|
fake_message_event,
|
||
|
|
)
|
||
|
|
await command._ack()
|
||
|
|
|
||
|
|
# Check that we attempted to create silences
|
||
|
|
self.fake_alertmanager.create_silence.assert_has_calls(
|
||
|
|
list(
|
||
|
|
call(
|
||
|
|
fingerprint,
|
||
|
|
"2d",
|
||
|
|
fake_message_event.sender,
|
||
|
|
)
|
||
|
|
for fingerprint in self.fake_fingerprints.return_value
|
||
|
|
)
|
||
|
|
)
|
||
|
|
fake_send_text_to_room.assert_called_once_with(
|
||
|
|
self.fake_client,
|
||
|
|
self.fake_room.room_id,
|
||
|
|
"Created 2 silences with a duration of 2d.",
|
||
|
|
)
|
||
|
|
|
||
|
|
@patch.object(matrix_alertbot.command, "send_text_to_room")
|
||
|
|
async def test_unack_in_reply(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
|
||
|
|
|
||
|
|
fake_message_event = Mock(spec=nio.RoomMessageText)
|
||
|
|
fake_message_event.sender = "@some_other_fake_user:example.com"
|
||
|
|
fake_message_event.body = ""
|
||
|
|
fake_message_event.source = self.fake_source_in_reply
|
||
|
|
|
||
|
|
command = Command(
|
||
|
|
self.fake_client,
|
||
|
|
self.fake_cache,
|
||
|
|
self.fake_alertmanager,
|
||
|
|
self.fake_config,
|
||
|
|
"unack",
|
||
|
|
self.fake_room,
|
||
|
|
fake_message_event,
|
||
|
|
)
|
||
|
|
await command._unack()
|
||
|
|
|
||
|
|
# Check that we attempted to create silences
|
||
|
|
self.fake_alertmanager.delete_silence.assert_has_calls(
|
||
|
|
list(
|
||
|
|
call(fingerprint) for fingerprint in self.fake_fingerprints.return_value
|
||
|
|
)
|
||
|
|
)
|
||
|
|
fake_send_text_to_room.assert_called_with(
|
||
|
|
self.fake_client, self.fake_room.room_id, "Removed 2 silences."
|
||
|
|
)
|
||
|
|
|
||
|
|
# @patch.object(matrix_alertbot.callback, "Command", autospec=True)
|
||
|
|
# async def test_message_without_prefix(self, fake_command: 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.assert_not_called()
|
||
|
|
|
||
|
|
# @patch.object(matrix_alertbot.callback, "Command", autospec=True)
|
||
|
|
# async def test_message_with_prefix(self, fake_command: 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_instance = fake_command.return_value
|
||
|
|
# fake_command_instance.process.side_effect = lambda: print("hello")
|
||
|
|
|
||
|
|
# fake_message_event = Mock(spec=nio.RoomMessageText)
|
||
|
|
# fake_message_event.sender = "@some_other_fake_user:example.com"
|
||
|
|
# fake_message_event.body = "!alert help"
|
||
|
|
|
||
|
|
# # 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.assert_called_once_with(
|
||
|
|
# self.fake_client,
|
||
|
|
# self.fake_alertmanager,
|
||
|
|
# self.fake_cache,
|
||
|
|
# self.fake_config,
|
||
|
|
# "help",
|
||
|
|
# self.fake_room,
|
||
|
|
# fake_message_event,
|
||
|
|
# )
|
||
|
|
# fake_command_instance.process.assert_called_once()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|