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

Fixes #23097: Prevent duplicate Cable Paths when Cable Terminations are unchanged (#23100)

* fix(dcim): Prevent path rebuild when Cable Terminations unchanged

Compare Cable Terminations against stored values instead of the empty
cache when checking for modifications, so a freshly loaded Cable that is
resaved with the same terminations no longer rebuilds its paths. Raise
the flag whenever update_terminations() force-recreates an end, since
the edit form warms the cache that gated it and a profile change then
tore every path down without rebuilding it. Add regression tests for
both.

Fixes #23097

* 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 22 часов назад
Родитель
Сommit
1745a7d9aa

+ 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

+ 19 - 4
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,7 +254,11 @@ class Cable(PrimaryModel):
                 ct.termination for ct in CableTermination.objects.filter(pk__in=value).prefetch_related('termination')
             ]
 
-        if not self.pk or getattr(self, _attr, []) != 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)
@@ -510,6 +524,10 @@ class Cable(PrimaryModel):
         force_a = force or self._connectors_reassigned(a_terminations, self.a_terminations)
         force_b = force or self._connectors_reassigned(b_terminations, self.b_terminations)
 
+        # Recreating either end's terminations invalidates its paths, even when the endpoints are unchanged
+        if force_a or force_b:
+            self._terminations_modified = True
+
         # When force-recreating terminations (e.g. after a profile change), cache the termination objects
         # from the database before deleting, so they are available for recreation. Without this, the
         # a_terminations/b_terminations properties would query the DB after deletion and return empty lists.
@@ -518,9 +536,6 @@ class Cable(PrimaryModel):
         if force_b and not hasattr(self, '_b_terminations'):
             self._b_terminations = list(b_terminations.keys())
 
-            # Recreating terminations invalidates existing paths, even when the endpoints are unchanged
-            self._terminations_modified = True
-
         # Delete any stale CableTerminations
         for termination, ct in a_terminations.items():
             if force_a or (termination.pk and termination not in self.a_terminations):

+ 46 - 0
netbox/dcim/tests/test_cablepaths.py

@@ -2892,6 +2892,52 @@ class LegacyCablePathTestCase(BaseCablePathTestCase):
         interface3.refresh_from_db()
         self.assertPathIsNotSet(interface3)
 
+    def test_304_resave_cable_with_unchanged_terminations(self):
+        """
+        [IF1] --C1-- [IF2]
+        """
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        interface2 = Interface.objects.create(device=self.device, name='Interface 2')
+
+        cable1 = Cable(
+            a_terminations=[interface1],
+            b_terminations=[interface2]
+        )
+        cable1.save()
+
+        path_pks = set(CablePath.objects.values_list('pk', flat=True))
+        termination_pks = set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True))
+        self.assertEqual(len(path_pks), 2)
+        self.assertEqual(len(termination_pks), 2)
+
+        # Reassign the same terminations on a freshly loaded instance
+        cable1 = Cable.objects.get(pk=cable1.pk)
+        cable1.a_terminations = [interface1]
+        cable1.b_terminations = [interface2]
+        cable1.label = 'Renamed'
+        cable1.save()
+
+        self.assertEqual(set(CablePath.objects.values_list('pk', flat=True)), path_pks)
+        self.assertEqual(
+            set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)),
+            termination_pks
+        )
+
+        path1 = self.assertPathExists(
+            (interface1, cable1, interface2),
+            is_complete=True,
+            is_active=True
+        )
+        path2 = self.assertPathExists(
+            (interface2, cable1, interface1),
+            is_complete=True,
+            is_active=True
+        )
+        interface1.refresh_from_db()
+        interface2.refresh_from_db()
+        self.assertPathIsSet(interface1, path1)
+        self.assertPathIsSet(interface2, path2)
+
     def test_401_exclude_midspan_devices(self):
         """
         [IF1] --C1-- [FP1][Test Device][RP1] --C2-- [RP2][Test Device][FP2] --C3-- [IF2]

+ 46 - 0
netbox/dcim/tests/test_cablepaths2.py

@@ -2785,3 +2785,49 @@ class CablePathTestCase(BaseCablePathTestCase):
             set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)),
             termination_pks
         )
+
+    def test_311_change_cable_profile_after_reassigning_unchanged_terminations(self):
+        """
+        [IF1] --C1-- [IF2]
+
+        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'),
+            Interface.objects.create(device=self.device, name='Interface 2'),
+        ]
+
+        # Create cable 1 without a profile
+        cable1 = Cable(
+            a_terminations=[interfaces[0]],
+            b_terminations=[interfaces[1]],
+        )
+        cable1.clean()
+        cable1.save()
+        self.assertEqual(CablePath.objects.count(), 2)
+
+        # 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]]
+        self.assertFalse(cable1._terminations_modified)
+
+        cable1.profile = CableProfileChoices.SINGLE_1C1P
+        cable1.full_clean()
+        cable1.save()
+
+        path1 = self.assertPathExists(
+            (interfaces[0], cable1, interfaces[1]),
+            is_complete=True,
+            is_active=True
+        )
+        path2 = self.assertPathExists(
+            (interfaces[1], cable1, interfaces[0]),
+            is_complete=True,
+            is_active=True
+        )
+        self.assertEqual(CablePath.objects.count(), 2)
+        interfaces[0].refresh_from_db()
+        interfaces[1].refresh_from_db()
+        self.assertPathIsSet(interfaces[0], path1)
+        self.assertPathIsSet(interfaces[1], path2)

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

@@ -2413,6 +2413,50 @@ class CableTestCase(TestCase):
         with self.assertRaises(ValidationError):
             cable.clean()
 
+    def test_reassigning_unchanged_terminations_does_not_flag_a_change(self):
+        """
+        Assigning the stored terminations to a freshly loaded cable must leave them unflagged.
+        """
+        interface1 = Interface.objects.get(device__name='TestDevice1', name='eth0')
+        interface2 = Interface.objects.get(device__name='TestDevice2', name='eth0')
+
+        # A cable loaded from the database has no cached terminations
+        cable = Cable.objects.first()
+        cable.a_terminations = [interface1]
+        cable.b_terminations = [interface2]
+
+        self.assertFalse(cable._terminations_modified)
+
+    def test_reassigning_different_terminations_flags_a_change(self):
+        """
+        Assigning a different termination to a freshly loaded cable must flag the change.
+        """
+        interface1 = Interface.objects.get(device__name='TestDevice1', name='eth0')
+        interface3 = Interface.objects.get(device__name='TestDevice2', name='eth1')
+
+        cable = Cable.objects.first()
+        cable.a_terminations = [interface1]
+        cable.b_terminations = [interface3]
+
+        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.

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

@@ -5271,6 +5271,117 @@ class CableTestCase(
             [(1, interfaces[1]), (2, interfaces[0])]
         )
 
+    @tag('regression')  # Issue #23097
+    def test_edit_with_unchanged_terminations_preserves_paths(self):
+        """Editing a cable without changing its terminations must leave its paths in place."""
+        # 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]
+        path_pks = set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True))
+        self.assertEqual(len(path_pks), 2)
+
+        data = {
+            'a_terminations': [interface_a.pk],
+            'b_terminations': [interface_b.pk],
+            'type': CableTypeChoices.TYPE_CAT6,
+            'status': LinkStatusChoices.STATUS_CONNECTED,
+            'label': 'Renamed',
+            'color': 'c0c0c0',
+        }
+        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(
+            set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True)),
+            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