from __future__ import annotations import json from sys import implementation import unittest from typing import Any from unittest.mock import MagicMock, Mock, patch import aiotools import aiohttp from aiohttp import web, web_request import aiohttp.test_utils import nio from diskcache import Cache from matrix_alertbot.alertmanager import AlertmanagerClient from matrix_alertbot.errors import ( AlertNotFoundError, AlertmanagerError, SilenceNotFoundError, ) class AbstractFakeAlertmanagerServer: def __init__(self) -> None: self.app = web.Application() self.app.router.add_routes( [ web.get("/api/v2/alerts", self.get_alerts), web.post("/api/v2/silences", self.create_silence), web.delete("/api/v2/silence/{silence}", self.delete_silence), ] ) self.runner = web.AppRunner(self.app) async def __aenter__(self) -> AbstractFakeAlertmanagerServer: await self.start() return self async def __aexit__(self, *args: Any) -> None: await self.stop() async def start(self) -> None: self.port = aiohttp.test_utils.unused_port() await self.runner.setup() site = web.TCPSite(self.runner, "localhost", self.port) await site.start() async def stop(self) -> None: await self.runner.cleanup() async def get_alerts(self, request: web_request.Request) -> web.Response: raise NotImplementedError async def create_silence(self, request: web_request.Request) -> web.Response: raise NotImplementedError async def delete_silence(self, request: web_request.Request) -> web.Response: raise NotImplementedError class FakeAlertmanagerServer(AbstractFakeAlertmanagerServer): async def get_alerts(self, request: web_request.Request) -> web.Response: return web.Response( body=json.dumps( [ { "fingerprint": "fingerprint1", "labels": {"alertname": "alert1"}, "status": {"state": "active"}, }, { "fingerprint": "fingerprint2", "labels": {"alertname": "alert2"}, "status": { "state": "suppressed", "silencedBy": ["silence1", "silence2"], }, }, ] ), content_type="application/json", ) async def create_silence(self, request: web_request.Request) -> web.Response: return web.Response( body=json.dumps({"silenceID": "silence1"}), content_type="application/json" ) async def delete_silence(self, request: web_request.Request) -> web.Response: return web.Response(status=200, content_type="application/json") class FakeAlertmanagerServerWithoutAlert(AbstractFakeAlertmanagerServer): async def get_alerts(self, request: web_request.Request) -> web.Response: return web.Response(body=json.dumps([]), content_type="application/json") class FakeAlertmanagerServerWithErrorAlerts(AbstractFakeAlertmanagerServer): async def get_alerts(self, request: web_request.Request) -> web.Response: return web.Response(status=500) class FakeAlertmanagerServerWithErrorCreateSilence(FakeAlertmanagerServer): async def create_silence(self, request: web_request.Request) -> web.Response: return web.Response(status=500) class FakeAlertmanagerServerWithErrorDeleteSilence(FakeAlertmanagerServer): async def delete_silence(self, request: web_request.Request) -> web.Response: return web.Response(status=500) class AlertmanagerClientTestCase(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: self.fake_fingerprints = Mock(return_value=["fingerprint1", "fingerprint2"]) self.fake_cache = MagicMock(spec=Cache) self.fake_cache.__getitem__ = self.fake_fingerprints async def test_get_alerts_happy(self) -> None: async with FakeAlertmanagerServer() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager = AlertmanagerClient( f"http://localhost:{port}", self.fake_cache ) async with aiotools.closing_async(alertmanager) as alertmanager: alerts = await alertmanager.get_alerts() self.assertEqual( [ { "fingerprint": "fingerprint1", "labels": {"alertname": "alert1"}, "status": {"state": "active"}, }, { "fingerprint": "fingerprint2", "labels": {"alertname": "alert2"}, "status": { "state": "suppressed", "silencedBy": ["silence1", "silence2"], }, }, ], alerts, ) async def test_get_alerts_empty(self) -> None: async with FakeAlertmanagerServerWithoutAlert() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager = AlertmanagerClient( f"http://localhost:{port}", self.fake_cache ) async with aiotools.closing_async(alertmanager) as alertmanager: alerts = await alertmanager.get_alerts() self.assertEqual([], alerts) async def test_get_alerts_raise_alertmanager_error(self) -> None: async with FakeAlertmanagerServerWithErrorAlerts() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager = AlertmanagerClient( f"http://localhost:{port}", self.fake_cache ) async with aiotools.closing_async(alertmanager) as alertmanager: with self.assertRaises(AlertmanagerError): await alertmanager.get_alerts() async def test_get_alert_happy(self) -> None: async with FakeAlertmanagerServer() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager = AlertmanagerClient( f"http://localhost:{port}", self.fake_cache ) async with aiotools.closing_async(alertmanager) as alertmanager: alert = await alertmanager.get_alert("fingerprint1") self.assertEqual( { "fingerprint": "fingerprint1", "labels": {"alertname": "alert1"}, "status": {"state": "active"}, }, alert, ) async def test_get_alert_raise_alert_not_found(self) -> None: async with FakeAlertmanagerServerWithoutAlert() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager = AlertmanagerClient( f"http://localhost:{port}", self.fake_cache ) async with aiotools.closing_async(alertmanager) as alertmanager: with self.assertRaises(AlertNotFoundError): await alertmanager.get_alert("fingerprint1") async def test_get_alert_raise_alertmanager_error(self) -> None: async with FakeAlertmanagerServerWithErrorAlerts() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager = AlertmanagerClient( f"http://localhost:{port}", self.fake_cache ) async with aiotools.closing_async(alertmanager) as alertmanager: with self.assertRaises(AlertmanagerError): await alertmanager.get_alert("fingerprint1") async def test_create_silence_happy(self) -> None: async with FakeAlertmanagerServer() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager = AlertmanagerClient( f"http://localhost:{port}", self.fake_cache ) async with aiotools.closing_async(alertmanager) as alertmanager: silence = await alertmanager.create_silence( "fingerprint1", "1d", "user" ) self.assertEqual("silence1", silence) async def test_create_silence_raise_alert_not_found(self) -> None: async with FakeAlertmanagerServerWithoutAlert() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager = AlertmanagerClient( f"http://localhost:{port}", self.fake_cache ) async with aiotools.closing_async(alertmanager) as alertmanager: with self.assertRaises(AlertNotFoundError): await alertmanager.create_silence("fingerprint1", "1d", "user") async def test_create_silence_raise_alertmanager_error(self) -> None: async with FakeAlertmanagerServerWithErrorCreateSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager = AlertmanagerClient( f"http://localhost:{port}", self.fake_cache ) async with aiotools.closing_async(alertmanager) as alertmanager: await alertmanager.get_alert("fingerprint1") with self.assertRaises(AlertmanagerError): await alertmanager.create_silence("fingerprint1", "1d", "user") async def test_delete_silences_happy(self) -> None: async with FakeAlertmanagerServer() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager = AlertmanagerClient( f"http://localhost:{port}", self.fake_cache ) async with aiotools.closing_async(alertmanager) as alertmanager: silences = await alertmanager.delete_silences("fingerprint2") self.assertEqual(["silence1", "silence2"], silences) async def test_delete_silences_raise_silence_not_found(self) -> None: async with FakeAlertmanagerServer() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager = AlertmanagerClient( f"http://localhost:{port}", self.fake_cache ) async with aiotools.closing_async(alertmanager) as alertmanager: with self.assertRaises(SilenceNotFoundError): await alertmanager.delete_silences("fingerprint1") async def test_delete_silences_raise_alert_not_found(self) -> None: async with FakeAlertmanagerServerWithoutAlert() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager = AlertmanagerClient( f"http://localhost:{port}", self.fake_cache ) async with aiotools.closing_async(alertmanager) as alertmanager: with self.assertRaises(AlertNotFoundError): await alertmanager.delete_silences("fingerprint2") async def test_delete_silences_raise_alertmanager_error(self) -> None: async with FakeAlertmanagerServerWithErrorDeleteSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager = AlertmanagerClient( f"http://localhost:{port}", self.fake_cache ) async with aiotools.closing_async(alertmanager) as alertmanager: await alertmanager.get_alert("fingerprint1") with self.assertRaises(AlertmanagerError): await alertmanager.delete_silences("fingerprint2") async def test_find_alert_happy(self) -> None: alertmanager = AlertmanagerClient(f"http://localhost", self.fake_cache) alert = alertmanager._find_alert( "fingerprint1", [{"fingerprint": "fingerprint1"}] ) self.assertEqual({"fingerprint": "fingerprint1"}, alert) async def test_find_alert_raise_alert_not_found(self) -> None: alertmanager = AlertmanagerClient(f"http://localhost", self.fake_cache) with self.assertRaises(AlertNotFoundError): alertmanager._find_alert("fingerprint2", [{"fingerprint": "fingerprint1"}]) # fake_session_get.assert_called_once_with("http://localhost:9093/api/v2/alerts") # async def test_get_alerts_not_empty(self) -> None: # alerts = await self.alertmanager.get_alerts() # self.assertEqual(["alert1", "alert2"], alerts) # # fake_session_get.assert_called_once_with("http://localhost:9093/api/v2/alerts") # async def test_get_alerts_raise_alertmanager_error(self) -> None: # with self.assertRaises(AlertmanagerError): # await self.alertmanager.get_alerts() # # fake_session_get.assert_called_once_with("http://localhost:9093/api/v2/alerts") # @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." # ) if __name__ == "__main__": unittest.main()