matrix-alertbot/tests/test_alertmanager.py

from __future__ import annotations

import json
import unittest
from datetime import datetime
from typing import Any, List
from unittest.mock import MagicMock, Mock, patch

import aiohttp
import aiohttp.test_utils
import aiotools
from aiohttp import web, web_request
from diskcache import Cache

from matrix_alertbot.alertmanager import AlertmanagerClient
from matrix_alertbot.errors import (
    AlertmanagerServerError,
    AlertMismatchError,
    AlertNotFoundError,
    SilenceNotFoundError,
)
from matrix_alertbot.matcher import AlertMatcher, AlertRegexMatcher


class FakeTimeDelta:
    def __init__(self, seconds: int) -> None:
        self.seconds = seconds

    def __radd__(self, other: Any) -> datetime:
        return datetime.utcfromtimestamp(self.seconds)


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):
                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):
                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):
                with self.assertRaises(AlertmanagerServerError):
                    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):
                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):
                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):
                with self.assertRaises(AlertmanagerServerError):
                    await alertmanager.get_alert("fingerprint1")

    @patch("matrix_alertbot.alertmanager.timedelta", side_effect=FakeTimeDelta)
    async def test_create_silence_without_matchers(self, fake_timedelta: Mock) -> 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):
                silence = await alertmanager.create_silence(
                    "fingerprint1", 86400, "user", []
                )

        self.assertEqual("silence1", silence)
        fake_timedelta.assert_called_once_with(seconds=86400)

    @patch("matrix_alertbot.alertmanager.timedelta", side_effect=FakeTimeDelta)
    async def test_create_silence_with_matchers(self, fake_timedelta: Mock) -> None:
        matchers = [AlertMatcher(label="alertname", value="alert1")]

        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):
                silence = await alertmanager.create_silence(
                    "fingerprint1",
                    86400,
                    "user",
                    matchers,
                )

        self.assertEqual("silence1", silence)
        fake_timedelta.assert_called_once_with(seconds=86400)

    @patch("matrix_alertbot.alertmanager.timedelta", side_effect=FakeTimeDelta)
    async def test_create_silence_with_regex_matchers(
        self, fake_timedelta: Mock
    ) -> None:
        matchers: List[AlertMatcher] = [
            AlertRegexMatcher(label="alertname", regex=r"alert\d+")
        ]

        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):
                silence = await alertmanager.create_silence(
                    "fingerprint1",
                    86400,
                    "user",
                    matchers,
                )

        self.assertEqual("silence1", silence)
        fake_timedelta.assert_called_once_with(seconds=86400)

    async def test_create_silence_raise_missing_label(self) -> None:
        matchers = [
            AlertMatcher(label="alertname", value="alert1"),
            AlertMatcher(label="severity", value="critical"),
        ]

        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):
                with self.assertRaises(AlertMismatchError):
                    await alertmanager.create_silence(
                        "fingerprint1",
                        86400,
                        "user",
                        matchers,
                    )

    async def test_create_silence_raise_mismatch_label(self) -> None:
        matchers = [AlertMatcher(label="alertname", value="alert2")]

        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):
                with self.assertRaises(AlertMismatchError):
                    await alertmanager.create_silence(
                        "fingerprint1",
                        86400,
                        "user",
                        matchers,
                    )

    async def test_create_silence_raise_mismatch_regex_label(self) -> None:
        matchers: List[AlertMatcher] = [
            AlertRegexMatcher(label="alertname", regex=r"alert[^\d]+")
        ]

        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):
                with self.assertRaises(AlertMismatchError):
                    await alertmanager.create_silence(
                        "fingerprint1",
                        86400,
                        "user",
                        matchers,
                    )

    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):
                with self.assertRaises(AlertNotFoundError):
                    await alertmanager.create_silence("fingerprint1", 86400, "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):
                await alertmanager.get_alert("fingerprint1")

                with self.assertRaises(AlertmanagerServerError):
                    await alertmanager.create_silence("fingerprint1", 86400, "user", [])

    async def test_delete_silences_without_matchers(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):
                silences = await alertmanager.delete_silences("fingerprint2", [])

        self.assertEqual(["silence1", "silence2"], silences)

    async def test_delete_silences_with_matchers(self) -> None:
        matchers = [AlertMatcher(label="alertname", value="alert2")]

        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):
                silences = await alertmanager.delete_silences("fingerprint2", matchers)

        self.assertEqual(["silence1", "silence2"], silences)

    async def test_delete_silences_with_regex_matchers(self) -> None:
        matchers: List[AlertMatcher] = [
            AlertRegexMatcher(label="alertname", regex=r"alert\d+")
        ]

        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):
                silences = await alertmanager.delete_silences("fingerprint2", matchers)

        self.assertEqual(["silence1", "silence2"], silences)

    async def test_delete_silences_raise_missing_label(self) -> None:
        matchers = [
            AlertMatcher(label="alertname", value="alert2"),
            AlertMatcher(label="severity", value="critical"),
        ]

        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):
                with self.assertRaises(AlertMismatchError):
                    await alertmanager.delete_silences("fingerprint2", matchers)

    async def test_delete_silences_raise_mismatch_label(self) -> None:
        matchers = [
            AlertMatcher(label="alertname", value="alert1"),
        ]

        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):
                with self.assertRaises(AlertMismatchError):
                    await alertmanager.delete_silences("fingerprint2", matchers)

    async def test_delete_silences_raise_mismatch_regex_label(self) -> None:
        matchers: List[AlertMatcher] = [
            AlertRegexMatcher(label="alertname", regex=r"alert[^\d]+"),
        ]

        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):
                with self.assertRaises(AlertMismatchError):
                    await alertmanager.delete_silences("fingerprint2", matchers)

    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):
                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):
                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):
                await alertmanager.get_alert("fingerprint1")

                with self.assertRaises(AlertmanagerServerError):
                    await alertmanager.delete_silences("fingerprint2", [])

    async def test_find_alert_happy(self) -> None:
        alertmanager = AlertmanagerClient("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("http://localhost", self.fake_cache)

        with self.assertRaises(AlertNotFoundError):
            alertmanager._find_alert("fingerprint2", [{"fingerprint": "fingerprint1"}])


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