Parcourir la source

Fixes #16808: Correct event type of webhooks emitted upon object deletion immediately following a modification

Jeremy Stretch il y a 1 an
Parent
commit
7564f6f538
2 fichiers modifiés avec 29 ajouts et 3 suppressions
  1. 3 0
      netbox/extras/events.py
  2. 26 3
      netbox/extras/tests/test_event_rules.py

+ 3 - 0
netbox/extras/events.py

@@ -63,6 +63,9 @@ def enqueue_object(queue, instance, user, request_id, action):
     if key in queue:
         queue[key]['data'] = serialize_for_event(instance)
         queue[key]['snapshots']['postchange'] = get_snapshots(instance, action)['postchange']
+        # If the object is being deleted, update any prior "update" event to "delete"
+        if action == ObjectChangeActionChoices.ACTION_DELETE:
+            queue[key]['event'] = action
     else:
         queue[key] = {
             'content_type': ContentType.objects.get_for_model(instance),

+ 26 - 3
netbox/extras/tests/test_event_rules.py

@@ -390,13 +390,36 @@ class EventRuleTest(APITestCase):
         request.id = uuid.uuid4()
         request.user = self.user
 
-        self.assertEqual(self.queue.count, 0, msg="Unexpected jobs found in queue")
-
+        # Test create & update
         with event_tracking(request):
             site = Site(name='Site 1', slug='site-1')
             site.save()
+            site.description = 'foo'
+            site.save()
+        self.assertEqual(self.queue.count, 1, msg="Duplicate jobs found in queue")
+        job = self.queue.get_jobs()[0]
+        self.assertEqual(job.kwargs['event'], ObjectChangeActionChoices.ACTION_CREATE)
+        self.queue.empty()
 
-            # Save the site a second time
+        # Test multiple updates
+        site = Site.objects.create(name='Site 2', slug='site-2')
+        with event_tracking(request):
+            site.description = 'foo'
+            site.save()
+            site.description = 'bar'
             site.save()
+        self.assertEqual(self.queue.count, 1, msg="Duplicate jobs found in queue")
+        job = self.queue.get_jobs()[0]
+        self.assertEqual(job.kwargs['event'], ObjectChangeActionChoices.ACTION_UPDATE)
+        self.queue.empty()
 
+        # Test update & delete
+        site = Site.objects.create(name='Site 3', slug='site-3')
+        with event_tracking(request):
+            site.description = 'foo'
+            site.save()
+            site.delete()
         self.assertEqual(self.queue.count, 1, msg="Duplicate jobs found in queue")
+        job = self.queue.get_jobs()[0]
+        self.assertEqual(job.kwargs['event'], ObjectChangeActionChoices.ACTION_DELETE)
+        self.queue.empty()