From 7de5a62451d90cb103babde66a72acd644ab9237 Mon Sep 17 00:00:00 2001 From: Jeremy Stretch Date: Mon, 10 Aug 2026 16:15:55 -0400 Subject: [PATCH] Flag duplicate object IDs in bulk operations --- netbox/dcim/tests/test_api.py | 45 ++++++++++++++++ netbox/netbox/api/viewsets/mixins.py | 62 ++++++++++++++++++++- netbox/utilities/testing/api.py | 81 ++++++++++++++++++++++++++++ 3 files changed, 186 insertions(+), 2 deletions(-) diff --git a/netbox/dcim/tests/test_api.py b/netbox/dcim/tests/test_api.py index 406803d91..4e5f4a919 100644 --- a/netbox/dcim/tests/test_api.py +++ b/netbox/dcim/tests/test_api.py @@ -467,6 +467,51 @@ class SiteTestCase(APIViewTestCases.APIViewTestCase): response = self.client.patch(url, data, format='json', **self.header) self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + def test_bulk_update_objects_duplicate_id_invalid_entry(self): + """ + PATCH a set of objects in which one object is named twice, once with invalid data and once + with valid data. The invalid entry must not be discarded in favor of the valid one. + """ + obj_perm = ObjectPermission(name='Test permission', actions=['change']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) + + site = Site.objects.get(slug='site-1') + data = [ + {'id': site.pk, 'name': ''}, # Invalid: name is required + {'id': site.pk, 'name': 'Renamed Site'}, + ] + response = self.client.patch(self._get_list_url(), data, format='json', **self.header) + + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + self.assertEqual([e['id'] for e in response.data['errors']], [site.pk]) + + # The valid entry must not have been applied + site.refresh_from_db() + self.assertEqual(site.name, 'Site 1') + + def test_bulk_delete_objects_duplicate_id_changelog_message(self): + """ + DELETE a set of objects in which one object is named twice with differing changelog + messages. The request must be rejected rather than recording only one of the messages. + """ + obj_perm = ObjectPermission(name='Test permission', actions=['delete']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) + + site = Site.objects.get(slug='site-1') + data = [ + {'id': site.pk, 'changelog_message': 'First message'}, + {'id': site.pk, 'changelog_message': 'Second message'}, + ] + response = self.client.delete(self._get_list_url(), data, format='json', **self.header) + + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + self.assertEqual([e['id'] for e in response.data['errors']], [site.pk]) + self.assertTrue(Site.objects.filter(pk=site.pk).exists()) + def test_bulk_create_objects_conflicting(self): """ POST a set of objects in which two conflict with one another. Objects are created one at a diff --git a/netbox/netbox/api/viewsets/mixins.py b/netbox/netbox/api/viewsets/mixins.py index e31cbcc7e..2d80e6a68 100644 --- a/netbox/netbox/api/viewsets/mixins.py +++ b/netbox/netbox/api/viewsets/mixins.py @@ -1,4 +1,5 @@ import warnings +from collections import Counter from contextlib import contextmanager from django.core.exceptions import ObjectDoesNotExist @@ -32,10 +33,53 @@ __all__ = ( 'ObjectValidationMixin', 'SequentialBulkCreatesMixin', 'discard_events_on_rollback', + 'get_duplicate_objects_response', 'get_missing_objects_response', ) +def get_duplicate_objects_response(object_ids): + """ + Return a structured error Response naming each of the given object IDs which appears more than + once, or None if they are all distinct. + + A bulk operation identifies its objects by ID, so listing one twice is ambiguous. For an update, + only one of the two sets of attributes can be applied, and the discarded entry is never even + validated: a request pairing an invalid entry with a valid one for the same object would + otherwise report success while silently ignoring the invalid data. For a delete, the repetition + is meaningless, but it likewise causes the response to report on fewer objects than were named. + Rather than guess at the intent, such a request is rejected. + """ + errors = [ + { + 'id': object_id, + 'errors': { + 'id': [ + _("Each object may be specified only once; ID {id} is listed {count} times").format( + id=object_id, count=count + ), + ], + }, + } + # Counter preserves the order in which each ID was first seen + for object_id, count in Counter(object_ids).items() + if count > 1 + ] + if not errors: + return None + + return Response( + { + 'detail': _('{failed_count} of {total} objects are listed more than once.').format( + failed_count=len(errors), + total=len(object_ids), + ), + 'errors': errors, + }, + status=status.HTTP_400_BAD_REQUEST, + ) + + def get_missing_objects_response(object_ids, queryset): """ Return a structured error Response naming each of the given object IDs which the queryset does @@ -60,8 +104,10 @@ def get_missing_objects_response(object_ids, queryset): 'id': [_("Object with ID {id} does not exist").format(id=object_id)], }, } - # dict.fromkeys() de-duplicates while preserving the order of first appearance, so an ID - # repeated in the request is reported once rather than once per occurrence. + # NetBox's bulk actions reject a repeated ID before reaching this point (see + # get_duplicate_objects_response), but de-duplicate anyway so that any other caller reports + # such an ID once rather than once per occurrence. dict.fromkeys() preserves the order of + # first appearance. for object_id in dict.fromkeys(object_ids) if object_id not in found_pks ] @@ -376,6 +422,12 @@ class BulkUpdateModelMixin: serializer.is_valid(raise_exception=True) object_ids = [o['id'] for o in serializer.validated_data] + + # Reject the batch if any object is named more than once, rather than applying only one of + # the entries given for it. + if (response := get_duplicate_objects_response(object_ids)) is not None: + return response + qs = self.get_bulk_update_queryset().filter(pk__in=object_ids) # Reject the batch if any of the objects to be updated could not be found, rather than @@ -485,6 +537,12 @@ class BulkDestroyModelMixin: serializer.is_valid(raise_exception=True) object_ids = [o['id'] for o in serializer.validated_data] + + # Reject the batch if any object is named more than once, rather than ignoring the + # repetition (and any changelog message attached to it) and reporting success. + if (response := get_duplicate_objects_response(object_ids)) is not None: + return response + qs = self.get_bulk_destroy_queryset().filter(pk__in=object_ids) # Reject the batch if any of the objects to be deleted could not be found, rather than diff --git a/netbox/utilities/testing/api.py b/netbox/utilities/testing/api.py index a978c306e..4e77d0831 100644 --- a/netbox/utilities/testing/api.py +++ b/netbox/utilities/testing/api.py @@ -661,6 +661,52 @@ class APIViewTestCases: f'sibling ID', ) + def test_bulk_update_objects_duplicate_id(self): + """ + PATCH a set of objects in which the same object is named twice. The request must be + rejected rather than applying only one of the entries given for that object. + """ + if self.bulk_update_data is None: + self.skipTest('Bulk update data not set') + + obj_perm = ObjectPermission(name='Test permission', actions=['change']) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) + + id_list = list(self._get_queryset().values_list('id', flat=True)[:2]) + self.assertEqual(len(id_list), 2, 'Insufficient number of objects to test bulk update') + + # Repeat the first ID at the end of the request + data = [{'id': id, **self.bulk_update_data} for id in (*id_list, id_list[0])] + + # Snapshot the objects which would otherwise have been updated + instances_before = list(self._get_queryset().filter(pk__in=id_list)) + + response = self.client.patch(self._get_list_url(), data, format='json', **self.header) + + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + self.assertIn('detail', response.data) + self.assertIn('errors', response.data) + + # The repeated ID must be reported once, not once per occurrence + self.assertEqual(len(response.data['errors']), 1) + self.assertEqual(response.data['errors'][0]['id'], id_list[0]) + self.assertIn('id', response.data['errors'][0]['errors']) + + # No object named in the request may have been updated, including the one named only once + for instance_before in instances_before: + instance_after = self._get_queryset().get(pk=instance_before.pk) + for field in self.bulk_update_data: + if field in ('changelog_message', 'add_tags', 'remove_tags'): + continue + self.assertEqual( + getattr(instance_after, field, None), + getattr(instance_before, field, None), + f'Field {field!r} of object {instance_before.pk} was modified despite a duplicated ' + f'sibling ID', + ) + class DeleteObjectViewTestCase(APITestCase): def test_delete_object_without_permission(self): @@ -782,6 +828,41 @@ class APIViewTestCases: # The objects named alongside the missing one must not have been deleted self.assertEqual(self._get_queryset().count(), initial_count) + def test_bulk_delete_objects_duplicate_id(self): + """ + DELETE a set of objects in which the same object is named twice. The request must be + rejected rather than reporting success for a batch it only partly acted on. + """ + obj_perm = ObjectPermission( + name='Test permission', + actions=['delete'] + ) + obj_perm.save() + obj_perm.users.add(self.user) + obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model)) + + # Target the most recently created objects to avoid triggering recursive deletions + id_list = list(self._get_queryset().order_by('-id').values_list('id', flat=True)[:2]) + self.assertEqual(len(id_list), 2, 'Insufficient number of objects to test bulk deletion') + + # Repeat the first ID at the end of the request + data = [{'id': id} for id in (*id_list, id_list[0])] + + initial_count = self._get_queryset().count() + response = self.client.delete(self._get_list_url(), data, format='json', **self.header) + + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + self.assertIn('detail', response.data) + self.assertIn('errors', response.data) + + # The repeated ID must be reported once, not once per occurrence + self.assertEqual(len(response.data['errors']), 1) + self.assertEqual(response.data['errors'][0]['id'], id_list[0]) + self.assertIn('id', response.data['errors'][0]['errors']) + + # No object named in the request may have been deleted + self.assertEqual(self._get_queryset().count(), initial_count) + class GraphQLTestCase(APITestCase): graphql_auto_filter_tests = True graphql_auto_filter_exclude = ()