add tests for grafana formatter

tests/start.py

@@ -25,10 +25,10 @@ parser.add_argument(
 )
 
 
-def bot_req(req=None, key=None, room_id=None, key_as_param=False):
+def bot_req(req=None, key=None, room_id=None, params=None, key_as_param=False):
     """Bot requests boilerplate."""
+    if params is None:
         params = {}
-
     if key is not None:
         if key_as_param:
             params["key"] = key

tests/test_grafana.py

@@ -0,0 +1,66 @@
+"""Test module for grafana formatter."""
+
+import unittest
+
+import httpx
+import nio
+
+from .start import BOT_URL, FULL_ID, KEY, MATRIX_ID, MATRIX_PW, MATRIX_URL
+
+# ref https://grafana.com/docs/grafana/latest/alerting/old-alerting/notifications/#webhook
+EXAMPLE_GRAFANA_REQUEST = """
+{
+  "dashboardId":1,
+  "evalMatches":[
+    {
+      "value":1,
+      "metric":"Count",
+      "tags":{}
+    }
+  ],
+  "imageUrl":"https://grafana.com/assets/img/blog/mixed_styles.png",
+  "message":"Notification Message",
+  "orgId":1,
+  "panelId":2,
+  "ruleId":1,
+  "ruleName":"Panel Title alert",
+  "ruleUrl":"http://localhost:3000/d/hZ7BuVbWz/test-dashboard?fullscreen\u0026edit\u0026tab=alert\u0026panelId=2\u0026orgId=1",
+  "state":"alerting",
+  "tags":{
+    "tag name":"tag value"
+  },
+  "title":"[Alerting] Panel Title alert"
+}
+"""
+
+
+class GrafanaFormatterTest(unittest.IsolatedAsyncioTestCase):
+    """Grafana formatter test class."""
+
+    async def test_grafana_body(self):
+        """Send a markdown message, and check the result."""
+        messages = []
+        client = nio.AsyncClient(MATRIX_URL, MATRIX_ID)
+
+        await client.login(MATRIX_PW)
+        room = await client.room_create()
+
+        self.assertEqual(
+            httpx.post(
+                f"{BOT_URL}/{room.room_id}",
+                params={"formatter": "grafana", "key": KEY},
+                content=EXAMPLE_GRAFANA_REQUEST,
+            ).json(),
+            {"status": 200, "ret": "OK"},
+        )
+
+        sync = await client.sync()
+        messages = await client.room_messages(room.room_id, sync.next_batch)
+        await client.close()
+
+        message = messages.chunk[0]
+        self.assertEqual(message.sender, FULL_ID)
+        self.assertEqual(
+            message.body,
+            "### [Alerting] Panel Title alert\nNotification Message\n\n* Count: 1\n",
+        )

tests/tests.py

@@ -24,6 +24,10 @@ class BotTest(unittest.IsolatedAsyncioTestCase):
             bot_req({"body": 3}, "wrong_key", key_as_param=True),
             {"status": 401, "ret": "Invalid API key"},
         )
+        self.assertEqual(
+            bot_req({"body": 3}, KEY, params={"formatter": "wrong_formatter"}),
+            {"status": 400, "ret": "Unknown formatter"},
+        )
         # TODO: if the client from matrix_webhook has olm support, this won't be a 403 from synapse,
         # but a LocalProtocolError from matrix_webhook
         self.assertEqual(