From 68d39065ceef8f8804c1d75811bc54bb0fe7def2 Mon Sep 17 00:00:00 2001 From: Jeremy Stretch Date: Fri, 25 Jul 2025 14:35:27 -0400 Subject: [PATCH] Finish adding tests for changelog message functionality --- netbox/utilities/testing/api.py | 110 ++++++++++++++++++++++++++---- netbox/utilities/testing/views.py | 21 ++++-- 2 files changed, 109 insertions(+), 22 deletions(-) diff --git a/netbox/utilities/testing/api.py b/netbox/utilities/testing/api.py index 12e38a27f..ba704bd40 100644 --- a/netbox/utilities/testing/api.py +++ b/netbox/utilities/testing/api.py @@ -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): diff --git a/netbox/utilities/testing/views.py b/netbox/utilities/testing/views.py index 165456095..7141e61c0 100644 --- a/netbox/utilities/testing/views.py +++ b/netbox/utilities/testing/views.py @@ -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)