Address review findings

netbox/dcim/models/cables.py

@@ -993,8 +993,8 @@ class CablePath(models.Model):
         is_complete = False
         is_active = True
         is_split = False
-        # Path length and active state where the trace fanned out across a multi-position front port
-        fanout = None
+        # Whether the trace fanned out across a multi-position front port without positional context
+        fanout = False
         diverged = False
 
         logger.debug(f'Tracing cable path from {terminations}...')
@@ -1015,7 +1015,7 @@ class CablePath(models.Model):
                 not isinstance(terminations[0].parent_object, Circuit) and
                 not all(t.parent_object == terminations[0].parent_object for t in terminations[1:])
             ):
-                if fanout is None:
+                if not fanout:
                     raise UnsupportedCablePath(_("All mid-span terminations must have the same parent object"))
                 # Fanned-out branches crossing different devices are not followed, even if they reconverge later
                 diverged = True
@@ -1186,8 +1186,8 @@ class CablePath(models.Model):
                     port_mappings = PortMapping.objects.filter(q_filter)
                 else:
                     # Without positional context, follow all mappings; branches must converge on one endpoint
-                    if remote_terminations[0].positions > 1 and fanout is None:
-                        fanout = (len(path), is_active)
+                    if remote_terminations[0].positions > 1:
+                        fanout = True
                     port_mappings = PortMapping.objects.filter(front_port__in=remote_terminations)
                 if not port_mappings:
                     break
@@ -1260,7 +1260,7 @@ class CablePath(models.Model):
 
             else:
                 # Check for non-symmetric path
-                if fanout is not None and len(remote_terminations) > 1:
+                if fanout and len(remote_terminations) > 1:
                     diverged = True
                 elif all(isinstance(t, type(remote_terminations[0])) for t in remote_terminations[1:]):
                     is_complete = True
@@ -1274,10 +1274,8 @@ class CablePath(models.Model):
                 break
 
         if diverged:
-            # Branches from a fanned-out front port reached different endpoints; truncate as a split path
+            # Keep the full trace so changes to any branch retrace this path
             logger.debug('Front port branches diverged; treating path as split.')
-            path = path[:fanout[0]]
-            is_active = fanout[1]
             is_complete = False
             is_split = True
 
@@ -1408,6 +1406,9 @@ class CablePath(models.Model):
             return [
                 ct.get_peer_termination() for ct in nodes
             ]
+        # Front port branches diverging to different endpoints
+        if isinstance(nodes[0], PathEndpoint):
+            return nodes
         return []
 
     def get_asymmetric_nodes(self):

netbox/dcim/tests/test_cablepaths.py

@@ -2840,11 +2840,15 @@ class LegacyCablePathTestCase(BaseCablePathTestCase):
         cable3.save()
 
         # Branches reaching different endpoints are still treated as a split
-        self.assertPathExists(
-            (interface1, cable1, frontport1),
+        path1 = self.assertPathExists(
+            (
+                interface1, cable1, frontport1, (rearport1, rearport2), (cable2, cable3),
+                (interface2, interface3)
+            ),
             is_complete=False,
             is_split=True
         )
+        self.assertEqual(path1.get_split_nodes(), [interface2, interface3])
         self.assertPathExists(
             (interface2, cable2, rearport1, frontport1, cable1, interface1),
             is_complete=True,
@@ -2973,11 +2977,11 @@ class LegacyCablePathTestCase(BaseCablePathTestCase):
         cable3 = Cable(a_terminations=[rearport2], b_terminations=[rearport4])
         cable3.save()
 
-        # Branches crossing different devices are truncated at the front port, keeping its active state
+        # Branches crossing different devices are not followed past the remote rear ports
         path1 = self.assertPathExists(
-            (interface1, cable1, frontport1),
+            (interface1, cable1, frontport1, (rearport1, rearport2), (cable2, cable3), (rearport3, rearport4)),
             is_complete=False,
-            is_active=True,
+            is_active=False,
             is_split=True
         )
         self.assertEqual(CablePath.objects.count(), 1)
@@ -3041,6 +3045,91 @@ class LegacyCablePathTestCase(BaseCablePathTestCase):
         self.assertPathIsSet(interface2, path2)
         self.assertEqual(interface1.connected_endpoints, [interface2])
 
+    def test_231_multiposition_frontport_rewire_diverged_to_converged(self):
+        """
+        [IF1] --C1-- [FP1:1] [RP1] --C2-- [RP3] [FP2] --C4-- [IF2]
+                     [FP1:2] [RP2] --C3-- [RP4] [FP3] --C5-- [IF3]
+        Rewired so that FP2 and FP3 both connect to IF2, then back again.
+        """
+        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')
+        rearport1 = RearPort.objects.create(device=self.device, name='Rear Port 1')
+        rearport2 = RearPort.objects.create(device=self.device, name='Rear Port 2')
+        rearport3 = RearPort.objects.create(device=self.device, name='Rear Port 3')
+        rearport4 = RearPort.objects.create(device=self.device, name='Rear Port 4')
+        frontport1 = FrontPort.objects.create(device=self.device, name='Front Port 1', positions=2)
+        frontport2 = FrontPort.objects.create(device=self.device, name='Front Port 2')
+        frontport3 = FrontPort.objects.create(device=self.device, name='Front Port 3')
+        PortMapping.objects.bulk_create([
+            PortMapping(
+                device=self.device, front_port=frontport1, front_port_position=1,
+                rear_port=rearport1, rear_port_position=1,
+            ),
+            PortMapping(
+                device=self.device, front_port=frontport1, front_port_position=2,
+                rear_port=rearport2, rear_port_position=1,
+            ),
+            PortMapping(
+                device=self.device, front_port=frontport2, front_port_position=1,
+                rear_port=rearport3, rear_port_position=1,
+            ),
+            PortMapping(
+                device=self.device, front_port=frontport3, front_port_position=1,
+                rear_port=rearport4, rear_port_position=1,
+            ),
+        ])
+
+        def assert_path_is_current(interface):
+            interface.refresh_from_db()
+            stored = interface.path
+            traced = CablePath.from_origin([interface])
+            self.assertEqual(
+                (stored.path, stored.is_complete, stored.is_active, stored.is_split),
+                (traced.path, traced.is_complete, traced.is_active, traced.is_split),
+            )
+            return stored
+
+        cable1 = Cable(a_terminations=[interface1], b_terminations=[frontport1])
+        cable1.save()
+        cable2 = Cable(a_terminations=[rearport1], b_terminations=[rearport3])
+        cable2.save()
+        cable3 = Cable(a_terminations=[rearport2], b_terminations=[rearport4])
+        cable3.save()
+        cable4 = Cable(a_terminations=[frontport2], b_terminations=[interface2])
+        cable4.save()
+        cable5 = Cable(a_terminations=[frontport3], b_terminations=[interface3])
+        cable5.save()
+        path1 = assert_path_is_current(interface1)
+        self.assertTrue(path1.is_split)
+        self.assertFalse(path1.is_complete)
+
+        # Removing one branch's endpoint leaves the other branch complete (and asymmetric)
+        cable5.delete()
+        path1 = assert_path_is_current(interface1)
+        self.assertTrue(path1.is_complete)
+
+        # Converge both branches on a single endpoint
+        cable4.delete()
+        cable6 = Cable(a_terminations=[frontport2, frontport3], b_terminations=[interface2])
+        cable6.save()
+        path1 = assert_path_is_current(interface1)
+        self.assertTrue(path1.is_complete)
+        self.assertFalse(path1.is_split)
+        self.assertEqual(Interface.objects.get(pk=interface1.pk).connected_endpoints, [interface2])
+        self.assertEqual(Interface.objects.get(pk=interface2.pk).connected_endpoints, [interface1])
+
+        # Diverge the branches again
+        cable6.delete()
+        cable7 = Cable(a_terminations=[frontport2], b_terminations=[interface2])
+        cable7.save()
+        cable8 = Cable(a_terminations=[frontport3], b_terminations=[interface3])
+        cable8.save()
+        path1 = assert_path_is_current(interface1)
+        self.assertTrue(path1.is_split)
+        self.assertFalse(path1.is_complete)
+        self.assertEqual(Interface.objects.get(pk=interface1.pk).connected_endpoints, [])
+
     def test_301_create_path_via_existing_cable(self):
         """
         [IF1] --C1-- [FP1] [RP1] --C2-- [RP2] [FP2] --C3-- [IF2]