Просмотр исходного кода

fix(dcim): Preserve cable end order when terminations unchanged

Compare cable terminations against stored values instead of potentially
stale prefetched relations when checking for modifications. Skip setting
terminations in the form's clean() when a saved cable's members are
unchanged, preserving the connector order assigned by the profile.
Martin Hauser 2 месяцев назад
Родитель
Сommit
fac9601216

+ 5 - 3
netbox/dcim/forms/connections.py

@@ -142,8 +142,10 @@ def get_cable_form(a_type, b_type):
         def clean(self):
             super().clean()
 
-            # Set the A/B terminations on the Cable instance
-            self.instance.a_terminations = self.cleaned_data.get('a_terminations', [])
-            self.instance.b_terminations = self.cleaned_data.get('b_terminations', [])
+            # The field discards submission order, so a saved cable's end is assigned only when its members changed
+            for field_name in ('a_terminations', 'b_terminations'):
+                value = self.cleaned_data.get(field_name, [])
+                if not self.instance.pk or set(value) != set(self.initial.get(field_name, [])):
+                    setattr(self.instance, field_name, value)
 
     return _CableForm

+ 15 - 2
netbox/dcim/models/cables.py

@@ -229,6 +229,16 @@ class Cable(PrimaryModel):
             ct.termination for ct in self.terminations.all() if ct.cable_end == side
         ]
 
+    def _cache_stored_terminations(self):
+        """
+        Fill each cold termination cache from the CableTermination rows, in their stored order.
+        """
+        a_terminations, b_terminations = self.get_terminations()
+        if not hasattr(self, '_a_terminations'):
+            self._a_terminations = list(a_terminations.keys())
+        if not hasattr(self, '_b_terminations'):
+            self._b_terminations = list(b_terminations.keys())
+
     def _set_x_terminations(self, side, value):
         """
         Set the terminating objects for the given cable end (A or B).
@@ -244,8 +254,11 @@ class Cable(PrimaryModel):
                 ct.termination for ct in CableTermination.objects.filter(pk__in=value).prefetch_related('termination')
             ]
 
-        # Compare against the cached terminations, falling back to the stored rows on a freshly loaded cable
-        if not self.pk or self._get_x_terminations(side) != list(value):
+        # Compare a saved cable against its stored rows, not against a possibly stale prefetch of self.terminations
+        if self.pk and not hasattr(self, _attr):
+            self._cache_stored_terminations()
+
+        if not self.pk or getattr(self, _attr) != list(value):
             self._terminations_modified = True
 
         setattr(self, _attr, value)

+ 1 - 1
netbox/dcim/tests/test_cablepaths.py

@@ -2910,7 +2910,7 @@ class LegacyCablePathTestCase(BaseCablePathTestCase):
         self.assertEqual(len(path_pks), 2)
         self.assertEqual(len(termination_pks), 2)
 
-        # Reassign the same terminations, as the edit form does on every submission
+        # Reassign the same terminations on a freshly loaded instance
         cable1 = Cable.objects.get(pk=cable1.pk)
         cable1.a_terminations = [interface1]
         cable1.b_terminations = [interface2]

+ 2 - 2
netbox/dcim/tests/test_cablepaths2.py

@@ -2790,7 +2790,7 @@ class CablePathTestCase(BaseCablePathTestCase):
         """
         [IF1] --C1-- [IF2]
 
-        Applying a profile after the edit form re-assigned the unchanged terminations rebuilds the paths.
+        Applying a profile after both termination caches have been populated must still rebuild the paths.
         """
         interfaces = [
             Interface.objects.create(device=self.device, name='Interface 1'),
@@ -2806,7 +2806,7 @@ class CablePathTestCase(BaseCablePathTestCase):
         cable1.save()
         self.assertEqual(CablePath.objects.count(), 2)
 
-        # Reload, then re-assign the stored terminations as the edit form does on every submission
+        # Reload and populate both termination caches by reassigning their stored values
         cable1 = Cable.objects.get(pk=cable1.pk)
         cable1.a_terminations = [interfaces[0]]
         cable1.b_terminations = [interfaces[1]]

+ 17 - 0
netbox/dcim/tests/test_models.py

@@ -2440,6 +2440,23 @@ class CableTestCase(TestCase):
 
         self.assertTrue(cable._terminations_modified)
 
+    def test_reassigning_stale_prefetched_terminations_flags_a_change(self):
+        """
+        A stale prefetched relation must not hide a real termination change.
+        """
+        cable = Cable.objects.prefetch_related('terminations__termination').first()
+        stale_termination = cable.b_terminations[0]
+        current_termination = Interface.objects.get(device__name='TestDevice2', name='eth1')
+
+        # Moving the B end through a second instance leaves the prefetch above stale
+        moved = Cable.objects.get(pk=cable.pk)
+        moved.b_terminations = [current_termination]
+        moved.save()
+
+        # The value matches the stale prefetch but not the stored row
+        cable.b_terminations = [stale_termination]
+        self.assertTrue(cable._terminations_modified)
+
     def test_partial_save_does_not_apply_an_unwritten_profile(self):
         """
         A save excluding profile must leave the terminations alone but keep the change pending.

+ 76 - 0
netbox/dcim/tests/test_views.py

@@ -5306,6 +5306,82 @@ class CableTestCase(
             path_pks
         )
 
+    @tag('regression')  # Issue #23097
+    def test_edit_with_unchanged_terminations_preserves_connector_order(self):
+        """A label-only edit must keep the connectors of an end whose stored order differs from the form's."""
+        # The form's termination fields are restricted by view permission
+        self.add_permissions('dcim.change_cable', 'dcim.view_interface')
+
+        interface_a = Interface.objects.get(device__name='Device 3', name='Interface 1')
+        interfaces = list(Interface.objects.filter(device__name='Device 4').order_by('name')[:2])
+        cable = Cable(
+            a_terminations=[interface_a],
+            b_terminations=[interfaces[1], interfaces[0]],
+            profile=CableProfileChoices.BREAKOUT_1C2P_2C1P,
+        )
+        cable.save()
+
+        def b_terminations():
+            return list(
+                CableTermination.objects.filter(cable=cable, cable_end=CableEndChoices.SIDE_B)
+                .values_list('pk', 'connector', 'termination_id')
+            )
+
+        terminations = b_terminations()
+        self.assertEqual([t[1:] for t in terminations], [(1, interfaces[1].pk), (2, interfaces[0].pk)])
+        path_pks = set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True))
+
+        data = {
+            'a_terminations': [interface_a.pk],
+            'b_terminations': [interfaces[0].pk, interfaces[1].pk],
+            'profile': CableProfileChoices.BREAKOUT_1C2P_2C1P,
+            'status': LinkStatusChoices.STATUS_CONNECTED,
+            'label': 'Renamed',
+        }
+        request = {
+            'path': self._get_url('edit', cable),
+            'data': post_data(data),
+        }
+        self.assertHttpStatus(self.client.post(**request), 302)
+
+        cable.refresh_from_db()
+        self.assertEqual(cable.label, 'Renamed')
+        self.assertEqual(b_terminations(), terminations)
+        self.assertEqual(
+            set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True)),
+            path_pks
+        )
+
+    def test_edit_with_changed_terminations_rewires_the_end(self):
+        """Replacing a termination through the edit form must still rewrite that end."""
+        # The form's termination fields are restricted by view permission
+        self.add_permissions('dcim.change_cable', 'dcim.view_interface')
+
+        interface_a = Interface.objects.get(
+            device__name='Device 1', device__site__name='Site 1', name='Interface 1'
+        )
+        cable = interface_a.cable
+        interface_b = cable.b_terminations[0]
+        new_interface_b = Interface.objects.get(device__name='Device 4', name='Interface 3')
+
+        data = {
+            'a_terminations': [interface_a.pk],
+            'b_terminations': [new_interface_b.pk],
+            'type': CableTypeChoices.TYPE_CAT6,
+            'status': LinkStatusChoices.STATUS_CONNECTED,
+        }
+        request = {
+            'path': self._get_url('edit', cable),
+            'data': post_data(data),
+        }
+        self.assertHttpStatus(self.client.post(**request), 302)
+
+        self.assertEqual(Cable.objects.get(pk=cable.pk).b_terminations, [new_interface_b])
+        interface_b.refresh_from_db()
+        self.assertIsNone(interface_b.cable)
+        new_interface_b.refresh_from_db()
+        self.assertEqual(new_interface_b.cable, cable)
+
 
 #
 # Connections