diff --git a/netbox/dcim/tests/test_views.py b/netbox/dcim/tests/test_views.py index 6a34df6528..9103befdb0 100644 --- a/netbox/dcim/tests/test_views.py +++ b/netbox/dcim/tests/test_views.py @@ -2322,6 +2322,32 @@ def test_device_inventory(self): url = reverse('dcim:device_inventory', kwargs={'pk': device.pk}) self.assertHttpStatus(self.client.get(url), 200) + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*']) + def test_bulk_import_duplicate_ids_error_message(self): + device = Device.objects.first() + csv_data = ( + "id,role", + f"{device.pk},Device Role 1", + f"{device.pk},Device Role 2", + ) + + self.add_permissions('dcim.add_device', 'dcim.change_device') + response = self.client.post( + self._get_url('bulk_import'), + { + 'data': '\n'.join(csv_data), + 'format': ImportFormatChoices.CSV, + 'csv_delimiter': CSVDelimiterChoices.AUTO, + }, + follow=True + ) + + self.assertEqual(response.status_code, 200) + self.assertIn( + f'Duplicate objects found: Device with ID(s) {device.pk} appears multiple times', + response.content.decode('utf-8') + ) + class ModuleTestCase( # Module does not support bulk renaming (no name field) or diff --git a/netbox/netbox/views/generic/bulk_views.py b/netbox/netbox/views/generic/bulk_views.py index 1d8d6b2987..b8d70e1122 100644 --- a/netbox/netbox/views/generic/bulk_views.py +++ b/netbox/netbox/views/generic/bulk_views.py @@ -1,5 +1,6 @@ import logging import re +from collections import Counter from copy import deepcopy from django.contrib import messages @@ -33,6 +34,7 @@ from utilities.permissions import get_permission_for_model from utilities.query import reapply_model_ordering from utilities.request import safe_for_redirect +from utilities.string import title from utilities.tables import get_table_configs from utilities.views import GetReturnURLMixin, get_action_url from .base import BaseMultiObjectView @@ -443,6 +445,18 @@ def create_and_update_objects(self, form, request): # Prefetch objects to be updated, if any prefetch_ids = [int(record['id']) for record in records if record.get('id')] + + # check for duplicate IDs + duplicate_pks = [pk for pk, count in Counter(prefetch_ids).items() if count > 1] + if duplicate_pks: + error_msg = _( + "Duplicate objects found: {model} with ID(s) {ids} appears multiple times" + ).format( + model=title(self.queryset.model._meta.verbose_name), + ids=', '.join(str(pk) for pk in sorted(duplicate_pks)) + ) + raise ValidationError(error_msg) + prefetched_objects = { obj.pk: obj for obj in self.queryset.model.objects.filter(id__in=prefetch_ids)