22752 - Restore rear port fields on front port bulk import (#22776)

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(

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',