Closes #3952: Add test for webhooks_worker; introduce generate_signature()

netbox/extras/tests/test_webhooks.py

@@ -1,11 +1,19 @@
+import json
+import uuid
+from unittest.mock import patch
+
 import django_rq
 from django.contrib.contenttypes.models import ContentType
+from django.http import HttpResponse
 from django.urls import reverse
+from requests import Session
 from rest_framework import status
 
 from dcim.models import Site
 from extras.choices import ObjectChangeActionChoices
 from extras.models import Webhook
+from extras.webhooks import enqueue_webhooks, generate_signature
+from extras.webhooks_worker import process_webhook
 from utilities.testing import APITestCase
 
 
@@ -22,11 +30,13 @@ class WebhookTest(APITestCase):
     def setUpTestData(cls):
 
         site_ct = ContentType.objects.get_for_model(Site)
-        PAYLOAD_URL = "http://localhost/"
+        DUMMY_URL = "http://localhost/"
+        DUMMY_SECRET = "LOOKATMEIMASECRETSTRING"
+
         webhooks = Webhook.objects.bulk_create((
-            Webhook(name='Site Create Webhook', type_create=True, payload_url=PAYLOAD_URL),
-            Webhook(name='Site Update Webhook', type_update=True, payload_url=PAYLOAD_URL),
-            Webhook(name='Site Delete Webhook', type_delete=True, payload_url=PAYLOAD_URL),
+            Webhook(name='Site Create Webhook', type_create=True, payload_url=DUMMY_URL, secret=DUMMY_SECRET, additional_headers={'X-Foo': 'Bar'}),
+            Webhook(name='Site Update Webhook', type_update=True, payload_url=DUMMY_URL, secret=DUMMY_SECRET),
+            Webhook(name='Site Delete Webhook', type_delete=True, payload_url=DUMMY_URL, secret=DUMMY_SECRET),
         ))
         for webhook in webhooks:
             webhook.obj_type.set([site_ct])
@@ -87,3 +97,47 @@ class WebhookTest(APITestCase):
         self.assertEqual(job.args[1]['id'], site.pk)
         self.assertEqual(job.args[2], 'site')
         self.assertEqual(job.args[3], ObjectChangeActionChoices.ACTION_DELETE)
+
+    def test_webhooks_worker(self):
+
+        request_id = uuid.uuid4()
+
+        def dummy_send(_, request):
+            """
+            A dummy implementation of Session.send() to be used for testing.
+            Always returns a 200 HTTP response.
+            """
+            webhook = Webhook.objects.get(type_create=True)
+            signature = generate_signature(request.body, webhook.secret)
+
+            # Validate the outgoing request headers
+            self.assertEqual(request.headers['Content-Type'], webhook.http_content_type)
+            self.assertEqual(request.headers['X-Hook-Signature'], signature)
+            self.assertEqual(request.headers['X-Foo'], 'Bar')
+
+            # Validate the outgoing request body
+            body = json.loads(request.body)
+            self.assertEqual(body['event'], 'created')
+            self.assertEqual(body['timestamp'], job.args[4])
+            self.assertEqual(body['model'], 'site')
+            self.assertEqual(body['username'], 'testuser')
+            self.assertEqual(body['request_id'], str(request_id))
+            self.assertEqual(body['data']['name'], 'Site 1')
+
+            return HttpResponse()
+
+        # Enqueue a webhook for processing
+        site = Site.objects.create(name='Site 1', slug='site-1')
+        enqueue_webhooks(
+            instance=site,
+            user=self.user,
+            request_id=request_id,
+            action=ObjectChangeActionChoices.ACTION_CREATE
+        )
+
+        # Retrieve the job from queue
+        job = self.queue.jobs[0]
+
+        # Patch the Session object with our dummy_send() method, then process the webhook for sending
+        with patch.object(Session, 'send', dummy_send) as mock_send:
+            process_webhook(*job.args)