From 7472c5d0675617e78a02bbf09f160294113e3222 Mon Sep 17 00:00:00 2001 From: Arthur Hanson Date: Tue, 28 Jul 2026 07:02:41 -0700 Subject: [PATCH] 22752 - Restore rear port fields on front port bulk import (#22776) --- netbox/dcim/forms/bulk_import.py | 90 +++++++++++++++++++++++++++++++- netbox/dcim/tests/test_views.py | 51 ++++++++++++++++-- 2 files changed, 136 insertions(+), 5 deletions(-) diff --git a/netbox/dcim/forms/bulk_import.py b/netbox/dcim/forms/bulk_import.py index f756672fe..af38ff6fb 100644 --- a/netbox/dcim/forms/bulk_import.py +++ b/netbox/dcim/forms/bulk_import.py @@ -8,6 +8,7 @@ from django.utils.translation import gettext_lazy as _ from dcim.choices import * from dcim.constants import * from dcim.models import * +from dcim.utils import reconcile_port_mappings from extras.models import ConfigTemplate from ipam.choices import VLANQinQRoleChoices from ipam.models import VLAN, VRF, IPAddress, VLANGroup @@ -1118,13 +1119,100 @@ class FrontPortImportForm(OwnerCSVMixin, NetBoxModelImportForm): choices=PortTypeChoices, help_text=_('Physical medium classification') ) + rear_port = CSVModelChoiceField( + label=_('Rear port'), + queryset=RearPort.objects.all(), + to_field_name='name', + help_text=_('Corresponding rear port (mapped to the front port\'s first position)') + ) + rear_port_position = forms.IntegerField( + label=_('Rear port position'), + required=False, + help_text=_('Mapped position on the corresponding rear port (defaults to 1)') + ) class Meta: model = FrontPort fields = ( - 'device', 'name', 'label', 'type', 'color', 'mark_connected', 'positions', 'description', 'owner', 'tags' + 'device', 'name', 'label', 'type', 'color', 'mark_connected', 'positions', 'rear_port', + 'rear_port_position', 'description', 'owner', 'tags' ) + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + + # Limit RearPort choices to those belonging to this device (or VC master) + if self.is_bound and 'device' in self.data: + try: + device = self.fields['device'].to_python(self.data['device']) + except forms.ValidationError: + device = None + else: + try: + device = self.instance.device + except Device.DoesNotExist: + device = None + + if device: + self.fields['rear_port'].queryset = RearPort.objects.filter( + device__in=[device, device.get_vc_master()] + ) + else: + self.fields['rear_port'].queryset = RearPort.objects.none() + + def clean(self): + super().clean() + + rear_port = self.cleaned_data.get('rear_port') + rear_port_position = self.cleaned_data.get('rear_port_position') or 1 + if not rear_port: + return + + # Validate the rear port position against the selected rear port + if rear_port_position > rear_port.positions: + raise forms.ValidationError({ + 'rear_port_position': _( + "Invalid rear port position ({rear_port_position}): Rear port {name} has only {positions} " + "positions." + ).format( + rear_port_position=rear_port_position, + name=rear_port.name, + positions=rear_port.positions + ) + }) + + # Ensure the target rear port position isn't already occupied. reconcile_port_mappings() creates the + # mapping via create() (bypassing validate_unique()), so without this check a collision would surface + # as an uncaught IntegrityError (HTTP 500) rather than a row-level validation error. + occupied = PortMapping.objects.filter( + rear_port=rear_port, rear_port_position=rear_port_position + ).exclude(front_port=self.instance.pk) + if occupied.exists(): + raise forms.ValidationError({ + 'rear_port_position': _( + "Rear port {name} position {rear_port_position} is already occupied." + ).format( + name=rear_port.name, + rear_port_position=rear_port_position + ) + }) + + def _save_m2m(self): + super()._save_m2m() + + # Map the front port's first position to the specified rear port & position + if rear_port := self.cleaned_data.get('rear_port'): + reconcile_port_mappings( + PortMapping, + parent_field='front_port', + parent=self.instance, + desired=[{ + 'front_port_position': 1, + 'rear_port_id': rear_port.pk, + 'rear_port_position': self.cleaned_data.get('rear_port_position') or 1, + }], + ) + class RearPortImportForm(OwnerCSVMixin, NetBoxModelImportForm): device = CSVModelChoiceField( diff --git a/netbox/dcim/tests/test_views.py b/netbox/dcim/tests/test_views.py index 875226a98..cf5d8ce20 100644 --- a/netbox/dcim/tests/test_views.py +++ b/netbox/dcim/tests/test_views.py @@ -3522,10 +3522,10 @@ class FrontPortTestCase(ViewTestCases.DeviceComponentViewTestCase): } cls.csv_data = ( - "device,name,type,positions", - "Device 1,Front Port 4,8p8c,1", - "Device 1,Front Port 5,8p8c,1", - "Device 1,Front Port 6,8p8c,1", + "device,name,type,positions,rear_port,rear_port_position", + "Device 1,Front Port 4,8p8c,1,Rear Port 4,1", + "Device 1,Front Port 5,8p8c,1,Rear Port 5,1", + "Device 1,Front Port 6,8p8c,1,Rear Port 6,1", ) cls.csv_update_data = ( @@ -3535,6 +3535,49 @@ class FrontPortTestCase(ViewTestCases.DeviceComponentViewTestCase): f"{front_ports[2].pk},Front Port 9,New description9", ) + def test_bulk_import_objects_with_permission(self): + # Importing front ports with a rear_port (and position) should create the corresponding PortMapping + def check_port_mappings(scenario_name): + front_port = FrontPort.objects.get(name='Front Port 4') + mapping = PortMapping.objects.get(front_port=front_port) + self.assertEqual(mapping.rear_port.name, 'Rear Port 4') + self.assertEqual(mapping.front_port_position, 1) + self.assertEqual(mapping.rear_port_position, 1) + + super().test_bulk_import_objects_with_permission(post_import_callback=check_port_mappings) + + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[]) + def test_bulk_import_rear_port_position_exceeds_capacity(self): + # A rear_port_position beyond the rear port's capacity is rejected without creating the front port + self.add_permissions('dcim.add_frontport') + csv_data = ( + "device,name,type,positions,rear_port,rear_port_position", + "Device 1,Front Port 10,8p8c,1,Rear Port 4,2", + ) + response = self.client.post(self._get_url('bulk_import'), { + 'data': '\n'.join(csv_data), + 'format': ImportFormatChoices.CSV, + 'csv_delimiter': CSVDelimiterChoices.AUTO, + }) + self.assertEqual(response.status_code, 200) + self.assertFalse(FrontPort.objects.filter(name='Front Port 10').exists()) + + @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[]) + def test_bulk_import_rear_port_position_occupied(self): + # An already-occupied rear port position is rejected (rather than raising an IntegrityError) + self.add_permissions('dcim.add_frontport') + csv_data = ( + "device,name,type,positions,rear_port,rear_port_position", + "Device 1,Front Port 10,8p8c,1,Rear Port 1,1", + ) + response = self.client.post(self._get_url('bulk_import'), { + 'data': '\n'.join(csv_data), + 'format': ImportFormatChoices.CSV, + 'csv_delimiter': CSVDelimiterChoices.AUTO, + }) + self.assertEqual(response.status_code, 200) + self.assertFalse(FrontPort.objects.filter(name='Front Port 10').exists()) + def test_trace(self): self.add_permissions( 'dcim.view_frontport',