mirror of
https://github.com/netbox-community/netbox.git
synced 2026-01-09 13:22:18 -06:00
Fixes #20011: UI Error msg for duplicate IDs in bulk import
This commit is contained in:
@@ -2322,6 +2322,32 @@ class DeviceTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|||||||
url = reverse('dcim:device_inventory', kwargs={'pk': device.pk})
|
url = reverse('dcim:device_inventory', kwargs={'pk': device.pk})
|
||||||
self.assertHttpStatus(self.client.get(url), 200)
|
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(
|
class ModuleTestCase(
|
||||||
# Module does not support bulk renaming (no name field) or
|
# Module does not support bulk renaming (no name field) or
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
import logging
|
import logging
|
||||||
import re
|
import re
|
||||||
|
from collections import Counter
|
||||||
from copy import deepcopy
|
from copy import deepcopy
|
||||||
|
|
||||||
from django.contrib import messages
|
from django.contrib import messages
|
||||||
@@ -33,6 +34,7 @@ from utilities.jobs import is_background_request, process_request_as_job
|
|||||||
from utilities.permissions import get_permission_for_model
|
from utilities.permissions import get_permission_for_model
|
||||||
from utilities.query import reapply_model_ordering
|
from utilities.query import reapply_model_ordering
|
||||||
from utilities.request import safe_for_redirect
|
from utilities.request import safe_for_redirect
|
||||||
|
from utilities.string import title
|
||||||
from utilities.tables import get_table_configs
|
from utilities.tables import get_table_configs
|
||||||
from utilities.views import GetReturnURLMixin, get_action_url
|
from utilities.views import GetReturnURLMixin, get_action_url
|
||||||
from .base import BaseMultiObjectView
|
from .base import BaseMultiObjectView
|
||||||
@@ -443,6 +445,18 @@ class BulkImportView(GetReturnURLMixin, BaseMultiObjectView):
|
|||||||
|
|
||||||
# Prefetch objects to be updated, if any
|
# Prefetch objects to be updated, if any
|
||||||
prefetch_ids = [int(record['id']) for record in records if record.get('id')]
|
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 = {
|
prefetched_objects = {
|
||||||
obj.pk: obj
|
obj.pk: obj
|
||||||
for obj in self.queryset.model.objects.filter(id__in=prefetch_ids)
|
for obj in self.queryset.model.objects.filter(id__in=prefetch_ids)
|
||||||
|
|||||||
Reference in New Issue
Block a user