Added function hooks and reworked functionWebhook

Author: TCA166

gitWebhook/functionWebhook.py

@@ -1,11 +1,12 @@
 from .webhook import webhookBlueprint
 from typing import Callable, Any
 from logging import Logger
+import json
 
 class functionWebhookBlueprint(webhookBlueprint):
     """A subclass of webhookBlueprint that processes the webhook data using a list of functions. The functions should return True if the webhook data is valid, and False otherwise. If the function returns a string, it will be included in the output."""
 
-    def __init__(self, webhookToken: str | None, functions: list[Callable[[dict[str, Any]], bool | str]], log:Logger | None = None, name:str="webhook", github:bool=True, gitlab:bool=True, gitea:bool=True, ipWhitelist:list[str] | None = None, *args, **kwargs):
+    def __init__(self, webhookToken: str | None, functions: list[Callable[[dict[str, Any]], bool | Any]], log:Logger | None = None, name:str="webhook", github:bool=True, gitlab:bool=True, gitea:bool=True, ipWhitelist:list[str] | None = None, *args, **kwargs):
         """Initialize the webhook blueprint with a list of functions to process the webhook data.
 
         Args:
@@ -38,24 +39,36 @@ def processWebhook(self, data: dict[str, Any]) -> tuple[int, str]:
         """
         if self.log is not None:
             self.log.debug(f"Processing webhook: {data}")
-        output = []
+        success = True
+        output:dict[str, str | bool] = {}
         for function in self.functions:
-            res = function(data)
-            if isinstance(res, str):
+            try:
+                res = function(data)
+            except Exception as e:
+                output[function.__name__] = str(e)
                 if self.log is not None:
-                    self.log.debug(f"Function {function.__name__} returned a string")
-                output.append(res)
-            elif isinstance(res, bool):
+                    self.log.error(f"Function {function.__name__} raised an exception: {e}")
+                success = False
+                continue
+            if isinstance(res, bool):
                 if not res:
                     if self.log is not None:
                         self.log.error(f"Function {function.__name__} returned false")
-                    return 400, f"Function {function.__name__} returned false"
+                    success = False
                 else:
                     if self.log is not None:
                         self.log.debug(f"Function {function.__name__} returned true")
-                    output.append(str(res))
+                    output[function.__name__] = res
             else:
                 if self.log is not None:
-                    self.log.error(f"Function {function.__name__} returned an invalid type")
-                return 500, f"Function {function.__name__} returned an invalid type"
-        return 200, str(output)
+                    self.log.debug(f"Function {function.__name__} returned a string")
+                try:
+                    output[function.__name__] = str(res)
+                except Exception as e:
+                    if self.log is not None:
+                        self.log.error(f"Function {function.__name__} returned an invalid type")
+                    return 500, f"Function {function.__name__} returned an invalid type"
+        if success:
+            return 200, str(output)
+        else:
+            return 400, json.dumps(output)

gitWebhook/webhook.py

@@ -1,7 +1,7 @@
 from flask import Blueprint, request, Response, abort, Request
 from hashlib import sha256
 from hmac import new as hmacNew
-from typing import Any
+from typing import Any, Callable
 from logging import Logger
 from .abstractWebhook import gitWebhookBlueprintABC
 
@@ -56,6 +56,7 @@ def __init__(self, webhookToken:str | None, log:Logger | None = None, name:str="
         self.gitlab = gitlab
         self.gitea = gitea
         self.ipWhitelist = ipWhitelist
+        self.hooks = []
         self.route("/", methods=["POST"])(self.receiveWebhook)
 
     def receiveWebhook(self) -> Response:
@@ -96,6 +97,8 @@ def receiveWebhook(self) -> Response:
                     self.log.warning("A request with no signature found")
                 abort(401) #no feedback, in case somebody is trying to guess the token
         #logs beforehand were warnings, so that messages regarding unauthorized requests can be filtered
+        for hook in self.hooks:
+            hook()
         data = request.json
         if data is None or not isinstance(data, dict):
             if self.log is not None:
@@ -117,3 +120,18 @@ def processWebhook(self, data:dict[str, Any]) -> tuple[int, str]:
             tuple[int, str]: HTTP return code with a message
         """
         return (200, "OK")
+    
+    def hook(self, func:Callable[..., Any]) -> Callable[..., Any]:
+        """Adds a function to the list of functions that will be called when a webhook is received.
+        This is different from what functionWebhook functions do as this function is called before the processWebhook method and the return value is not checked.
+        
+        Args:
+            func (Callable): The function to add to the list of functions that will be called when a webhook is received.
+        """
+        self.hooks.append(func)
+        def inner(*args, **kwargs):
+            print("test")
+            if self.log is not None:
+                self.log.debug("Received a POST request to a hooked function")
+            return func(*args, **kwargs)
+        return inner

test.py

@@ -143,5 +143,24 @@ def testReceiveWebhookInvalidCheck(self):
     def testProcessWebhook(self):
         self.assertEqual(self.webhook.processWebhook({"test":"test"}), (400, "Function <lambda> returned false"))
 
+class TestHooks(unittest.TestCase):
+    def setUp(self) -> None:
+        self.webhook = webhookBlueprint(VALID_TOKEN, name="valid")
+        self.app = Flask(__name__)
+        self.app.register_blueprint(self.webhook, url_prefix="/valid")
+        self.app.config.update({"TESTING": True})
+        self.client = self.app.test_client()
+        self.hooked = False
+        return super().setUp()
+    
+    def testHook(self):
+        @self.webhook.hook
+        def hook():
+            self.hooked = True
+        request = testingRequest(VALID_TOKEN)
+        request.headers["Content-Type"] = "application/json"
+        resp = self.client.post("/valid/", headers=request.headers, data=request.data)
+        self.assertTrue(self.hooked)
+
 if __name__ == "__main__":
     unittest.main()