RomanNum3ral
/
netbox
mirror of https://github.com/netbox-community/netbox.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(dcim): Support multiple Terminations per side in Cable bulk import (#22641) 

Enable comma-separated Device, Power Panel, and Termination name lists
in Cable CSV/JSON/YAML imports. Each side accepts either one parent for
all terminations or one parent per name, preserving submission order for
connector assignment.

Add validation for duplicate terminations, empty names, parent count
mismatches, and MultipleObjectsReturned cases. Change side_a/b_device
and side_a/b_power_panel fields from CSVModelChoiceField to
CSVModelMultipleChoiceField with updated help text.

Fixes #18645
This commit is contained in:
Martin Hauser 2026-07-10 19:27:31 +02:00 committed by GitHub
parent 517804758f
commit 8e525c89fb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 549 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

219
netbox/dcim/forms/bulk_import.py
View File

@ -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(
'{} <code>{}</code>',
_('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(
'{} <code>{}</code>',
_('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(
'{} <code>{}</code>',
_('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(
'{} <code>{}</code>',
_('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(
'{} <code>{}</code>',
_('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(
'{} <code>{}</code>',
_('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):
"""

334
netbox/dcim/tests/test_forms.py
View File

@ -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):
"""

47
netbox/dcim/tests/test_views.py
View File

@ -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