mirror/netbox-community-netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2024-05-10 07:54:54 +00:00
Code Activity

Fixes #5583: Eliminate redundant change records when adding/removing tags

Browse Source
This commit is contained in:
jeremystretch
2021-04-13 10:14:25 -04:00
parent 9bda2a44ae
commit e5bbf47ab9
5 changed files with 96 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
docs/release-notes/version-2.11.md
View File

@ -13,6 +13,7 @@
### Bug Fixes (from Beta)
* [#5583](https://github.com/netbox-community/netbox/issues/5583) - Eliminate redundant change records when adding/removing tags
* [#6100](https://github.com/netbox-community/netbox/issues/6100) - Fix VM interfaces table "add interfaces" link
* [#6104](https://github.com/netbox-community/netbox/issues/6104) - Fix location column on racks table
* [#6105](https://github.com/netbox-community/netbox/issues/6105) - Hide checkboxes for VMs under cluster VMs view

14
netbox/extras/signals.py
View File

@ -22,19 +22,31 @@ def _handle_changed_object(request, sender, instance, **kwargs):
"""
Fires when an object is created or updated.
"""
# Queue the object for processing once the request completes
m2m_changed = False
# Determine the type of change being made
if kwargs.get('created'):
action = ObjectChangeActionChoices.ACTION_CREATE
elif 'created' in kwargs:
action = ObjectChangeActionChoices.ACTION_UPDATE
elif kwargs.get('action') in ['post_add', 'post_remove'] and kwargs['pk_set']:
# m2m_changed with objects added or removed
m2m_changed = True
action = ObjectChangeActionChoices.ACTION_UPDATE
else:
return
# Record an ObjectChange if applicable
if hasattr(instance, 'to_objectchange'):
if m2m_changed:
ObjectChange.objects.filter(
changed_object_type=ContentType.objects.get_for_model(instance),
changed_object_id=instance.pk,
request_id=request.id
).update(
postchange_data=instance.to_objectchange(action).postchange_data
)
else:
objectchange = instance.to_objectchange(action)
objectchange.user = request.user
objectchange.request_id = request.id

42
netbox/extras/tests/test_changelog.py
View File

@ -56,19 +56,18 @@ class ChangeLogViewTest(ModelViewTestCase):
response = self.client.post(**request)
self.assertHttpStatus(response, 302)
# Verify the creation of a new ObjectChange record
site = Site.objects.get(name='Site 1')
# First OC is the creation; second is the tags update
oc_list = ObjectChange.objects.filter(
oc = ObjectChange.objects.get(
changed_object_type=ContentType.objects.get_for_model(Site),
changed_object_id=site.pk
).order_by('pk')
self.assertEqual(oc_list[0].changed_object, site)
self.assertEqual(oc_list[0].action, ObjectChangeActionChoices.ACTION_CREATE)
self.assertEqual(oc_list[0].prechange_data, None)
self.assertEqual(oc_list[0].postchange_data['custom_fields']['my_field'], form_data['cf_my_field'])
self.assertEqual(oc_list[0].postchange_data['custom_fields']['my_field_select'], form_data['cf_my_field_select'])
self.assertEqual(oc_list[1].action, ObjectChangeActionChoices.ACTION_UPDATE)
self.assertEqual(oc_list[1].postchange_data['tags'], ['Tag 1', 'Tag 2'])
)
self.assertEqual(oc.changed_object, site)
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_CREATE)
self.assertEqual(oc.prechange_data, None)
self.assertEqual(oc.postchange_data['custom_fields']['my_field'], form_data['cf_my_field'])
self.assertEqual(oc.postchange_data['custom_fields']['my_field_select'], form_data['cf_my_field_select'])
self.assertEqual(oc.postchange_data['tags'], ['Tag 1', 'Tag 2'])
def test_update_object(self):
site = Site(name='Site 1', slug='site-1')
@ -93,8 +92,8 @@ class ChangeLogViewTest(ModelViewTestCase):
response = self.client.post(**request)
self.assertHttpStatus(response, 302)
# Verify the creation of a new ObjectChange record
site.refresh_from_db()
# Get only the most recent OC
oc = ObjectChange.objects.filter(
changed_object_type=ContentType.objects.get_for_model(Site),
changed_object_id=site.pk
@ -259,17 +258,15 @@ class ChangeLogAPITest(APITestCase):
self.assertHttpStatus(response, status.HTTP_201_CREATED)
site = Site.objects.get(pk=response.data['id'])
# First OC is the creation; second is the tags update
oc_list = ObjectChange.objects.filter(
oc = ObjectChange.objects.get(
changed_object_type=ContentType.objects.get_for_model(Site),
changed_object_id=site.pk
).order_by('pk')
self.assertEqual(oc_list[0].changed_object, site)
self.assertEqual(oc_list[0].action, ObjectChangeActionChoices.ACTION_CREATE)
self.assertEqual(oc_list[0].prechange_data, None)
self.assertEqual(oc_list[0].postchange_data['custom_fields'], data['custom_fields'])
self.assertEqual(oc_list[1].action, ObjectChangeActionChoices.ACTION_UPDATE)
self.assertEqual(oc_list[1].postchange_data['tags'], ['Tag 1', 'Tag 2'])
)
self.assertEqual(oc.changed_object, site)
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_CREATE)
self.assertEqual(oc.prechange_data, None)
self.assertEqual(oc.postchange_data['custom_fields'], data['custom_fields'])
self.assertEqual(oc.postchange_data['tags'], ['Tag 1', 'Tag 2'])
def test_update_object(self):
site = Site(name='Site 1', slug='site-1')
@ -294,11 +291,10 @@ class ChangeLogAPITest(APITestCase):
self.assertHttpStatus(response, status.HTTP_200_OK)
site = Site.objects.get(pk=response.data['id'])
# Get only the most recent OC
oc = ObjectChange.objects.filter(
oc = ObjectChange.objects.get(
changed_object_type=ContentType.objects.get_for_model(Site),
changed_object_id=site.pk
).first()
)
self.assertEqual(oc.changed_object, site)
self.assertEqual(oc.action, ObjectChangeActionChoices.ACTION_UPDATE)
self.assertEqual(oc.postchange_data['custom_fields'], data['custom_fields'])

32
netbox/utilities/testing/api.py
View File

@ -6,6 +6,8 @@ from django.test import override_settings
from rest_framework import status
from rest_framework.test import APIClient
from extras.choices import ObjectChangeActionChoices
from extras.models import ObjectChange
from users.models import ObjectPermission, Token
from .utils import disable_warnings
from .views import ModelTestCase
@ -223,13 +225,23 @@ class APIViewTestCases:
response = self.client.post(self._get_list_url(), self.create_data[0], 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'])
self.assertInstanceEqual(
self._get_queryset().get(pk=response.data['id']),
instance,
self.create_data[0],
exclude=self.validation_excluded_fields,
api=True
)
# Verify ObjectChange creation
if hasattr(self.model, 'to_objectchange'):
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)
def test_bulk_create_objects(self):
"""
POST a set of objects in a single request.
@ -304,6 +316,15 @@ class APIViewTestCases:
api=True
)
# Verify ObjectChange creation
if hasattr(self.model, 'to_objectchange'):
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)
def test_bulk_update_objects(self):
"""
PATCH a set of objects in a single request.
@ -367,6 +388,15 @@ class APIViewTestCases:
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(
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)
def test_bulk_delete_objects(self):
"""
DELETE a set of objects in a single request.

30
netbox/utilities/testing/views.py
View File

@ -10,7 +10,8 @@ from django.utils.text import slugify
from netaddr import IPNetwork
from taggit.managers import TaggableManager
from extras.models import Tag
from extras.choices import ObjectChangeActionChoices
from extras.models import ObjectChange, Tag
from users.models import ObjectPermission
from utilities.permissions import resolve_permission_ct
from .utils import disable_warnings, extract_form_failures, post_data
@ -323,7 +324,16 @@ class ViewTestCases:
}
self.assertHttpStatus(self.client.post(**request), 302)
self.assertEqual(initial_count + 1, self._get_queryset().count())
self.assertInstanceEqual(self._get_queryset().order_by('pk').last(), self.form_data)
instance = self._get_queryset().order_by('pk').last()
self.assertInstanceEqual(instance, self.form_data)
# Verify ObjectChange creation
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)
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_create_object_with_constrained_permission(self):
@ -410,6 +420,14 @@ class ViewTestCases:
self.assertHttpStatus(self.client.post(**request), 302)
self.assertInstanceEqual(self._get_queryset().get(pk=instance.pk), self.form_data)
# Verify ObjectChange creation
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)
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_edit_object_with_constrained_permission(self):
instance1, instance2 = self._get_queryset().all()[:2]
@ -489,6 +507,14 @@ class ViewTestCases:
with self.assertRaises(ObjectDoesNotExist):
self._get_queryset().get(pk=instance.pk)
# Verify ObjectChange creation
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)
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_delete_object_with_constrained_permission(self):
instance1, instance2 = self._get_queryset().all()[:2]