Improve webhook tests

netbox/extras/tests/test_webhooks.py

@@ -9,11 +9,12 @@ from django.urls import reverse
 from requests import Session
 from rest_framework import status
 
+from dcim.choices import SiteStatusChoices
 from dcim.models import Site
 from extras.choices import ObjectChangeActionChoices
 from extras.models import Tag, Webhook
-from extras.webhooks import enqueue_object, flush_webhooks, generate_signature
-from extras.webhooks_worker import process_webhook
+from extras.webhooks import enqueue_object, flush_webhooks, generate_signature, serialize_for_webhook
+from extras.webhooks_worker import eval_conditions, process_webhook
 from utilities.testing import APITestCase
 
 
@@ -251,6 +252,37 @@ class WebhookTest(APITestCase):
             self.assertEqual(job.kwargs['snapshots']['prechange']['name'], sites[i].name)
             self.assertEqual(job.kwargs['snapshots']['prechange']['tags'], ['Bar', 'Foo'])
 
+    def test_webhook_conditions(self):
+        # Create a conditional Webhook
+        webhook = Webhook(
+            name='Conditional Webhook',
+            type_create=True,
+            type_update=True,
+            payload_url='http://localhost/',
+            conditions={
+                'and': [
+                    {
+                        'attr': 'status.value',
+                        'value': 'active',
+                    }
+                ]
+            }
+        )
+
+        # Create a Site to evaluate
+        site = Site.objects.create(name='Site 1', slug='site-1', status=SiteStatusChoices.STATUS_STAGING)
+        data = serialize_for_webhook(site)
+
+        # Evaluate the conditions (status='staging')
+        self.assertFalse(eval_conditions(webhook, data))
+
+        # Change the site's status
+        site.status = SiteStatusChoices.STATUS_ACTIVE
+        data = serialize_for_webhook(site)
+
+        # Evaluate the conditions (status='active')
+        self.assertTrue(eval_conditions(webhook, data))
+
     def test_webhooks_worker(self):
 
         request_id = uuid.uuid4()

netbox/extras/webhooks_worker.py

@@ -12,15 +12,29 @@ from .webhooks import generate_signature
 logger = logging.getLogger('netbox.webhooks_worker')
 
 
+def eval_conditions(webhook, data):
+    """
+    Test whether the given data meets the conditions of the webhook (if any). Return True
+    if met or no conditions are specified.
+    """
+    if not webhook.conditions:
+        return True
+
+    logger.debug(f'Evaluating webhook conditions: {webhook.conditions}')
+    if ConditionSet(webhook.conditions).eval(data):
+        return True
+
+    return False
+
+
 @job('default')
 def process_webhook(webhook, model_name, event, data, snapshots, timestamp, username, request_id):
     """
     Make a POST request to the defined Webhook
     """
     # Evaluate webhook conditions (if any)
-    if webhook.conditions:
-        if not ConditionSet(webhook.conditions).eval(data):
-            return
+    if not eval_conditions(webhook, data):
+        return
 
     # Prepare context data for headers & body templates
     context = {

netbox/templates/extras/webhook.html

@@ -132,6 +132,18 @@
         </table>
       </div>
     </div>
+    <div class="card">
+      <h5 class="card-header">
+        Conditions
+      </h5>
+      <div class="card-body">
+        {% if object.conditions %}
+          <pre>{{ object.conditions|render_json }}</pre>
+        {% else %}
+          <p class="text-muted">None</p>
+        {% endif %}
+      </div>
+    </div>
     <div class="card">
       <h5 class="card-header">
         Additional Headers