Closes #5610: Add REST API endpoint for webhooks

docs/release-notes/version-2.11.md

@@ -44,6 +44,7 @@ The ObjectChange model (which is used to record the creation, modification, and
 * [#5401](https://github.com/netbox-community/netbox/issues/5401) - Extend custom field support to device component models
 * [#5451](https://github.com/netbox-community/netbox/issues/5451) - Add support for multiple-selection custom fields
 * [#5608](https://github.com/netbox-community/netbox/issues/5608) - Add REST API endpoint for custom links
+* [#5610](https://github.com/netbox-community/netbox/issues/5610) - Add REST API endpoint for webhooks
 * [#5894](https://github.com/netbox-community/netbox/issues/5894) - Use primary keys when filtering object lists by related objects in the UI
 * [#5895](https://github.com/netbox-community/netbox/issues/5895) - Rename RackGroup to Location
 * [#5901](https://github.com/netbox-community/netbox/issues/5901) - Add `created` and `last_updated` fields to device component models
@@ -89,3 +90,5 @@ The ObjectChange model (which is used to record the creation, modification, and
 * extras.ObjectChange
   * Added the `prechange_data` field
   * Renamed `object_data` to `postchange_data`
+* extras.Webhook
+  * Added the `/api/extras/webhooks/` endpoint

netbox/extras/api/nested_serializers.py

@@ -12,9 +12,18 @@ __all__ = [
     'NestedImageAttachmentSerializer',
     'NestedJobResultSerializer',
     'NestedTagSerializer',
+    'NestedWebhookSerializer',
 ]
 
 
+class NestedWebhookSerializer(WritableNestedSerializer):
+    url = serializers.HyperlinkedIdentityField(view_name='extras-api:webhook-detail')
+
+    class Meta:
+        model = models.Webhook
+        fields = ['id', 'url', 'name']
+
+
 class NestedCustomFieldSerializer(WritableNestedSerializer):
     url = serializers.HyperlinkedIdentityField(view_name='extras-api:customfield-detail')
 

netbox/extras/api/serializers.py

@@ -9,9 +9,7 @@ from dcim.api.nested_serializers import (
 )
 from dcim.models import Device, DeviceRole, Platform, Rack, Region, Site, SiteGroup
 from extras.choices import *
-from extras.models import (
+from extras.models import *
-    ConfigContext, CustomField, CustomLink, ExportTemplate, ImageAttachment, ObjectChange, JobResult, Tag,
-)
 from extras.utils import FeatureQuery
 from netbox.api import ChoiceField, ContentTypeField, SerializedPKRelatedField, ValidatedModelSerializer
 from netbox.api.exceptions import SerializerNotFound
@@ -24,6 +22,48 @@ from virtualization.models import Cluster, ClusterGroup
 from .nested_serializers import *
 
 
+__all__ = (
+    'ConfigContextSerializer',
+    'ContentTypeSerializer',
+    'CustomFieldSerializer',
+    'CustomLinkSerializer',
+    'ExportTemplateSerializer',
+    'ImageAttachmentSerializer',
+    'JobResultSerializer',
+    'ObjectChangeSerializer',
+    'ReportDetailSerializer',
+    'ReportSerializer',
+    'ScriptDetailSerializer',
+    'ScriptInputSerializer',
+    'ScriptLogMessageSerializer',
+    'ScriptOutputSerializer',
+    'ScriptSerializer',
+    'TagSerializer',
+    'TaggedObjectSerializer',
+    'WebhookSerializer',
+)
+
+
+#
+# Webhooks
+#
+
+class WebhookSerializer(ValidatedModelSerializer):
+    url = serializers.HyperlinkedIdentityField(view_name='extras-api:webhook-detail')
+    content_types = ContentTypeField(
+        queryset=ContentType.objects.filter(FeatureQuery('webhooks').get_query()),
+        many=True
+    )
+
+    class Meta:
+        model = Webhook
+        fields = [
+            'id', 'url', 'content_types', 'name', 'type_create', 'type_update', 'type_delete', 'payload_url', 'enabled',
+            'http_method', 'http_content_type', 'additional_headers', 'body_template', 'secret', 'ssl_verification',
+            'ca_file_path',
+        ]
+
+
 #
 # Custom fields
 #

netbox/extras/api/urls.py

@@ -5,6 +5,9 @@ from . import views
 router = OrderedDefaultRouter()
 router.APIRootView = views.ExtrasRootView
 
+# Webhooks
+router.register('webhooks', views.WebhookViewSet)
+
 # Custom fields
 router.register('custom-fields', views.CustomFieldViewSet)
 

netbox/extras/api/views.py

@@ -11,9 +11,7 @@ from rq import Worker
 
 from extras import filters
 from extras.choices import JobResultStatusChoices
-from extras.models import (
+from extras.models import *
-    ConfigContext, CustomLink, ExportTemplate, ImageAttachment, ObjectChange, JobResult, Tag, TaggedItem,
-)
 from extras.models import CustomField
 from extras.reports import get_report, get_reports, run_report
 from extras.scripts import get_script, get_scripts, run_script
@@ -55,6 +53,17 @@ class ConfigContextQuerySetMixin:
         return queryset.annotate_config_context_data()
 
 
+#
+# Webhooks
+#
+
+class WebhookViewSet(ModelViewSet):
+    metadata_class = ContentTypeMetadata
+    queryset = Webhook.objects.all()
+    serializer_class = serializers.WebhookSerializer
+    filterset_class = filters.WebhookFilterSet
+
+
 #
 # Custom fields
 #

netbox/extras/filters.py

@@ -9,9 +9,7 @@ from tenancy.models import Tenant, TenantGroup
 from utilities.filters import BaseFilterSet, ContentTypeFilter
 from virtualization.models import Cluster, ClusterGroup
 from .choices import *
-from .models import (
+from .models import *
-    ConfigContext, CustomField, CustomLink, ExportTemplate, ImageAttachment, JobResult, ObjectChange, Tag,
-)
 
 
 __all__ = (
@@ -26,6 +24,7 @@ __all__ = (
     'LocalConfigContextFilterSet',
     'ObjectChangeFilterSet',
     'TagFilterSet',
+    'WebhookFilterSet',
 )
 
 EXACT_FILTER_TYPES = (
@@ -36,6 +35,20 @@ EXACT_FILTER_TYPES = (
 )
 
 
+class WebhookFilterSet(BaseFilterSet):
+    content_types = ContentTypeFilter()
+    http_method = django_filters.MultipleChoiceFilter(
+        choices=WebhookHttpMethodChoices
+    )
+
+    class Meta:
+        model = Webhook
+        fields = [
+            'id', 'content_types', 'name', 'type_create', 'type_update', 'type_delete', 'payload_url', 'enabled',
+            'http_method', 'http_content_type', 'secret', 'ssl_verification', 'ca_file_path',
+        ]
+
+
 class CustomFieldFilter(django_filters.Filter):
     """
     Filter objects by the presence of a CustomFieldValue. The filter's name is used as the CustomField name.

netbox/extras/models/models.py

@@ -21,6 +21,19 @@ from utilities.querysets import RestrictedQuerySet
 from utilities.utils import deepmerge, render_jinja2
 
 
+__all__ = (
+    'ConfigContext',
+    'ConfigContextModel',
+    'CustomLink',
+    'ExportTemplate',
+    'ImageAttachment',
+    'JobResult',
+    'Report',
+    'Script',
+    'Webhook',
+)
+
+
 #
 # Webhooks
 #
@@ -109,6 +122,8 @@ class Webhook(BigIDModel):
                   'Leave blank to use the system defaults.'
     )
 
+    objects = RestrictedQuerySet.as_manager()
+
     class Meta:
         ordering = ('name',)
         unique_together = ('payload_url', 'type_create', 'type_update', 'type_delete',)

netbox/extras/tests/test_api.py

@@ -11,7 +11,7 @@ from rq import Worker
 
 from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Rack, Location, RackRole, Site
 from extras.api.views import ReportViewSet, ScriptViewSet
-from extras.models import ConfigContext, CustomField, CustomLink, ExportTemplate, ImageAttachment, Tag
+from extras.models import *
 from extras.reports import Report
 from extras.scripts import BooleanVar, IntegerVar, Script, StringVar
 from utilities.testing import APITestCase, APIViewTestCases
@@ -30,6 +30,60 @@ class AppTest(APITestCase):
         self.assertEqual(response.status_code, 200)
 
 
+class WebhookTest(APIViewTestCases.APIViewTestCase):
+    model = Webhook
+    brief_fields = ['id', 'name', 'url']
+    create_data = [
+        {
+            'content_types': ['dcim.device', 'dcim.devicetype'],
+            'name': 'Webhook 4',
+            'type_create': True,
+            'payload_url': 'http://example.com/?4',
+        },
+        {
+            'content_types': ['dcim.device', 'dcim.devicetype'],
+            'name': 'Webhook 5',
+            'type_update': True,
+            'payload_url': 'http://example.com/?5',
+        },
+        {
+            'content_types': ['dcim.device', 'dcim.devicetype'],
+            'name': 'Webhook 6',
+            'type_delete': True,
+            'payload_url': 'http://example.com/?6',
+        },
+    ]
+    bulk_update_data = {
+        'ssl_verification': False,
+    }
+
+    @classmethod
+    def setUpTestData(cls):
+        site_ct = ContentType.objects.get_for_model(Site)
+        rack_ct = ContentType.objects.get_for_model(Rack)
+
+        webhooks = (
+            Webhook(
+                name='Webhook 1',
+                type_create=True,
+                payload_url='http://example.com/?1',
+            ),
+            Webhook(
+                name='Webhook 2',
+                type_update=True,
+                payload_url='http://example.com/?1',
+            ),
+            Webhook(
+                name='Webhook 3',
+                type_delete=True,
+                payload_url='http://example.com/?1',
+            ),
+        )
+        Webhook.objects.bulk_create(webhooks)
+        for webhook in webhooks:
+            webhook.content_types.add(site_ct, rack_ct)
+
+
 class CustomFieldTest(APIViewTestCases.APIViewTestCase):
     model = CustomField
     brief_fields = ['id', 'name', 'url']

netbox/extras/tests/test_filters.py

@@ -7,12 +7,88 @@ from django.test import TestCase
 from dcim.models import DeviceRole, Platform, Rack, Region, Site, SiteGroup
 from extras.choices import ObjectChangeActionChoices
 from extras.filters import *
-from extras.models import ConfigContext, CustomLink, ExportTemplate, ImageAttachment, ObjectChange, Tag
+from extras.models import *
 from ipam.models import IPAddress
 from tenancy.models import Tenant, TenantGroup
 from virtualization.models import Cluster, ClusterGroup, ClusterType
 
 
+class WebhookTestCase(TestCase):
+    queryset = Webhook.objects.all()
+    filterset = WebhookFilterSet
+
+    @classmethod
+    def setUpTestData(cls):
+        content_types = ContentType.objects.filter(model__in=['site', 'rack', 'device'])
+
+        webhooks = (
+            Webhook(
+                name='Webhook 1',
+                type_create=True,
+                payload_url='http://example.com/?1',
+                enabled=True,
+                http_method='GET',
+                ssl_verification=True,
+            ),
+            Webhook(
+                name='Webhook 2',
+                type_update=True,
+                payload_url='http://example.com/?2',
+                enabled=True,
+                http_method='POST',
+                ssl_verification=True,
+            ),
+            Webhook(
+                name='Webhook 3',
+                type_delete=True,
+                payload_url='http://example.com/?3',
+                enabled=False,
+                http_method='PATCH',
+                ssl_verification=False,
+            ),
+        )
+        Webhook.objects.bulk_create(webhooks)
+        webhooks[0].content_types.add(content_types[0])
+        webhooks[1].content_types.add(content_types[1])
+        webhooks[2].content_types.add(content_types[2])
+
+    def test_id(self):
+        params = {'id': self.queryset.values_list('pk', flat=True)[:2]}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+
+    def test_name(self):
+        params = {'name': ['Webhook 1', 'Webhook 2']}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+
+    def test_content_types(self):
+        params = {'content_types': 'dcim.site'}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
+
+    def test_type_create(self):
+        params = {'type_create': True}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
+
+    def test_type_update(self):
+        params = {'type_update': True}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
+
+    def test_type_delete(self):
+        params = {'type_delete': True}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
+
+    def test_enabled(self):
+        params = {'enabled': True}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+
+    def test_http_method(self):
+        params = {'http_method': ['GET', 'POST']}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+
+    def test_ssl_verification(self):
+        params = {'ssl_verification': True}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+
+
 class CustomLinkTestCase(TestCase):
     queryset = CustomLink.objects.all()
     filterset = CustomLinkFilterSet