Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions netbox/dcim/tests/test_views.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
14 changes: 14 additions & 0 deletions netbox/netbox/views/generic/bulk_views.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import logging
import re
from collections import Counter
from copy import deepcopy

from django.contrib import messages
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down
Loading