From 1745a7d9aa676243ce28449df14e7ccc30241c46 Mon Sep 17 00:00:00 2001 From: Martin Hauser Date: Fri, 4 Sep 2026 20:02:59 +0200 Subject: [PATCH] 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. --- netbox/dcim/forms/connections.py | 8 +- netbox/dcim/models/cables.py | 23 +++++- netbox/dcim/tests/test_cablepaths.py | 46 +++++++++++ netbox/dcim/tests/test_cablepaths2.py | 46 +++++++++++ netbox/dcim/tests/test_models.py | 44 ++++++++++ netbox/dcim/tests/test_views.py | 111 ++++++++++++++++++++++++++ 6 files changed, 271 insertions(+), 7 deletions(-) diff --git a/netbox/dcim/forms/connections.py b/netbox/dcim/forms/connections.py index b79b22bdf..a73f70196 100644 --- a/netbox/dcim/forms/connections.py +++ b/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 diff --git a/netbox/dcim/models/cables.py b/netbox/dcim/models/cables.py index f26cf1e97..0eaa01263 100644 --- a/netbox/dcim/models/cables.py +++ b/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): diff --git a/netbox/dcim/tests/test_cablepaths.py b/netbox/dcim/tests/test_cablepaths.py index ec7de6020..a42a527ac 100644 --- a/netbox/dcim/tests/test_cablepaths.py +++ b/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] diff --git a/netbox/dcim/tests/test_cablepaths2.py b/netbox/dcim/tests/test_cablepaths2.py index 5732917a4..6f4355a90 100644 --- a/netbox/dcim/tests/test_cablepaths2.py +++ b/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) diff --git a/netbox/dcim/tests/test_models.py b/netbox/dcim/tests/test_models.py index 3c870f4ec..82a6bff67 100644 --- a/netbox/dcim/tests/test_models.py +++ b/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. diff --git a/netbox/dcim/tests/test_views.py b/netbox/dcim/tests/test_views.py index 374ad4a40..18e910940 100644 --- a/netbox/dcim/tests/test_views.py +++ b/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