mirror/netbox-community-netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2024-05-10 07:54:54 +00:00
Code Activity

Extend webhook create/update/delete tests

Browse Source
This commit is contained in:
jeremystretch
2021-06-01 09:04:01 -04:00
parent 3d1e4fde81
commit c88dcef900
2 changed files with 30 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
netbox/extras/signals.py
View File

@ -60,6 +60,7 @@ def _handle_changed_object(request, webhook_queue, sender, instance, **kwargs):
if m2m_changed and webhook_queue:
# TODO: Need more validation here
# TODO: Need to account for snapshot changes
instance.refresh_from_db() # Ensure that we're working with fresh M2M assignments
webhook_queue[-1]['data'] = serialize_for_webhook(instance)
else:
enqueue_object(webhook_queue, instance, request.user, request.id, action)

39
netbox/extras/tests/test_webhooks.py
View File

@ -11,7 +11,7 @@ from rest_framework import status
from dcim.models import Site
from extras.choices import ObjectChangeActionChoices
from extras.models import Webhook
from extras.models import Tag, Webhook
from extras.webhooks import enqueue_object, generate_signature
from extras.webhooks_worker import process_webhook
from utilities.testing import APITestCase
@ -20,11 +20,10 @@ from utilities.testing import APITestCase
class WebhookTest(APITestCase):
def setUp(self):
super().setUp()
self.queue = django_rq.get_queue('default')
self.queue.empty() # Begin each test with an empty queue
self.queue.empty()
@classmethod
def setUpTestData(cls):
@ -34,38 +33,55 @@ class WebhookTest(APITestCase):
DUMMY_SECRET = "LOOKATMEIMASECRETSTRING"
webhooks = Webhook.objects.bulk_create((
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),
Webhook(name='Webhook 1', type_create=True, payload_url=DUMMY_URL, secret=DUMMY_SECRET, additional_headers='X-Foo: Bar'),
Webhook(name='Webhook 2', type_update=True, payload_url=DUMMY_URL, secret=DUMMY_SECRET),
Webhook(name='Webhook 3', type_delete=True, payload_url=DUMMY_URL, secret=DUMMY_SECRET),
))
for webhook in webhooks:
webhook.content_types.set([site_ct])
Tag.objects.bulk_create((
Tag(name='Foo', slug='foo'),
Tag(name='Bar', slug='bar'),
Tag(name='Baz', slug='baz'),
))
def test_enqueue_webhook_create(self):
# Create an object via the REST API
data = {
'name': 'Test Site',
'slug': 'test-site',
'name': 'Site 1',
'slug': 'site-1',
'tags': [
{'name': 'Foo'},
{'name': 'Bar'},
]
}
url = reverse('dcim-api:site-list')
self.add_permissions('dcim.add_site')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(Site.objects.count(), 1)
self.assertEqual(Site.objects.first().tags.count(), 2)
# Verify that a job was queued for the object creation webhook
self.assertEqual(self.queue.count, 1)
job = self.queue.jobs[0]
self.assertEqual(job.kwargs['webhook'], Webhook.objects.get(type_create=True))
self.assertEqual(job.kwargs['data']['id'], response.data['id'])
self.assertEqual(len(job.kwargs['data']['tags']), len(response.data['tags']))
self.assertEqual(job.kwargs['model_name'], 'site')
self.assertEqual(job.kwargs['event'], ObjectChangeActionChoices.ACTION_CREATE)
def test_enqueue_webhook_update(self):
# Update an object via the REST API
site = Site.objects.create(name='Site 1', slug='site-1')
site.tags.set(*Tag.objects.filter(name__in=['Foo', 'Bar']))
# Update an object via the REST API
data = {
'comments': 'Updated the site',
'tags': [
{'name': 'Baz'}
]
}
url = reverse('dcim-api:site-detail', kwargs={'pk': site.pk})
self.add_permissions('dcim.change_site')
@ -77,12 +93,15 @@ class WebhookTest(APITestCase):
job = self.queue.jobs[0]
self.assertEqual(job.kwargs['webhook'], Webhook.objects.get(type_update=True))
self.assertEqual(job.kwargs['data']['id'], site.pk)
self.assertEqual(len(job.kwargs['data']['tags']), len(response.data['tags']))
self.assertEqual(job.kwargs['model_name'], 'site')
self.assertEqual(job.kwargs['event'], ObjectChangeActionChoices.ACTION_UPDATE)
def test_enqueue_webhook_delete(self):
# Delete an object via the REST API
site = Site.objects.create(name='Site 1', slug='site-1')
site.tags.set(*Tag.objects.filter(name__in=['Foo', 'Bar']))
# Delete an object via the REST API
url = reverse('dcim-api:site-detail', kwargs={'pk': site.pk})
self.add_permissions('dcim.delete_site')
response = self.client.delete(url, **self.header)