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