Remove many-to-one profiles

Author: jeremystretch

netbox/dcim/cable_profiles.py

@@ -12,10 +12,6 @@ class BaseCableProfile:
     # Number of A & B terminations must match
     symmetrical = True
 
-    # Whether terminations on either side of the cable have a numeric position
-    a_side_numbered = True
-    b_side_numbered = True
-
     def clean(self, cable):
         if self.a_max_connections and len(cable.a_terminations) > self.a_max_connections:
             raise ValidationError({
@@ -54,24 +50,21 @@ def get_mapped_position(self, side, position):
 
     def get_peer_terminations(self, terminations, position_stack):
         local_end = terminations[0].cable_end
-        position = None
-
-        # Pop the position stack if necessary
-        if (local_end == 'A' and self.b_side_numbered) or (local_end == 'B' and self.a_side_numbered):
-            try:
-                position = position_stack.pop()[0]
-            except IndexError:
-                # TODO: Should this raise an error?
-                # Bottomed out of stack
-                pass
-
         qs = CableTermination.objects.filter(
             cable=terminations[0].cable,
             cable_end=terminations[0].opposite_cable_end
         )
-        if position is not None:
-            qs = qs.filter(position=self.get_mapped_position(local_end, position))
-        return qs
+
+        # TODO: Optimize this to use a single query under any condition
+        if position_stack:
+            # Attempt to find a peer termination at the same position currently in the stack. Pop the stack only if
+            # we find one. Otherwise, return any peer terminations with a null position.
+            position = self.get_mapped_position(local_end, position_stack[-1][0])
+            if peers := qs.filter(position=position):
+                position_stack.pop()
+                return peers
+
+        return qs.filter(position=None)
 
 
 class StraightSingleCableProfile(BaseCableProfile):
@@ -84,20 +77,6 @@ class StraightMultiCableProfile(BaseCableProfile):
     b_max_connections = None
 
 
-class AToManyCableProfile(BaseCableProfile):
-    a_max_connections = 1
-    b_max_connections = None
-    symmetrical = False
-    a_side_numbered = False
-
-
-class BToManyCableProfile(BaseCableProfile):
-    a_max_connections = None
-    b_max_connections = 1
-    symmetrical = False
-    b_side_numbered = False
-
-
 class Shuffle2x2MPO8CableProfile(BaseCableProfile):
     a_max_connections = 8
     b_max_connections = 8
@@ -129,7 +108,7 @@ class Shuffle4x4MPO8CableProfile(BaseCableProfile):
         7: 6,
         8: 8,
     }
-    # B side to A side position mapping
+    # B side to A side position mapping (reverse of _a_mapping)
     _b_mapping = {v: k for k, v in _a_mapping.items()}
 
     def get_mapped_position(self, side, position):

netbox/dcim/choices.py

@@ -1720,17 +1720,12 @@ class PortTypeChoices(ChoiceSet):
 class CableProfileChoices(ChoiceSet):
     STRAIGHT_SINGLE = 'straight-single'
     STRAIGHT_MULTI = 'straight-multi'
-    A_TO_MANY = 'a-to-many'
-    B_TO_MANY = 'b-to-many'
     SHUFFLE_2X2_MPO8 = 'shuffle-2x2-mpo8'
     SHUFFLE_4X4_MPO8 = 'shuffle-4x4-mpo8'
 
     CHOICES = (
         (STRAIGHT_SINGLE, _('Straight (single position)')),
         (STRAIGHT_MULTI, _('Straight (multi-position)')),
-        # TODO: Better names for many-to-one profiles?
-        (A_TO_MANY, _('A to many')),
-        (B_TO_MANY, _('B to many')),
         (SHUFFLE_2X2_MPO8, _('Shuffle (2x2 MPO8)')),
         (SHUFFLE_4X4_MPO8, _('Shuffle (4x4 MPO8)')),
     )

netbox/dcim/models/cables.py

@@ -21,7 +21,7 @@
 from utilities.querysets import RestrictedQuerySet
 from utilities.serialization import deserialize_object, serialize_object
 from wireless.models import WirelessLink
-from .device_components import FrontPort, RearPort, PathEndpoint
+from .device_components import FrontPort, PathEndpoint, RearPort
 
 __all__ = (
     'Cable',
@@ -136,8 +136,6 @@ def profile_class(self):
         return {
             CableProfileChoices.STRAIGHT_SINGLE: cable_profiles.StraightSingleCableProfile,
             CableProfileChoices.STRAIGHT_MULTI: cable_profiles.StraightMultiCableProfile,
-            CableProfileChoices.A_TO_MANY: cable_profiles.AToManyCableProfile,
-            CableProfileChoices.B_TO_MANY: cable_profiles.BToManyCableProfile,
             CableProfileChoices.SHUFFLE_2X2_MPO8: cable_profiles.Shuffle2x2MPO8CableProfile,
             CableProfileChoices.SHUFFLE_4X4_MPO8: cable_profiles.Shuffle4x4MPO8CableProfile,
         }.get(self.profile)
@@ -328,7 +326,6 @@ def update_terminations(self):
         Create/delete CableTerminations for this Cable to reflect its current state.
         """
         a_terminations, b_terminations = self.get_terminations()
-        profile = self.profile_class if self.profile else None
 
         # Delete any stale CableTerminations
         for termination, ct in a_terminations.items():
@@ -341,11 +338,11 @@ def update_terminations(self):
         # Save any new CableTerminations
         for i, termination in enumerate(self.a_terminations, start=1):
             if not termination.pk or termination not in a_terminations:
-                position = i if profile and profile.a_side_numbered else None
+                position = i if self.profile and isinstance(termination, PathEndpoint) else None
                 CableTermination(cable=self, cable_end='A', position=position, termination=termination).save()
         for i, termination in enumerate(self.b_terminations, start=1):
             if not termination.pk or termination not in b_terminations:
-                position = i if profile and profile.b_side_numbered else None
+                position = i if self.profile and isinstance(termination, PathEndpoint) else None
                 CableTermination(cable=self, cable_end='B', position=position, termination=termination).save()
 
 

netbox/dcim/tests/test_cablepaths.py

@@ -2191,6 +2191,55 @@ def test_222_single_path_via_multiple_singleposition_rear_ports(self):
         CableTraceSVG(interface1).render()
         CableTraceSVG(interface2).render()
 
+    def test_223_single_path_via_multiple_pass_throughs_with_breakouts(self):
+        """
+        [IF1] --C1-- [FP1] [RP1] --C2-- [IF3]
+        [IF2]        [FP2] [RP2]        [IF4]
+        """
+        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')
+        interface4 = Interface.objects.create(device=self.device, name='Interface 4')
+        rearport1 = RearPort.objects.create(device=self.device, name='Rear Port 1', positions=1)
+        rearport2 = RearPort.objects.create(device=self.device, name='Rear Port 2', positions=1)
+        frontport1 = FrontPort.objects.create(
+            device=self.device, name='Front Port 1', rear_port=rearport1, rear_port_position=1
+        )
+        frontport2 = FrontPort.objects.create(
+            device=self.device, name='Front Port 2', rear_port=rearport2, rear_port_position=1
+        )
+
+        # Create cables
+        cable1 = Cable(
+            a_terminations=[interface1, interface2],
+            b_terminations=[frontport1, frontport2]
+        )
+        cable1.save()
+        cable2 = Cable(
+            a_terminations=[rearport1, rearport2],
+            b_terminations=[interface3, interface4]
+        )
+        cable2.save()
+
+        # Validate paths
+        self.assertPathExists(
+            (
+                [interface1, interface2], cable1, [frontport1, frontport2],
+                [rearport1, rearport2], cable2, [interface3, interface4],
+            ),
+            is_complete=True,
+            is_active=True
+        )
+        self.assertPathExists(
+            (
+                [interface3, interface4], cable2, [rearport1, rearport2],
+                [frontport1, frontport2], cable1, [interface1, interface2],
+            ),
+            is_complete=True,
+            is_active=True
+        )
+        self.assertEqual(CablePath.objects.count(), 2)
+
     def test_301_create_path_via_existing_cable(self):
         """
         [IF1] --C1-- [FP1] [RP1] --C2-- [RP2] [FP2] --C3-- [IF2]