mirror of
https://github.com/netbox-community/netbox.git
synced 2025-07-27 02:48:38 -06:00
Finish adding tests for changelog message functionality
This commit is contained in:
Jeremy Stretch
parent
7470147c1a
commit
68d39065ce
@ -1,3 +1,4 @@
|
||||
import copy
|
||||
import inspect
|
||||
import json
|
||||
|
||||
@ -15,10 +16,11 @@ from strawberry.types.union import StrawberryUnion
|
||||
from core.choices import ObjectChangeActionChoices
|
||||
from core.models import ObjectChange, ObjectType
|
||||
from ipam.graphql.types import IPAddressFamilyType
|
||||
from netbox.models.features import ChangeLoggingMixin
|
||||
from users.models import ObjectPermission, Token, User
|
||||
from utilities.api import get_graphql_type_for_model
|
||||
from .base import ModelTestCase
|
||||
from .utils import disable_logging, disable_warnings
|
||||
from .utils import disable_logging, disable_warnings, get_random_string
|
||||
|
||||
__all__ = (
|
||||
'APITestCase',
|
||||
@ -223,8 +225,14 @@ class APIViewTestCases:
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
|
||||
|
||||
data = copy.deepcopy(self.create_data[0])
|
||||
|
||||
# If supported, add a changelog message
|
||||
if issubclass(self.model, ChangeLoggingMixin):
|
||||
data['changelog_message'] = get_random_string(10)
|
||||
|
||||
initial_count = self._get_queryset().count()
|
||||
response = self.client.post(self._get_list_url(), self.create_data[0], format='json', **self.header)
|
||||
response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
||||
self.assertEqual(self._get_queryset().count(), initial_count + 1)
|
||||
instance = self._get_queryset().get(pk=response.data['id'])
|
||||
@ -236,13 +244,13 @@ class APIViewTestCases:
|
||||
)
|
||||
|
||||
# Verify ObjectChange creation
|
||||
if hasattr(self.model, 'to_objectchange'):
|
||||
objectchanges = ObjectChange.objects.filter(
|
||||
if issubclass(self.model, ChangeLoggingMixin):
|
||||
objectchange = ObjectChange.objects.get(
|
||||
changed_object_type=ContentType.objects.get_for_model(instance),
|
||||
changed_object_id=instance.pk
|
||||
)
|
||||
self.assertEqual(len(objectchanges), 1)
|
||||
self.assertEqual(objectchanges[0].action, ObjectChangeActionChoices.ACTION_CREATE)
|
||||
self.assertEqual(objectchange.action, ObjectChangeActionChoices.ACTION_CREATE)
|
||||
self.assertEqual(objectchange.message, data['changelog_message'])
|
||||
|
||||
def test_bulk_create_objects(self):
|
||||
"""
|
||||
@ -257,6 +265,12 @@ class APIViewTestCases:
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
|
||||
|
||||
# If supported, add a changelog message
|
||||
changelog_message = get_random_string(10)
|
||||
if issubclass(self.model, ChangeLoggingMixin):
|
||||
for obj_data in self.create_data:
|
||||
obj_data['changelog_message'] = changelog_message
|
||||
|
||||
initial_count = self._get_queryset().count()
|
||||
response = self.client.post(self._get_list_url(), self.create_data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
||||
@ -264,6 +278,9 @@ class APIViewTestCases:
|
||||
self.assertEqual(self._get_queryset().count(), initial_count + len(self.create_data))
|
||||
for i, obj in enumerate(response.data):
|
||||
for field in self.create_data[i]:
|
||||
if field == 'changelog_message':
|
||||
# Write-only field
|
||||
continue
|
||||
if field not in self.validation_excluded_fields:
|
||||
self.assertIn(field, obj, f"Bulk create field '{field}' missing from object {i} in response")
|
||||
for i, obj in enumerate(response.data):
|
||||
@ -274,6 +291,20 @@ class APIViewTestCases:
|
||||
api=True
|
||||
)
|
||||
|
||||
# Verify ObjectChange creation
|
||||
if issubclass(self.model, ChangeLoggingMixin):
|
||||
id_list = [
|
||||
obj['id'] for obj in response.data
|
||||
]
|
||||
objectchanges = ObjectChange.objects.filter(
|
||||
changed_object_type=ContentType.objects.get_for_model(self.model),
|
||||
changed_object_id__in=id_list
|
||||
)
|
||||
self.assertEqual(len(objectchanges), len(self.create_data))
|
||||
for oc in objectchanges:
|
||||
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_CREATE)
|
||||
self.assertEqual(oc.message, changelog_message)
|
||||
|
||||
class UpdateObjectViewTestCase(APITestCase):
|
||||
update_data = {}
|
||||
bulk_update_data = None
|
||||
@ -308,24 +339,30 @@ class APIViewTestCases:
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
|
||||
|
||||
response = self.client.patch(url, update_data, format='json', **self.header)
|
||||
data = copy.deepcopy(update_data)
|
||||
|
||||
# If supported, add a changelog message
|
||||
if issubclass(self.model, ChangeLoggingMixin):
|
||||
data['changelog_message'] = get_random_string(10)
|
||||
|
||||
response = self.client.patch(url, data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
instance.refresh_from_db()
|
||||
self.assertInstanceEqual(
|
||||
instance,
|
||||
update_data,
|
||||
data,
|
||||
exclude=self.validation_excluded_fields,
|
||||
api=True
|
||||
)
|
||||
|
||||
# Verify ObjectChange creation
|
||||
if hasattr(self.model, 'to_objectchange'):
|
||||
objectchanges = ObjectChange.objects.filter(
|
||||
objectchange = ObjectChange.objects.get(
|
||||
changed_object_type=ContentType.objects.get_for_model(instance),
|
||||
changed_object_id=instance.pk
|
||||
)
|
||||
self.assertEqual(len(objectchanges), 1)
|
||||
self.assertEqual(objectchanges[0].action, ObjectChangeActionChoices.ACTION_UPDATE)
|
||||
self.assertEqual(objectchange.action, ObjectChangeActionChoices.ACTION_UPDATE)
|
||||
self.assertEqual(objectchange.message, data['changelog_message'])
|
||||
|
||||
def test_bulk_update_objects(self):
|
||||
"""
|
||||
@ -349,14 +386,34 @@ class APIViewTestCases:
|
||||
{'id': id, **self.bulk_update_data} for id in id_list
|
||||
]
|
||||
|
||||
# If supported, add a changelog message
|
||||
changelog_message = get_random_string(10)
|
||||
if issubclass(self.model, ChangeLoggingMixin):
|
||||
for obj_data in data:
|
||||
obj_data['changelog_message'] = changelog_message
|
||||
|
||||
response = self.client.patch(self._get_list_url(), data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
for i, obj in enumerate(response.data):
|
||||
for field in self.bulk_update_data:
|
||||
if field == 'changelog_data':
|
||||
# Write-only field
|
||||
continue
|
||||
self.assertIn(field, obj, f"Bulk update field '{field}' missing from object {i} in response")
|
||||
for instance in self._get_queryset().filter(pk__in=id_list):
|
||||
self.assertInstanceEqual(instance, self.bulk_update_data, api=True)
|
||||
|
||||
# Verify ObjectChange creation
|
||||
if issubclass(self.model, ChangeLoggingMixin):
|
||||
objectchanges = ObjectChange.objects.filter(
|
||||
changed_object_type=ContentType.objects.get_for_model(self.model),
|
||||
changed_object_id__in=id_list
|
||||
)
|
||||
self.assertEqual(len(objectchanges), len(data))
|
||||
for oc in objectchanges:
|
||||
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_UPDATE)
|
||||
self.assertEqual(oc.message, changelog_message)
|
||||
|
||||
class DeleteObjectViewTestCase(APITestCase):
|
||||
|
||||
def test_delete_object_without_permission(self):
|
||||
@ -386,18 +443,24 @@ class APIViewTestCases:
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
|
||||
|
||||
response = self.client.delete(url, **self.header)
|
||||
data = {}
|
||||
|
||||
# If supported, add a changelog message
|
||||
if issubclass(self.model, ChangeLoggingMixin):
|
||||
data['changelog_message'] = get_random_string(10)
|
||||
|
||||
response = self.client.delete(url, data, **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
|
||||
self.assertFalse(self._get_queryset().filter(pk=instance.pk).exists())
|
||||
|
||||
# Verify ObjectChange creation
|
||||
if hasattr(self.model, 'to_objectchange'):
|
||||
objectchanges = ObjectChange.objects.filter(
|
||||
objectchange = ObjectChange.objects.get(
|
||||
changed_object_type=ContentType.objects.get_for_model(instance),
|
||||
changed_object_id=instance.pk
|
||||
)
|
||||
self.assertEqual(len(objectchanges), 1)
|
||||
self.assertEqual(objectchanges[0].action, ObjectChangeActionChoices.ACTION_DELETE)
|
||||
self.assertEqual(objectchange.action, ObjectChangeActionChoices.ACTION_DELETE)
|
||||
self.assertEqual(objectchange.message, data['changelog_message'])
|
||||
|
||||
def test_bulk_delete_objects(self):
|
||||
"""
|
||||
@ -418,11 +481,28 @@ class APIViewTestCases:
|
||||
self.assertEqual(len(id_list), 3, "Insufficient number of objects to test bulk deletion")
|
||||
data = [{"id": id} for id in id_list]
|
||||
|
||||
# If supported, add a changelog message
|
||||
changelog_message = get_random_string(10)
|
||||
if issubclass(self.model, ChangeLoggingMixin):
|
||||
for obj_data in data:
|
||||
obj_data['changelog_message'] = changelog_message
|
||||
|
||||
initial_count = self._get_queryset().count()
|
||||
response = self.client.delete(self._get_list_url(), data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
|
||||
self.assertEqual(self._get_queryset().count(), initial_count - 3)
|
||||
|
||||
# Verify ObjectChange creation
|
||||
if issubclass(self.model, ChangeLoggingMixin):
|
||||
objectchanges = ObjectChange.objects.filter(
|
||||
changed_object_type=ContentType.objects.get_for_model(self.model),
|
||||
changed_object_id__in=id_list
|
||||
)
|
||||
self.assertEqual(len(objectchanges), len(data))
|
||||
for oc in objectchanges:
|
||||
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_DELETE)
|
||||
self.assertEqual(oc.message, changelog_message)
|
||||
|
||||
class GraphQLTestCase(APITestCase):
|
||||
|
||||
def _get_graphql_base_name(self):
|
||||
|
||||
@ -653,7 +653,10 @@ class ViewTestCases:
|
||||
if issubclass(self.model, ChangeLoggingMixin):
|
||||
request_id = response.headers.get('X-Request-ID')
|
||||
self.assertIsNotNone(request_id, "Unable to determine request ID from response")
|
||||
objectchanges = ObjectChange.objects.filter(request_id=request_id)
|
||||
objectchanges = ObjectChange.objects.filter(
|
||||
changed_object_type=ContentType.objects.get_for_model(self.model),
|
||||
request_id=request_id
|
||||
)
|
||||
self.assertEqual(len(objectchanges), len(self.csv_data) - 1)
|
||||
for oc in objectchanges:
|
||||
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_CREATE)
|
||||
@ -786,7 +789,10 @@ class ViewTestCases:
|
||||
if issubclass(self.model, ChangeLoggingMixin):
|
||||
request_id = response.headers.get('X-Request-ID')
|
||||
self.assertIsNotNone(request_id, "Unable to determine request ID from response")
|
||||
objectchanges = ObjectChange.objects.filter(request_id=request_id)
|
||||
objectchanges = ObjectChange.objects.filter(
|
||||
changed_object_type=ContentType.objects.get_for_model(self.model),
|
||||
changed_object_id__in=pk_list
|
||||
)
|
||||
self.assertEqual(len(objectchanges), len(pk_list))
|
||||
for oc in objectchanges:
|
||||
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_UPDATE)
|
||||
@ -852,7 +858,7 @@ class ViewTestCases:
|
||||
self.assertHttpStatus(self.client.post(self._get_url('bulk_delete'), data), 403)
|
||||
|
||||
def test_bulk_delete_objects_with_permission(self):
|
||||
pk_list = self._get_queryset().values_list('pk', flat=True)
|
||||
pk_list = list(self._get_queryset().values_list('pk', flat=True))
|
||||
data = {
|
||||
'pk': pk_list,
|
||||
'confirm': True,
|
||||
@ -875,13 +881,14 @@ class ViewTestCases:
|
||||
# Try POST with model-level permission
|
||||
response = self.client.post(self._get_url('bulk_delete'), data)
|
||||
self.assertHttpStatus(response, 302)
|
||||
self.assertEqual(self._get_queryset().count(), 0)
|
||||
self.assertFalse(self._get_queryset().filter(pk__in=pk_list).exists())
|
||||
|
||||
# Verify ObjectChange creation
|
||||
if issubclass(self.model, ChangeLoggingMixin):
|
||||
request_id = response.headers.get('X-Request-ID')
|
||||
self.assertIsNotNone(request_id, "Unable to determine request ID from response")
|
||||
objectchanges = ObjectChange.objects.filter(request_id=request_id)
|
||||
objectchanges = ObjectChange.objects.filter(
|
||||
changed_object_type=ContentType.objects.get_for_model(self.model),
|
||||
changed_object_id__in=pk_list
|
||||
)
|
||||
self.assertEqual(len(objectchanges), len(pk_list))
|
||||
for oc in objectchanges:
|
||||
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_DELETE)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user