From c1d8ff1216758a599786f96f752b46ca1aaf6e15 Mon Sep 17 00:00:00 2001 From: Arthur Hanson Date: Tue, 14 Jul 2026 14:20:33 -0700 Subject: [PATCH] #22644 Add ObjectChange to PortMapping (#22645) --- netbox/dcim/api/serializers_/base.py | 36 +- netbox/dcim/forms/mixins.py | 56 +-- .../0239_add_portmapping_objectchange.py | 31 ++ netbox/dcim/models/base.py | 5 + .../dcim/models/device_component_templates.py | 7 +- netbox/dcim/models/device_components.py | 7 +- netbox/dcim/tests/test_port_mappings.py | 361 ++++++++++++++++++ netbox/dcim/utils.py | 49 +++ 8 files changed, 496 insertions(+), 56 deletions(-) create mode 100644 netbox/dcim/migrations/0239_add_portmapping_objectchange.py create mode 100644 netbox/dcim/tests/test_port_mappings.py diff --git a/netbox/dcim/api/serializers_/base.py b/netbox/dcim/api/serializers_/base.py index 9120ec109..8bcee54c3 100644 --- a/netbox/dcim/api/serializers_/base.py +++ b/netbox/dcim/api/serializers_/base.py @@ -3,6 +3,7 @@ from drf_spectacular.utils import extend_schema_field from rest_framework import serializers from dcim.models import FrontPort, FrontPortTemplate, PortMapping, PortTemplateMapping, RearPort, RearPortTemplate +from dcim.utils import reconcile_port_mappings from utilities.api import get_serializer_for_model __all__ = ( @@ -68,17 +69,25 @@ class PortSerializer(serializers.ModelSerializer): return PortTemplateMapping, 'rear_port' raise ValueError(f"Could not determine mapping details for {self.__class__}") + def _reconcile_mappings(self, instance, mappings): + mapping_model, fk_name = self._mapper + other_field = 'rear_port' if fk_name == 'front_port' else 'front_port' + + # Normalize the opposite-port FK from a model instance to its id so the mappings can be + # reconciled by value. + desired = [] + for attrs in mappings: + attrs = dict(attrs) + if other_field in attrs: + attrs[f'{other_field}_id'] = attrs.pop(other_field).pk + desired.append(attrs) + + reconcile_port_mappings(mapping_model, parent_field=fk_name, parent=instance, desired=desired) + def create(self, validated_data): mappings = validated_data.pop('mappings', []) instance = super().create(validated_data) - - # Create port mappings - mapping_model, fk_name = self._mapper - for attrs in mappings: - mapping_model.objects.create(**{ - fk_name: instance, - **attrs, - }) + self._reconcile_mappings(instance, mappings) return instance @@ -86,14 +95,9 @@ class PortSerializer(serializers.ModelSerializer): mappings = validated_data.pop('mappings', None) instance = super().update(instance, validated_data) + # Only reconcile when the client supplied rear_ports; a PATCH that omits it leaves the + # existing mappings untouched. if mappings is not None: - # Update port mappings - mapping_model, fk_name = self._mapper - mapping_model.objects.filter(**{fk_name: instance}).delete() - for attrs in mappings: - mapping_model.objects.create(**{ - fk_name: instance, - **attrs, - }) + self._reconcile_mappings(instance, mappings) return instance diff --git a/netbox/dcim/forms/mixins.py b/netbox/dcim/forms/mixins.py index 709d61222..6e0e82846 100644 --- a/netbox/dcim/forms/mixins.py +++ b/netbox/dcim/forms/mixins.py @@ -1,12 +1,11 @@ from django import forms from django.contrib.contenttypes.models import ContentType from django.core.exceptions import ObjectDoesNotExist, ValidationError -from django.db import connection -from django.db.models.signals import post_save from django.utils.translation import gettext_lazy as _ from dcim.constants import LOCATION_SCOPE_TYPES -from dcim.models import PortMapping, PortTemplateMapping, Site +from dcim.models import Site +from dcim.utils import reconcile_port_mappings from utilities.forms import get_field_value from utilities.forms.fields import ( ContentTypeChoiceField, @@ -204,43 +203,24 @@ class FrontPortFormMixin(forms.Form): def _save_m2m(self): super()._save_m2m() - # TODO: Can this be made more efficient? - # Delete existing rear port mappings - self.port_mapping_model.objects.filter(front_port_id=self.instance.pk).delete() - - # Create new rear port mappings - mappings = [] - if self.port_mapping_model is PortTemplateMapping: - params = { - 'device_type_id': self.instance.device_type_id, - 'module_type_id': self.instance.module_type_id, - } - else: - params = { - 'device_id': self.instance.device_id, - } + # Build the desired set of mappings from the submitted rear port pairs, assigning front port + # positions in order. reconcile_port_mappings() then writes only the difference, so re-saving + # a front port without changing its wiring produces no writes (and no changelog churn). + desired = [] for i, rp_position in enumerate(self.cleaned_data['rear_ports'], start=1): rear_port_id, rear_port_position = rp_position.split(':') - mappings.append( - self.port_mapping_model(**{ - **params, - 'front_port_id': self.instance.pk, - 'front_port_position': i, - 'rear_port_id': rear_port_id, - 'rear_port_position': rear_port_position, - }) - ) - self.port_mapping_model.objects.bulk_create(mappings) - # Send post_save signals - for mapping in mappings: - post_save.send( - sender=PortMapping, - instance=mapping, - created=True, - raw=False, - using=connection, - update_fields=None - ) + desired.append({ + 'front_port_position': i, + 'rear_port_id': int(rear_port_id), + 'rear_port_position': int(rear_port_position), + }) + + reconcile_port_mappings( + self.port_mapping_model, + parent_field='front_port', + parent=self.instance, + desired=desired, + ) def _get_rear_port_choices(self, parent_filter, front_port): """ diff --git a/netbox/dcim/migrations/0239_add_portmapping_objectchange.py b/netbox/dcim/migrations/0239_add_portmapping_objectchange.py new file mode 100644 index 000000000..2df74ef7e --- /dev/null +++ b/netbox/dcim/migrations/0239_add_portmapping_objectchange.py @@ -0,0 +1,31 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("dcim", "0238_alter_cable__abs_length"), + ] + + operations = [ + migrations.AddField( + model_name="portmapping", + name="created", + field=models.DateTimeField(auto_now_add=True, null=True), + ), + migrations.AddField( + model_name="portmapping", + name="last_updated", + field=models.DateTimeField(auto_now=True, null=True), + ), + migrations.AddField( + model_name="porttemplatemapping", + name="created", + field=models.DateTimeField(auto_now_add=True, null=True), + ), + migrations.AddField( + model_name="porttemplatemapping", + name="last_updated", + field=models.DateTimeField(auto_now=True, null=True), + ), + ] diff --git a/netbox/dcim/models/base.py b/netbox/dcim/models/base.py index f8021d4db..f80a037e2 100644 --- a/netbox/dcim/models/base.py +++ b/netbox/dcim/models/base.py @@ -29,6 +29,8 @@ class PortMappingBase(models.Model): ), ) + # Change-logged but private: no public API/URL, and the delete-time UPDATE cascade onto the + # parent ports stays suppressed (see #21390, #22270). _netbox_private = True class Meta: @@ -44,6 +46,9 @@ class PortMappingBase(models.Model): ), ) + def __str__(self): + return f'{self.front_port}:{self.front_port_position} to {self.rear_port}:{self.rear_port_position}' + def clean(self): super().clean() diff --git a/netbox/dcim/models/device_component_templates.py b/netbox/dcim/models/device_component_templates.py index dec8758a9..26abb36d8 100644 --- a/netbox/dcim/models/device_component_templates.py +++ b/netbox/dcim/models/device_component_templates.py @@ -11,6 +11,7 @@ from dcim.models.base import PortMappingBase from dcim.models.mixins import InterfaceValidationMixin from dcim.utils import get_module_bay_positions, resolve_module_placeholder from netbox.models import ChangeLoggedModel +from netbox.models.features import ChangeLoggingMixin from utilities.fields import ColorField, NaturalOrderingField from utilities.mptt import TreeManager from utilities.ordering import naturalize_interface @@ -538,7 +539,7 @@ class InterfaceTemplate(InterfaceValidationMixin, ModularComponentTemplateModel) } -class PortTemplateMapping(PortMappingBase): +class PortTemplateMapping(ChangeLoggingMixin, PortMappingBase): """ Maps a FrontPortTemplate & position to a RearPortTemplate & position. """ @@ -567,6 +568,10 @@ class PortTemplateMapping(PortMappingBase): related_name='mappings', ) + class Meta(PortMappingBase.Meta): + # Inherit the unique constraints from PortMappingBase.Meta. + pass + def clean(self): super().clean() diff --git a/netbox/dcim/models/device_components.py b/netbox/dcim/models/device_components.py index 717bb5de4..25006344e 100644 --- a/netbox/dcim/models/device_components.py +++ b/netbox/dcim/models/device_components.py @@ -15,6 +15,7 @@ from dcim.models.base import PortMappingBase from dcim.models.mixins import InterfaceValidationMixin from netbox.choices import ColorChoices from netbox.models import NetBoxModel, OrganizationalModel +from netbox.models.features import ChangeLoggingMixin from netbox.models.mixins import OwnerMixin from utilities.fields import ColorField, NaturalOrderingField from utilities.mptt import TreeManager @@ -1196,7 +1197,7 @@ class Interface( # Pass-through ports # -class PortMapping(PortMappingBase): +class PortMapping(ChangeLoggingMixin, PortMappingBase): """ Maps a FrontPort & position to a RearPort & position. """ @@ -1216,6 +1217,10 @@ class PortMapping(PortMappingBase): related_name='mappings', ) + class Meta(PortMappingBase.Meta): + # Inherit the unique constraints from PortMappingBase.Meta. + pass + def clean(self): super().clean() diff --git a/netbox/dcim/tests/test_port_mappings.py b/netbox/dcim/tests/test_port_mappings.py new file mode 100644 index 000000000..622d53f5a --- /dev/null +++ b/netbox/dcim/tests/test_port_mappings.py @@ -0,0 +1,361 @@ +import uuid + +from django.contrib.contenttypes.models import ContentType +from django.test import RequestFactory, TestCase, tag +from django.urls import reverse +from rest_framework import status + +from core.choices import ObjectChangeActionChoices +from core.models import ObjectChange +from dcim.choices import PortTypeChoices +from dcim.models import ( + Device, + DeviceRole, + DeviceType, + FrontPort, + FrontPortTemplate, + Manufacturer, + PortMapping, + PortTemplateMapping, + RearPort, + RearPortTemplate, + Site, +) +from dcim.utils import reconcile_port_mappings +from netbox.context_managers import event_tracking +from users.models import User +from utilities.testing import APITestCase + + +def _build_request(user): + request = RequestFactory().get('/') + request.id = uuid.uuid4() + request.user = user + return request + + +class ReconcilePortMappingsTestCase(TestCase): + """ + Exercise dcim.utils.reconcile_port_mappings and confirm that PortMapping now participates in + change logging (#22644): only the difference is written, so unchanged mappings keep their PK and + emit no ObjectChange. + """ + @classmethod + def setUpTestData(cls): + cls.user = User.objects.create_user(username='testuser', password='pw') + + manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1') + device_type = DeviceType.objects.create(manufacturer=manufacturer, model='Device Type 1', slug='device-type-1') + role = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1', color='ff0000') + site = Site.objects.create(name='Site 1', slug='site-1') + cls.device = Device.objects.create(device_type=device_type, role=role, name='Device 1', site=site) + + cls.front_port = FrontPort.objects.create( + device=cls.device, name='Front Port 1', type=PortTypeChoices.TYPE_8P8C, positions=4 + ) + cls.rear_ports = [ + RearPort.objects.create( + device=cls.device, name=f'Rear Port {i}', type=PortTypeChoices.TYPE_8P8C, positions=4 + ) + for i in range(1, 4) + ] + + def _desired(self, *pairs): + # Each pair is (front_port_position, rear_port, rear_port_position). + return [ + {'front_port_position': fpp, 'rear_port_id': rp.pk, 'rear_port_position': rpp} + for fpp, rp, rpp in pairs + ] + + def _reconcile(self, desired): + request = _build_request(self.user) + with event_tracking(request): + reconcile_port_mappings(PortMapping, parent_field='front_port', parent=self.front_port, desired=desired) + + def _mapping_changes(self, action=None): + changes = ObjectChange.objects.filter(changed_object_type=ContentType.objects.get_for_model(PortMapping)) + if action is not None: + changes = changes.filter(action=action) + return changes + + def _current_pks(self): + return set(PortMapping.objects.filter(front_port=self.front_port).values_list('pk', flat=True)) + + def test_create_records_objectchange(self): + self._reconcile(self._desired( + (1, self.rear_ports[0], 1), + (2, self.rear_ports[1], 1), + )) + + self.assertEqual(PortMapping.objects.filter(front_port=self.front_port).count(), 2) + self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_CREATE).count(), 2) + + @tag('regression') # Ref: #22644 + def test_noop_resave_writes_nothing(self): + # Re-saving a front port without changing its wiring must not mint new PKs or emit changelog + # entries — this is what previously broke branch merges (colliding on the unique constraint + # when both sides replayed DELETE(old-pk) + CREATE(new-pk)). + desired = self._desired( + (1, self.rear_ports[0], 1), + (2, self.rear_ports[1], 1), + ) + self._reconcile(desired) + original_pks = self._current_pks() + ObjectChange.objects.all().delete() + + self._reconcile(desired) + + self.assertEqual(self._mapping_changes().count(), 0) + self.assertEqual(self._current_pks(), original_pks) + + def test_repointing_a_slot_deletes_and_creates(self): + self._reconcile(self._desired((1, self.rear_ports[0], 1))) + original_pk = self._current_pks().pop() + ObjectChange.objects.all().delete() + + # Same front port position, different rear port: the slot is re-pointed. + self._reconcile(self._desired((1, self.rear_ports[1], 1))) + + mappings = PortMapping.objects.filter(front_port=self.front_port) + self.assertEqual(mappings.count(), 1) + mapping = mappings.first() + self.assertEqual(mapping.rear_port_id, self.rear_ports[1].pk) + self.assertNotEqual(mapping.pk, original_pk) + self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_DELETE).count(), 1) + self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_CREATE).count(), 1) + + def test_unchanged_rows_survive_alongside_changed_rows(self): + self._reconcile(self._desired( + (1, self.rear_ports[0], 1), + (2, self.rear_ports[1], 1), + )) + unchanged_pk = PortMapping.objects.get(front_port=self.front_port, front_port_position=1).pk + ObjectChange.objects.all().delete() + + # Position 1 is untouched; position 2 is re-pointed to a third rear port. + self._reconcile(self._desired( + (1, self.rear_ports[0], 1), + (2, self.rear_ports[2], 1), + )) + + self.assertEqual(PortMapping.objects.get(front_port=self.front_port, front_port_position=1).pk, unchanged_pk) + self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_CREATE).count(), 1) + self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_DELETE).count(), 1) + + def test_removing_a_mapping_records_delete(self): + self._reconcile(self._desired( + (1, self.rear_ports[0], 1), + (2, self.rear_ports[1], 1), + )) + ObjectChange.objects.all().delete() + + self._reconcile(self._desired((1, self.rear_ports[0], 1))) + + self.assertEqual(PortMapping.objects.filter(front_port=self.front_port).count(), 1) + self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_DELETE).count(), 1) + self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_CREATE).count(), 0) + + def test_swapping_positions_preserves_constraints(self): + # Two front port positions pointing at the same rear port's positions 1 and 2. + self._reconcile(self._desired( + (1, self.rear_ports[0], 1), + (2, self.rear_ports[0], 2), + )) + + # Swap their rear port positions. Reconcile deletes both changed rows before recreating, so + # the transient state never violates the (rear_port, rear_port_position) unique constraint. + self._reconcile(self._desired( + (1, self.rear_ports[0], 2), + (2, self.rear_ports[0], 1), + )) + + self.assertEqual( + PortMapping.objects.get(front_port=self.front_port, front_port_position=1).rear_port_position, 2 + ) + self.assertEqual( + PortMapping.objects.get(front_port=self.front_port, front_port_position=2).rear_port_position, 1 + ) + + def test_direct_delete_records_objectchange(self): + self._reconcile(self._desired((1, self.rear_ports[0], 1))) + mapping = PortMapping.objects.get(front_port=self.front_port) + mapping_pk = mapping.pk + ObjectChange.objects.all().delete() + + request = _build_request(self.user) + with event_tracking(request): + mapping.delete() + + self.assertTrue( + self._mapping_changes(ObjectChangeActionChoices.ACTION_DELETE).filter(changed_object_id=mapping_pk).exists() + ) + + +class ReconcilePortTemplateMappingsTestCase(TestCase): + """ + Confirm reconcile_port_mappings and change logging behave identically for PortTemplateMapping. + """ + @classmethod + def setUpTestData(cls): + cls.user = User.objects.create_user(username='testuser', password='pw') + + manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1') + cls.device_type = DeviceType.objects.create( + manufacturer=manufacturer, model='Device Type 1', slug='device-type-1' + ) + + cls.front_port = FrontPortTemplate.objects.create( + device_type=cls.device_type, name='Front Port 1', type=PortTypeChoices.TYPE_8P8C, positions=4 + ) + cls.rear_ports = [ + RearPortTemplate.objects.create( + device_type=cls.device_type, name=f'Rear Port {i}', type=PortTypeChoices.TYPE_8P8C, positions=4 + ) + for i in range(1, 3) + ] + + def _desired(self, *pairs): + return [ + {'front_port_position': fpp, 'rear_port_id': rp.pk, 'rear_port_position': rpp} + for fpp, rp, rpp in pairs + ] + + def _reconcile(self, desired): + request = _build_request(self.user) + with event_tracking(request): + reconcile_port_mappings( + PortTemplateMapping, + parent_field='front_port', + parent=self.front_port, + desired=desired, + ) + + def _mapping_changes(self, action=None): + changes = ObjectChange.objects.filter( + changed_object_type=ContentType.objects.get_for_model(PortTemplateMapping) + ) + if action is not None: + changes = changes.filter(action=action) + return changes + + def test_create_records_objectchange(self): + self._reconcile(self._desired((1, self.rear_ports[0], 1))) + + mapping = PortTemplateMapping.objects.get(front_port=self.front_port) + self.assertEqual(mapping.device_type_id, self.device_type.pk) + self.assertEqual(self._mapping_changes(ObjectChangeActionChoices.ACTION_CREATE).count(), 1) + + @tag('regression') # Ref: #22644 + def test_noop_resave_writes_nothing(self): + desired = self._desired((1, self.rear_ports[0], 1)) + self._reconcile(desired) + original_pk = PortTemplateMapping.objects.get(front_port=self.front_port).pk + ObjectChange.objects.all().delete() + + self._reconcile(desired) + + self.assertEqual(self._mapping_changes().count(), 0) + self.assertEqual(PortTemplateMapping.objects.get(front_port=self.front_port).pk, original_pk) + + +class PortMappingAPITestCase(APITestCase): + """ + Exercise the reconcile behaviour through PortSerializer.create()/update() over the REST API, + covering the model-instance -> _id normalization in PortSerializer._reconcile_mappings. + """ + @classmethod + def setUpTestData(cls): + manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1') + device_type = DeviceType.objects.create(manufacturer=manufacturer, model='Device Type 1', slug='device-type-1') + role = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1', color='ff0000') + site = Site.objects.create(name='Site 1', slug='site-1') + cls.device = Device.objects.create(device_type=device_type, role=role, name='Device 1', site=site) + + cls.rear_ports = [ + RearPort.objects.create( + device=cls.device, name=f'Rear Port {i}', type=PortTypeChoices.TYPE_8P8C, positions=4 + ) + for i in range(1, 4) + ] + + # An existing front port with two mappings, for the update/PATCH tests. + cls.front_port = FrontPort.objects.create( + device=cls.device, name='Front Port 1', type=PortTypeChoices.TYPE_8P8C, positions=2 + ) + PortMapping.objects.bulk_create([ + PortMapping( + device=cls.device, front_port=cls.front_port, front_port_position=1, + rear_port=cls.rear_ports[0], rear_port_position=1, + ), + PortMapping( + device=cls.device, front_port=cls.front_port, front_port_position=2, + rear_port=cls.rear_ports[1], rear_port_position=1, + ), + ]) + + def _mapping_pks(self, front_port): + return set(PortMapping.objects.filter(front_port=front_port).values_list('pk', flat=True)) + + def _mapping_creates(self): + return ObjectChange.objects.filter( + changed_object_type=ContentType.objects.get_for_model(PortMapping), + action=ObjectChangeActionChoices.ACTION_CREATE, + ) + + def test_create_records_mappings_and_changelog(self): + # Confirms PortSerializer.create() + _reconcile_mappings normalization (rear_port instance -> + # rear_port_id) writes the mappings and records their ObjectChanges. + self.add_permissions('dcim.add_frontport', 'dcim.view_frontport', 'dcim.view_rearport', 'dcim.view_device') + data = { + 'device': self.device.pk, + 'name': 'Front Port 2', + 'type': PortTypeChoices.TYPE_8P8C, + 'positions': 2, + 'rear_ports': [ + {'position': 1, 'rear_port': self.rear_ports[2].pk, 'rear_port_position': 1}, + {'position': 2, 'rear_port': self.rear_ports[2].pk, 'rear_port_position': 2}, + ], + } + response = self.client.post(reverse('dcim-api:frontport-list'), data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_201_CREATED) + + front_port = FrontPort.objects.get(pk=response.data['id']) + mappings = PortMapping.objects.filter(front_port=front_port).order_by('front_port_position') + self.assertEqual(mappings.count(), 2) + self.assertEqual(mappings[0].rear_port_id, self.rear_ports[2].pk) + self.assertEqual(mappings[1].rear_port_position, 2) + self.assertEqual( + self._mapping_creates().filter(changed_object_id__in=mappings.values_list('pk', flat=True)).count(), 2 + ) + + def test_update_omitting_rear_ports_preserves_mappings(self): + # A PATCH that omits rear_ports must leave the existing mappings untouched. + self.add_permissions('dcim.change_frontport', 'dcim.view_frontport') + url = reverse('dcim-api:frontport-detail', kwargs={'pk': self.front_port.pk}) + original_pks = self._mapping_pks(self.front_port) + + response = self.client.patch(url, {'description': 'Updated'}, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + + self.assertEqual(self._mapping_pks(self.front_port), original_pks) + + def test_update_rewires_mappings(self): + # Re-point both slots to new rear port pairs; reconcile deletes the old rows and creates the + # replacements. + self.add_permissions('dcim.change_frontport', 'dcim.view_frontport', 'dcim.view_rearport', 'dcim.view_device') + url = reverse('dcim-api:frontport-detail', kwargs={'pk': self.front_port.pk}) + original_pks = self._mapping_pks(self.front_port) + + data = { + 'rear_ports': [ + {'position': 1, 'rear_port': self.rear_ports[2].pk, 'rear_port_position': 3}, + {'position': 2, 'rear_port': self.rear_ports[2].pk, 'rear_port_position': 4}, + ], + } + response = self.client.patch(url, data, format='json', **self.header) + self.assertHttpStatus(response, status.HTTP_200_OK) + + mappings = PortMapping.objects.filter(front_port=self.front_port).order_by('front_port_position') + self.assertEqual([m.rear_port_id for m in mappings], [self.rear_ports[2].pk, self.rear_ports[2].pk]) + self.assertEqual([m.rear_port_position for m in mappings], [3, 4]) + self.assertTrue(self._mapping_pks(self.front_port).isdisjoint(original_pks)) diff --git a/netbox/dcim/utils.py b/netbox/dcim/utils.py index ce4dd15ef..24547390f 100644 --- a/netbox/dcim/utils.py +++ b/netbox/dcim/utils.py @@ -168,4 +168,53 @@ def create_port_mappings(device, device_or_module_type, module=None): rear_port_position=template.rear_port_position, ) ) + # Bulk-created (no per-mapping ObjectChange) to match how every other component is instantiated. PortMapping.objects.bulk_create(mappings) + + +def reconcile_port_mappings(mapping_model, parent_field, parent, desired): + """ + Reconcile a parent port's mappings against `desired`, writing only the difference so unchanged + mappings keep their PK (and emit no changelog entry). Changed/removed rows are deleted before + replacements are created, all in one transaction, so position swaps don't trip the unique + constraint. Per-row create()/delete() let the change-logging signals fire naturally. + + Args: + mapping_model: PortMapping or PortTemplateMapping. + parent_field: 'front_port' or 'rear_port' — the side being edited; its '_position' + is each mapping's stable identity within the set. + parent: the parent instance (FrontPort/RearPort or their templates). + desired: iterable of dicts of mapping field values EXCLUDING the parent FK, using '_id' + for the opposite-port FK, e.g. {'front_port_position': 1, 'rear_port_id': 5, + 'rear_port_position': 2}. save() derives device/device_type/module_type from the front port. + """ + key_field = f'{parent_field}_position' + other_field = 'rear_port' if parent_field == 'front_port' else 'front_port' + value_fields = (f'{other_field}_id', f'{other_field}_position') + + def target(source): + # The comparable "value" of a mapping: the opposite port and its position. Two mappings with + # the same parent-side position but a different target represent a re-pointing of that slot. + get = source.get if isinstance(source, dict) else lambda f: getattr(source, f) + return tuple(get(f) for f in value_fields) + + desired_by_key = {d[key_field]: d for d in desired} + + with transaction.atomic(using=router.db_for_write(mapping_model)): + # Lock the parent's existing mappings for the duration of the reconcile. Two requests editing + # the same port would otherwise read the same snapshot and race, the second colliding on a + # unique constraint when it recreates rows the first has already committed. + existing = { + getattr(m, key_field): m + for m in mapping_model.objects.filter(**{parent_field: parent}).select_for_update() + } + + # Delete rows that no longer exist or whose target changed (before creating, to free the slots) + for key, mapping in existing.items(): + if key not in desired_by_key or target(mapping) != target(desired_by_key[key]): + mapping.delete() + + # Create rows that are new or whose target changed + for key, attrs in desired_by_key.items(): + if key not in existing or target(existing[key]) != target(attrs): + mapping_model.objects.create(**{parent_field: parent, **attrs})