starred/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2026-04-27 03:09:31 +02:00
Code Issues 273 Packages Projects Releases 10 Wiki Activity

Closes #20003: Introduce mechanism to register callbacks for webhook context (#20025)

Browse Source 
* Closes #20003: Introduce mechanism to register callbacks for webhook context

* Swap ContentType with ObjectType

* Add plugin dev documentation for webhook callbacks

* Fix tests

* Add note about namespacing webhook data
This commit is contained in:
Jeremy Stretch
2025-08-07 16:28:53 -04:00
committed by GitHub
parent 33d891e67b
commit 37d6c160b9
10 changed files with 153 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
netbox/extras/tests/test_event_rules.py
View File

@@ -135,7 +135,7 @@ class EventRuleTest(APITestCase):
job = self.queue.jobs[0]
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(name='Event Rule 1'))
self.assertEqual(job.kwargs['event_type'], OBJECT_CREATED)
self.assertEqual(job.kwargs['model_name'], 'site')
self.assertEqual(job.kwargs['object_type'], ObjectType.objects.get_for_model(Site))
self.assertEqual(job.kwargs['data']['id'], response.data['id'])
self.assertEqual(job.kwargs['data']['foo'], 1)
self.assertEqual(len(job.kwargs['data']['tags']), len(response.data['tags']))
@@ -186,7 +186,7 @@ class EventRuleTest(APITestCase):
for i, job in enumerate(self.queue.jobs):
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(name='Event Rule 1'))
self.assertEqual(job.kwargs['event_type'], OBJECT_CREATED)
self.assertEqual(job.kwargs['model_name'], 'site')
self.assertEqual(job.kwargs['object_type'], ObjectType.objects.get_for_model(Site))
self.assertEqual(job.kwargs['data']['id'], response.data[i]['id'])
self.assertEqual(job.kwargs['data']['foo'], 1)
self.assertEqual(len(job.kwargs['data']['tags']), len(response.data[i]['tags']))
@@ -218,7 +218,7 @@ class EventRuleTest(APITestCase):
job = self.queue.jobs[0]
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(name='Event Rule 2'))
self.assertEqual(job.kwargs['event_type'], OBJECT_UPDATED)
self.assertEqual(job.kwargs['model_name'], 'site')
self.assertEqual(job.kwargs['object_type'], ObjectType.objects.get_for_model(Site))
self.assertEqual(job.kwargs['data']['id'], site.pk)
self.assertEqual(job.kwargs['data']['foo'], 2)
self.assertEqual(len(job.kwargs['data']['tags']), len(response.data['tags']))
@@ -275,7 +275,7 @@ class EventRuleTest(APITestCase):
for i, job in enumerate(self.queue.jobs):
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(name='Event Rule 2'))
self.assertEqual(job.kwargs['event_type'], OBJECT_UPDATED)
self.assertEqual(job.kwargs['model_name'], 'site')
self.assertEqual(job.kwargs['object_type'], ObjectType.objects.get_for_model(Site))
self.assertEqual(job.kwargs['data']['id'], data[i]['id'])
self.assertEqual(job.kwargs['data']['foo'], 2)
self.assertEqual(len(job.kwargs['data']['tags']), len(response.data[i]['tags']))
@@ -302,7 +302,7 @@ class EventRuleTest(APITestCase):
job = self.queue.jobs[0]
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(name='Event Rule 3'))
self.assertEqual(job.kwargs['event_type'], OBJECT_DELETED)
self.assertEqual(job.kwargs['model_name'], 'site')
self.assertEqual(job.kwargs['object_type'], ObjectType.objects.get_for_model(Site))
self.assertEqual(job.kwargs['data']['id'], site.pk)
self.assertEqual(job.kwargs['data']['foo'], 3)
self.assertEqual(job.kwargs['snapshots']['prechange']['name'], 'Site 1')
@@ -336,7 +336,7 @@ class EventRuleTest(APITestCase):
for i, job in enumerate(self.queue.jobs):
self.assertEqual(job.kwargs['event_rule'], EventRule.objects.get(name='Event Rule 3'))
self.assertEqual(job.kwargs['event_type'], OBJECT_DELETED)
self.assertEqual(job.kwargs['model_name'], 'site')
self.assertEqual(job.kwargs['object_type'], ObjectType.objects.get_for_model(Site))
self.assertEqual(job.kwargs['data']['id'], sites[i].pk)
self.assertEqual(job.kwargs['data']['foo'], 3)
self.assertEqual(job.kwargs['snapshots']['prechange']['name'], sites[i].name)
@@ -368,18 +368,23 @@ class EventRuleTest(APITestCase):
self.assertEqual(body['request_id'], str(request_id))
self.assertEqual(body['data']['name'], 'Site 1')
self.assertEqual(body['data']['foo'], 1)
self.assertEqual(body['context']['foo'], 123) # From netbox.tests.dummy_plugin
return HttpResponse()
# Create a dummy request
request = RequestFactory().get(reverse('dcim:site_add'))
request.id = request_id
request.user = self.user
# Enqueue a webhook for processing
webhooks_queue = {}
site = Site.objects.create(name='Site 1', slug='site-1')
enqueue_event(
webhooks_queue,
instance=site,
user=self.user,
request_id=request_id,
event_type=OBJECT_CREATED
request=request,
event_type=OBJECT_CREATED,
)
flush_events(list(webhooks_queue.values()))