from __future__ import annotations import json import unittest from datetime import datetime, timedelta from typing import Any, Dict, Optional, Tuple from unittest.mock import Mock, patch import aiohttp import aiohttp.test_utils import aiotools from aiohttp import web, web_request from diskcache import Cache from freezegun import freeze_time import matrix_alertbot.alertmanager from matrix_alertbot.alertmanager import AlertmanagerClient from matrix_alertbot.errors import ( AlertmanagerServerError, AlertNotFoundError, SilenceExpiredError, SilenceExtendError, SilenceNotFoundError, ) async def update_silence_raise_silence_not_found( fingerprint: str, user: str, duration_seconds: int, *, force: bool = False ) -> str: raise SilenceNotFoundError class FakeCache: def __init__(self, cache_dict: Optional[Dict] = None) -> None: if cache_dict is None: cache_dict = {} self.cache = cache_dict def get( self, key: str, expire_time: bool = False ) -> Optional[Tuple[str, Optional[int]] | str]: return self.cache.get(key) def set(self, key: str, value: str, expire: int) -> None: self.cache[key] = value, expire class AbstractFakeAlertmanagerServer: def __init__(self) -> None: self.app = web.Application() self.app.router.add_routes( [ web.get("/api/v2/alerts", self.get_alerts), web.get("/api/v2/silences", self.get_silences), web.post("/api/v2/silences", self.create_silence), web.delete("/api/v2/silence/{silence}", self.delete_silence), ] ) self.app["silences"] = [ { "id": "silence1", "createdBy": "user1", "status": {"state": "active"}, "matchers": [], }, { "id": "silence2", "createdBy": "user2", "status": {"state": "expired"}, "matchers": [], }, ] 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, "127.0.0.1", 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 get_silences(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 get_silences(self, request: web_request.Request) -> web.Response: return web.Response( body=json.dumps(self.app["silences"]), content_type="application/json" ) async def create_silence(self, request: web_request.Request) -> web.Response: silences = self.app["silences"] silence = await request.json() silence_count = len(silences) + 1 silence["id"] = f"silence{silence_count}" silence["status"] = {"state": "active"} silences.append(silence) return web.Response( body=json.dumps({"silenceID": silence["id"]}), content_type="application/json", ) async def delete_silence(self, request: web_request.Request) -> web.Response: silence_id = request.match_info["silence"] for i, silence in enumerate(self.app["silences"]): if silence["id"] == silence_id: del self.app["silences"][i] break return web.Response(status=200, content_type="application/json") class FakeAlertmanagerServerWithoutAlert(FakeAlertmanagerServer): async def get_alerts(self, request: web_request.Request) -> web.Response: return web.Response(body=json.dumps([]), content_type="application/json") class FakeAlertmanagerServerWithErrorAlerts(FakeAlertmanagerServer): async def get_alerts(self, request: web_request.Request) -> web.Response: return web.Response(status=500) class FakeAlertmanagerServerWithoutSilence(FakeAlertmanagerServer): def __init__(self) -> None: super().__init__() self.app["silences"] = [] class FakeAlertmanagerServerWithErrorSilences(FakeAlertmanagerServer): async def get_silences(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"]) async def test_get_alerts_happy(self) -> None: fake_cache = FakeCache() async with FakeAlertmanagerServer() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): alerts = await alertmanager_client.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: fake_cache = FakeCache() async with FakeAlertmanagerServerWithoutAlert() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): alerts = await alertmanager_client.get_alerts() self.assertEqual([], alerts) async def test_get_alerts_raise_alertmanager_error(self) -> None: fake_cache = FakeCache() async with FakeAlertmanagerServerWithErrorAlerts() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): with self.assertRaises(AlertmanagerServerError): await alertmanager_client.get_alerts() async def test_get_silences_happy(self) -> None: fake_cache = FakeCache() async with FakeAlertmanagerServer() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): silences = await alertmanager_client.get_silences() self.assertEqual( [ { "id": "silence1", "createdBy": "user1", "status": {"state": "active"}, "matchers": [], }, { "id": "silence2", "createdBy": "user2", "status": {"state": "expired"}, "matchers": [], }, ], silences, ) async def test_get_silences_empty(self) -> None: fake_cache = FakeCache() async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): silences = await alertmanager_client.get_silences() self.assertEqual([], silences) async def test_get_silences_raise_alertmanager_error(self) -> None: fake_cache = FakeCache() async with FakeAlertmanagerServerWithErrorSilences() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): with self.assertRaises(AlertmanagerServerError): await alertmanager_client.get_silences() async def test_get_alert_happy(self) -> None: fake_cache = FakeCache() async with FakeAlertmanagerServer() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): alert = await alertmanager_client.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: fake_cache = FakeCache() async with FakeAlertmanagerServerWithoutAlert() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): with self.assertRaises(AlertNotFoundError): await alertmanager_client.get_alert("fingerprint1") async def test_get_alert_raise_alertmanager_error(self) -> None: fake_cache = FakeCache() async with FakeAlertmanagerServerWithErrorAlerts() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): with self.assertRaises(AlertmanagerServerError): await alertmanager_client.get_alert("fingerprint1") async def test_get_silence_happy(self) -> None: fake_cache = FakeCache() async with FakeAlertmanagerServer() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): silence1 = await alertmanager_client.get_silence("silence1") silence2 = await alertmanager_client.get_silence("silence2") self.assertEqual( { "id": "silence1", "createdBy": "user1", "status": {"state": "active"}, "matchers": [], }, silence1, ) self.assertEqual( { "id": "silence2", "createdBy": "user2", "status": {"state": "expired"}, "matchers": [], }, silence2, ) async def test_get_silence_raise_silence_not_found(self) -> None: fake_cache = FakeCache() async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): with self.assertRaises(SilenceNotFoundError): await alertmanager_client.get_silence("silence1") async def test_get_silence_raise_alertmanager_error(self) -> None: fake_cache = FakeCache() async with FakeAlertmanagerServerWithErrorSilences() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): with self.assertRaises(AlertmanagerServerError): await alertmanager_client.get_silence("silence1") @freeze_time(datetime.utcfromtimestamp(0)) async def test_create_silence_with_duration(self) -> None: fake_cache = Mock(return_value=FakeCache()) async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): silence_id = await alertmanager_client.create_silence( "fingerprint1", "user", 86400 ) silence = await alertmanager_client.get_silence("silence1") self.assertEqual("silence1", silence_id) self.assertEqual( { "id": "silence1", "status": {"state": "active"}, "matchers": [ { "name": "alertname", "value": "alert1", "isRegex": False, "isEqual": True, } ], "createdBy": "user", "startsAt": "1970-01-01T00:00:00", "endsAt": "1970-01-02T00:00:00", "comment": "Acknowledge alert from Matrix", }, silence, ) fake_cache.set.assert_called_once_with("fingerprint1", "silence1", expire=86400) @freeze_time(datetime.utcfromtimestamp(0)) async def test_update_silence_raise_extend_error(self) -> None: fake_cache = FakeCache() async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): await alertmanager_client.create_silence("fingerprint1", "user", 86400) with self.assertRaises(SilenceExtendError): await alertmanager_client.update_silence("fingerprint1") with self.assertRaises(SilenceNotFoundError): await alertmanager_client.get_silence("silence2") self.assertEqual({"fingerprint1": ("silence1", 86400)}, fake_cache.cache) @freeze_time(datetime.utcfromtimestamp(0)) async def test_update_silence_remove_duration(self) -> None: fake_cache = FakeCache() async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): silence_id1 = await alertmanager_client.create_silence( "fingerprint1", "user", 86400 ) silence_id2 = await alertmanager_client.update_silence( "fingerprint1", force=True ) silence2 = await alertmanager_client.get_silence("silence2") self.assertEqual("silence1", silence_id1) self.assertEqual("silence2", silence_id2) self.assertEqual( { "id": "silence2", "status": {"state": "active"}, "matchers": [ { "name": "alertname", "value": "alert1", "isRegex": False, "isEqual": True, } ], "createdBy": "user", "startsAt": "1970-01-01T00:00:00", "endsAt": "1970-01-01T03:00:00", "comment": "Acknowledge alert from Matrix", }, silence2, ) self.assertEqual({"fingerprint1": ("silence2", None)}, fake_cache.cache) @freeze_time(datetime.utcfromtimestamp(0)) async def test_update_silence_override_user_and_duration(self) -> None: fake_cache = FakeCache() async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): await alertmanager_client.create_silence("fingerprint1", "user1", 86400) silence_id2 = await alertmanager_client.update_silence( "fingerprint1", "user2", 864000 ) silence2 = await alertmanager_client.get_silence("silence2") self.assertEqual("silence2", silence_id2) self.assertEqual( { "id": "silence2", "status": {"state": "active"}, "matchers": [ { "name": "alertname", "value": "alert1", "isRegex": False, "isEqual": True, } ], "createdBy": "user2", "startsAt": "1970-01-01T00:00:00", "endsAt": "1970-01-11T00:00:00", "comment": "Acknowledge alert from Matrix", }, silence2, ) self.assertEqual({"fingerprint1": ("silence2", 864000)}, fake_cache.cache) @patch.object(matrix_alertbot.alertmanager.AlertmanagerClient, "update_silence") @patch.object(matrix_alertbot.alertmanager.AlertmanagerClient, "create_silence") async def test_create_or_update_silence_with_duration_and_silence_not_found( self, fake_create_silence: Mock, fake_update_silence: Mock ) -> None: fake_cache = Mock(spec=Cache) fake_update_silence.side_effect = update_silence_raise_silence_not_found fake_create_silence.return_value = "silence1" alertmanager_client = AlertmanagerClient("http://localhost", fake_cache) async with aiotools.closing_async(alertmanager_client): silence_id1 = await alertmanager_client.create_or_update_silence( "fingerprint1", "user", 86400 ) self.assertEqual("silence1", silence_id1) fake_update_silence.assert_called_once_with( "fingerprint1", "user", 86400, force=False ) fake_create_silence.assert_called_once_with("fingerprint1", "user", 86400) @patch.object(matrix_alertbot.alertmanager.AlertmanagerClient, "update_silence") @patch.object(matrix_alertbot.alertmanager.AlertmanagerClient, "create_silence") async def test_create_or_update_silence_with_duration_and_silence_found( self, fake_create_silence: Mock, fake_update_silence: Mock ) -> None: fake_cache = Mock(spec=Cache) fake_update_silence.return_value = "silence1" alertmanager_client = AlertmanagerClient("http://localhost", fake_cache) async with aiotools.closing_async(alertmanager_client): silence_id1 = await alertmanager_client.create_or_update_silence( "fingerprint1", "user", 86400 ) self.assertEqual("silence1", silence_id1) fake_update_silence.assert_called_once_with( "fingerprint1", "user", 86400, force=False ) fake_create_silence.assert_not_called() @freeze_time(datetime.utcfromtimestamp(0)) async def test_create_silence_without_duration(self) -> None: fake_cache = Mock(spec=Cache) fake_cache.get.return_value = None async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): silence_id = await alertmanager_client.create_silence( "fingerprint1", "user" ) silence = await alertmanager_client.get_silence("silence1") self.assertEqual("silence1", silence_id) self.assertEqual( { "id": "silence1", "status": {"state": "active"}, "matchers": [ { "name": "alertname", "value": "alert1", "isRegex": False, "isEqual": True, } ], "createdBy": "user", "startsAt": "1970-01-01T00:00:00", "endsAt": "1970-01-01T03:00:00", "comment": "Acknowledge alert from Matrix", }, silence, ) fake_cache.set.assert_called_once_with("fingerprint1", "silence1", expire=None) @freeze_time(datetime.utcfromtimestamp(0)) async def test_update_silence_without_duration(self) -> None: fake_cache = FakeCache() async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): silence1_id = await alertmanager_client.create_silence( "fingerprint1", "user" ) silence2_id = await alertmanager_client.update_silence("fingerprint1") silence = await alertmanager_client.get_silence("silence2") self.assertEqual("silence1", silence1_id) self.assertEqual("silence2", silence2_id) self.assertEqual( { "id": "silence2", "status": {"state": "active"}, "matchers": [ { "name": "alertname", "value": "alert1", "isRegex": False, "isEqual": True, } ], "createdBy": "user", "startsAt": "1970-01-01T00:00:00", "endsAt": "1970-01-01T03:00:00", "comment": "Acknowledge alert from Matrix", }, silence, ) self.assertEqual({"fingerprint1": ("silence2", None)}, fake_cache.cache) @patch.object(matrix_alertbot.alertmanager.AlertmanagerClient, "update_silence") @patch.object(matrix_alertbot.alertmanager.AlertmanagerClient, "create_silence") async def test_create_or_update_silence_without_duration_and_silence_not_found( self, fake_create_silence: Mock, fake_update_silence: Mock ) -> None: fake_cache = Mock(spec=Cache) fake_update_silence.side_effect = update_silence_raise_silence_not_found fake_create_silence.return_value = "silence1" alertmanager_client = AlertmanagerClient("http://localhost", fake_cache) async with aiotools.closing_async(alertmanager_client): silence_id1 = await alertmanager_client.create_or_update_silence( "fingerprint1", "user" ) self.assertEqual("silence1", silence_id1) fake_update_silence.assert_called_once_with( "fingerprint1", "user", None, force=False ) fake_create_silence.assert_called_once_with("fingerprint1", "user", None) @patch.object(matrix_alertbot.alertmanager.AlertmanagerClient, "update_silence") @patch.object(matrix_alertbot.alertmanager.AlertmanagerClient, "create_silence") async def test_create_or_update_silence_without_duration_and_silence_found( self, fake_create_silence: Mock, fake_update_silence: Mock ) -> None: fake_cache = Mock(spec=Cache) fake_update_silence.return_value = "silence1" alertmanager_client = AlertmanagerClient("http://localhost", fake_cache) async with aiotools.closing_async(alertmanager_client): silence_id1 = await alertmanager_client.create_or_update_silence( "fingerprint1", "user" ) self.assertEqual("silence1", silence_id1) fake_update_silence.assert_called_once_with( "fingerprint1", "user", None, force=False ) fake_create_silence.assert_not_called() @freeze_time(datetime.utcfromtimestamp(0)) async def test_create_silence_with_max_duration(self) -> None: fake_cache = Mock(spec=Cache) fake_cache.get.return_value = None async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): silence_id = await alertmanager_client.create_silence( "fingerprint1", "user", int(timedelta.max.total_seconds()) + 1 ) silence = await alertmanager_client.get_silence("silence1") self.assertEqual("silence1", silence_id) self.assertEqual( { "id": "silence1", "status": {"state": "active"}, "matchers": [ { "name": "alertname", "value": "alert1", "isRegex": False, "isEqual": True, } ], "createdBy": "user", "startsAt": "1970-01-01T00:00:00", "endsAt": "1980-01-01T00:00:00", "comment": "Acknowledge alert from Matrix", }, silence, ) async def test_create_silence_raise_alert_not_found(self) -> None: fake_cache = Mock(spec=Cache) fake_cache.get.return_value = None async with FakeAlertmanagerServerWithoutAlert() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): with self.assertRaises(AlertNotFoundError): await alertmanager_client.create_silence("fingerprint1", "user") async def test_create_silence_raise_alertmanager_error(self) -> None: fake_cache = Mock(spec=Cache) fake_cache.get.return_value = None async with FakeAlertmanagerServerWithErrorCreateSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): await alertmanager_client.get_alert("fingerprint1") with self.assertRaises(AlertmanagerServerError): await alertmanager_client.create_silence("fingerprint1", "user") async def test_update_silence_raise_silence_not_found(self) -> None: fake_cache = FakeCache({"fingerprint1": ("silence1", None)}) async with FakeAlertmanagerServerWithoutSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): with self.assertRaises(SilenceNotFoundError): await alertmanager_client.update_silence("fingerprint1") with self.assertRaises(SilenceNotFoundError): await alertmanager_client.update_silence("fingerprint2") async def test_update_silence_raise_silence_extend_error(self) -> None: fake_cache = FakeCache({"fingerprint1": ("silence1", 86400)}) async with FakeAlertmanagerServerWithoutAlert() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): with self.assertRaises(SilenceExtendError): await alertmanager_client.update_silence("fingerprint1") async def test_update_silence_raise_alertmanager_error(self) -> None: fake_cache = FakeCache({"fingerprint1": ("silence1", None)}) async with FakeAlertmanagerServerWithErrorCreateSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): await alertmanager_client.get_alert("fingerprint1") with self.assertRaises(AlertmanagerServerError): await alertmanager_client.update_silence("fingerprint1") async def test_delete_silence(self) -> None: fake_cache = Mock(spec=Cache) async with FakeAlertmanagerServer() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): await alertmanager_client.delete_silence("silence1") silences = await alertmanager_client.get_silences() self.assertEqual( [ { "id": "silence2", "createdBy": "user2", "status": {"state": "expired"}, "matchers": [], } ], silences, ) async def test_delete_silence_raise_silence_expired(self) -> None: fake_cache = Mock(spec=Cache) async with FakeAlertmanagerServer() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): with self.assertRaises(SilenceExpiredError): await alertmanager_client.delete_silence("silence2") silences = await alertmanager_client.get_silences() self.assertEqual(2, len(silences)) async def test_delete_silence_raise_alertmanager_error(self) -> None: fake_cache = Mock(spec=Cache) async with FakeAlertmanagerServerWithErrorDeleteSilence() as fake_alertmanager_server: port = fake_alertmanager_server.port alertmanager_client = AlertmanagerClient( f"http://localhost:{port}", fake_cache ) async with aiotools.closing_async(alertmanager_client): await alertmanager_client.get_alert("fingerprint1") with self.assertRaises(AlertmanagerServerError): await alertmanager_client.delete_silence("silence1") async def test_find_alert_happy(self) -> None: fake_cache = Mock(spec=Cache) alertmanager_client = AlertmanagerClient("http://localhost", fake_cache) alert = alertmanager_client._find_alert( "fingerprint1", [{"fingerprint": "fingerprint1"}] ) self.assertEqual({"fingerprint": "fingerprint1"}, alert) async def test_find_alert_raise_alert_not_found(self) -> None: fake_cache = Mock(spec=Cache) alertmanager_client = AlertmanagerClient("http://localhost", fake_cache) with self.assertRaises(AlertNotFoundError): alertmanager_client._find_alert("fingerprint1", []) with self.assertRaises(AlertNotFoundError): alertmanager_client._find_alert( "fingerprint2", [{"fingerprint": "fingerprint1"}] ) async def test_find_silence_happy(self) -> None: fake_cache = Mock(spec=Cache) alertmanager_client = AlertmanagerClient("http://localhost", fake_cache) silence = alertmanager_client._find_silence("silence1", [{"id": "silence1"}]) self.assertEqual({"id": "silence1"}, silence) async def test_find_silence_raise_silence_not_found(self) -> None: fake_cache = Mock(spec=Cache) alertmanager_client = AlertmanagerClient("http://localhost", fake_cache) with self.assertRaises(SilenceNotFoundError): alertmanager_client._find_silence("silence1", []) with self.assertRaises(SilenceNotFoundError): alertmanager_client._find_silence("silence2", [{"id": "silence1"}]) if __name__ == "__main__": unittest.main()