Add Matrix webhook backend

CHANGELOG.md

@@ -2,7 +2,7 @@
 
 ## Under development
 
- -
+ - Add Matrix webhook backend
 
 ## Version 0.4
 

README.md

@@ -193,6 +193,11 @@ The config file:
     priority = 1 very low
     footer = Ticket automatically created by Wartungsplan
 
+### Matrix webhook ###
+
+Add the correct webhook URL the config file and select the webhook backend.
+Done!
+
 # Examples
 
 ```

plan.conf.sample

@@ -23,6 +23,9 @@ password = AiX3sheeIyahf8aaQuah2wio
 #priority = 5 very high
 #footer = Ticket automatically created by Wartungsplan
 
+[webhook]
+url = https://matrix.example.org/appservice-webhooks/api/v1/matrix/hook/96..34kh
+
 [headers]
 # Configure here the available (allowed) headers with their
 # default value (defaults overwrite backend options)

src/Wartungsplan.py

@@ -32,12 +32,15 @@
 import datetime
 import sys
 import logging
+import json
 import configparser
 import smtplib
 import re
+import time
 import importlib.metadata
 from email.message import EmailMessage
 
+import requests
 import dateutil.parser
 import icalendar
 # under active development, few issues, nothing major
@@ -280,6 +283,75 @@ def _perform_action(self, actions_data):
         return True
 
 
+class CallWebhook(Backend):
+    """ Sends events to a webhook:
+        this is specific to Matrix webhooks and might not work with other
+        webhooks.
+        Once established this will become a parent class and _prepare_event
+        will be overwritten by child classes. """
+    def _prepare_event(self, headers, text, event):
+        summary = event['summary']
+        message_data = {
+            "text": f"{summary}: {text}",
+            "format": "text",
+            "displayName": "Wartungsplan"
+        }
+        return message_data
+
+    def _perform_action(self, actions_data, retries=3):
+        webhook_url = self.config["webhook"]["url"]
+
+        # Check if this is a dry run.
+        messages = actions_data
+        if self.dry_run:
+            logger.info("This is a Dry run!")
+            for msg in messages:
+                logger.debug("Calling webhook")
+                print(msg)
+                return True
+        else:
+            logger.info("We are calling the webhook")
+
+            # Convert the message data to JSON
+            for msg in messages:
+                message_json = json.dumps(msg)
+                logger.debug("Sending data to webhook \"%s\"", message_json)
+
+                # Set the headers for the HTTP request
+                headers = {
+                    "Content-Type": "application/json",
+                }
+
+                # Send the message using an HTTP POST request
+                error = None
+                for iteration in range(0, retries+1):
+                    time.sleep(iteration)
+                    try:
+                        response = requests.post(webhook_url, data=message_json,
+                                                 headers=headers, verify=False,
+                                                 timeout=5)
+                        response.raise_for_status()  # Check for any HTTP errors
+                        logger.info("Message sent successfully!")
+                    except requests.Timeout as err:
+                        logger.error("Connection timeout: %s", err)
+                        error = err
+                        continue
+                    except requests.exceptions.ConnectionError as err:
+                        logger.error("Connection error: %s", err)
+                        error = err
+                        continue
+                    except requests.exceptions.RequestException as err:
+                        logger.error("An error occurred: %s", err)
+                        # this can't be reasonably retried
+                        raise err
+                    # it worked so clear error
+                    error = None
+                    break
+                # it didn't work -> raise the error
+                if error:
+                    raise error
+
+
 class Wartungsplan:
     """ Builds the events for the given range and allow to call
         into the backend """
@@ -344,7 +416,7 @@ def main():
     # list: List installed jobs
     # send: To call the SendEmail backend
     # This list will grow with more backends
-    actions = ['version', 'list', 'send', 'otrs']
+    actions = ['version', 'list', 'send', 'otrs', 'webhook']
     parser.add_argument('action', choices=actions, help="Just print the version "\
                         "or select the desired action.")
     args = parser.parse_args()
@@ -406,6 +478,9 @@ def main():
     if args.action == 'otrs':
         backend = OtrsApi({"otrs":config["otrs"],
                            "headers":config["headers"]}, args.dry_run)
+    if args.action == 'webhook':
+        backend = CallWebhook({"webhook":config["webhook"],
+                           "headers":config["headers"]}, args.dry_run)
 
     if not backend:
         raise NameError("Action not found")

test/test.py

@@ -33,6 +33,7 @@
 import unittest
 import tempfile
 import icalendar
+import requests
 
 # Add Wartungsplan to PYTHONPATH
 TESTSDIR = os.path.dirname(os.path.abspath(__file__))
@@ -248,6 +249,46 @@ def test_config_option_priority(self):
         self.assertEqual(article, a1)
 
 
+class TestWebhook(unittest.TestCase):
+    """ Test Webhook backend """
+    @classmethod
+    def setUpClass(cls):
+        """ Set up common test case resources. """
+        cls.config = {"webhook": {"url": "http://127.0.0.1:1234/I/don't/get/it"}}
+
+    def test_data_structure(self):
+        """ Set up common test case resources. """
+        message_data = {
+            "text": "Summary: Some dummy text",
+            "format": "text",
+            "displayName": "Wartungsplan"
+        }
+        webhook_backend = Wartungsplan.CallWebhook(self.config, False)
+
+        message_data2 = webhook_backend._prepare_event(None,"Some dummy text",
+                                                       {'summary': "Summary"})
+        self.assertTrue(message_data == message_data2)
+
+        message_data2 = webhook_backend._prepare_event(None, "Some dummy\n text",
+                                                       {'summary': "Summary"})
+        self.assertTrue(len(message_data2["text"].splitlines()) == 2)
+
+    def test_action(self):
+        """ Check for correct error handling. """
+        webhook_backend = Wartungsplan.CallWebhook(self.config, False)
+        message_data = webhook_backend._prepare_event(None, "Some dummy text",
+                                                      {'summary': "Summary"})
+        with self.assertRaises(requests.exceptions.ConnectionError):
+            webhook_backend._perform_action(message_data, 0)
+
+        config = {"webhook": {"url": None}}
+        webhook_backend = Wartungsplan.CallWebhook(config, False)
+        message_data = webhook_backend._prepare_event(None, "Some dummy text",
+                                                      {'summary': "Summary"})
+        with self.assertRaises(requests.exceptions.MissingSchema):
+            webhook_backend._perform_action(message_data, 0)
+
+
 class TestAddEventToIcal(unittest.TestCase):
     """ Test tool to add event or create new calendar """
     @classmethod