diff --git a/netbox/dcim/models/cables.py b/netbox/dcim/models/cables.py index 0eaa01263..00113449f 100644 --- a/netbox/dcim/models/cables.py +++ b/netbox/dcim/models/cables.py @@ -229,16 +229,6 @@ 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). @@ -254,11 +244,8 @@ class Cable(PrimaryModel): ct.termination for ct in CableTermination.objects.filter(pk__in=value).prefetch_related('termination') ] - # 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): + # A cold end always counts as assigned: replay writes the rows first, so equal rows do not imply current paths + if not self.pk or not hasattr(self, _attr) or getattr(self, _attr) != list(value): self._terminations_modified = True setattr(self, _attr, value) @@ -517,6 +504,12 @@ class Cable(PrimaryModel): """ a_terminations, b_terminations = self.get_terminations() + # An end the caller did not assign is reconciled against its stored rows, not a possibly stale prefetch + 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()) + # A CableTermination's connector is derived from its position within its end's list of terminating # objects, so reordering that list (or removing an object from the middle of it) rewires the Cable # without changing which objects it connects. Recreate the affected end's CableTerminations so that @@ -528,14 +521,6 @@ class Cable(PrimaryModel): 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. - if force_a and not hasattr(self, '_a_terminations'): - self._a_terminations = list(a_terminations.keys()) - if force_b and not hasattr(self, '_b_terminations'): - self._b_terminations = list(b_terminations.keys()) - # 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_api.py b/netbox/dcim/tests/test_api.py index 5ba406ee5..29ceb0c32 100644 --- a/netbox/dcim/tests/test_api.py +++ b/netbox/dcim/tests/test_api.py @@ -4719,6 +4719,108 @@ class CableTestCase(APIViewTestCases.APIViewTestCase): }, ] + def _cable_topology(self, cable): + """ + Return the termination rows, path rows and endpoint references of a cable, for equality checks. + """ + paths = CablePath.objects.filter(_nodes__contains=cable).order_by('pk') + return ( + list(CableTermination.objects.filter(cable=cable).values_list('pk', 'cable_end', 'termination_id')), + list(paths.values_list('pk', 'path', 'is_complete', 'is_active')), + list(Interface.objects.filter(cable=cable).values_list('pk', 'cable_end', '_path_id')), + ) + + def test_patch_without_terminations_leaves_the_topology_alone(self): + """ + A PATCH that omits both termination lists must not touch the cable's rows or its endpoints. + """ + self.add_permissions('dcim.change_cable') + cable = Cable.objects.get(label='Cable 1') + topology = self._cable_topology(cable) + + response = self.client.patch(self._get_detail_url(cable), {'label': 'Renamed'}, format='json', **self.header) + + self.assertHttpStatus(response, status.HTTP_200_OK) + self.assertEqual(Cable.objects.get(pk=cable.pk).label, 'Renamed') + self.assertEqual(self._cable_topology(cable), topology) + + def test_patch_clearing_an_end_keeps_the_other_end(self): + """ + A PATCH with an empty termination list must detach that end only and keep the other end's row. + """ + self.add_permissions('dcim.change_cable') + for label, attr, cleared_side, kept_side in ( + ('Cable 1', 'a_terminations', CableEndChoices.SIDE_A, CableEndChoices.SIDE_B), + ('Cable 2', 'b_terminations', CableEndChoices.SIDE_B, CableEndChoices.SIDE_A), + ): + with self.subTest(attr=attr): + cable = Cable.objects.get(label=label) + cleared = Interface.objects.get(cable=cable, cable_end=cleared_side) + kept = Interface.objects.get(cable=cable, cable_end=kept_side) + kept_row_pk = CableTermination.objects.get(cable=cable, cable_end=kept_side).pk + + response = self.client.patch(self._get_detail_url(cable), {attr: []}, format='json', **self.header) + + self.assertHttpStatus(response, status.HTTP_200_OK) + self.assertEqual( + list( + CableTermination.objects.filter(cable=cable) + .values_list('pk', 'cable_end', 'termination_id') + ), + [(kept_row_pk, kept_side, kept.pk)] + ) + cleared.refresh_from_db() + self.assertIsNone(cleared.cable) + self.assertIsNone(cleared._path_id) + kept.refresh_from_db() + self.assertEqual(kept.cable, cable) + self.assertFalse(kept._path.is_complete) + + def test_patch_replacing_one_end_keeps_the_other_end_row(self): + """ + A PATCH that replaces one end must rewire it, keep the other end's row and detach the old endpoint. + """ + self.add_permissions('dcim.change_cable') + cable = Cable.objects.get(label='Cable 1') + interface_a = Interface.objects.get(cable=cable, cable_end=CableEndChoices.SIDE_A) + old_b = Interface.objects.get(cable=cable, cable_end=CableEndChoices.SIDE_B) + new_b = Interface.objects.get(device__name='Device 2', name='eth3') + a_row_pk = CableTermination.objects.get(cable=cable, cable_end=CableEndChoices.SIDE_A).pk + data = {'b_terminations': [{'object_type': 'dcim.interface', 'object_id': new_b.pk}]} + + response = self.client.patch(self._get_detail_url(cable), data, format='json', **self.header) + + self.assertHttpStatus(response, status.HTTP_200_OK) + self.assertEqual(CableTermination.objects.get(cable=cable, cable_end=CableEndChoices.SIDE_A).pk, a_row_pk) + for interface, peer in ((interface_a, new_b), (new_b, interface_a)): + interface = Interface.objects.get(pk=interface.pk) + self.assertEqual(interface.cable, cable) + self.assertTrue(interface._path.is_complete) + self.assertEqual(interface.connected_endpoints, [peer]) + old_b.refresh_from_db() + self.assertIsNone(old_b.cable) + self.assertIsNone(old_b._path_id) + + def test_rejected_patch_leaves_both_cables_alone(self): + """ + A PATCH rejected for an occupied endpoint must persist nothing, not even the accompanying label. + """ + self.add_permissions('dcim.change_cable') + cable = Cable.objects.get(label='Cable 1') + other = Cable.objects.get(label='Cable 2') + occupied = Interface.objects.get(cable=other, cable_end=CableEndChoices.SIDE_B) + topology = (self._cable_topology(cable), self._cable_topology(other)) + data = { + 'label': 'Must not persist', + 'b_terminations': [{'object_type': 'dcim.interface', 'object_id': occupied.pk}], + } + + response = self.client.patch(self._get_detail_url(cable), data, format='json', **self.header) + + self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST) + self.assertEqual(Cable.objects.get(pk=cable.pk).label, 'Cable 1') + self.assertEqual((self._cable_topology(cable), self._cable_topology(other)), topology) + def test_graphql_cable_termination_cached_filters(self): """ Validate filtering cables by cached CableTermination relations via GraphQL: diff --git a/netbox/dcim/tests/test_cablepaths.py b/netbox/dcim/tests/test_cablepaths.py index a42a527ac..f1bfb26b1 100644 --- a/netbox/dcim/tests/test_cablepaths.py +++ b/netbox/dcim/tests/test_cablepaths.py @@ -1,5 +1,5 @@ from circuits.models import * -from dcim.choices import LinkStatusChoices +from dcim.choices import CableEndChoices, LinkStatusChoices from dcim.models import * from dcim.svg import CableTraceSVG from dcim.tests.utils import BaseCablePathTestCase @@ -2892,9 +2892,52 @@ class LegacyCablePathTestCase(BaseCablePathTestCase): interface3.refresh_from_db() self.assertPathIsNotSet(interface3) - def test_304_resave_cable_with_unchanged_terminations(self): + def test_304_replayed_termination_move_rebuilds_paths(self): + """ + [IF1] --C1-- [IF2] becomes [IF1] --C1-- [IF3] + + Assigning terminations whose rows were already replaced must rebuild the paths from those rows. + """ + interface1 = Interface.objects.create(device=self.device, name='Interface 1') + interface2 = Interface.objects.create(device=self.device, name='Interface 2') + interface3 = Interface.objects.create(device=self.device, name='Interface 3') + + cable1 = Cable( + a_terminations=[interface1], + b_terminations=[interface2] + ) + cable1.save() + + # Replace the B end's row directly, as change replay does before it saves the cable itself + CableTermination.objects.get(cable=cable1, cable_end=CableEndChoices.SIDE_B).delete() + CableTermination(cable=cable1, cable_end=CableEndChoices.SIDE_B, termination=interface3).save() + termination_pks = set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)) + interface1.refresh_from_db() + interface3.refresh_from_db() + self.assertFalse(interface1._path.is_complete) + self.assertPathIsNotSet(interface3) + + data = cable1.serialize_object() + cable1 = Cable.objects.get(pk=cable1.pk) + cable1.a_terminations = data['a_terminations'] + cable1.b_terminations = data['b_terminations'] + cable1.save() + + self.assertCurrentPathExists((interface1, cable1, interface3), is_complete=True, is_active=True) + self.assertCurrentPathExists((interface3, cable1, interface1), is_complete=True, is_active=True) + self.assertEqual( + set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)), + termination_pks + ) + interface2.refresh_from_db() + self.assertIsNone(interface2.cable) + self.assertPathIsNotSet(interface2) + + def test_305_reassigning_terminations_restores_missing_paths(self): """ [IF1] --C1-- [IF2] + + Assigning a fresh cable's own terminations, as objects or as serialized IDs, must recreate removed paths. """ interface1 = Interface.objects.create(device=self.device, name='Interface 1') interface2 = Interface.objects.create(device=self.device, name='Interface 2') @@ -2904,39 +2947,109 @@ class LegacyCablePathTestCase(BaseCablePathTestCase): 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) + serialized = cable1.serialize_object() - # Reassign the same terminations on a freshly loaded instance + for value in ({'a_terminations': [interface1], 'b_terminations': [interface2]}, serialized): + with self.subTest(value_type=type(value['a_terminations'][0]).__name__): + # Delete each path on its own so the origins' _path references are cleared with it + for path in list(CablePath.objects.all()): + path.delete() + interface1.refresh_from_db() + interface2.refresh_from_db() + self.assertPathIsNotSet(interface1) + self.assertPathIsNotSet(interface2) + + cable1 = Cable.objects.get(pk=cable1.pk) + cable1.a_terminations = value['a_terminations'] + cable1.b_terminations = value['b_terminations'] + cable1.save() + + self.assertCurrentPathExists((interface1, cable1, interface2), is_complete=True, is_active=True) + self.assertCurrentPathExists((interface2, cable1, interface1), is_complete=True, is_active=True) + self.assertEqual(CablePath.objects.count(), 2) + self.assertEqual( + set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)), + termination_pks + ) + + def test_306_replaying_one_end_rebuilds_paths(self): + """ + [IF1] --C1-- [IF2] becomes [IF1] --C1-- [IF3] or [IF3] --C1-- [IF2] + + Assigning only the replaced end's serialized terminations must rebuild both paths and keep the other end's row. + """ + for side in (CableEndChoices.SIDE_A, CableEndChoices.SIDE_B): + with self.subTest(side=side): + interface1 = Interface.objects.create(device=self.device, name=f'Interface {side}1') + interface2 = Interface.objects.create(device=self.device, name=f'Interface {side}2') + interface3 = Interface.objects.create(device=self.device, name=f'Interface {side}3') + cable1 = Cable(a_terminations=[interface1], b_terminations=[interface2]) + cable1.save() + old, kept = (interface1, interface2) if side == CableEndChoices.SIDE_A else (interface2, interface1) + + # Replace one end's row directly, as change replay does before it saves the cable itself + CableTermination.objects.get(cable=cable1, cable_end=side).delete() + CableTermination(cable=cable1, cable_end=side, termination=interface3).save() + termination_pks = set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)) + kept.refresh_from_db() + interface3.refresh_from_db() + self.assertFalse(kept._path.is_complete) + self.assertPathIsNotSet(interface3) + + attr = f'{side.lower()}_terminations' + data = cable1.serialize_object() + cable1 = Cable.objects.get(pk=cable1.pk) + setattr(cable1, attr, data[attr]) + cable1.save() + + self.assertCurrentPathExists((kept, cable1, interface3), is_complete=True, is_active=True) + self.assertCurrentPathExists((interface3, cable1, kept), is_complete=True, is_active=True) + self.assertEqual( + set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)), + termination_pks + ) + old.refresh_from_db() + self.assertIsNone(old.cable) + self.assertPathIsNotSet(old) + + def test_307_replaying_a_termination_move_leaves_other_cables_alone(self): + """ + [IF1] --C1-- [IF2] becomes [IF1] --C1-- [IF3], while [IF4] --C2-- [IF5] stays as it is + + Rebuilding one cable's paths must not touch another cable's termination or path rows. + """ + interfaces = [Interface.objects.create(device=self.device, name=f'Interface {i}') for i in range(1, 6)] + cable1 = Cable(a_terminations=[interfaces[0]], b_terminations=[interfaces[1]]) + cable1.save() + cable2 = Cable(a_terminations=[interfaces[3]], b_terminations=[interfaces[4]]) + cable2.save() + + def other_rows(): + return ( + list(CableTermination.objects.filter(cable=cable2).values_list('pk', 'cable_end', 'termination_id')), + list( + CablePath.objects.filter(_nodes__contains=cable2) + .order_by('pk') + .values_list('pk', 'path', 'is_complete', 'is_active') + ), + ) + + expected = other_rows() + self.assertEqual(len(expected[1]), 2) + + CableTermination.objects.get(cable=cable1, cable_end=CableEndChoices.SIDE_B).delete() + CableTermination(cable=cable1, cable_end=CableEndChoices.SIDE_B, termination=interfaces[2]).save() + data = cable1.serialize_object() cable1 = Cable.objects.get(pk=cable1.pk) - cable1.a_terminations = [interface1] - cable1.b_terminations = [interface2] - cable1.label = 'Renamed' + cable1.b_terminations = data['b_terminations'] 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) + self.assertCurrentPathExists((interfaces[0], cable1, interfaces[2]), is_complete=True, is_active=True) + self.assertCurrentPathExists((interfaces[2], cable1, interfaces[0]), is_complete=True, is_active=True) + self.assertEqual(other_rows(), expected) + self.assertCurrentPathExists((interfaces[3], cable2, interfaces[4]), is_complete=True, is_active=True) + self.assertCurrentPathExists((interfaces[4], cable2, interfaces[3]), is_complete=True, is_active=True) def test_401_exclude_midspan_devices(self): """ diff --git a/netbox/dcim/tests/test_cablepaths2.py b/netbox/dcim/tests/test_cablepaths2.py index 6f4355a90..df0572d8b 100644 --- a/netbox/dcim/tests/test_cablepaths2.py +++ b/netbox/dcim/tests/test_cablepaths2.py @@ -1,7 +1,7 @@ from unittest import skip from circuits.models import Circuit, CircuitTermination, ProviderNetwork -from dcim.choices import CableLengthUnitChoices, CableProfileChoices +from dcim.choices import CableEndChoices, CableLengthUnitChoices, CableProfileChoices from dcim.models import * from dcim.svg import CableTraceSVG from dcim.tests.utils import BaseCablePathTestCase @@ -2786,48 +2786,159 @@ class CablePathTestCase(BaseCablePathTestCase): termination_pks ) - def test_311_change_cable_profile_after_reassigning_unchanged_terminations(self): + def test_311_change_cable_profile_on_a_warm_instance_rebuilds_paths(self): """ [IF1] --C1-- [IF2] - Applying a profile after both termination caches have been populated must still rebuild the paths. + Applying a profile to an instance whose termination caches are warm must recreate the rows and 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 + # Creating the cable warms both caches and saving it resets the flag cable1 = Cable( a_terminations=[interfaces[0]], b_terminations=[interfaces[1]], ) cable1.clean() cable1.save() + termination_pks = set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)) 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.assertTrue(hasattr(cable1, '_a_terminations') and hasattr(cable1, '_b_terminations')) 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.assertCurrentPathExists((interfaces[0], cable1, interfaces[1]), is_complete=True, is_active=True) + self.assertCurrentPathExists((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) + new_termination_pks = set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)) + self.assertTrue(termination_pks.isdisjoint(new_termination_pks)) + self.assertEqual( + sorted(CableTermination.objects.filter(cable=cable1).values_list('cable_end', 'connector', 'positions')), + [(CableEndChoices.SIDE_A, 1, [1]), (CableEndChoices.SIDE_B, 1, [1])] + ) + data = cable1.serialize_object() + self.assertEqual(set(data['a_terminations'] + data['b_terminations']), new_termination_pks) + + def test_312_replayed_midspan_terminations_complete_existing_paths(self): + """ + [IF1] --C1-- [FP1][RP1] --C3-- [RP2][FP2] --C2-- [IF2] + + Assigning a mid-span cable's stored terminations must complete the paths that stop at its rear ports. + """ + 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() + 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 = tuple(reversed(nodes_a_to_b)) + self.assertCurrentPathExists(nodes_a_to_b, is_complete=True, is_active=True) + self.assertCurrentPathExists(nodes_b_to_a, is_complete=True, is_active=True) + + # Recreate the B end's row directly, which leaves both paths incomplete at the rear ports + CableTermination.objects.get(cable=cable3, cable_end=CableEndChoices.SIDE_B).delete() + CableTermination(cable=cable3, cable_end=CableEndChoices.SIDE_B, termination=rear_ports[1]).save() + termination_pks = set(CableTermination.objects.filter(cable=cable3).values_list('pk', flat=True)) + for interface in interfaces: + interface.refresh_from_db() + self.assertFalse(interface._path.is_complete) + + data = cable3.serialize_object() + cable3 = Cable.objects.get(pk=cable3.pk) + cable3.a_terminations = data['a_terminations'] + cable3.b_terminations = data['b_terminations'] + cable3.save() + + self.assertCurrentPathExists(nodes_a_to_b, is_complete=True, is_active=True) + self.assertCurrentPathExists(nodes_b_to_a, is_complete=True, is_active=True) + self.assertEqual(CablePath.objects.count(), 2) + self.assertEqual( + set(CableTermination.objects.filter(cable=cable3).values_list('pk', flat=True)), + termination_pks + ) + + def test_313_reordering_trunk_terminations_rewires_connectors_and_paths(self): + """ + [IF1] --1 C1 1-- [IF3] becomes [IF2] --1 C1 1-- [IF3] + [IF2] --2 2-- [IF4] [IF1] --2 2-- [IF4] + + Reordering one end's members on a fresh instance must recreate that end's rows and rebuild the paths. + """ + interfaces = [ + Interface.objects.create(device=self.device, name=f'Interface {i}') for i in range(1, 5) + ] + + cable1 = Cable( + profile=CableProfileChoices.TRUNK_2C1P, + a_terminations=[interfaces[0], interfaces[1]], + b_terminations=[interfaces[2], interfaces[3]], + ) + cable1.full_clean() + cable1.save() + self.assertCurrentPathExists((interfaces[0], cable1, interfaces[2]), is_complete=True, is_active=True) + self.assertCurrentPathExists((interfaces[1], cable1, interfaces[3]), is_complete=True, is_active=True) + a_termination_pks = set( + CableTermination.objects.filter(cable=cable1, cable_end=CableEndChoices.SIDE_A).values_list('pk', flat=True) + ) + b_terminations = list( + CableTermination.objects.filter(cable=cable1, cable_end=CableEndChoices.SIDE_B) + .values_list('pk', 'connector', 'termination_id') + ) + + cable1 = Cable.objects.get(pk=cable1.pk) + cable1.a_terminations = [interfaces[1], interfaces[0]] + cable1.full_clean() + cable1.save() + + self.assertCurrentPathExists((interfaces[1], cable1, interfaces[2]), is_complete=True, is_active=True) + self.assertCurrentPathExists((interfaces[0], cable1, interfaces[3]), is_complete=True, is_active=True) + self.assertCurrentPathExists((interfaces[2], cable1, interfaces[1]), is_complete=True, is_active=True) + self.assertCurrentPathExists((interfaces[3], cable1, interfaces[0]), is_complete=True, is_active=True) + a_terminations = CableTermination.objects.filter(cable=cable1, cable_end=CableEndChoices.SIDE_A) + self.assertEqual( + list(a_terminations.order_by('connector').values_list('connector', 'termination_id')), + [(1, interfaces[1].pk), (2, interfaces[0].pk)] + ) + self.assertTrue(a_termination_pks.isdisjoint(a_terminations.values_list('pk', flat=True))) + self.assertEqual( + list( + CableTermination.objects.filter(cable=cable1, cable_end=CableEndChoices.SIDE_B) + .values_list('pk', 'connector', 'termination_id') + ), + b_terminations + ) diff --git a/netbox/dcim/tests/test_models.py b/netbox/dcim/tests/test_models.py index 82a6bff67..2b4663ea4 100644 --- a/netbox/dcim/tests/test_models.py +++ b/netbox/dcim/tests/test_models.py @@ -1,7 +1,7 @@ from decimal import Decimal from django.core.exceptions import ValidationError -from django.db.models import ProtectedError +from django.db.models import F, ProtectedError from django.db.models.signals import post_save from django.test import TestCase, tag @@ -2413,49 +2413,185 @@ class CableTestCase(TestCase): with self.assertRaises(ValidationError): cable.clean() - def test_reassigning_unchanged_terminations_does_not_flag_a_change(self): + def test_assigning_terminations_to_a_fresh_instance_flags_a_change(self): """ - Assigning the stored terminations to a freshly loaded cable must leave them unflagged. + Assigning either end of a freshly loaded cable must flag a change, even with its stored terminations. """ 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] + for attr, interface in (('a_terminations', interface1), ('b_terminations', interface2)): + with self.subTest(attr=attr): + cable = Cable.objects.first() + self.assertFalse(cable._terminations_modified) + setattr(cable, attr, [interface]) + self.assertTrue(cable._terminations_modified) + def test_assigning_a_different_end_to_a_warm_instance_flags_a_change(self): + """ + Assigning a different termination to an end already held in memory must flag a change. + """ + interface2 = Interface.objects.get(device__name='TestDevice2', name='eth0') + interface3 = Interface.objects.get(device__name='TestDevice2', name='eth1') + + # Assigning warms the B end's cache and the save resets the flag + cable = Cable.objects.first() + cable.b_terminations = [interface2] + cable.save() self.assertFalse(cable._terminations_modified) - def test_reassigning_different_terminations_flags_a_change(self): + cable.b_terminations = [interface3] + self.assertTrue(cable._terminations_modified) + + def test_assigning_one_end_reconciles_the_other_against_its_stored_rows(self): """ - Assigning a different termination to a freshly loaded cable must flag the change. + Assigning one end must leave the other end's stored rows alone, even when a prefetch of them is stale. + """ + interface2 = Interface.objects.get(device__name='TestDevice2', name='eth0') + interface3 = Interface.objects.get(device__name='TestDevice2', name='eth1') + cable = Cable.objects.prefetch_related('terminations__termination').first() + + # Moving the A end through a second instance leaves the prefetch above stale + moved = Cable.objects.get(pk=cable.pk) + moved.a_terminations = [interface3] + moved.save() + a_rows = CableTermination.objects.filter(cable=cable, cable_end=CableEndChoices.SIDE_A) + a_row_pk = a_rows.get().pk + + cable.b_terminations = [interface2] + cable.save() + + self.assertEqual(list(a_rows.values_list('pk', 'termination_id')), [(a_row_pk, interface3.pk)]) + + def test_assigning_serialized_termination_ids_flags_a_change(self): + """ + Serialized CableTermination IDs assigned to a fresh instance must resolve to the endpoints and flag a change. + """ + interface1 = Interface.objects.get(device__name='TestDevice1', name='eth0') + interface2 = Interface.objects.get(device__name='TestDevice2', name='eth0') + data = Cable.objects.first().serialize_object() + + for attr, interface in (('a_terminations', interface1), ('b_terminations', interface2)): + with self.subTest(attr=attr): + cable = Cable.objects.first() + setattr(cable, attr, data[attr]) + self.assertEqual(getattr(cable, attr), [interface]) + self.assertTrue(cable._terminations_modified) + + def test_serialize_object_includes_termination_ids(self): + """ + Serialization must carry each end's CableTermination IDs, not the IDs of the terminating objects. + """ + cable = Cable.objects.first() + # Move the rows above every existing ID so matching endpoint IDs cannot satisfy the assertion + offset = max( + Interface.objects.order_by('-pk').first().pk, + CableTermination.objects.order_by('-pk').first().pk, + ) + 1 + CableTermination.objects.filter(cable=cable).update(id=F('id') + offset) + + data = cable.serialize_object() + + for side, attr in ((CableEndChoices.SIDE_A, 'a_terminations'), (CableEndChoices.SIDE_B, 'b_terminations')): + with self.subTest(side=side): + self.assertEqual(data[attr], [CableTermination.objects.get(cable=cable, cable_end=side).pk]) + + def test_clearing_an_end_of_a_fresh_instance_removes_its_terminations(self): + """ + Assigning an empty list to an end of a freshly loaded cable must delete that end's terminations only. + """ + interface1 = Interface.objects.get(device__name='TestDevice1', name='eth0') + interface2 = Interface.objects.get(device__name='TestDevice2', name='eth0') + cable_pk = Cable.objects.first().pk + + for attr, kept_side, cleared, kept in ( + ('a_terminations', CableEndChoices.SIDE_B, interface1, interface2), + ('b_terminations', CableEndChoices.SIDE_A, interface2, interface1), + ): + with self.subTest(attr=attr): + cable = Cable.objects.get(pk=cable_pk) + kept_row_pk = CableTermination.objects.get(cable=cable, cable_end=kept_side).pk + setattr(cable, attr, []) + self.assertTrue(cable._terminations_modified) + cable.full_clean() + cable.save() + + self.assertEqual( + list(CableTermination.objects.filter(cable=cable).values_list('pk', 'cable_end')), + [(kept_row_pk, kept_side)] + ) + cleared.refresh_from_db() + self.assertIsNone(cleared.cable) + self.assertIsNone(cleared._path_id) + kept.refresh_from_db() + self.assertEqual(kept.cable, cable) + self.assertFalse(kept._path.is_complete) + + # Reconnect the cleared end so the other side starts from a complete cable + cable = Cable.objects.get(pk=cable_pk) + setattr(cable, attr, [cleared]) + cable.save() + + cable = Cable.objects.get(pk=cable_pk) + cable.a_terminations = [] + cable.save() + cable = Cable.objects.get(pk=cable_pk) + cable.b_terminations = [] + cable.save() + + self.assertFalse(CableTermination.objects.filter(cable=cable).exists()) + for interface in (interface1, interface2): + interface.refresh_from_db() + self.assertIsNone(interface.cable) + self.assertIsNone(interface._path_id) + + def test_reassigning_an_unchanged_end_on_a_warm_instance_does_not_flag_a_change(self): + """ + Assigning the value an end already holds in memory must not flag a change or clear a pending one. """ interface1 = Interface.objects.get(device__name='TestDevice1', name='eth0') interface3 = Interface.objects.get(device__name='TestDevice2', name='eth1') + # Assigning warms the A end's cache and the save resets the flag cable = Cable.objects.first() cable.a_terminations = [interface1] + cable.save() + self.assertFalse(cable._terminations_modified) + termination_pks = set(CableTermination.objects.filter(cable=cable).values_list('pk', flat=True)) + path_pks = set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True)) + + cable.a_terminations = [interface1] + self.assertFalse(cable._terminations_modified) + cable.save() + self.assertEqual( + set(CableTermination.objects.filter(cable=cable).values_list('pk', flat=True)), + termination_pks + ) + self.assertEqual(set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True)), path_pks) + cable.b_terminations = [interface3] - + cable.a_terminations = [interface1] self.assertTrue(cable._terminations_modified) - def test_reassigning_stale_prefetched_terminations_flags_a_change(self): + def test_saving_a_fresh_instance_without_an_assignment_leaves_terminations_and_paths(self): """ - A stale prefetched relation must not hide a real termination change. + A save on a freshly loaded cable that assigns no end must keep its termination and path rows. """ - cable = Cable.objects.prefetch_related('terminations__termination').first() - stale_termination = cable.b_terminations[0] - current_termination = Interface.objects.get(device__name='TestDevice2', name='eth1') + cable = Cable.objects.first() + termination_pks = set(CableTermination.objects.filter(cable=cable).values_list('pk', flat=True)) + path_pks = set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True)) + self.assertEqual(len(path_pks), 2) - # 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() + cable = Cable.objects.get(pk=cable.pk) + cable.label = 'Renamed' + cable.full_clean() + cable.save() - # The value matches the stale prefetch but not the stored row - cable.b_terminations = [stale_termination] - self.assertTrue(cable._terminations_modified) + self.assertEqual( + set(CableTermination.objects.filter(cable=cable).values_list('pk', flat=True)), + termination_pks + ) + self.assertEqual(set(CablePath.objects.filter(_nodes__contains=cable).values_list('pk', flat=True)), path_pks) def test_partial_save_does_not_apply_an_unwritten_profile(self): """ diff --git a/netbox/dcim/tests/utils.py b/netbox/dcim/tests/utils.py index 5fe270b34..229c5b3f1 100644 --- a/netbox/dcim/tests/utils.py +++ b/netbox/dcim/tests/utils.py @@ -64,6 +64,18 @@ class BaseCablePathTestCase(TestCase): cablepath = self._get_cablepath(nodes, **kwargs) self.assertIsNone(cablepath, msg='Unexpected CablePath found') + def assertCurrentPathExists(self, nodes, **kwargs): + """ + Assert that the first node references a CablePath with the given route via _path, and return it. + + :param nodes: Iterable of steps, the first being the originating path endpoint object + """ + origin = type(nodes[0]).objects.get(pk=nodes[0].pk) + self.assertIsNotNone(origin._path_id, msg=f'No path set on originating endpoint {origin}') + cablepath = self._get_cablepath(nodes, pk=origin._path_id, **kwargs) + self.assertIsNotNone(cablepath, msg=f'Path #{origin._path_id} on {origin} does not match the expected route') + return cablepath + def assertPathIsSet(self, origin, cablepath, msg=None): """ Assert that a specific CablePath instance is set as the path on the origin.