fix(dcim): Retire superseded Cable Paths on Termination changes

Delete any path already originating at a node before tracing a fresh
one from it, so that replacing, extending or recreating a Cable's
Terminations no longer leaves the rows it supersedes behind. Repeated
changes now converge on the paths the current terminations trace.

Retire from the expanded origins so that a channelized end retires its
channels' paths rather than the parent's, which has none, and write the
replacements under a transaction so a failed trace cannot leave an
origin with neither its old path nor a new one.

Add regression coverage for replaced, extended and channelized ends and
for repeated identical REST writes.

Fixes #23121

netbox/dcim/tests/test_api.py

@@ -4719,6 +4719,29 @@ class CableTestCase(APIViewTestCases.APIViewTestCase):
             },
         ]
 
+    def test_repeated_put_does_not_accumulate_paths(self):
+        """
+        Repeating an identical PUT must leave the cable with the two paths its terminations trace.
+        """
+        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)
+        interface_b = Interface.objects.get(cable=cable, cable_end=CableEndChoices.SIDE_B)
+        data = {
+            'status': cable.status,
+            'a_terminations': [{'object_type': 'dcim.interface', 'object_id': interface_a.pk}],
+            'b_terminations': [{'object_type': 'dcim.interface', 'object_id': interface_b.pk}],
+        }
+
+        for attempt in range(3):
+            with self.subTest(attempt=attempt):
+                response = self.client.put(self._get_detail_url(cable), data, format='json', **self.header)
+
+                self.assertHttpStatus(response, status.HTTP_200_OK)
+                for interface in (interface_a, interface_b):
+                    self.assertTrue(Interface.objects.get(pk=interface.pk)._path.is_complete)
+                self.assertEqual(CablePath.objects.filter(_nodes__contains=cable).count(), 2)
+
     def test_graphql_cable_termination_cached_filters(self):
         """
         Validate filtering cables by cached CableTermination relations via GraphQL:

netbox/dcim/tests/test_cablepaths.py

@@ -2891,6 +2891,65 @@ class LegacyCablePathTestCase(BaseCablePathTestCase):
         # Verify _path is cleared on removed interface (#21127)
         interface3.refresh_from_db()
         self.assertPathIsNotSet(interface3)
+        self.assertEqual(CablePath.objects.count(), 2)
+
+    def test_304_replacing_a_termination_retires_superseded_paths(self):
+        """
+        [IF1] --C1-- [IF2] becomes [IF1] --C1-- [IF3], and back again
+
+        Each replacement must leave only the two paths the cable's current terminations trace.
+        """
+        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()
+        self.assertEqual(CablePath.objects.count(), 2)
+
+        for peer, detached in ((interface3, interface2), (interface2, interface3)):
+            with self.subTest(peer=peer.name):
+                cable1 = Cable.objects.get(pk=cable1.pk)
+                cable1.b_terminations = [peer]
+                cable1.full_clean()
+                cable1.save()
+
+                self.assertCurrentPathExists((interface1, cable1, peer), is_complete=True, is_active=True)
+                self.assertCurrentPathExists((peer, cable1, interface1), is_complete=True, is_active=True)
+                self.assertEqual(CablePath.objects.count(), 2)
+                detached.refresh_from_db()
+                self.assertIsNone(detached.cable)
+                self.assertPathIsNotSet(detached)
+
+    def test_305_adding_a_termination_retires_superseded_paths(self):
+        """
+        [IF1] --C1-- [IF2] gains a second B-side termination [IF3]
+
+        Extending an end must retire the paths whose destinations the extension supersedes.
+        """
+        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()
+        self.assertEqual(CablePath.objects.count(), 2)
+
+        cable1 = Cable.objects.get(pk=cable1.pk)
+        cable1.b_terminations = [interface2, interface3]
+        cable1.full_clean()
+        cable1.save()
+
+        self.assertCurrentPathExists(
+            (interface1, cable1, [interface2, interface3]), is_complete=True, is_active=True
+        )
+        path2 = self.assertPathExists(
+            ([interface2, interface3], cable1, interface1), is_complete=True, is_active=True
+        )
+        for interface in (interface2, interface3):
+            interface.refresh_from_db()
+            self.assertPathIsSet(interface, path2)
+        self.assertEqual(CablePath.objects.count(), 2)
 
     def test_401_exclude_midspan_devices(self):
         """

netbox/dcim/tests/test_cablepaths2.py

@@ -2785,3 +2785,67 @@ class CablePathTestCase(BaseCablePathTestCase):
             set(CableTermination.objects.filter(cable=cable1).values_list('pk', flat=True)),
             termination_pks
         )
+
+    def test_311_moving_a_midspan_termination_preserves_far_end_paths(self):
+        """
+        [IF1] --C1-- [FP1][RP1] --C3-- [RP2][FP2] --C2-- [IF2]
+        becomes
+        [IF1] --C1-- [FP1][RP1] --C3-- [RP3][FP3] --C4-- [IF3]
+
+        Rear ports originate no path, so moving a mid-span cable's end must retrace the rows which traverse it
+        rather than delete them: their origins cannot be recovered from the cable's own terminations.
+        """
+        interfaces = [
+            Interface.objects.create(device=self.device, name=f'Interface {i}') for i in range(1, 4)
+        ]
+        rear_ports = [
+            RearPort.objects.create(device=self.device, name=f'Rear Port {i}') for i in range(1, 4)
+        ]
+        front_ports = [
+            FrontPort.objects.create(device=self.device, name=f'Front Port {i}') for i in range(1, 4)
+        ]
+        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()
+        cable4 = Cable(a_terminations=[front_ports[2]], b_terminations=[interfaces[2]])
+        cable4.clean()
+        cable4.save()
+
+        before = (
+            interfaces[0], cable1, front_ports[0], rear_ports[0], cable3, rear_ports[1], front_ports[1], cable2,
+            interfaces[1],
+        )
+        self.assertCurrentPathExists(before, is_complete=True, is_active=True)
+        self.assertCurrentPathExists(tuple(reversed(before)), is_complete=True, is_active=True)
+        # Two for the completed link, one for the third interface stopping at its own rear port
+        self.assertEqual(CablePath.objects.count(), 3)
+
+        cable3 = Cable.objects.get(pk=cable3.pk)
+        cable3.b_terminations = [rear_ports[2]]
+        cable3.full_clean()
+        cable3.save()
+
+        after = (
+            interfaces[0], cable1, front_ports[0], rear_ports[0], cable3, rear_ports[2], front_ports[2], cable4,
+            interfaces[2],
+        )
+        self.assertCurrentPathExists(after, is_complete=True, is_active=True)
+        self.assertCurrentPathExists(tuple(reversed(after)), is_complete=True, is_active=True)
+        interfaces[1].refresh_from_db()
+        self.assertFalse(interfaces[1]._path.is_complete)
+        self.assertEqual(CablePath.objects.count(), 3)

netbox/dcim/tests/test_channelization.py

@@ -468,6 +468,46 @@ class ChannelizedCablePathTestCase(BaseCablePathTestCase):
         self.assertIsNone(channel.cable_positions)
         self.assertPathIsNotSet(channel)
 
+    def test_114_replacing_a_far_end_retires_the_channels_superseded_paths(self):
+        """
+        Replacing one far-end interface of a breakout cable must leave one path per channel in each direction. The
+        channels are the real origins, so the rows the retrace supersedes can only be found after the expansion.
+        """
+        parent, channels = self._create_channelized_interface('et0', 4)
+        far = [
+            Interface.objects.create(device=self.device, name=f'xe{i}', type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS)
+            for i in range(4)
+        ]
+        replacement = Interface.objects.create(
+            device=self.device, name='xe4', type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS
+        )
+
+        cable = Cable(
+            profile=CableProfileChoices.BREAKOUT_1C4P_4C1P,
+            a_terminations=[parent],
+            b_terminations=far,
+        )
+        cable.clean()
+        cable.save()
+        self.assertEqual(CablePath.objects.count(), 8)
+
+        cable = Cable.objects.get(pk=cable.pk)
+        cable.b_terminations = [*far[:3], replacement]
+        cable.clean()
+        cable.save()
+
+        self.assertEqual(CablePath.objects.count(), 8)
+        for channel, far_iface in zip(channels, [*far[:3], replacement]):
+            channel.refresh_from_db()
+            far_iface.refresh_from_db()
+            forward = self.assertPathExists((channel, cable, far_iface), is_complete=True, is_active=True)
+            reverse = self.assertPathExists((far_iface, cable, channel), is_complete=True, is_active=True)
+            self.assertPathIsSet(channel, forward)
+            self.assertPathIsSet(far_iface, reverse)
+        far[3].refresh_from_db()
+        self.assertIsNone(far[3].cable)
+        self.assertPathIsNotSet(far[3])
+
 
 class ChannelizedInterfaceTestCase(TestCase):
     """

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.

netbox/dcim/utils.py

@@ -125,9 +125,26 @@ def path_node_to_object(repr):
     return ct.model_class().objects.filter(pk=object_id).first()
 
 
+def retire_superseded_paths(objects):
+    """
+    Delete any CablePath which originates at one of the given nodes, so that a fresh trace from them
+    does not leave the row it replaces behind.
+    """
+    from dcim.models import CablePath
+
+    nodes = {object_to_path_node(obj) for obj in objects}
+
+    # `overlap` takes the encoded nodes directly, and matches nothing for an empty set
+    for cp in CablePath.objects.filter(_nodes__overlap=list(nodes)):
+        # `_nodes` matches a node anywhere in a path, including as another path's destination
+        if cp.path and nodes.intersection(cp.path[0]):
+            cp.delete()
+
+
 def create_cablepaths(objects):
     """
-    Create CablePaths for all paths originating from the specified set of nodes.
+    Create CablePaths for all paths originating from the specified set of nodes, retiring any path which
+    already originates there.
 
     :param objects: Iterable of cabled objects (e.g. Interfaces)
     """
@@ -144,20 +161,28 @@ def create_cablepaths(objects):
         else:
             expanded.append(obj)
 
-    # Arrange objects by cable connector. All objects with a null connector are grouped together. Channel
-    # subinterfaces must each originate their own path, as sharing a connector would otherwise collapse a group of
-    # siblings into a single malformed path.
-    origins = defaultdict(list)
-    for obj in expanded:
-        if isinstance(obj, Interface) and obj.channel_id:
-            if cp := CablePath.from_origin([obj]):
-                cp.save()
-        else:
-            origins[obj.cable_connector].append(obj)
+    # The savepoint must stay: Cable.save() turns UnsupportedCablePath into AbortRequest and callers keep querying
+    with transaction.atomic(using=router.db_for_write(CablePath)):
 
-    for connector, objects in origins.items():
-        if cp := CablePath.from_origin(objects):
-            cp.save()
+        # Arrange objects by cable connector. All objects with a null connector are grouped together. Channel
+        # subinterfaces must each originate their own path, as sharing a connector would otherwise collapse a
+        # group of siblings into a single malformed path.
+        origins = defaultdict(list)
+        for obj in expanded:
+            if isinstance(obj, Interface) and obj.channel_id:
+                # Traced before the retirement so that an unsupported topology aborts without dropping a path
+                cp = CablePath.from_origin([obj])
+                retire_superseded_paths([obj])
+                if cp:
+                    cp.save()
+            else:
+                origins[obj.cable_connector].append(obj)
+
+        for connector, objects in origins.items():
+            cp = CablePath.from_origin(objects)
+            retire_superseded_paths(objects)
+            if cp:
+                cp.save()
 
 
 def rebuild_paths(terminations):