Add REST API endpoint for ObjectPermissions

netbox/netbox/urls.py

@@ -65,6 +65,7 @@ _patterns = [
     path('api/ipam/', include('ipam.api.urls')),
     path('api/secrets/', include('secrets.api.urls')),
     path('api/tenancy/', include('tenancy.api.urls')),
+    path('api/users/', include('users.api.urls')),
     path('api/virtualization/', include('virtualization.api.urls')),
     path('api/docs/', schema_view.with_ui('swagger'), name='api_docs'),
     path('api/redoc/', schema_view.with_ui('redoc'), name='api_redocs'),

netbox/netbox/views.py

@@ -343,5 +343,6 @@ class APIRootView(APIView):
             ('plugins', reverse('plugins-api:api-root', request=request, format=format)),
             ('secrets', reverse('secrets-api:api-root', request=request, format=format)),
             ('tenancy', reverse('tenancy-api:api-root', request=request, format=format)),
+            ('users', reverse('users-api:api-root', request=request, format=format)),
             ('virtualization', reverse('virtualization-api:api-root', request=request, format=format)),
         )))

netbox/users/api/nested_serializers.py

@@ -1,4 +1,4 @@
-from django.contrib.auth.models import User
+from django.contrib.auth.models import Group, User
 
 from utilities.api import WritableNestedSerializer
 
@@ -8,9 +8,16 @@ _all_ = [
 
 
 #
-# Users
+# Groups and users
 #
 
+class NestedGroupSerializer(WritableNestedSerializer):
+
+    class Meta:
+        model = Group
+        fields = ['id', 'name']
+
+
 class NestedUserSerializer(WritableNestedSerializer):
 
     class Meta:

netbox/users/api/serializers.py

@@ -1,4 +1,28 @@
+from django.contrib.contenttypes.models import ContentType
+
+from users.models import ObjectPermission
+from utilities.api import ContentTypeField, SerializedPKRelatedField, ValidatedModelSerializer
 from .nested_serializers import *
 
 
-# Placeholder for future serializers
+class ObjectPermissionSerializer(ValidatedModelSerializer):
+    object_types = ContentTypeField(
+        queryset=ContentType.objects.all(),
+        many=True
+    )
+    groups = SerializedPKRelatedField(
+        queryset=Group.objects.all(),
+        serializer=NestedGroupSerializer,
+        required=False,
+        many=True
+    )
+    users = SerializedPKRelatedField(
+        queryset=User.objects.all(),
+        serializer=NestedUserSerializer,
+        required=False,
+        many=True
+    )
+
+    class Meta:
+        model = ObjectPermission
+        fields = ('id', 'object_types', 'groups', 'users', 'actions', 'constraints')

netbox/users/api/urls.py

@@ -0,0 +1,21 @@
+from rest_framework import routers
+
+from . import views
+
+
+class UsersRootView(routers.APIRootView):
+    """
+    Users API root view
+    """
+    def get_view_name(self):
+        return 'Users'
+
+
+router = routers.DefaultRouter()
+router.APIRootView = UsersRootView
+
+# Permissions
+router.register('permissions', views.ObjectPermissionViewSet)
+
+app_name = 'users-api'
+urlpatterns = router.urls

netbox/users/api/views.py

@@ -0,0 +1,14 @@
+from utilities.api import ModelViewSet
+from . import serializers
+
+from users.models import ObjectPermission
+
+
+#
+# ObjectPermissions
+#
+
+class ObjectPermissionViewSet(ModelViewSet):
+    queryset = ObjectPermission.objects.prefetch_related('object_types', 'groups', 'users')
+    serializer_class = serializers.ObjectPermissionSerializer
+    # filterset_class = filters.ObjectPermissionFilterSet

netbox/users/models.py

@@ -233,16 +233,6 @@ class ObjectPermission(models.Model):
     A mapping of view, add, change, and/or delete permission for users and/or groups to an arbitrary set of objects
     identified by ORM query parameters.
     """
-    users = models.ManyToManyField(
-        to=User,
-        blank=True,
-        related_name='object_permissions'
-    )
-    groups = models.ManyToManyField(
-        to=Group,
-        blank=True,
-        related_name='object_permissions'
-    )
     object_types = models.ManyToManyField(
         to=ContentType,
         limit_choices_to={
@@ -252,15 +242,25 @@ class ObjectPermission(models.Model):
         },
         related_name='object_permissions'
     )
-    constraints = JSONField(
+    groups = models.ManyToManyField(
+        to=Group,
         blank=True,
-        null=True,
+        related_name='object_permissions'
-        help_text="Queryset filter matching the applicable objects of the selected type(s)"
+    )
+    users = models.ManyToManyField(
+        to=User,
+        blank=True,
+        related_name='object_permissions'
     )
     actions = ArrayField(
         base_field=models.CharField(max_length=30),
         help_text="The list of actions granted by this permission"
     )
+    constraints = JSONField(
+        blank=True,
+        null=True,
+        help_text="Queryset filter matching the applicable objects of the selected type(s)"
+    )
 
     class Meta:
         verbose_name = "Permission"

netbox/users/tests/test_api.py

@@ -0,0 +1,144 @@
+from django.contrib.auth.models import Group, User
+from django.contrib.contenttypes.models import ContentType
+from django.urls import reverse
+from rest_framework import status
+
+from users.models import ObjectPermission
+from utilities.testing import APITestCase
+
+
+class AppTest(APITestCase):
+
+    def test_root(self):
+
+        url = reverse('users-api:api-root')
+        response = self.client.get('{}?format=api'.format(url), **self.header)
+
+        self.assertEqual(response.status_code, 200)
+
+
+class ObjectPermissionTest(APITestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+
+        groups = (
+            Group(name='Group 1'),
+            Group(name='Group 2'),
+            Group(name='Group 3'),
+        )
+        Group.objects.bulk_create(groups)
+
+        users = (
+            User(username='User 1', is_active=True),
+            User(username='User 2', is_active=True),
+            User(username='User 3', is_active=True),
+        )
+        User.objects.bulk_create(users)
+
+        object_type = ContentType.objects.get(app_label='dcim', model='device')
+
+        for i in range(0, 3):
+            objectpermission = ObjectPermission(
+                actions=['view', 'add', 'change', 'delete'],
+                constraints={'name': f'TEST{i+1}'}
+            )
+            objectpermission.save()
+            objectpermission.object_types.add(object_type)
+            objectpermission.groups.add(groups[i])
+            objectpermission.users.add(users[i])
+
+    def test_get_objectpermission(self):
+        objectpermission = ObjectPermission.objects.first()
+        url = reverse('users-api:objectpermission-detail', kwargs={'pk': objectpermission.pk})
+        response = self.client.get(url, **self.header)
+
+        self.assertEqual(response.data['id'], objectpermission.pk)
+
+    def test_list_objectpermissions(self):
+        url = reverse('users-api:objectpermission-list')
+        response = self.client.get(url, **self.header)
+
+        self.assertEqual(response.data['count'], ObjectPermission.objects.count())
+
+    def test_create_objectpermission(self):
+        data = {
+            'object_types': ['dcim.site'],
+            'groups': [Group.objects.first().pk],
+            'users': [User.objects.first().pk],
+            'actions': ['view', 'add', 'change', 'delete'],
+            'constraints': {'name': 'TEST4'},
+        }
+
+        url = reverse('users-api:objectpermission-list')
+        response = self.client.post(url, data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_201_CREATED)
+        self.assertEqual(ObjectPermission.objects.count(), 4)
+        objectpermission = ObjectPermission.objects.get(pk=response.data['id'])
+        self.assertEqual(objectpermission.groups.first().pk, data['groups'][0])
+        self.assertEqual(objectpermission.users.first().pk, data['users'][0])
+        self.assertEqual(objectpermission.actions, data['actions'])
+        self.assertEqual(objectpermission.constraints, data['constraints'])
+
+    def test_create_objectpermission_bulk(self):
+        groups = Group.objects.all()[:3]
+        users = User.objects.all()[:3]
+        data = [
+            {
+                'object_types': ['dcim.site'],
+                'groups': [groups[0].pk],
+                'users': [users[0].pk],
+                'actions': ['view', 'add', 'change', 'delete'],
+                'constraints': {'name': 'TEST4'},
+            },
+            {
+                'object_types': ['dcim.site'],
+                'groups': [groups[1].pk],
+                'users': [users[1].pk],
+                'actions': ['view', 'add', 'change', 'delete'],
+                'constraints': {'name': 'TEST5'},
+            },
+            {
+                'object_types': ['dcim.site'],
+                'groups': [groups[2].pk],
+                'users': [users[2].pk],
+                'actions': ['view', 'add', 'change', 'delete'],
+                'constraints': {'name': 'TEST6'},
+            },
+        ]
+
+        url = reverse('users-api:objectpermission-list')
+        response = self.client.post(url, data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_201_CREATED)
+        self.assertEqual(ObjectPermission.objects.count(), 6)
+
+    def test_update_objectpermission(self):
+        objectpermission = ObjectPermission.objects.first()
+        data = {
+            'object_types': ['dcim.site', 'dcim.device'],
+            'groups': [g.pk for g in Group.objects.all()[:2]],
+            'users': [u.pk for u in User.objects.all()[:2]],
+            'actions': ['view'],
+            'constraints': {'name': 'TEST'},
+        }
+
+        url = reverse('users-api:objectpermission-detail', kwargs={'pk': objectpermission.pk})
+        response = self.client.put(url, data, format='json', **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertEqual(ObjectPermission.objects.count(), 3)
+        objectpermission = ObjectPermission.objects.get(pk=response.data['id'])
+        self.assertEqual(objectpermission.groups.first().pk, data['groups'][0])
+        self.assertEqual(objectpermission.users.first().pk, data['users'][0])
+        self.assertEqual(objectpermission.actions, data['actions'])
+        self.assertEqual(objectpermission.constraints, data['constraints'])
+
+    def test_delete_objectpermission(self):
+        objectpermission = ObjectPermission.objects.first()
+        url = reverse('users-api:objectpermission-detail', kwargs={'pk': objectpermission.pk})
+        response = self.client.delete(url, **self.header)
+
+        self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
+        self.assertEqual(ObjectPermission.objects.count(), 2)