{# Dynamic content goes here #}
diff --git a/netbox/utilities/forms/bulk_import.py b/netbox/utilities/forms/bulk_import.py
index b11f6688a50..abebe255fa0 100644
--- a/netbox/utilities/forms/bulk_import.py
+++ b/netbox/utilities/forms/bulk_import.py
@@ -8,12 +8,13 @@
from core.forms.mixins import SyncedDataMixin
from netbox.choices import CSVDelimiterChoices, ImportFormatChoices, ImportMethodChoices
+from netbox.forms.mixins import ChangeLoggingMixin
from utilities.constants import CSV_DELIMITERS
from utilities.forms.mixins import BackgroundJobMixin
from utilities.forms.utils import parse_csv
-class BulkImportForm(BackgroundJobMixin, SyncedDataMixin, forms.Form):
+class BulkImportForm(ChangeLoggingMixin, BackgroundJobMixin, SyncedDataMixin, forms.Form):
import_method = forms.ChoiceField(
choices=ImportMethodChoices,
required=False
diff --git a/netbox/utilities/forms/forms.py b/netbox/utilities/forms/forms.py
index 2192c5a9974..122107728d1 100644
--- a/netbox/utilities/forms/forms.py
+++ b/netbox/utilities/forms/forms.py
@@ -10,6 +10,7 @@
'BulkRenameForm',
'ConfirmationForm',
'CSVModelForm',
+ 'DeleteForm',
'FilterForm',
'TableConfigForm',
)
@@ -30,6 +31,16 @@ class ConfirmationForm(forms.Form):
)
+class DeleteForm(ConfirmationForm):
+ """
+ Confirm the deletion of an object, optionally providing a changelog message.
+ """
+ changelog_message = forms.CharField(
+ required=False,
+ max_length=200
+ )
+
+
class BulkEditForm(BackgroundJobMixin, forms.Form):
"""
Provides bulk edit support for objects.
diff --git a/netbox/utilities/testing/api.py b/netbox/utilities/testing/api.py
index 12e38a27f3c..ba704bd40c5 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 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 @@ def test_create_object(self):
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 @@ def test_create_object(self):
)
# 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 @@ def test_bulk_create_objects(self):
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 @@ def test_bulk_create_objects(self):
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 @@ def test_bulk_create_objects(self):
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 @@ def test_update_object(self):
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 @@ def test_bulk_update_objects(self):
{'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 @@ def test_delete_object(self):
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 @@ def test_bulk_delete_objects(self):
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/utils.py b/netbox/utilities/testing/utils.py
index 434f12b7b66..7715daa39f0 100644
--- a/netbox/utilities/testing/utils.py
+++ b/netbox/utilities/testing/utils.py
@@ -1,6 +1,8 @@
import json
import logging
+import random
import re
+import string
from contextlib import contextmanager
from django.contrib.auth.models import Permission
@@ -154,3 +156,15 @@ def add_custom_field_data(form_data, model):
f'cf_{k}': v if type(v) is str else json.dumps(v)
for k, v in DUMMY_CF_DATA.items()
})
+
+
+#
+# Misc utilities
+#
+
+def get_random_string(length, charset=None):
+ """
+ Return a random string of the given length.
+ """
+ characters = string.ascii_letters + string.digits # a-z, A-Z, 0-9
+ return ''.join(random.choice(characters) for __ in range(length))
diff --git a/netbox/utilities/testing/views.py b/netbox/utilities/testing/views.py
index c451649ff79..da8a870984f 100644
--- a/netbox/utilities/testing/views.py
+++ b/netbox/utilities/testing/views.py
@@ -14,7 +14,7 @@
from netbox.models.features import ChangeLoggingMixin, CustomFieldsMixin
from users.models import ObjectPermission
from .base import ModelTestCase
-from .utils import add_custom_field_data, disable_warnings, post_data
+from .utils import add_custom_field_data, disable_warnings, get_random_string, post_data
__all__ = (
'ModelViewTestCase',
@@ -169,6 +169,11 @@ def test_create_object_with_permission(self):
if issubclass(self.model, CustomFieldsMixin):
add_custom_field_data(self.form_data, self.model)
+ # If supported, add a changelog message
+ if issubclass(self.model, ChangeLoggingMixin):
+ if 'changelog_message' not in self.form_data:
+ self.form_data['changelog_message'] = get_random_string(10)
+
# Try POST with model-level permission
initial_count = self._get_queryset().count()
request = {
@@ -181,13 +186,14 @@ def test_create_object_with_permission(self):
self.assertInstanceEqual(instance, self.form_data, exclude=self.validation_excluded_fields)
# Verify ObjectChange creation
- if issubclass(instance.__class__, ChangeLoggingMixin):
+ if issubclass(self.model, ChangeLoggingMixin):
objectchanges = ObjectChange.objects.filter(
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(objectchanges[0].message, self.form_data['changelog_message'])
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
def test_create_object_with_constrained_permission(self):
@@ -272,6 +278,11 @@ def test_edit_object_with_permission(self):
if issubclass(self.model, CustomFieldsMixin):
add_custom_field_data(self.form_data, self.model)
+ # If supported, add a changelog message
+ if issubclass(self.model, ChangeLoggingMixin):
+ if 'changelog_message' not in self.form_data:
+ self.form_data['changelog_message'] = get_random_string(10)
+
# Try POST with model-level permission
request = {
'path': self._get_url('edit', instance),
@@ -282,13 +293,14 @@ def test_edit_object_with_permission(self):
self.assertInstanceEqual(instance, self.form_data, exclude=self.validation_excluded_fields)
# Verify ObjectChange creation
- if issubclass(instance.__class__, ChangeLoggingMixin):
+ if issubclass(self.model, ChangeLoggingMixin):
objectchanges = ObjectChange.objects.filter(
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(objectchanges[0].message, self.form_data['changelog_message'])
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
def test_edit_object_with_constrained_permission(self):
@@ -348,6 +360,7 @@ def test_delete_object_without_permission(self):
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_delete_object_with_permission(self):
instance = self._get_queryset().first()
+ form_data = {'confirm': True}
# Assign model-level permission
obj_perm = ObjectPermission(
@@ -361,23 +374,28 @@ def test_delete_object_with_permission(self):
# Try GET with model-level permission
self.assertHttpStatus(self.client.get(self._get_url('delete', instance)), 200)
+ # If supported, add a changelog message
+ if issubclass(self.model, ChangeLoggingMixin):
+ form_data['changelog_message'] = get_random_string(10)
+
# Try POST with model-level permission
request = {
'path': self._get_url('delete', instance),
- 'data': post_data({'confirm': True}),
+ 'data': post_data(form_data),
}
self.assertHttpStatus(self.client.post(**request), 302)
with self.assertRaises(ObjectDoesNotExist):
self._get_queryset().get(pk=instance.pk)
# Verify ObjectChange creation
- if issubclass(instance.__class__, ChangeLoggingMixin):
+ if issubclass(self.model, ChangeLoggingMixin):
objectchanges = ObjectChange.objects.filter(
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(objectchanges[0].message, form_data['changelog_message'])
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_delete_object_with_constrained_permission(self):
@@ -610,6 +628,10 @@ def test_bulk_import_objects_with_permission(self):
'csv_delimiter': CSVDelimiterChoices.AUTO,
}
+ # If supported, add a changelog message
+ if issubclass(self.model, ChangeLoggingMixin):
+ data['changelog_message'] = get_random_string(10)
+
# Assign model-level permission
obj_perm = ObjectPermission(
name='Test permission',
@@ -623,9 +645,23 @@ def test_bulk_import_objects_with_permission(self):
self.assertHttpStatus(self.client.get(self._get_url('bulk_import')), 200)
# Test POST with permission
- self.assertHttpStatus(self.client.post(self._get_url('bulk_import'), data), 302)
+ response = self.client.post(self._get_url('bulk_import'), data)
+ self.assertHttpStatus(response, 302)
self.assertEqual(self._get_queryset().count(), initial_count + len(self.csv_data) - 1)
+ # 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(
+ 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)
+ self.assertEqual(oc.message, data['changelog_message'])
+
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_bulk_update_objects_with_permission(self):
if not hasattr(self, 'csv_update_data'):
@@ -727,6 +763,10 @@ def test_bulk_edit_objects_with_permission(self):
'_apply': True, # Form button
}
+ # If supported, add a changelog message
+ if issubclass(self.model, ChangeLoggingMixin):
+ data['changelog_message'] = get_random_string(10)
+
# Append the form data to the request
data.update(post_data(self.bulk_edit_data))
@@ -740,10 +780,24 @@ def test_bulk_edit_objects_with_permission(self):
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
# Try POST with model-level permission
- self.assertHttpStatus(self.client.post(self._get_url('bulk_edit'), data), 302)
+ response = self.client.post(self._get_url('bulk_edit'), data)
+ self.assertHttpStatus(response, 302)
for i, instance in enumerate(self._get_queryset().filter(pk__in=pk_list)):
self.assertInstanceEqual(instance, self.bulk_edit_data)
+ # 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(
+ 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)
+ self.assertEqual(oc.message, data['changelog_message'])
+
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
def test_bulk_edit_objects_with_constrained_permission(self):
pk_list = list(self._get_queryset().values_list('pk', flat=True)[:3])
@@ -804,13 +858,17 @@ def test_bulk_delete_objects_without_permission(self):
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))[:3]
data = {
'pk': pk_list,
'confirm': True,
'_confirm': True, # Form button
}
+ # If supported, add a changelog message
+ if issubclass(self.model, ChangeLoggingMixin):
+ data['changelog_message'] = get_random_string(10)
+
# Assign unconstrained permission
obj_perm = ObjectPermission(
name='Test permission',
@@ -821,8 +879,20 @@ def test_bulk_delete_objects_with_permission(self):
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
# Try POST with model-level permission
- self.assertHttpStatus(self.client.post(self._get_url('bulk_delete'), data), 302)
- self.assertEqual(self._get_queryset().count(), 0)
+ response = self.client.post(self._get_url('bulk_delete'), data)
+ self.assertHttpStatus(response, 302)
+ self.assertFalse(self._get_queryset().filter(pk__in=pk_list).exists())
+
+ # 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=pk_list
+ )
+ self.assertEqual(len(objectchanges), len(pk_list))
+ for oc in objectchanges:
+ self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_DELETE)
+ self.assertEqual(oc.message, data['changelog_message'])
def test_bulk_delete_objects_with_constrained_permission(self):
pk_list = self._get_queryset().values_list('pk', flat=True)