Merge branch 'develop' into 568-csv-import-cf

netbox/extras/models.py

@@ -10,6 +10,7 @@ from django.db import models
 from django.http import HttpResponse
 from django.template import Template, Context
 from django.urls import reverse
+from django.utils.text import slugify
 from taggit.models import TagBase, GenericTaggedItemBase
 
 from utilities.fields import ColorField
@@ -952,6 +953,13 @@ class Tag(TagBase, ChangeLoggedModel):
     def get_absolute_url(self):
         return reverse('extras:tag', args=[self.slug])
 
+    def slugify(self, tag, i=None):
+        # Allow Unicode in Tag slugs (avoids empty slugs for Tags with all-Unicode names)
+        slug = slugify(tag, allow_unicode=True)
+        if i is not None:
+            slug += "_%d" % i
+        return slug
+
 
 class TaggedItem(GenericTaggedItemBase):
     tag = models.ForeignKey(

netbox/extras/tests/test_models.py

@@ -3,7 +3,7 @@ from django.test import TestCase
 
 from dcim.models import Site
 from extras.choices import TemplateLanguageChoices
-from extras.models import Graph
+from extras.models import Graph, Tag
 
 
 class GraphTest(TestCase):
@@ -44,3 +44,12 @@ class GraphTest(TestCase):
 
         self.assertEqual(graph.embed_url(self.site), RENDERED_TEXT)
         self.assertEqual(graph.embed_link(self.site), RENDERED_TEXT)
+
+
+class TagTest(TestCase):
+
+    def test_create_tag_unicode(self):
+        tag = Tag(name='Testing Unicode: 台灣')
+        tag.save()
+
+        self.assertEqual(tag.slug, 'testing-unicode-台灣')

netbox/extras/tests/test_webhooks.py

@@ -0,0 +1,89 @@
+import django_rq
+from django.contrib.contenttypes.models import ContentType
+from django.urls import reverse
+from rest_framework import status
+
+from dcim.models import Site
+from extras.choices import ObjectChangeActionChoices
+from extras.models import Webhook
+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
+
+    @classmethod
+    def setUpTestData(cls):
+
+        site_ct = ContentType.objects.get_for_model(Site)
+        PAYLOAD_URL = "http://localhost/"
+        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),
+        ))
+        for webhook in webhooks:
+            webhook.obj_type.set([site_ct])
+
+    def test_enqueue_webhook_create(self):
+
+        # Create an object via the REST API
+        data = {
+            'name': 'Test Site',
+            'slug': 'test-site',
+        }
+        url = reverse('dcim-api:site-list')
+        response = self.client.post(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_201_CREATED)
+        self.assertEqual(Site.objects.count(), 1)
+
+        # 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.args[0], Webhook.objects.get(type_create=True))
+        self.assertEqual(job.args[1]['id'], response.data['id'])
+        self.assertEqual(job.args[2], 'site')
+        self.assertEqual(job.args[3], ObjectChangeActionChoices.ACTION_CREATE)
+
+    def test_enqueue_webhook_update(self):
+
+        site = Site.objects.create(name='Site 1', slug='site-1')
+
+        # Update an object via the REST API
+        data = {
+            'comments': 'Updated the site',
+        }
+        url = reverse('dcim-api:site-detail', kwargs={'pk': site.pk})
+        response = self.client.patch(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+
+        # Verify that a job was queued for the object update webhook
+        self.assertEqual(self.queue.count, 1)
+        job = self.queue.jobs[0]
+        self.assertEqual(job.args[0], Webhook.objects.get(type_update=True))
+        self.assertEqual(job.args[1]['id'], site.pk)
+        self.assertEqual(job.args[2], 'site')
+        self.assertEqual(job.args[3], ObjectChangeActionChoices.ACTION_UPDATE)
+
+    def test_enqueue_webhook_delete(self):
+
+        site = Site.objects.create(name='Site 1', slug='site-1')
+
+        # Delete an object via the REST API
+        url = reverse('dcim-api:site-detail', kwargs={'pk': site.pk})
+        response = self.client.delete(url, **self.header)
+        self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
+
+        # Verify that a job was queued for the object update webhook
+        self.assertEqual(self.queue.count, 1)
+        job = self.queue.jobs[0]
+        self.assertEqual(job.args[0], Webhook.objects.get(type_delete=True))
+        self.assertEqual(job.args[1]['id'], site.pk)
+        self.assertEqual(job.args[2], 'site')
+        self.assertEqual(job.args[3], ObjectChangeActionChoices.ACTION_DELETE)

netbox/extras/webhooks_worker.py

@@ -6,8 +6,7 @@ import requests
 from django_rq import job
 from rest_framework.utils.encoders import JSONEncoder
 
-from .choices import ObjectChangeActionChoices
-from .constants import *
+from .choices import ObjectChangeActionChoices, WebhookContentTypeChoices
 
 
 @job('default')
@@ -35,9 +34,9 @@ def process_webhook(webhook, data, model_name, event, timestamp, username, reque
         'headers': headers
     }
 
-    if webhook.http_content_type == WEBHOOK_CT_JSON:
+    if webhook.http_content_type == WebhookContentTypeChoices.CONTENTTYPE_JSON:
         params.update({'data': json.dumps(payload, cls=JSONEncoder)})
-    elif webhook.http_content_type == WEBHOOK_CT_X_WWW_FORM_ENCODED:
+    elif webhook.http_content_type == WebhookContentTypeChoices.CONTENTTYPE_FORMDATA:
         params.update({'data': payload})
 
     prepared_request = requests.Request(**params).prepare()
@@ -61,5 +60,7 @@ def process_webhook(webhook, data, model_name, event, timestamp, username, reque
         return 'Status {} returned, webhook successfully processed.'.format(response.status_code)
     else:
         raise requests.exceptions.RequestException(
-            "Status {} returned with content '{}', webhook FAILED to process.".format(response.status_code, response.content)
+            "Status {} returned with content '{}', webhook FAILED to process.".format(
+                response.status_code, response.content
+            )
         )