review fixes

netbox/dcim/tests/test_channelization.py

@@ -548,6 +548,69 @@ class ChannelizedCablePathTestCase(BaseCablePathTestCase):
         )
         self.assertEqual(CablePath.objects.count(), 1)
 
+    def _move_channel_between_cabled_parents(self, old_name, new_name):
+        """
+        Cable two channelized parents, then move a channel subinterface from the first to the second. The
+        parents are retraced in name order, so the caller's naming decides which is processed first.
+        """
+        old_parent, old_channels = self._create_channelized_interface(old_name, 4)
+        new_parent, new_channels = self._create_channelized_interface(new_name, 4)
+        old_far = [
+            Interface.objects.create(device=self.device, name=f'xe{i}', type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS)
+            for i in range(4)
+        ]
+        new_far = [
+            Interface.objects.create(device=self.device, name=f'ye{i}', type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS)
+            for i in range(4)
+        ]
+        old_cable = Cable(
+            profile=CableProfileChoices.BREAKOUT_1C4P_4C1P, a_terminations=[old_parent], b_terminations=old_far
+        )
+        old_cable.clean()
+        old_cable.save()
+        new_cable = Cable(
+            profile=CableProfileChoices.BREAKOUT_1C4P_4C1P, a_terminations=[new_parent], b_terminations=new_far
+        )
+        new_cable.clean()
+        new_cable.save()
+
+        # Free position 4 on the new parent, then move the old parent's fourth channel onto it. Both the
+        # channel and its new parent must be refetched: the in-memory instances predate the cables.
+        new_channels[3].delete()
+        channel = Interface.objects.get(pk=old_channels[3].pk)
+        channel.parent = Interface.objects.get(pk=new_parent.pk)
+        channel.save()
+
+        return channel, old_cable, new_cable, old_far[3], new_far[3]
+
+    def _assert_single_origin_path(self, channel, cable, far):
+        """Assert the channel originates exactly one path, traced through the given cable."""
+        originating = [
+            cp for cp in CablePath.objects.filter(_nodes__contains=channel) if channel in cp.origins
+        ]
+        self.assertEqual(len(originating), 1, msg=f'{len(originating)} paths originate at {channel}; expected 1')
+        self.assertPathExists((channel, cable, far), is_complete=True, is_active=True)
+
+    def test_116_move_channel_between_cabled_parents_old_first(self):
+        """
+        Moving a channel subinterface between two cabled channelized parents, the old parent retraced first.
+        The channel's stale mirrored cable must not resurrect a path through the old cable.
+        """
+        channel, old_cable, new_cable, old_far, new_far = self._move_channel_between_cabled_parents('et0', 'et1')
+
+        self._assert_single_origin_path(channel, new_cable, new_far)
+        self.assertPathDoesNotExist((channel, old_cable, old_far))
+
+    def test_117_move_channel_between_cabled_parents_new_first(self):
+        """
+        The same move with the new parent retraced first: restoring the old path must not duplicate the one
+        already traced through the new cable.
+        """
+        channel, old_cable, new_cable, old_far, new_far = self._move_channel_between_cabled_parents('et1', 'et0')
+
+        self._assert_single_origin_path(channel, new_cable, new_far)
+        self.assertPathDoesNotExist((channel, old_cable, old_far))
+
 
 class ChannelizedInterfaceTestCase(TestCase):
     """

netbox/dcim/utils.py

@@ -201,9 +201,9 @@ def rebuild_cable_paths(cable):
             if not isinstance(termination, PathEndpoint):
                 affected.update({cp.pk: cp for cp in CablePath.objects.filter(_nodes__contains=termination)})
 
-        # Record each affected path's originating node(s) before deleting it. A path which merely passes through
-        # the Cable originates elsewhere, and can only be retraced from its own origins.
-        origins = {tuple(cp.path[0]): cp.origins for cp in affected.values()}
+        # Record each affected path's originating node(s) before deleting it. These are kept as compiled path
+        # nodes; resolving them to objects is deferred to the paths which actually need restoring.
+        origin_keys = {tuple(cp.path[0]) for cp in affected.values()}
 
         # Delete existing paths individually so each clears its `_path` back-reference on the originating endpoints.
         for cp in affected.values():
@@ -211,17 +211,28 @@ def rebuild_cable_paths(cable):
 
         # Trace from the Cable's own terminations first, so that a channelized origin is expanded into its channel
         # subinterfaces exactly once
-        retraced = set()
         for nodes in (a_terminations, b_terminations):
             if nodes and isinstance(nodes[0], PathEndpoint):
                 create_cablepaths(nodes)
-                retraced.add(tuple(object_to_path_node(node) for node in nodes))
+        retraced = {tuple(cp.path[0]) for cp in CablePath.objects.filter(_nodes__contains=cable)}
 
-        # Restore any affected path the tracing above did not reproduce
-        retraced |= {tuple(cp.path[0]) for cp in CablePath.objects.filter(_nodes__contains=cable)}
-        for key, nodes in origins.items():
-            if key not in retraced:
-                create_cablepaths(nodes)
+        # Restore the affected paths which merely passed through the Cable: those originate elsewhere, so the
+        # tracing above cannot reproduce them.
+        for key in origin_keys - retraced:
+            nodes = [obj for node in key if (obj := path_node_to_object(node))]
+            if not nodes:
+                continue
+
+            # An origin which terminates this Cable belongs to the tracing above: that it produced no path means
+            # the origin no longer has one (e.g. a channel subinterface moved to another parent).
+            if any(getattr(obj, 'cable_id', None) == cable.pk for obj in nodes):
+                continue
+
+            # Nor restore an origin whose path has already been traced through another Cable
+            if key in {tuple(cp.path[0]) for cp in CablePath.objects.filter(_nodes__contains=nodes[0])}:
+                continue
+
+            create_cablepaths(nodes)
 
 
 def update_interface_parents(device, interface_templates, module=None):