render alert from jinja2 templates

matrix_alertbot/alert.py

@@ -1,7 +1,15 @@
 from __future__ import annotations
 
 import logging
-from typing import Dict
+from typing import Dict, Optional
+
+from jinja2 import (
+    BaseLoader,
+    ChoiceLoader,
+    Environment,
+    FileSystemLoader,
+    PackageLoader,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -24,6 +32,7 @@ class Alert:
 
         self.labels = labels
         self.annotations = annotations
+        self.description = annotations["description"]
 
         if self.firing:
             self.status = self.labels["severity"]
@@ -48,20 +57,20 @@ class Alert:
     def color(self) -> str:
         return self.COLORS[self.status]
 
-    def plaintext(self) -> str:
-        alertname = self.labels["alertname"]
-        description = self.annotations["description"]
-        return f"[{self.emoji} {self.status.upper()}] {alertname}: {description}"
 
-    def html(self) -> str:
+class AlertRenderer:
-        alertname = self.labels["alertname"]
+    def __init__(self, template_dir: Optional[str] = None) -> None:
+        loader: BaseLoader = PackageLoader("matrix_alertbot", "resources/templates")
+        if template_dir is not None:
+            loader = ChoiceLoader([FileSystemLoader(template_dir), loader])
+        env = Environment(loader=loader, autoescape=True)
 
-        job = self.labels.get("job", "")
+        self.html_template = env.get_template("alert.html.j2")
-        if job:
+        self.text_template = env.get_template("alert.txt.j2")
-            job = f"({job})"
 
-        description = self.annotations["description"]
+    def render(self, alert: Alert, html: bool = True) -> str:
-        return (
+        if html:
-            f"<font color='#{self.color}'><b>[{self.emoji} {self.status.upper()}]</b></font> "
+            template = self.html_template
-            f"<a href='{self.url}'>{alertname}</a> {job}<br/>{description}"
+        else:
-        )
+            template = self.text_template
+        return template.render(vars(alert), color=alert.color, emoji=alert.emoji)

matrix_alertbot/callback.py

@@ -21,9 +21,6 @@ from matrix_alertbot.config import Config
 logger = logging.getLogger(__name__)
 
 
-REACTIONS = {"🤫", "😶", "🤐", "🙊", "🔇", "🔕"}
-
-
 class Callbacks:
     def __init__(
         self,
@@ -176,7 +173,7 @@ class Callbacks:
         reaction = event.source.get("content", {}).get("m.relates_to", {}).get("key")
         logger.debug(f"Got reaction {reaction} to {room.room_id} from {event.sender}.")
 
-        if reaction not in REACTIONS:
+        if reaction not in self.config.allowed_reactions:
             logger.warning(f"Uknown duration reaction {reaction}")
             return
 

matrix_alertbot/command.py

@@ -104,7 +104,9 @@ class AckAlertCommand(BaseAlertCommand):
                 )
                 return
             elif duration_seconds < 0:
-                logger.error(f"Unable to create silence: Duration must be positive, got '{duration}'")
+                logger.error(
+                    f"Unable to create silence: Duration must be positive, got '{duration}'"
+                )
                 await send_text_to_room(
                     self.client,
                     self.room.room_id,

matrix_alertbot/config.py

@@ -19,6 +19,9 @@ logging.getLogger("peewee").setLevel(
 )  # Prevent debug messages from peewee lib
 
 
+DEFAULT_REACTIONS = {"🤫", "😶", "🤐", "🙊", "🔇", "🔕"}
+
+
 class Config:
     """Creates a Config object from a YAML-encoded config file from a given filepath"""
 
@@ -75,6 +78,9 @@ class Config:
                     f"storage.path '{self.store_dir}' is not a directory"
                 )
 
+        # Template setup
+        self.template_dir: str = self._get_cfg(["template", "path"], required=False)
+
         # Cache setup
         self.cache_dir: str = self._get_cfg(["cache", "path"], required=True)
         expire_time: str = self._get_cfg(["cache", "expire_time"], default="1w")
@@ -105,6 +111,9 @@ class Config:
         self.allowed_rooms: list = self._get_cfg(
             ["matrix", "allowed_rooms"], required=True
         )
+        self.allowed_reactions = set(
+            self._get_cfg(["matrix", "allowed_reactions"], default=DEFAULT_REACTIONS)
+        )
 
         self.address: str = self._get_cfg(["webhook", "address"], required=False)
         self.port: int = self._get_cfg(["webhook", "port"], required=False)

matrix_alertbot/resources/templates/alert.html.j2

@@ -0,0 +1,5 @@
+<font color="#{{ color }}">
+  <b>[{{ emoji }} {{ status | upper }}]</b>
+</font> <a href="{{ url }}">{{ labels.alertname }}</a>
+{% if 'job' in labels %} ({{ labels.job }}){% endif %}<br/>
+{{ description }}

matrix_alertbot/resources/templates/alert.txt.j2

@@ -0,0 +1 @@
+[{{ emoji }} {{ status | upper }}] {{ labels.alertname }}: {{ description }}

matrix_alertbot/webhook.py

@@ -9,7 +9,7 @@ from aiohttp_prometheus_exporter.middleware import prometheus_middleware_factory
 from diskcache import Cache
 from nio import AsyncClient, LocalProtocolError
 
-from matrix_alertbot.alert import Alert
+from matrix_alertbot.alert import Alert, AlertRenderer
 from matrix_alertbot.chat_functions import send_text_to_room
 from matrix_alertbot.config import Config
 
@@ -31,10 +31,13 @@ async def create_alerts(request: web_request.Request) -> web.Response:
     client: AsyncClient = request.app["client"]
     config: Config = request.app["config"]
     cache: Cache = request.app["cache"]
+    alert_renderer: AlertRenderer = request.app["alert_renderer"]
 
     if room_id not in config.allowed_rooms:
         logger.error("Cannot send alerts to room ID {room_id}.")
-        return web.Response(status=401, body=f"Cannot send alerts to room ID {room_id}.")
+        return web.Response(
+            status=401, body=f"Cannot send alerts to room ID {room_id}."
+        )
 
     if "alerts" not in data:
         logger.error("Received data without 'alerts' key")
@@ -61,8 +64,8 @@ async def create_alerts(request: web_request.Request) -> web.Response:
             logger.error(f"Cannot parse alert dict: {e}")
             return web.Response(status=400, body=f"Invalid alert: {alert}.")
 
-        plaintext = alert.plaintext()
+        plaintext = alert_renderer.render(alert, html=False)
-        html = alert.html()
+        html = alert_renderer.render(alert, html=True)
 
         try:
             event = await send_text_to_room(
@@ -91,6 +94,7 @@ class Webhook:
         self.app["client"] = client
         self.app["config"] = config
         self.app["cache"] = cache
+        self.app["alert_renderer"] = AlertRenderer(config.template_dir)
         self.app.add_routes(routes)
 
         prometheus_registry = prometheus_client.CollectorRegistry(auto_describe=True)
@@ -112,10 +116,10 @@ class Webhook:
         site: web.BaseSite
         if self.address and self.port:
             site = web.TCPSite(self.runner, self.address, self.port)
-            logger.info(f"Listenning on {self.address}:{self.port}")
+            logger.info(f"Listening on {self.address}:{self.port}")
         elif self.socket:
             site = web.UnixSite(self.runner, self.socket)
-            logger.info(f"Listenning on unix://{self.socket}")
+            logger.info(f"Listening on unix://{self.socket}")
 
         await site.start()
 

setup.py

@@ -29,6 +29,7 @@ setup(
         "aiohttp-prometheus-exporter>=0.2.4",
         "aiotools>=1.5.9",
         "diskcache>=5.4.0",
+        "jinja2>=3.1.2",
         "matrix-nio>=0.19.0",
         "Markdown>=3.3.7",
         "pytimeparse2>=1.4.0",

tests/resources/config/config.full.yml

@@ -22,6 +22,7 @@ matrix:
   device_name: fake_device_name
   allowed_rooms:
     - "!abcdefgh:matrix.example.com"
+  allowed_reactions: [🤫, 😶, 🤐]
 
 webhook:
   socket: matrix-alertbot.socket
@@ -38,6 +39,9 @@ storage:
   # containing encryption keys, sync tokens, etc.
   path: "data/store"
 
+template:
+  path: "data/templates"
+
 # Logging setup
 logging:
   # Logging level

tests/resources/templates/alert.html.j2

@@ -0,0 +1 @@
+<b>hello world</b>

tests/resources/templates/alert.txt.j2

@@ -0,0 +1 @@
+hello world

tests/test_alert.py

@@ -1,7 +1,10 @@
+import os
 import unittest
 from typing import Dict
 
-from matrix_alertbot.alert import Alert
+from matrix_alertbot.alert import Alert, AlertRenderer
+
+TESTS_DIR = os.path.dirname(__file__)
 
 
 class AlertTestCase(unittest.TestCase):
@@ -40,62 +43,121 @@ class AlertTestCase(unittest.TestCase):
         self.assertEqual("resolved", alert.status)
         self.assertFalse(alert.firing)
 
-    def test_display_firing_critical_alert(self) -> None:
+
+class AlertRendererTestCase(unittest.TestCase):
+    def setUp(self) -> None:
+        self.alert_dict: Dict = {
+            "fingerprint": "fingerprint1",
+            "generatorURL": "http://example.com",
+            "status": "unknown",
+            "labels": {"alertname": "alert1", "severity": "critical", "job": "job1"},
+            "annotations": {"description": "some description"},
+        }
+        self.renderer = AlertRenderer()
+
+    def test_render_firing_critical_alert(self) -> None:
         self.alert_dict["status"] = "firing"
         alert = Alert.from_dict(self.alert_dict)
         alert.labels["severity"] = "critical"
 
-        html = alert.html()
+        html = self.renderer.render(alert, html=True)
         self.assertEqual(
-            "<font color='#dc3545'><b>[🔥 CRITICAL]</b></font> "
+            '<font color="#dc3545">\n  <b>[🔥 CRITICAL]</b>\n</font> '
-            "<a href='http://example.com'>alert1</a> (job1)<br/>"
+            '<a href="http://example.com">alert1</a>\n (job1)<br/>\n'
             "some description",
             html,
         )
 
-        plaintext = alert.plaintext()
+        plaintext = self.renderer.render(alert, html=False)
         self.assertEqual("[🔥 CRITICAL] alert1: some description", plaintext)
 
-    def test_display_firing_warning_alert(self) -> None:
+    def test_render_firing_warning_alert(self) -> None:
         self.alert_dict["status"] = "firing"
         self.alert_dict["labels"]["severity"] = "warning"
         alert = Alert.from_dict(self.alert_dict)
 
-        html = alert.html()
+        html = self.renderer.render(alert, html=True)
         self.assertEqual(
-            "<font color='#ffc107'><b>[⚠️ WARNING]</b></font> "
+            '<font color="#ffc107">\n  <b>[⚠️ WARNING]</b>\n</font> '
-            "<a href='http://example.com'>alert1</a> (job1)<br/>"
+            '<a href="http://example.com">alert1</a>\n (job1)<br/>\n'
             "some description",
             html,
         )
 
-        plaintext = alert.plaintext()
+        plaintext = self.renderer.render(alert, html=False)
         self.assertEqual("[⚠️ WARNING] alert1: some description", plaintext)
 
-    def test_display_firing_unknown_alert(self) -> None:
+    def test_render_firing_unknown_alert(self) -> None:
         self.alert_dict["status"] = "firing"
         self.alert_dict["labels"]["severity"] = "unknown"
         alert = Alert.from_dict(self.alert_dict)
 
         with self.assertRaisesRegex(KeyError, "unknown"):
-            alert.html()
+            self.renderer.render(alert, html=True)
 
         with self.assertRaisesRegex(KeyError, "unknown"):
-            alert.plaintext()
+            self.renderer.render(alert, html=False)
 
-    def test_display_resolved_alert(self) -> None:
+    def test_render_resolved_alert(self) -> None:
         self.alert_dict["status"] = "resolved"
         alert = Alert.from_dict(self.alert_dict)
 
-        html = alert.html()
+        html = self.renderer.render(alert, html=True)
         self.assertEqual(
-            "<font color='#33cc33'><b>[🥦 RESOLVED]</b></font> "
+            '<font color="#33cc33">\n  <b>[🥦 RESOLVED]</b>\n</font> '
-            "<a href='http://example.com'>alert1</a> (job1)<br/>"
+            '<a href="http://example.com">alert1</a>\n (job1)<br/>\n'
             "some description",
             html,
         )
 
-        plaintext = alert.plaintext()
+        plaintext = self.renderer.render(alert, html=False)
+        self.assertEqual("[🥦 RESOLVED] alert1: some description", plaintext)
+
+    def test_render_resolved_alert_without_job(self) -> None:
+        self.alert_dict["status"] = "resolved"
+        del self.alert_dict["labels"]["job"]
+        alert = Alert.from_dict(self.alert_dict)
+
+        html = self.renderer.render(alert, html=True)
+        self.assertEqual(
+            '<font color="#33cc33">\n  <b>[🥦 RESOLVED]</b>\n</font> '
+            '<a href="http://example.com">alert1</a>\n<br/>\n'
+            "some description",
+            html,
+        )
+
+        plaintext = self.renderer.render(alert, html=False)
+        self.assertEqual("[🥦 RESOLVED] alert1: some description", plaintext)
+
+    def test_render_with_existing_filesystem_template(self) -> None:
+        alert = Alert.from_dict(self.alert_dict)
+
+        template_dir = os.path.join(TESTS_DIR, "resources/templates")
+        renderer = AlertRenderer(template_dir)
+
+        html = renderer.render(alert, html=True)
+        self.assertEqual(
+            "<b>hello world</b>",
+            html,
+        )
+
+        plaintext = renderer.render(alert, html=False)
+        self.assertEqual("hello world", plaintext)
+
+    def test_render_with_inexistent_filesystem_template(self) -> None:
+        self.alert_dict["status"] = "resolved"
+        alert = Alert.from_dict(self.alert_dict)
+
+        renderer = AlertRenderer(TESTS_DIR)
+        html = renderer.render(alert, html=True)
+        self.assertEqual(
+            '<font color="#33cc33">\n  <b>[🥦 RESOLVED]</b>\n</font> '
+            '<a href="http://example.com">alert1</a>\n (job1)<br/>\n'
+            "some description",
+            html,
+        )
+
+        plaintext = renderer.render(alert, html=False)
         self.assertEqual("[🥦 RESOLVED] alert1: some description", plaintext)
 
 

tests/test_callback.py

@@ -31,6 +31,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
         # We don't spec config, as it doesn't currently have well defined attributes
         self.fake_config = Mock()
         self.fake_config.allowed_rooms = [self.fake_room.room_id]
+        self.fake_config.allowed_reactions = ["🤫"]
         self.fake_config.command_prefix = "!alert "
 
         self.callbacks = Callbacks(

tests/test_config.py

@@ -5,7 +5,7 @@ from unittest.mock import Mock, patch
 
 import yaml
 
-from matrix_alertbot.config import Config
+from matrix_alertbot.config import DEFAULT_REACTIONS, Config
 from matrix_alertbot.errors import (
     InvalidConfigError,
     ParseConfigError,
@@ -58,6 +58,7 @@ class ConfigTestCase(unittest.TestCase):
         self.assertEqual("matrix-alertbot", config.device_name)
         self.assertEqual("https://matrix.example.com", config.homeserver_url)
         self.assertEqual(["!abcdefgh:matrix.example.com"], config.allowed_rooms)
+        self.assertEqual(DEFAULT_REACTIONS, config.allowed_reactions)
 
         self.assertEqual("0.0.0.0", config.address)
         self.assertEqual(8080, config.port)
@@ -71,6 +72,8 @@ class ConfigTestCase(unittest.TestCase):
 
         self.assertEqual("data/store", config.store_dir)
 
+        self.assertIsNone(config.template_dir)
+
         self.assertEqual("!alert ", config.command_prefix)
 
     @patch("os.path.isdir")
@@ -96,6 +99,7 @@ class ConfigTestCase(unittest.TestCase):
         self.assertEqual("fake_device_name", config.device_name)
         self.assertEqual("https://matrix.example.com", config.homeserver_url)
         self.assertEqual(["!abcdefgh:matrix.example.com"], config.allowed_rooms)
+        self.assertEqual({"🤫", "😶", "🤐"}, config.allowed_reactions)
 
         self.assertIsNone(config.address)
         self.assertIsNone(config.port)
@@ -109,6 +113,8 @@ class ConfigTestCase(unittest.TestCase):
 
         self.assertEqual("data/store", config.store_dir)
 
+        self.assertEqual("data/templates", config.template_dir)
+
         self.assertEqual("!alert ", config.command_prefix)
 
     def test_read_config_raise_config_error(self) -> None:

tests/test_webhook.py

@@ -32,6 +32,7 @@ class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
         self.fake_config.socket = "webhook.sock"
         self.fake_config.allowed_rooms = [self.fake_room_id]
         self.fake_config.cache_expire_time = 0
+        self.fake_config.template_dir = None
 
         self.fake_alerts = {
             "alerts": [
@@ -77,8 +78,8 @@ class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
                     self.fake_client,
                     self.fake_room_id,
                     "[🔥 CRITICAL] alert1: some description1",
-                    "<font color='#dc3545'><b>[🔥 CRITICAL]</b></font> "
+                    '<font color="#dc3545">\n  <b>[🔥 CRITICAL]</b>\n</font> '
-                    "<a href='http://example.com/alert1'>alert1</a> (job1)<br/>"
+                    '<a href="http://example.com/alert1">alert1</a>\n (job1)<br/>\n'
                     "some description1",
                     notice=False,
                 ),
@@ -86,8 +87,8 @@ class WebhookApplicationTestCase(aiohttp.test_utils.AioHTTPTestCase):
                     self.fake_client,
                     self.fake_room_id,
                     "[🥦 RESOLVED] alert2: some description2",
-                    "<font color='#33cc33'><b>[🥦 RESOLVED]</b></font> "
+                    '<font color="#33cc33">\n  <b>[🥦 RESOLVED]</b>\n</font> '
-                    "<a href='http://example.com/alert2'>alert2</a> (job2)<br/>"
+                    '<a href="http://example.com/alert2">alert2</a>\n (job2)<br/>\n'
                     "some description2",
                     notice=False,
                 ),
@@ -212,6 +213,7 @@ class WebhookServerTestCase(unittest.IsolatedAsyncioTestCase):
         self.fake_config.address = "localhost"
         self.fake_config.socket = "webhook.sock"
         self.fake_config.cache_expire_time = 0
+        self.fake_config.template_dir = None
 
     @patch.object(matrix_alertbot.webhook.web, "TCPSite", autospec=True)
     async def test_webhook_start_address_port(self, fake_tcp_site: Mock) -> None: