diff --git a/netbox/dcim/forms/bulk_import.py b/netbox/dcim/forms/bulk_import.py index b7937745c..6e97c7efe 100644 --- a/netbox/dcim/forms/bulk_import.py +++ b/netbox/dcim/forms/bulk_import.py @@ -1,7 +1,8 @@ from django import forms from django.contrib.contenttypes.models import ContentType from django.contrib.postgres.forms.array import SimpleArrayField -from django.core.exceptions import ObjectDoesNotExist +from django.core.exceptions import MultipleObjectsReturned, ObjectDoesNotExist +from django.utils.html import format_html from django.utils.safestring import mark_safe from django.utils.translation import gettext_lazy as _ @@ -1440,19 +1441,29 @@ class CableImportForm(PrimaryModelImportForm): to_field_name='name', help_text=_('Site of parent device A (if any)'), ) - side_a_device = CSVModelChoiceField( + side_a_device = CSVModelMultipleChoiceField( label=_('Side A device'), queryset=Device.objects.all(), required=False, to_field_name='name', - help_text=_('Device name (for device component terminations)') + help_text=format_html( + '{} {}', + _('Device name(s) for device component terminations. Separate multiple values with commas, ' + 'encased with double quotes. Example:'), + '"device1,device2"' + ) ) - side_a_power_panel = CSVModelChoiceField( + side_a_power_panel = CSVModelMultipleChoiceField( label=_('Side A power panel'), queryset=PowerPanel.objects.all(), required=False, to_field_name='name', - help_text=_('Power panel name (for power feed terminations)') + help_text=format_html( + '{} {}', + _('Power panel name(s) for power feed terminations. Separate multiple values with commas, ' + 'encased with double quotes. Example:'), + '"panel1,panel2"' + ) ) side_a_type = CSVContentTypeField( label=_('Side A type'), @@ -1462,7 +1473,12 @@ class CableImportForm(PrimaryModelImportForm): ) side_a_name = forms.CharField( label=_('Side A name'), - help_text=_('Termination name') + help_text=format_html( + '{} {}', + _('Termination name(s). Separate multiple values with commas, encased with double quotes. ' + 'Example:'), + '"eth0,eth1"' + ) ) # Termination B @@ -1473,19 +1489,29 @@ class CableImportForm(PrimaryModelImportForm): to_field_name='name', help_text=_('Site of parent device B (if any)'), ) - side_b_device = CSVModelChoiceField( + side_b_device = CSVModelMultipleChoiceField( label=_('Side B device'), queryset=Device.objects.all(), required=False, to_field_name='name', - help_text=_('Device name (for device component terminations)') + help_text=format_html( + '{} {}', + _('Device name(s) for device component terminations. Separate multiple values with commas, ' + 'encased with double quotes. Example:'), + '"device1,device2"' + ) ) - side_b_power_panel = CSVModelChoiceField( + side_b_power_panel = CSVModelMultipleChoiceField( label=_('Side B power panel'), queryset=PowerPanel.objects.all(), required=False, to_field_name='name', - help_text=_('Power panel name (for power feed terminations)') + help_text=format_html( + '{} {}', + _('Power panel name(s) for power feed terminations. Separate multiple values with commas, ' + 'encased with double quotes. Example:'), + '"panel1,panel2"' + ) ) side_b_type = CSVContentTypeField( label=_('Side B type'), @@ -1495,7 +1521,12 @@ class CableImportForm(PrimaryModelImportForm): ) side_b_name = forms.CharField( label=_('Side B name'), - help_text=_('Termination name') + help_text=format_html( + '{} {}', + _('Termination name(s). Separate multiple values with commas, encased with double quotes. ' + 'Example:'), + '"eth0,eth1"' + ) ) # Cable attributes @@ -1577,6 +1608,63 @@ class CableImportForm(PrimaryModelImportForm): **side_b_parent_params ) + @staticmethod + def _split_side_values(value): + """ + Split a side_* cell into an ordered list of values, preserving duplicates and empty + entries. Accepts a comma-separated string (CSV) or a native list (JSON/YAML). + """ + if value in (None, ''): + return [] + if not isinstance(value, (list, tuple)): + value = str(value).split(',') + return ['' if item is None else str(item).strip() for item in value] + + def _resolve_side_parent_objects(self, field_name): + """ + Resolve a side's parent objects from the raw submitted values, preserving their order. + CSVModelMultipleChoiceField cleans to an unordered queryset, which cannot be used to pair + each parent with its corresponding termination name by position. Resolution errors are + reported on the parent field itself. + """ + if field_name not in self.cleaned_data: + # The parent field has already raised its own validation error + return None + field = self.fields[field_name] + to_field_name = field.to_field_name or 'pk' + + parents = [] + for value in self._split_side_values(self.data.get(field_name)): + try: + parents.append(field.queryset.get(**{to_field_name: value})) + except ObjectDoesNotExist: + self.add_error(field_name, _("Object not found: {value}").format(value=value)) + return None + except MultipleObjectsReturned: + self.add_error( + field_name, + _('"{value}" is not a unique value for this field; multiple objects were found').format( + value=value + ) + ) + return None + return parents + + @staticmethod + def _get_device_component_termination(model, device, name): + """ + Resolve a device component by its device and name. If the device is a virtual chassis + master and the component is not found on it, search all virtual chassis members. + """ + queryset = model.objects.filter(device=device, name=name) + if ( + device.virtual_chassis and + device.virtual_chassis.master == device and + not queryset.exists() + ): + queryset = model.objects.filter(device__in=device.virtual_chassis.members.all(), name=name) + return queryset.get() + def _clean_side(self, side): """ Derive a Cable's A/B termination objects. @@ -1586,60 +1674,89 @@ class CableImportForm(PrimaryModelImportForm): if side not in ('a', 'b'): raise ValueError(_("Invalid side designation: {side}").format(side=side)) - device = self.cleaned_data.get(f'side_{side}_device') - power_panel = self.cleaned_data.get(f'side_{side}_power_panel') content_type = self.cleaned_data.get(f'side_{side}_type') - name = self.cleaned_data.get(f'side_{side}_name') - if not content_type or not name: + # Native list values (JSON/YAML) bypass the CharField; strings use its cleaned value + names = self.data.get(f'side_{side}_name') + if not isinstance(names, (list, tuple)): + names = self.cleaned_data.get(f'side_{side}_name') + names = self._split_side_values(names) + if not content_type or not names: return None + if '' in names: + raise forms.ValidationError( + _("Side {side_upper}: Empty termination names are not permitted").format(side_upper=side.upper()) + ) + model = content_type.model_class() - # PowerFeed terminations reference a PowerPanel, not a Device + # Identify the parent field for the termination type. PowerFeed terminations reference a + # PowerPanel; all other supported types reference a Device. if content_type.model == 'powerfeed': - if not power_panel: - return None - try: - termination_object = model.objects.get(power_panel=power_panel, name=name) - if termination_object.cable is not None and termination_object.cable != self.instance: - raise forms.ValidationError( - _("Side {side_upper}: {power_panel} {termination_object} is already connected").format( - side_upper=side.upper(), power_panel=power_panel, termination_object=termination_object - ) - ) - except ObjectDoesNotExist: - raise forms.ValidationError( - _("{side_upper} side termination not found: {power_panel} {name}").format( - side_upper=side.upper(), power_panel=power_panel, name=name - ) - ) + parent_field_name = f'side_{side}_power_panel' + parent_label = _('power panel') + elif any(field.name == 'device' for field in model._meta.fields): + parent_field_name = f'side_{side}_device' + parent_label = _('device') else: - if not device: - return None + raise forms.ValidationError( + _("Bulk import does not support {type} terminations").format(type=content_type) + ) + + parents = self._resolve_side_parent_objects(parent_field_name) + if parents is None: + # The parent field has already raised its own validation error + return None + if not parents: + raise forms.ValidationError( + _("Side {side_upper}: Must specify a {parent} for the selected termination type").format( + side_upper=side.upper(), parent=parent_label + ) + ) + if len(parents) == 1: + parents = parents * len(names) + elif len(parents) != len(names): + raise forms.ValidationError( + _( + "Side {side_upper}: Must specify either one {parent} for all terminations or one {parent} " + "per termination name" + ).format(side_upper=side.upper(), parent=parent_label) + ) + + terminations = [] + for parent, name in zip(parents, names): try: - if ( - device.virtual_chassis and - device.virtual_chassis.master == device and - not model.objects.filter(device=device, name=name).exists() - ): - termination_object = model.objects.get(device__in=device.virtual_chassis.members.all(), name=name) + if content_type.model == 'powerfeed': + termination_object = model.objects.get(power_panel=parent, name=name) else: - termination_object = model.objects.get(device=device, name=name) - if termination_object.cable is not None and termination_object.cable != self.instance: - raise forms.ValidationError( - _("Side {side_upper}: {device} {termination_object} is already connected").format( - side_upper=side.upper(), device=device, termination_object=termination_object - ) - ) + termination_object = self._get_device_component_termination(model, parent, name) except ObjectDoesNotExist: raise forms.ValidationError( - _("{side_upper} side termination not found: {device} {name}").format( - side_upper=side.upper(), device=device, name=name + _("{side_upper} side termination not found: {parent} {name}").format( + side_upper=side.upper(), parent=parent, name=name ) ) + except MultipleObjectsReturned: + raise forms.ValidationError( + _("{side_upper} side termination not unique: {parent} {name}").format( + side_upper=side.upper(), parent=parent, name=name + ) + ) + if termination_object.cable is not None and termination_object.cable != self.instance: + raise forms.ValidationError( + _("Side {side_upper}: {parent} {termination_object} is already connected").format( + side_upper=side.upper(), parent=parent, termination_object=termination_object + ) + ) + terminations.append(termination_object) - setattr(self.instance, f'{side}_terminations', [termination_object]) - return termination_object + if len({termination.pk for termination in terminations}) != len(terminations): + raise forms.ValidationError( + _("Side {side_upper}: Duplicate termination specified").format(side_upper=side.upper()) + ) + + setattr(self.instance, f'{side}_terminations', terminations) + return terminations def _clean_color(self, color): """ diff --git a/netbox/dcim/tests/test_forms.py b/netbox/dcim/tests/test_forms.py index 7bf03075e..e501506b3 100644 --- a/netbox/dcim/tests/test_forms.py +++ b/netbox/dcim/tests/test_forms.py @@ -4,10 +4,13 @@ from django import forms from django.test import TestCase from dcim.choices import ( + CableEndChoices, + CableProfileChoices, DeviceFaceChoices, DeviceStatusChoices, InterfaceModeChoices, InterfaceTypeChoices, + LinkStatusChoices, PortTypeChoices, PowerOutletStatusChoices, ) @@ -553,12 +556,343 @@ class InterfaceTestCase(TestCase): class CableTestCase(TestCase): + @classmethod + def setUpTestData(cls): + cls.site = Site.objects.create(name='Site 1', slug='site-1') + cls.device_a = create_test_device('Device A', site=cls.site) + cls.device_b = create_test_device('Device B', site=cls.site) + cls.device_c = create_test_device('Device C', site=cls.site) + + cls.interfaces_a = ( + Interface(device=cls.device_a, name='et-0/0/0', type=InterfaceTypeChoices.TYPE_1GE_FIXED), + Interface(device=cls.device_a, name='et-0/0/1', type=InterfaceTypeChoices.TYPE_1GE_FIXED), + ) + cls.interfaces_b = ( + Interface(device=cls.device_b, name='et-0/0/0', type=InterfaceTypeChoices.TYPE_1GE_FIXED), + Interface(device=cls.device_b, name='et-0/0/1', type=InterfaceTypeChoices.TYPE_1GE_FIXED), + Interface(device=cls.device_b, name='et-0/0/2', type=InterfaceTypeChoices.TYPE_1GE_FIXED), + ) + cls.interface_c = Interface(device=cls.device_c, name='et-0/0/1', type=InterfaceTypeChoices.TYPE_1GE_FIXED) + Interface.objects.bulk_create([*cls.interfaces_a, *cls.interfaces_b, cls.interface_c]) + + cls.power_panel = PowerPanel.objects.create(site=cls.site, name='Power Panel 1') + cls.power_feeds = ( + PowerFeed(power_panel=cls.power_panel, name='Power Feed 1'), + PowerFeed(power_panel=cls.power_panel, name='Power Feed 2'), + ) + PowerFeed.objects.bulk_create(cls.power_feeds) + cls.power_ports = ( + PowerPort(device=cls.device_b, name='Power Port 1'), + PowerPort(device=cls.device_b, name='Power Port 2'), + ) + PowerPort.objects.bulk_create(cls.power_ports) + def test_invalid_side_designation_raises_value_error(self): """_clean_side rejects a side other than 'a' or 'b' with ValueError.""" form = CableImportForm.__new__(CableImportForm) with self.assertRaisesMessage(ValueError, "Invalid side designation: c"): form._clean_side('c') + def test_import_single_termination_cable(self): + """A single-value cell per side resolves one termination per side.""" + form = CableImportForm(data={ + 'side_a_site': 'Site 1', + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_site': 'Site 1', + 'side_b_device': 'Device B', + 'side_b_type': 'dcim.interface', + 'side_b_name': 'et-0/0/0', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertTrue(form.is_valid(), form.errors) + cable = form.save() + self.assertEqual(cable.a_terminations, [self.interfaces_a[0]]) + self.assertEqual(cable.b_terminations, [self.interfaces_b[0]]) + + def test_import_multiple_terminations_single_parent(self): + """A single parent value is reused for all comma-separated termination names.""" + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_device': 'Device B', + 'side_b_type': 'dcim.interface', + 'side_b_name': 'et-0/0/1, et-0/0/2', + 'status': LinkStatusChoices.STATUS_CONNECTED, + 'profile': CableProfileChoices.BREAKOUT_1C2P_2C1P, + }) + self.assertTrue(form.is_valid(), form.errors) + cable = form.save() + self.assertEqual(cable.a_terminations, [self.interfaces_a[0]]) + self.assertEqual(cable.b_terminations, [self.interfaces_b[1], self.interfaces_b[2]]) + + def test_import_multiple_terminations_multiple_parents_preserves_order(self): + """Pairwise parent/name lists resolve in submitted order, driving connector assignment.""" + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_device': 'Device C,Device B', + 'side_b_type': 'dcim.interface', + 'side_b_name': 'et-0/0/1,et-0/0/1', + 'status': LinkStatusChoices.STATUS_CONNECTED, + 'profile': CableProfileChoices.BREAKOUT_1C2P_2C1P, + }) + self.assertTrue(form.is_valid(), form.errors) + cable = form.save() + self.assertEqual(cable.b_terminations, [self.interface_c, self.interfaces_b[1]]) + + cable_terminations = CableTermination.objects.filter( + cable=cable, cable_end=CableEndChoices.SIDE_B + ).order_by('connector') + self.assertEqual([ct.termination for ct in cable_terminations], [self.interface_c, self.interfaces_b[1]]) + self.assertEqual([ct.connector for ct in cable_terminations], [1, 2]) + + def test_import_multiple_terminations_parent_count_mismatch(self): + """A parent list that is neither one value nor one per termination name is rejected.""" + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_device': 'Device B,Device C', + 'side_b_type': 'dcim.interface', + 'side_b_name': 'et-0/0/1,et-0/0/1,et-0/0/2', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertFalse(form.is_valid()) + self.assertIn('Must specify either one device', str(form.errors.get('side_b_name'))) + + def test_import_multiple_terminations_duplicate_termination(self): + """The same termination cannot be listed twice on one cable end.""" + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_device': 'Device B', + 'side_b_type': 'dcim.interface', + 'side_b_name': 'et-0/0/1,et-0/0/1', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertFalse(form.is_valid()) + self.assertIn('Duplicate termination', str(form.errors.get('side_b_name'))) + + def test_import_multiple_terminations_empty_name(self): + """A trailing comma produces an empty termination name and is rejected.""" + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_device': 'Device B', + 'side_b_type': 'dcim.interface', + 'side_b_name': 'et-0/0/1,', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertFalse(form.is_valid()) + self.assertIn('Empty termination names', str(form.errors.get('side_b_name'))) + + def test_import_multiple_terminations_connected_termination(self): + """An already-cabled termination in a multi-value list is rejected.""" + cable = Cable(a_terminations=[self.interfaces_a[1]], b_terminations=[self.interfaces_b[1]]) + cable.save() + + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_device': 'Device B', + 'side_b_type': 'dcim.interface', + 'side_b_name': 'et-0/0/1,et-0/0/2', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertFalse(form.is_valid()) + self.assertIn('already connected', str(form.errors.get('side_b_name'))) + + def test_import_multiple_terminations_power_feeds(self): + """Multiple power feeds import from a single broadcast power panel.""" + form = CableImportForm(data={ + 'side_a_power_panel': 'Power Panel 1', + 'side_a_type': 'dcim.powerfeed', + 'side_a_name': 'Power Feed 1,Power Feed 2', + 'side_b_device': 'Device B', + 'side_b_type': 'dcim.powerport', + 'side_b_name': 'Power Port 1,Power Port 2', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertTrue(form.is_valid(), form.errors) + cable = form.save() + self.assertEqual(cable.a_terminations, list(self.power_feeds)) + self.assertEqual(cable.b_terminations, list(self.power_ports)) + + def test_import_multiple_terminations_repeated_parent_values(self): + """A repeated parent in a pairwise list resolves per position, not deduplicated.""" + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_device': 'Device B,Device C,Device B', + 'side_b_type': 'dcim.interface', + 'side_b_name': 'et-0/0/1,et-0/0/1,et-0/0/2', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertTrue(form.is_valid(), form.errors) + cable = form.save() + self.assertEqual( + cable.b_terminations, + [self.interfaces_b[1], self.interface_c, self.interfaces_b[2]] + ) + + def test_import_multiple_terminations_native_lists(self): + """Native list values (JSON/YAML import) resolve like comma-separated cells.""" + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_device': ['Device C', 'Device B'], + 'side_b_type': 'dcim.interface', + 'side_b_name': ['et-0/0/1', 'et-0/0/1'], + 'status': LinkStatusChoices.STATUS_CONNECTED, + 'profile': CableProfileChoices.BREAKOUT_1C2P_2C1P, + }) + self.assertTrue(form.is_valid(), form.errors) + cable = form.save() + self.assertEqual(cable.b_terminations, [self.interface_c, self.interfaces_b[1]]) + + def test_import_multiple_terminations_unknown_parent(self): + """An unknown parent in a multi-value cell errors on the parent field only.""" + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_device': 'Device B,Device X', + 'side_b_type': 'dcim.interface', + 'side_b_name': 'et-0/0/1,et-0/0/2', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertFalse(form.is_valid()) + self.assertIn('Object not found: Device X', str(form.errors.get('side_b_device'))) + self.assertNotIn('side_b_name', form.errors) + + def test_import_multiple_terminations_missing_parent(self): + """A device component termination type without a device value is rejected.""" + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_type': 'dcim.interface', + 'side_b_name': 'et-0/0/1,et-0/0/2', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertFalse(form.is_valid()) + self.assertIn('Must specify a device', str(form.errors.get('side_b_name'))) + + def test_import_unsupported_termination_type(self): + """Termination types without a supported parent field are rejected.""" + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_device': 'Device B', + 'side_b_type': 'circuits.circuittermination', + 'side_b_name': 'Termination X', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertFalse(form.is_valid()) + self.assertIn('Bulk import does not support', str(form.errors.get('side_b_name'))) + + def test_import_unknown_termination_type(self): + """An unresolvable termination type errors on the type field only.""" + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_device': 'Device B', + 'side_b_type': 'dcim.nosuchmodel', + 'side_b_name': 'et-0/0/1', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertFalse(form.is_valid()) + self.assertIn('side_b_type', form.errors) + self.assertNotIn('side_b_name', form.errors) + + def test_import_multiple_terminations_unknown_name(self): + """An unknown termination name in a multi-value list is rejected.""" + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_device': 'Device B', + 'side_b_type': 'dcim.interface', + 'side_b_name': 'et-0/0/1,et-0/0/9', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertFalse(form.is_valid()) + self.assertIn('side termination not found', str(form.errors.get('side_b_name'))) + + def test_import_multiple_terminations_ambiguous_parent(self): + """A parent name matching multiple objects errors on the parent field.""" + site_2 = Site.objects.create(name='Site 2', slug='site-2') + create_test_device('Device D', site=self.site) + create_test_device('Device D', site=site_2) + + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_device': 'Device D', + 'side_b_type': 'dcim.interface', + 'side_b_name': 'et-0/0/1', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertFalse(form.is_valid()) + self.assertIn('is not a unique value', str(form.errors.get('side_b_device'))) + self.assertNotIn('side_b_name', form.errors) + + def test_import_multiple_terminations_site_filtered_parent_queryset(self): + """Parent resolution honors side_x_site queryset filtering for multi-value parents.""" + site_2 = Site.objects.create(name='Site 2', slug='site-2') + device_x = create_test_device('Device X', site=site_2) + Interface.objects.create(device=device_x, name='et-0/0/1', type=InterfaceTypeChoices.TYPE_1GE_FIXED) + + form = CableImportForm(data={ + 'side_a_site': 'Site 1', + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_site': 'Site 1', + 'side_b_device': 'Device B,Device X', + 'side_b_type': 'dcim.interface', + 'side_b_name': 'et-0/0/1,et-0/0/1', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertFalse(form.is_valid()) + self.assertIn('Object not found: Device X', str(form.errors.get('side_b_device'))) + self.assertNotIn('side_b_name', form.errors) + + def test_import_ambiguous_vc_component(self): + """A component name found on multiple VC members produces a form error.""" + vc = VirtualChassis.objects.create(name='Virtual Chassis 1') + master = create_test_device('VC Master', site=self.site, virtual_chassis=vc, vc_position=1) + member_2 = create_test_device('VC Member 2', site=self.site, virtual_chassis=vc, vc_position=2) + member_3 = create_test_device('VC Member 3', site=self.site, virtual_chassis=vc, vc_position=3) + vc.master = master + vc.save() + Interface.objects.create(device=member_2, name='vc-eth0', type=InterfaceTypeChoices.TYPE_1GE_FIXED) + Interface.objects.create(device=member_3, name='vc-eth0', type=InterfaceTypeChoices.TYPE_1GE_FIXED) + + form = CableImportForm(data={ + 'side_a_device': 'Device A', + 'side_a_type': 'dcim.interface', + 'side_a_name': 'et-0/0/0', + 'side_b_device': 'VC Master', + 'side_b_type': 'dcim.interface', + 'side_b_name': 'vc-eth0', + 'status': LinkStatusChoices.STATUS_CONNECTED, + }) + self.assertFalse(form.is_valid()) + self.assertIn('side termination not unique', str(form.errors.get('side_b_name'))) + class SiteFormTestCase(TestCase): """ diff --git a/netbox/dcim/tests/test_views.py b/netbox/dcim/tests/test_views.py index 3572a9c11..fc46b6190 100644 --- a/netbox/dcim/tests/test_views.py +++ b/netbox/dcim/tests/test_views.py @@ -4080,6 +4080,15 @@ class CableTestCase( "Power Panel 1,dcim.powerfeed,Power Feed 2,Device 4,dcim.powerport,Power Port 2", "Power Panel 1,dcim.powerfeed,Power Feed 3,Device 4,dcim.powerport,Power Port 3", ), + 'multi-termination': ( + # Ensure that a comma-separated cell imports multiple terminations per cable end, + # both with a single broadcast parent and with one parent per termination name. + "side_a_device,side_a_type,side_a_name,side_b_device,side_b_type,side_b_name,profile", + 'Device 3,dcim.interface,Interface 1,Device 4,dcim.interface,' + '"Interface 1,Interface 2",breakout-1c2p-2c1p', + 'Device 3,dcim.interface,Interface 2,"Device 4,Device 5",dcim.interface,' + '"Interface 3,Interface 1",breakout-1c2p-2c1p', + ), } cls.csv_update_data = ( @@ -4109,6 +4118,44 @@ class CableTestCase( return data + def test_bulk_import_unquoted_multi_value_cell(self): + """An unquoted multi-value cell is rejected with a column-count error.""" + self.add_permissions('dcim.add_cable') + csv_data = ( + "side_a_device,side_a_type,side_a_name,side_b_device,side_b_type,side_b_name,profile", + "Device 3,dcim.interface,Interface 1,Device 4,dcim.interface,Interface 1,Interface 2,breakout-1c2p-2c1p", + ) + initial_count = self._get_queryset().count() + data = { + 'data': '\n'.join(csv_data), + 'format': ImportFormatChoices.CSV, + 'csv_delimiter': CSVDelimiterChoices.AUTO, + } + + response = self.client.post(self._get_url('bulk_import'), data) + self.assertHttpStatus(response, 200) + self.assertIn('Expected 7 columns but found 8', response.content.decode()) + self.assertEqual(self._get_queryset().count(), initial_count) + + def test_bulk_import_unquoted_multi_value_cell_shifted_columns(self): + """An unquoted multi-value cell matching the column count is rejected by field validation.""" + self.add_permissions('dcim.add_cable') + csv_data = ( + "side_a_device,side_a_type,side_a_name,side_b_device,side_b_type,side_b_name,profile", + "Device 3,dcim.interface,Interface 1,Device 4,dcim.interface,Interface 1,Interface 2", + ) + initial_count = self._get_queryset().count() + data = { + 'data': '\n'.join(csv_data), + 'format': ImportFormatChoices.CSV, + 'csv_delimiter': CSVDelimiterChoices.AUTO, + } + + response = self.client.post(self._get_url('bulk_import'), data) + self.assertHttpStatus(response, 200) + self.assertIn('not one of the available choices', response.content.decode()) + self.assertEqual(self._get_queryset().count(), initial_count) + class VirtualChassisTestCase(ViewTestCases.PrimaryObjectViewTestCase): model = VirtualChassis