diff --git a/netbox/dcim/models/cables.py b/netbox/dcim/models/cables.py index 73b023913..f58654e0c 100644 --- a/netbox/dcim/models/cables.py +++ b/netbox/dcim/models/cables.py @@ -23,6 +23,7 @@ from dcim.utils import decompile_path_node, object_to_path_node from netbox.choices import ColorChoices from netbox.models import ChangeLoggedModel, PrimaryModel from utilities.conversion import to_meters +from utilities.data import normalize_update_fields from utilities.exceptions import AbortRequest from utilities.fields import ColorField, GenericArrayForeignKey from utilities.querysets import RestrictedQuerySet @@ -321,6 +322,11 @@ class Cable(PrimaryModel): def save(self, *args, force_insert=False, force_update=False, using=None, update_fields=None): _created = self.pk is None + save_kwargs = { + 'using': using, + 'update_fields': update_fields, + } + update_fields = normalize_update_fields(save_kwargs) # Store the given length (if any) in meters for use in database ordering if self.length is not None and self.length_unit: @@ -332,24 +338,35 @@ class Cable(PrimaryModel): if self.length is None: self.length_unit = None + # A field counts as changed only when this save actually writes it + status_written = update_fields is None or 'status' in update_fields + profile_written = update_fields is None or 'profile' in update_fields + # If this is a new Cable, save it before attempting to create its CableTerminations if self._state.adding: - super().save(*args, force_insert=True, using=using, update_fields=update_fields) + super().save(*args, force_insert=True, **save_kwargs) # Update the private PK used in __str__() self._pk = self.pk - if self._orig_profile != self.profile: + if profile_written and self._orig_profile != self.profile: self.update_terminations(force=True) elif self._terminations_modified: self.update_terminations() - super().save(*args, force_update=True, using=using, update_fields=update_fields) + super().save(*args, force_update=True, **save_kwargs) try: trace_paths.send(Cable, instance=self, created=_created) except UnsupportedCablePath as e: raise AbortRequest(e) + # Reset change tracking for the next save of this instance + if status_written: + self._orig_status = self.status + if profile_written: + self._orig_profile = self.profile + self._terminations_modified = False + def delete(self, *args, **kwargs): # Track this Cable as being deleted so the post_delete signal handler # for cascaded CableTerminations can skip redundant path retracing; @@ -477,6 +494,9 @@ class Cable(PrimaryModel): if 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 or (termination.pk and termination not in self.a_terminations): diff --git a/netbox/dcim/tests/test_cablepaths2.py b/netbox/dcim/tests/test_cablepaths2.py index 725cfccda..5732917a4 100644 --- a/netbox/dcim/tests/test_cablepaths2.py +++ b/netbox/dcim/tests/test_cablepaths2.py @@ -15,7 +15,7 @@ class CablePathTestCase(BaseCablePathTestCase): Tests are numbered as follows: 1XX: Test direct connections using each profile 2XX: Topology tests replicated from the legacy test case and adapted to use profiles - 3XX: Dynamic port mapping and termination changes + 3XX: Dynamic port mapping, profile and termination changes """ def test_101_cable_profile_single_1c1p(self): @@ -2512,3 +2512,276 @@ class CablePathTestCase(BaseCablePathTestCase): is_complete=True, is_active=True ) + + def test_307_change_cable_profile_rebuilds_paths(self): + """ + [IF1] --C1-- [IF2] + + Applying a profile to an existing cable rebuilds its 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.assertPathExists( + (interfaces[0], cable1, interfaces[1]), + is_complete=True, + is_active=True + ) + self.assertPathExists( + (interfaces[1], cable1, interfaces[0]), + is_complete=True, + is_active=True + ) + + # Reload so _terminations_modified starts False, as it does in a bulk edit + cable1 = Cable.objects.get(pk=cable1.pk) + 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) + + def test_308_change_cable_profile_regroups_trunk_paths(self): + """ + [IF1] --C1-- [IF3] + [IF2] [IF4] + + Applying a trunk profile to an existing cable regroups its paths by connector, and + clearing it again collapses them. + """ + interfaces = [ + Interface.objects.create(device=self.device, name='Interface 1'), + Interface.objects.create(device=self.device, name='Interface 2'), + Interface.objects.create(device=self.device, name='Interface 3'), + Interface.objects.create(device=self.device, name='Interface 4'), + ] + + # Create cable 1 without a profile + cable1 = Cable( + a_terminations=[interfaces[0], interfaces[1]], + b_terminations=[interfaces[2], interfaces[3]], + ) + cable1.clean() + cable1.save() + + # Without a profile both terminations on each end share a single path + self.assertPathExists( + ((interfaces[0], interfaces[1]), cable1, (interfaces[2], interfaces[3])), + is_complete=True, + is_active=True + ) + self.assertPathExists( + ((interfaces[2], interfaces[3]), cable1, (interfaces[0], interfaces[1])), + is_complete=True, + is_active=True + ) + self.assertEqual(CablePath.objects.count(), 2) + + # Reload so _terminations_modified starts False, as it does in a bulk edit + cable1 = Cable.objects.get(pk=cable1.pk) + cable1.profile = CableProfileChoices.TRUNK_2C1P + cable1.full_clean() + cable1.save() + + path1 = self.assertPathExists( + (interfaces[0], cable1, interfaces[2]), + is_complete=True, + is_active=True + ) + path2 = self.assertPathExists( + (interfaces[1], cable1, interfaces[3]), + is_complete=True, + is_active=True + ) + path3 = self.assertPathExists( + (interfaces[2], cable1, interfaces[0]), + is_complete=True, + is_active=True + ) + path4 = self.assertPathExists( + (interfaces[3], cable1, interfaces[1]), + is_complete=True, + is_active=True + ) + self.assertEqual(CablePath.objects.count(), 4) + + for interface in interfaces: + interface.refresh_from_db() + self.assertPathIsSet(interfaces[0], path1) + self.assertPathIsSet(interfaces[1], path2) + self.assertPathIsSet(interfaces[2], path3) + self.assertPathIsSet(interfaces[3], path4) + self.assertEqual(interfaces[0].cable_connector, 1) + self.assertEqual(interfaces[1].cable_connector, 2) + self.assertEqual(interfaces[2].cable_connector, 1) + self.assertEqual(interfaces[3].cable_connector, 2) + for interface in interfaces: + self.assertEqual(interface.cable_positions, [1]) + + # Clearing the profile is a bulk-edit action in its own right + cable1 = Cable.objects.get(pk=cable1.pk) + cable1.profile = '' + cable1.full_clean() + cable1.save() + + path5 = self.assertPathExists( + ((interfaces[0], interfaces[1]), cable1, (interfaces[2], interfaces[3])), + is_complete=True, + is_active=True + ) + path6 = self.assertPathExists( + ((interfaces[2], interfaces[3]), cable1, (interfaces[0], interfaces[1])), + is_complete=True, + is_active=True + ) + self.assertEqual(CablePath.objects.count(), 2) + + for interface in interfaces: + interface.refresh_from_db() + self.assertIsNone(interface.cable_connector) + self.assertPathIsSet(interfaces[0], path5) + self.assertPathIsSet(interfaces[1], path5) + self.assertPathIsSet(interfaces[2], path6) + self.assertPathIsSet(interfaces[3], path6) + + def test_309_change_midspan_cable_profile_rebuilds_paths(self): + """ + [IF1] --C1-- [FP1][RP1] --C3-- [RP2][FP2] --C2-- [IF2] + + Applying a profile to a cable which terminates on pass-through ports rebuilds the + paths traversing it. The rear ports are not path origins, so a missing rebuild + truncates those paths rather than deleting them. + """ + interfaces = [ + Interface.objects.create(device=self.device, name='Interface 1'), + Interface.objects.create(device=self.device, name='Interface 2'), + ] + rear_ports = [ + RearPort.objects.create(device=self.device, name='Rear Port 1'), + RearPort.objects.create(device=self.device, name='Rear Port 2'), + ] + front_ports = [ + FrontPort.objects.create(device=self.device, name='Front Port 1'), + FrontPort.objects.create(device=self.device, name='Front Port 2'), + ] + for front_port, rear_port in zip(front_ports, rear_ports): + PortMapping.objects.create( + device=self.device, + front_port=front_port, + front_port_position=1, + rear_port=rear_port, + rear_port_position=1 + ) + + cable1 = Cable(a_terminations=[interfaces[0]], b_terminations=[front_ports[0]]) + cable1.clean() + cable1.save() + cable2 = Cable(a_terminations=[front_ports[1]], b_terminations=[interfaces[1]]) + cable2.clean() + cable2.save() + + # Create the mid-span cable without a profile + cable3 = Cable(a_terminations=[rear_ports[0]], b_terminations=[rear_ports[1]]) + cable3.clean() + cable3.save() + + nodes_a_to_b = ( + interfaces[0], cable1, front_ports[0], rear_ports[0], cable3, rear_ports[1], front_ports[1], cable2, + interfaces[1], + ) + nodes_b_to_a = ( + interfaces[1], cable2, front_ports[1], rear_ports[1], cable3, rear_ports[0], front_ports[0], cable1, + interfaces[0], + ) + self.assertPathExists(nodes_a_to_b, is_complete=True, is_active=True) + self.assertPathExists(nodes_b_to_a, is_complete=True, is_active=True) + self.assertEqual(CablePath.objects.count(), 2) + + # Reload so _terminations_modified starts False, as it does in a bulk edit + cable3 = Cable.objects.get(pk=cable3.pk) + cable3.profile = CableProfileChoices.SINGLE_1C1P + cable3.full_clean() + cable3.save() + + path1 = self.assertPathExists(nodes_a_to_b, is_complete=True, is_active=True) + path2 = self.assertPathExists(nodes_b_to_a, is_complete=True, is_active=True) + self.assertEqual(CablePath.objects.count(), 2) + + for interface in interfaces: + interface.refresh_from_db() + self.assertPathIsSet(interfaces[0], path1) + self.assertPathIsSet(interfaces[1], path2) + + def test_310_repeat_save_does_not_recreate_paths(self): + """ + [IF1] --C1-- [IF2] + + Saving an unchanged cable again leaves its terminations and paths untouched. + """ + interfaces = [ + Interface.objects.create(device=self.device, name='Interface 1'), + Interface.objects.create(device=self.device, name='Interface 2'), + ] + + cable1 = Cable( + a_terminations=[interfaces[0]], + b_terminations=[interfaces[1]], + ) + cable1.clean() + 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) + + # Saving the same instance again must not duplicate its paths + 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 + ) + + # Reload so _terminations_modified starts False, as it does in a bulk edit + cable1 = Cable.objects.get(pk=cable1.pk) + cable1.profile = CableProfileChoices.SINGLE_1C1P + cable1.full_clean() + 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) + + # The profile change is applied once, so a second save must not recreate anything + 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 + ) diff --git a/netbox/dcim/tests/test_models.py b/netbox/dcim/tests/test_models.py index 06f07f99c..f81e88f1c 100644 --- a/netbox/dcim/tests/test_models.py +++ b/netbox/dcim/tests/test_models.py @@ -2133,6 +2133,33 @@ class CableTestCase(TestCase): with self.assertRaises(ValidationError): cable.clean() + 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. + """ + cable = Cable.objects.first() + interface1 = Interface.objects.get(device__name='TestDevice1', name='eth0') + termination_pks = set(CableTermination.objects.filter(cable=cable).values_list('pk', flat=True)) + + cable.profile = CableProfileChoices.SINGLE_1C1P + cable.save(update_fields=['label']) + + interface1.refresh_from_db() + # Requery rather than refresh, so the pending profile stays on the instance under test + self.assertEqual(Cable.objects.get(pk=cable.pk).profile, '') + self.assertIsNone(interface1.cable_connector) + self.assertEqual( + set(CableTermination.objects.filter(cable=cable).values_list('pk', flat=True)), + termination_pks + ) + + # _orig_profile was not advanced, so the pending change still applies here + cable.save() + + interface1.refresh_from_db() + self.assertEqual(Cable.objects.get(pk=cable.pk).profile, CableProfileChoices.SINGLE_1C1P) + self.assertEqual(interface1.cable_connector, 1) + def test_cable_profile_change_preserves_terminations(self): """ When a Cable's profile is changed via save() without explicitly setting terminations (as happens during diff --git a/netbox/dcim/tests/test_signals.py b/netbox/dcim/tests/test_signals.py index 2d0b1bee3..348c6ec0b 100644 --- a/netbox/dcim/tests/test_signals.py +++ b/netbox/dcim/tests/test_signals.py @@ -796,8 +796,7 @@ class CableSignalTestCase(TestCase): cable.save() self.assertTrue(all(cp.is_active for cp in CablePath.objects.all())) - # Reload the cable so _orig_status reflects the persisted value and - # _terminations_modified resets to False. + # Reload to exercise status tracking on a freshly loaded instance, as a request does cable = Cable.objects.get(pk=cable.pk) cable.status = LinkStatusChoices.STATUS_PLANNED cable.save() @@ -821,6 +820,42 @@ class CableSignalTestCase(TestCase): self.assertTrue(all(cp.is_active for cp in CablePath.objects.all())) + def test_toggling_cable_status_on_one_instance_reactivates_paths(self): + interface_a = Interface.objects.create(device=self.device, name='Interface A') + interface_b = Interface.objects.create(device=self.device, name='Interface B') + cable = Cable(a_terminations=[interface_a], b_terminations=[interface_b]) + cable.save() + + # Reuse the same instance for both changes, as a script would + cable = Cable.objects.get(pk=cable.pk) + cable.status = LinkStatusChoices.STATUS_PLANNED + cable.save() + self.assertFalse(any(cp.is_active for cp in CablePath.objects.all())) + + cable.status = LinkStatusChoices.STATUS_CONNECTED + cable.save() + self.assertTrue(all(cp.is_active for cp in CablePath.objects.all())) + + def test_partial_save_does_not_consume_an_unwritten_status_change(self): + interface_a = Interface.objects.create(device=self.device, name='Interface A') + interface_b = Interface.objects.create(device=self.device, name='Interface B') + cable = Cable( + a_terminations=[interface_a], + b_terminations=[interface_b], + status=LinkStatusChoices.STATUS_PLANNED, + ) + cable.save() + self.assertFalse(any(cp.is_active for cp in CablePath.objects.all())) + + # A save that excludes status must not advance the status snapshot + cable.status = LinkStatusChoices.STATUS_CONNECTED + cable.save(update_fields=['label']) + self.assertFalse(any(cp.is_active for cp in CablePath.objects.all())) + + # _orig_status was not advanced, so the change must still be detected + cable.save() + self.assertTrue(all(cp.is_active for cp in CablePath.objects.all())) + def test_deleting_cable_retraces_paths(self): interface_a = Interface.objects.create(device=self.device, name='Interface A') interface_b = Interface.objects.create(device=self.device, name='Interface B')