diff --git a/netbox/utilities/testing/api.py b/netbox/utilities/testing/api.py index d9688b6e2..fb8b0d795 100644 --- a/netbox/utilities/testing/api.py +++ b/netbox/utilities/testing/api.py @@ -1,10 +1,12 @@ from django.contrib.auth.models import User +from django.contrib.contenttypes.models import ContentType from django.urls import reverse from django.test import override_settings from rest_framework import status from rest_framework.test import APIClient -from users.models import Token +from users.models import ObjectPermission, Token +from .utils import disable_warnings from .views import TestCase @@ -26,7 +28,9 @@ class APITestCase(TestCase): """ Create a superuser and token for API calls. """ - self.user = User.objects.create(username='testuser', is_superuser=True) + # Create the test user and assign permissions + self.user = User.objects.create_user(username='testuser') + self.add_permissions(*self.user_permissions) self.token = Token.objects.create(user=self.user) self.header = {'HTTP_AUTHORIZATION': 'Token {}'.format(self.token.key)} @@ -43,32 +47,70 @@ class APIViewTestCases: class GetObjectViewTestCase(APITestCase): + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) + def test_get_object_anonymous(self): + """ + GET a single object as an unauthenticated user. + """ + url = self._get_detail_url(self.model.objects.first()) + response = self.client.get(url, **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + + @override_settings(EXEMPT_VIEW_PERMISSIONS=[]) + def test_get_object_without_permission(self): + """ + GET a single object as an authenticated user without the required permission. + """ + url = self._get_detail_url(self.model.objects.first()) + + # Try GET without permission + with disable_warnings('django.request'): + self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN) + @override_settings(EXEMPT_VIEW_PERMISSIONS=[]) def test_get_object(self): """ - GET a single object identified by its numeric ID. + GET a single object as an authenticated user with permission to view the object. """ - instance = self.model.objects.first() - url = self._get_detail_url(instance) - response = self.client.get(url, **self.header) + self.assertGreaterEqual(self.model.objects.count(), 2, + f"Test requires the creation of at least two {self.model} instances") + instance1, instance2 = self.model.objects.all()[:2] - self.assertEqual(response.data['id'], instance.pk) + # Add object-level permission + obj_perm = ObjectPermission( + constraints={'pk': instance1.pk}, + actions=['view'] + ) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ContentType.objects.get_for_model(self.model)) + + # Try GET to permitted object + url = self._get_detail_url(instance1) + self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_200_OK) + + # Try GET to non-permitted object + url = self._get_detail_url(instance2) + self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_404_NOT_FOUND) class ListObjectsViewTestCase(APITestCase): brief_fields = [] - def test_list_objects(self): + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) + def test_list_objects_anonymous(self): """ - GET a list of objects. + GET a list of objects as an unauthenticated user. """ url = self._get_list_url() response = self.client.get(url, **self.header) self.assertEqual(len(response.data['results']), self.model.objects.count()) + self.assertHttpStatus(response, status.HTTP_200_OK) + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) def test_list_objects_brief(self): """ - GET a list of objects using the "brief" parameter. + GET a list of objects using the "brief" parameter as an unauthenticated user. """ url = f'{self._get_list_url()}?brief=1' response = self.client.get(url, **self.header) @@ -76,35 +118,108 @@ class APIViewTestCases: self.assertEqual(len(response.data['results']), self.model.objects.count()) self.assertEqual(sorted(response.data['results'][0]), self.brief_fields) + @override_settings(EXEMPT_VIEW_PERMISSIONS=[]) + def test_list_objects_without_permission(self): + """ + GET a list of objects as an authenticated user without the required permission. + """ + url = self._get_list_url() + + # Try GET without permission + with disable_warnings('django.request'): + self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_403_FORBIDDEN) + + @override_settings(EXEMPT_VIEW_PERMISSIONS=[]) + def test_list_objects(self): + """ + GET a list of objects as an authenticated user with permission to view the objects. + """ + self.assertGreaterEqual(self.model.objects.count(), 3, + f"Test requires the creation of at least three {self.model} instances") + instance1, instance2 = self.model.objects.all()[:2] + + # Add object-level permission + obj_perm = ObjectPermission( + constraints={'pk__in': [instance1.pk, instance2.pk]}, + actions=['view'] + ) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ContentType.objects.get_for_model(self.model)) + + # Try GET to permitted objects + response = self.client.get(self._get_list_url(), **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + self.assertEqual(len(response.data['results']), 2) + class CreateObjectViewTestCase(APITestCase): create_data = [] + def test_create_object_without_permission(self): + """ + POST a single object without permission. + """ + url = self._get_list_url() + + # Try POST without permission + with disable_warnings('django.request'): + response = self.client.post(url, self.create_data[0], format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN) + def test_create_object(self): """ - POST a single object. + POST a single object with permission. """ initial_count = self.model.objects.count() url = self._get_list_url() - response = self.client.post(url, self.create_data[0], format='json', **self.header) + # Add object-level permission + obj_perm = ObjectPermission( + actions=['add'] + ) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ContentType.objects.get_for_model(self.model)) + + response = self.client.post(url, self.create_data[0], format='json', **self.header) self.assertHttpStatus(response, status.HTTP_201_CREATED) self.assertEqual(self.model.objects.count(), initial_count + 1) self.assertInstanceEqual(self.model.objects.get(pk=response.data['id']), self.create_data[0], api=True) - def test_bulk_create_object(self): + def test_bulk_create_objects(self): """ POST a set of objects in a single request. """ initial_count = self.model.objects.count() url = self._get_list_url() - response = self.client.post(url, self.create_data, format='json', **self.header) + # Add object-level permission + obj_perm = ObjectPermission( + actions=['add'] + ) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ContentType.objects.get_for_model(self.model)) + + response = self.client.post(url, self.create_data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_201_CREATED) self.assertEqual(self.model.objects.count(), initial_count + len(self.create_data)) class UpdateObjectViewTestCase(APITestCase): update_data = {} + def test_update_object_without_permission(self): + """ + PATCH a single object without permission. + """ + url = self._get_detail_url(self.model.objects.first()) + update_data = self.update_data or getattr(self, 'create_data')[0] + + # Try PATCH without permission + with disable_warnings('django.request'): + response = self.client.patch(url, update_data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN) + def test_update_object(self): """ PATCH a single object identified by its numeric ID. @@ -112,22 +227,49 @@ class APIViewTestCases: instance = self.model.objects.first() url = self._get_detail_url(instance) update_data = self.update_data or getattr(self, 'create_data')[0] - response = self.client.patch(url, update_data, format='json', **self.header) + # Add object-level permission + obj_perm = ObjectPermission( + actions=['change'] + ) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ContentType.objects.get_for_model(self.model)) + + response = self.client.patch(url, update_data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_200_OK) instance.refresh_from_db() self.assertInstanceEqual(instance, self.update_data, api=True) class DeleteObjectViewTestCase(APITestCase): + def test_delete_object_without_permission(self): + """ + DELETE a single object without permission. + """ + url = self._get_detail_url(self.model.objects.first()) + + # Try DELETE without permission + with disable_warnings('django.request'): + response = self.client.delete(url, **self.header) + self.assertHttpStatus(response, status.HTTP_403_FORBIDDEN) + def test_delete_object(self): """ DELETE a single object identified by its numeric ID. """ instance = self.model.objects.first() url = self._get_detail_url(instance) - response = self.client.delete(url, **self.header) + # Add object-level permission + obj_perm = ObjectPermission( + actions=['delete'] + ) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ContentType.objects.get_for_model(self.model)) + + response = self.client.delete(url, **self.header) self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT) self.assertFalse(self.model.objects.filter(pk=instance.pk).exists())