Fixes #22565: Include Circuit distance in Cable Path length calculations (#22666)

netbox/dcim/models/cables.py

@@ -1169,10 +1169,14 @@ class CablePath(models.Model):
 
     def get_total_length(self):
         """
-        Return a tuple containing the sum of the length of each cable in the path
-        and a flag indicating whether the length is definitive.
+        Return a tuple containing the sum of the length of each cable and the distance of each circuit
+        crossed by the path, and a flag indicating whether the length is definitive.
         """
-        cable_ct = ObjectType.objects.get_for_model(Cable).pk
+        from circuits.models import CircuitTermination
+
+        object_types = ObjectType.objects.get_for_models(Cable, CircuitTermination)
+        cable_ct = object_types[Cable].pk
+        circuit_termination_ct = object_types[CircuitTermination].pk
 
         # Pre-cache cable lengths by ID
         cable_ids = self.get_cable_ids()
@@ -1181,20 +1185,51 @@ class CablePath(models.Model):
             for cable in Cable.objects.filter(id__in=cable_ids, _abs_length__isnull=False).values('pk', '_abs_length')
         }
 
+        # Pre-cache the circuit terminations within the path, along with their circuits
+        circuit_termination_ids = []
+        for node in self._nodes:
+            ct, pk = decompile_path_node(node)
+            if ct == circuit_termination_ct:
+                circuit_termination_ids.append(pk)
+        circuit_terminations = CircuitTermination.objects.select_related('circuit').in_bulk(circuit_termination_ids)
+
         # Iterate through each set of nodes in the path. For cables, add the length of the longest cable to the total
-        # length of the path.
+        # length of the path. Also map each set of nodes to its circuit terminations, keyed by circuit ID.
         total_length = 0
+        circuit_hops = []
         for node_set in self.path:
             hop_length = 0
+            hop_terminations = {}
             for node in node_set:
                 ct, pk = decompile_path_node(node)
-                if ct != cable_ct:
-                    break  # Not a cable
-                if pk in cables and cables[pk] > hop_length:
-                    hop_length = cables[pk]
+                if ct == cable_ct:
+                    if pk in cables and cables[pk] > hop_length:
+                        hop_length = cables[pk]
+                elif ct == circuit_termination_ct:
+                    termination = circuit_terminations.get(pk)
+                    if termination is not None:
+                        hop_terminations[termination.circuit_id] = termination
+                else:
+                    break  # Neither a cable nor a circuit termination
             total_length += hop_length
-
-        is_definitive = len(cables) == len(cable_ids)
+            circuit_hops.append(hop_terminations)
+
+        # Unresolvable circuit terminations may conceal a crossing, so they render the total non-definitive
+        is_definitive = len(cables) == len(cable_ids) and len(circuit_terminations) == len(set(circuit_termination_ids))
+
+        # A circuit crossing appears as two adjacent sets of opposing terminations of the same circuit. For each
+        # crossing, add the longest distance among the circuits crossed, mirroring the handling of parallel cables.
+        for near_hop, far_hop in itertools.pairwise(circuit_hops):
+            crossing_distance = 0
+            for circuit_id in near_hop.keys() & far_hop.keys():
+                if near_hop[circuit_id].term_side == far_hop[circuit_id].term_side:
+                    continue
+                distance = near_hop[circuit_id].circuit._abs_distance
+                if distance is None:
+                    is_definitive = False
+                elif distance > crossing_distance:
+                    crossing_distance = distance
+            total_length += crossing_distance
 
         return total_length, is_definitive
 

netbox/dcim/tests/test_cablepaths2.py

@@ -1,10 +1,11 @@
 from unittest import skip
 
-from circuits.models import CircuitTermination
-from dcim.choices import CableProfileChoices
+from circuits.models import Circuit, CircuitTermination, ProviderNetwork
+from dcim.choices import CableLengthUnitChoices, CableProfileChoices
 from dcim.models import *
 from dcim.svg import CableTraceSVG
 from dcim.tests.utils import BaseCablePathTestCase
+from netbox.choices import DistanceUnitChoices
 
 
 class CablePathTestCase(BaseCablePathTestCase):
@@ -1954,6 +1955,362 @@ class CablePathTestCase(BaseCablePathTestCase):
         CableTraceSVG(interfaces[1]).render()
         CableTraceSVG(interfaces[2]).render()
 
+    def test_226_total_length_via_circuit(self):
+        """
+        [IF1] --C1-- [CT1] [CT2] --C2-- [IF2]
+        """
+        self.circuit.distance = 10
+        self.circuit.distance_unit = DistanceUnitChoices.UNIT_KILOMETER
+        self.circuit.save()
+
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        interface2 = Interface.objects.create(device=self.device, name='Interface 2')
+        circuittermination1 = CircuitTermination.objects.create(
+            circuit=self.circuit,
+            termination=self.site,
+            term_side='A'
+        )
+        circuittermination2 = CircuitTermination.objects.create(
+            circuit=self.circuit,
+            termination=self.site,
+            term_side='Z'
+        )
+
+        # Create cables
+        cable1 = Cable(
+            profile=CableProfileChoices.SINGLE_1C1P,
+            length=200,
+            length_unit=CableLengthUnitChoices.UNIT_METER,
+            a_terminations=[interface1],
+            b_terminations=[circuittermination1]
+        )
+        cable1.clean()
+        cable1.save()
+        cable2 = Cable(
+            profile=CableProfileChoices.SINGLE_1C1P,
+            length=200,
+            length_unit=CableLengthUnitChoices.UNIT_METER,
+            a_terminations=[circuittermination2],
+            b_terminations=[interface2]
+        )
+        cable2.clean()
+        cable2.save()
+
+        # Check for complete paths in both directions
+        paths = [
+            self.assertPathExists(
+                (interface1, cable1, circuittermination1, circuittermination2, cable2, interface2),
+                is_complete=True,
+                is_active=True
+            ),
+            self.assertPathExists(
+                (interface2, cable2, circuittermination2, circuittermination1, cable1, interface1),
+                is_complete=True,
+                is_active=True
+            ),
+        ]
+
+        # The crossed circuit's distance counts toward the total path length
+        for path in paths:
+            self.assertEqual(path.get_total_length(), (10400, True))
+
+        # An unset circuit distance makes the total length non-definitive
+        self.circuit.distance = None
+        self.circuit.save()
+        for path in paths:
+            self.assertEqual(path.get_total_length(), (400, False))
+
+        # A circuit distance of zero is a known value and remains definitive
+        self.circuit.distance = 0
+        self.circuit.distance_unit = DistanceUnitChoices.UNIT_KILOMETER
+        self.circuit.save()
+        for path in paths:
+            self.assertEqual(path.get_total_length(), (400, True))
+
+    def test_227_total_length_via_circuit_without_peer_termination(self):
+        """
+        [IF1] --C1-- [CT1]
+        """
+        self.circuit.distance = 10
+        self.circuit.distance_unit = DistanceUnitChoices.UNIT_KILOMETER
+        self.circuit.save()
+
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        circuittermination1 = CircuitTermination.objects.create(
+            circuit=self.circuit,
+            termination=self.site,
+            term_side='A'
+        )
+        cable1 = Cable(
+            profile=CableProfileChoices.SINGLE_1C1P,
+            length=200,
+            length_unit=CableLengthUnitChoices.UNIT_METER,
+            a_terminations=[interface1],
+            b_terminations=[circuittermination1]
+        )
+        cable1.clean()
+        cable1.save()
+
+        # The distance of a circuit which the path does not cross is excluded
+        path = self.assertPathExists(
+            (interface1, cable1, circuittermination1),
+            is_complete=False
+        )
+        self.assertEqual(path.get_total_length(), (200, True))
+
+    def test_228_total_length_via_multiple_circuits(self):
+        """
+        [IF1] --C1-- [CT1] [CT2] --C2-- [CT3] [CT4] --C3-- [IF2]
+        """
+        self.circuit.distance = 10
+        self.circuit.distance_unit = DistanceUnitChoices.UNIT_KILOMETER
+        self.circuit.save()
+        circuit2 = Circuit.objects.create(
+            provider=self.circuit.provider,
+            type=self.circuit.type,
+            cid='Circuit 2',
+            distance=5,
+            distance_unit=DistanceUnitChoices.UNIT_KILOMETER
+        )
+
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        interface2 = Interface.objects.create(device=self.device, name='Interface 2')
+        circuittermination1 = CircuitTermination.objects.create(
+            circuit=self.circuit,
+            termination=self.site,
+            term_side='A'
+        )
+        circuittermination2 = CircuitTermination.objects.create(
+            circuit=self.circuit,
+            termination=self.site,
+            term_side='Z'
+        )
+        circuittermination3 = CircuitTermination.objects.create(
+            circuit=circuit2,
+            termination=self.site,
+            term_side='A'
+        )
+        circuittermination4 = CircuitTermination.objects.create(
+            circuit=circuit2,
+            termination=self.site,
+            term_side='Z'
+        )
+
+        # Create cables
+        cable1 = Cable(
+            profile=CableProfileChoices.SINGLE_1C1P,
+            length=200,
+            length_unit=CableLengthUnitChoices.UNIT_METER,
+            a_terminations=[interface1],
+            b_terminations=[circuittermination1]
+        )
+        cable1.clean()
+        cable1.save()
+        cable2 = Cable(
+            profile=CableProfileChoices.SINGLE_1C1P,
+            length=100,
+            length_unit=CableLengthUnitChoices.UNIT_METER,
+            a_terminations=[circuittermination2],
+            b_terminations=[circuittermination3]
+        )
+        cable2.clean()
+        cable2.save()
+        cable3 = Cable(
+            profile=CableProfileChoices.SINGLE_1C1P,
+            length=200,
+            length_unit=CableLengthUnitChoices.UNIT_METER,
+            a_terminations=[circuittermination4],
+            b_terminations=[interface2]
+        )
+        cable3.clean()
+        cable3.save()
+
+        # Check for complete paths in both directions
+        paths = [
+            self.assertPathExists(
+                (
+                    interface1, cable1, circuittermination1, circuittermination2, cable2, circuittermination3,
+                    circuittermination4, cable3, interface2,
+                ),
+                is_complete=True,
+                is_active=True
+            ),
+            self.assertPathExists(
+                (
+                    interface2, cable3, circuittermination4, circuittermination3, cable2, circuittermination2,
+                    circuittermination1, cable1, interface1,
+                ),
+                is_complete=True,
+                is_active=True
+            ),
+        ]
+
+        # Each crossed circuit adds its distance to the total path length
+        for path in paths:
+            self.assertEqual(path.get_total_length(), (15500, True))
+
+    def test_229_total_length_via_circuit_to_site(self):
+        """
+        [IF1] --C1-- [CT1] [CT2] --> [Site2]
+        """
+        self.circuit.distance = 10
+        self.circuit.distance_unit = DistanceUnitChoices.UNIT_KILOMETER
+        self.circuit.save()
+
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        site2 = Site.objects.create(name='Site 2', slug='site-2')
+        circuittermination1 = CircuitTermination.objects.create(
+            circuit=self.circuit,
+            termination=self.site,
+            term_side='A'
+        )
+        circuittermination2 = CircuitTermination.objects.create(
+            circuit=self.circuit,
+            termination=site2,
+            term_side='Z'
+        )
+        cable1 = Cable(
+            profile=CableProfileChoices.SINGLE_1C1P,
+            length=200,
+            length_unit=CableLengthUnitChoices.UNIT_METER,
+            a_terminations=[interface1],
+            b_terminations=[circuittermination1]
+        )
+        cable1.clean()
+        cable1.save()
+
+        # The distance of a circuit crossed to reach its far-side site counts toward the total length
+        path = self.assertPathExists(
+            (interface1, cable1, circuittermination1, circuittermination2, site2),
+            is_active=True
+        )
+        self.assertEqual(path.get_total_length(), (10200, True))
+
+    def test_230_total_length_via_circuit_to_providernetwork(self):
+        """
+        [IF1] --C1-- [CT1] [CT2] --> [PN1]
+        """
+        self.circuit.distance = 10
+        self.circuit.distance_unit = DistanceUnitChoices.UNIT_KILOMETER
+        self.circuit.save()
+
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        providernetwork = ProviderNetwork.objects.create(name='Provider Network 1', provider=self.circuit.provider)
+        circuittermination1 = CircuitTermination.objects.create(
+            circuit=self.circuit,
+            termination=self.site,
+            term_side='A'
+        )
+        circuittermination2 = CircuitTermination.objects.create(
+            circuit=self.circuit,
+            termination=providernetwork,
+            term_side='Z'
+        )
+        cable1 = Cable(
+            profile=CableProfileChoices.SINGLE_1C1P,
+            length=200,
+            length_unit=CableLengthUnitChoices.UNIT_METER,
+            a_terminations=[interface1],
+            b_terminations=[circuittermination1]
+        )
+        cable1.clean()
+        cable1.save()
+
+        # The distance of a circuit crossed to reach a provider network counts toward the total length
+        path = self.assertPathExists(
+            (interface1, cable1, circuittermination1, circuittermination2, providernetwork),
+            is_complete=True,
+            is_active=True
+        )
+        self.assertEqual(path.get_total_length(), (10200, True))
+
+    def test_231_total_length_via_parallel_circuits(self):
+        """
+        [IF1] --C1-- [CT1_A] [CT1_Z] --C2-- [IF2]
+                     [CT2_A] [CT2_Z]
+        """
+        self.circuit.distance = 10
+        self.circuit.distance_unit = DistanceUnitChoices.UNIT_KILOMETER
+        self.circuit.save()
+        circuit2 = Circuit.objects.create(
+            provider=self.circuit.provider,
+            type=self.circuit.type,
+            cid='Circuit 2',
+            distance=5,
+            distance_unit=DistanceUnitChoices.UNIT_KILOMETER
+        )
+
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        interface2 = Interface.objects.create(device=self.device, name='Interface 2')
+        circuittermination1_A = CircuitTermination.objects.create(
+            circuit=self.circuit,
+            termination=self.site,
+            term_side='A'
+        )
+        circuittermination1_Z = CircuitTermination.objects.create(
+            circuit=self.circuit,
+            termination=self.site,
+            term_side='Z'
+        )
+        circuittermination2_A = CircuitTermination.objects.create(
+            circuit=circuit2,
+            termination=self.site,
+            term_side='A'
+        )
+        circuittermination2_Z = CircuitTermination.objects.create(
+            circuit=circuit2,
+            termination=self.site,
+            term_side='Z'
+        )
+
+        # Create cables
+        cable1 = Cable(
+            length=200,
+            length_unit=CableLengthUnitChoices.UNIT_METER,
+            a_terminations=[interface1],
+            b_terminations=[circuittermination1_A, circuittermination2_A]
+        )
+        cable1.clean()
+        cable1.save()
+        cable2 = Cable(
+            length=200,
+            length_unit=CableLengthUnitChoices.UNIT_METER,
+            a_terminations=[circuittermination1_Z, circuittermination2_Z],
+            b_terminations=[interface2]
+        )
+        cable2.clean()
+        cable2.save()
+
+        # Check for complete paths in both directions
+        paths = [
+            self.assertPathExists(
+                (
+                    interface1, cable1, (circuittermination1_A, circuittermination2_A),
+                    (circuittermination1_Z, circuittermination2_Z), cable2, interface2,
+                ),
+                is_complete=True,
+                is_active=True
+            ),
+            self.assertPathExists(
+                (
+                    interface2, cable2, (circuittermination1_Z, circuittermination2_Z),
+                    (circuittermination1_A, circuittermination2_A), cable1, interface1,
+                ),
+                is_complete=True,
+                is_active=True
+            ),
+        ]
+
+        # Parallel circuits crossed in the same hop count only the longest distance
+        for path in paths:
+            self.assertEqual(path.get_total_length(), (10400, True))
+
+        # An unset distance on one parallel circuit keeps the longest known distance but is non-definitive
+        circuit2.distance = None
+        circuit2.save()
+        for path in paths:
+            self.assertEqual(path.get_total_length(), (10400, False))
+
     def test_304_add_port_mapping_between_connected_ports(self):
         """
         [IF1] --C1-- [FP1] [RP1] --C2-- [IF2]