From aa7eedac4247a90a13a9201f576a5dfa7e5e9744 Mon Sep 17 00:00:00 2001 From: Jeremy Stretch Date: Mon, 17 Nov 2025 12:14:06 -0500 Subject: [PATCH] Remove many-to-one profiles --- netbox/dcim/cable_profiles.py | 45 ++---- netbox/dcim/choices.py | 5 - netbox/dcim/models/cables.py | 9 +- netbox/dcim/tests/test_cablepaths.py | 49 ++++++ netbox/dcim/tests/test_cablepaths2.py | 224 +++++++++----------------- 5 files changed, 140 insertions(+), 192 deletions(-) diff --git a/netbox/dcim/cable_profiles.py b/netbox/dcim/cable_profiles.py index 7a7f4953c..818c90fa8 100644 --- a/netbox/dcim/cable_profiles.py +++ b/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 @@ class BaseCableProfile: 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): diff --git a/netbox/dcim/choices.py b/netbox/dcim/choices.py index e1c090093..0656d96aa 100644 --- a/netbox/dcim/choices.py +++ b/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)')), ) diff --git a/netbox/dcim/models/cables.py b/netbox/dcim/models/cables.py index d7777870e..94ebc7570 100644 --- a/netbox/dcim/models/cables.py +++ b/netbox/dcim/models/cables.py @@ -21,7 +21,7 @@ from utilities.fields import ColorField, GenericArrayForeignKey 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 @@ class Cable(PrimaryModel): 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 @@ class Cable(PrimaryModel): 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 @@ class Cable(PrimaryModel): # 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() diff --git a/netbox/dcim/tests/test_cablepaths.py b/netbox/dcim/tests/test_cablepaths.py index 874b68340..d3a7cfc5e 100644 --- a/netbox/dcim/tests/test_cablepaths.py +++ b/netbox/dcim/tests/test_cablepaths.py @@ -2191,6 +2191,55 @@ class LegacyCablePathTests(CablePathTestCase): 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] diff --git a/netbox/dcim/tests/test_cablepaths2.py b/netbox/dcim/tests/test_cablepaths2.py index 54da8c1ad..8e9a4961f 100644 --- a/netbox/dcim/tests/test_cablepaths2.py +++ b/netbox/dcim/tests/test_cablepaths2.py @@ -1,3 +1,5 @@ +from unittest import skipIf + from circuits.models import CircuitTermination from dcim.choices import CableProfileChoices from dcim.models import * @@ -126,143 +128,7 @@ class CablePathTests(CablePathTestCase): # Check that all CablePaths have been deleted self.assertEqual(CablePath.objects.count(), 0) - def test_103_cable_profile_a_to_many(self): - """ - [IF1] --C1-- [IF2] - [IF3] - [IF4] - - Cable profile: A to many - """ - interfaces = [ - Interface.objects.create(device=self.device, name='Interface 1'), - Interface.objects.create(device=self.device, name='Interface 2'), - Interface.objects.create(device=self.device, name='Interface 3'), - Interface.objects.create(device=self.device, name='Interface 4'), - ] - - # Create cable 1 - cable1 = Cable( - profile=CableProfileChoices.A_TO_MANY, - a_terminations=[interfaces[0]], - b_terminations=[interfaces[1], interfaces[2], interfaces[3]], - ) - cable1.clean() - cable1.save() - - # A-to-B path leads to all interfaces - path1 = self.assertPathExists( - (interfaces[0], cable1, [interfaces[1], interfaces[2], interfaces[3]]), - is_complete=True, - is_active=True - ) - # B-to-A paths all lead to Interface 1 - path2 = self.assertPathExists( - (interfaces[1], cable1, interfaces[0]), - is_complete=True, - is_active=True - ) - path3 = self.assertPathExists( - (interfaces[2], cable1, interfaces[0]), - is_complete=True, - is_active=True - ) - path4 = self.assertPathExists( - (interfaces[3], cable1, interfaces[0]), - is_complete=True, - is_active=True - ) - self.assertEqual(CablePath.objects.count(), 4) - - for interface in interfaces: - interface.refresh_from_db() - self.assertPathIsSet(interfaces[0], path1) - self.assertPathIsSet(interfaces[1], path2) - self.assertPathIsSet(interfaces[2], path3) - self.assertPathIsSet(interfaces[3], path4) - self.assertIsNone(interfaces[0].cable_position) - self.assertEqual(interfaces[1].cable_position, 1) - self.assertEqual(interfaces[2].cable_position, 2) - self.assertEqual(interfaces[3].cable_position, 3) - - # Test SVG generation - CableTraceSVG(interfaces[0]).render() - - # Delete cable 1 - cable1.delete() - - # Check that all CablePaths have been deleted - self.assertEqual(CablePath.objects.count(), 0) - - def test_104_cable_profile_b_to_many(self): - """ - [IF1] --C1-- [IF4] - [IF2] - [IF3] - - Cable profile: B to many - """ - interfaces = [ - Interface.objects.create(device=self.device, name='Interface 1'), - Interface.objects.create(device=self.device, name='Interface 2'), - Interface.objects.create(device=self.device, name='Interface 3'), - Interface.objects.create(device=self.device, name='Interface 4'), - ] - - # Create cable 1 - cable1 = Cable( - profile=CableProfileChoices.B_TO_MANY, - a_terminations=[interfaces[0], interfaces[1], interfaces[2]], - b_terminations=[interfaces[3]], - ) - cable1.clean() - cable1.save() - - # A-to-B paths all lead to Interface 4 - path1 = self.assertPathExists( - (interfaces[0], cable1, interfaces[3]), - is_complete=True, - is_active=True - ) - path2 = self.assertPathExists( - (interfaces[1], cable1, interfaces[3]), - is_complete=True, - is_active=True - ) - path3 = self.assertPathExists( - (interfaces[2], cable1, interfaces[3]), - is_complete=True, - is_active=True - ) - # B-to-A path leads to all interfaces - path4 = self.assertPathExists( - (interfaces[3], cable1, [interfaces[0], interfaces[1], interfaces[2]]), - is_complete=True, - is_active=True - ) - self.assertEqual(CablePath.objects.count(), 4) - - for interface in interfaces: - interface.refresh_from_db() - self.assertPathIsSet(interfaces[0], path1) - self.assertPathIsSet(interfaces[1], path2) - self.assertPathIsSet(interfaces[2], path3) - self.assertPathIsSet(interfaces[3], path4) - self.assertEqual(interfaces[0].cable_position, 1) - self.assertEqual(interfaces[1].cable_position, 2) - self.assertEqual(interfaces[2].cable_position, 3) - self.assertIsNone(interfaces[3].cable_position) - - # Test SVG generation - CableTraceSVG(interfaces[0]).render() - - # Delete cable 1 - cable1.delete() - - # Check that all CablePaths have been deleted - self.assertEqual(CablePath.objects.count(), 0) - - def test_105_cable_profile_2x2_mpo8(self): + def test_103_cable_profile_2x2_mpo8(self): """ [IF1:1] --C1-- [IF3:1] [IF1:2] [IF3:2] @@ -374,7 +240,7 @@ class CablePathTests(CablePathTestCase): # Check that all CablePaths have been deleted self.assertEqual(CablePath.objects.count(), 0) - def test_106_cable_profile_4x4_mpo8(self): + def test_104_cable_profile_4x4_mpo8(self): """ [IF1:1] --C1-- [IF3:1] [IF1:2] [IF3:2] @@ -507,13 +373,13 @@ class CablePathTests(CablePathTestCase): # Create cables cable1 = Cable( - profile=CableProfileChoices.B_TO_MANY, + profile=CableProfileChoices.STRAIGHT_MULTI, a_terminations=[interfaces[0], interfaces[1]], b_terminations=[frontport1], ) cable1.save() cable2 = Cable( - profile=CableProfileChoices.A_TO_MANY, + profile=CableProfileChoices.STRAIGHT_MULTI, a_terminations=[rearport1], b_terminations=[interfaces[2], interfaces[3]] ) @@ -586,13 +452,13 @@ class CablePathTests(CablePathTestCase): # Create cables cable1 = Cable( - profile=CableProfileChoices.B_TO_MANY, + profile=CableProfileChoices.STRAIGHT_MULTI, a_terminations=[interfaces[0], interfaces[1]], b_terminations=[frontport1_1] ) cable1.save() cable2 = Cable( - profile=CableProfileChoices.B_TO_MANY, + profile=CableProfileChoices.STRAIGHT_MULTI, a_terminations=[interfaces[2], interfaces[3]], b_terminations=[frontport1_2] ) @@ -604,13 +470,13 @@ class CablePathTests(CablePathTestCase): ) cable3.save() cable4 = Cable( - profile=CableProfileChoices.A_TO_MANY, + profile=CableProfileChoices.STRAIGHT_MULTI, a_terminations=[frontport2_1], b_terminations=[interfaces[4], interfaces[5]] ) cable4.save() cable5 = Cable( - profile=CableProfileChoices.A_TO_MANY, + profile=CableProfileChoices.STRAIGHT_MULTI, a_terminations=[frontport2_2], b_terminations=[interfaces[6], interfaces[7]] ) @@ -722,13 +588,13 @@ class CablePathTests(CablePathTestCase): # Create cables cable1 = Cable( - profile=CableProfileChoices.B_TO_MANY, + profile=CableProfileChoices.STRAIGHT_MULTI, a_terminations=[interfaces[0], interfaces[1]], b_terminations=[circuittermination1] ) cable1.save() cable2 = Cable( - profile=CableProfileChoices.A_TO_MANY, + profile=CableProfileChoices.STRAIGHT_MULTI, a_terminations=[circuittermination2], b_terminations=[interfaces[2], interfaces[3]] ) @@ -801,7 +667,7 @@ class CablePathTests(CablePathTestCase): # Create cables cable1 = Cable( - profile=CableProfileChoices.A_TO_MANY, + profile=CableProfileChoices.STRAIGHT_MULTI, a_terminations=[interfaces[0]], b_terminations=[front_ports[0], front_ports[1]] ) @@ -812,7 +678,7 @@ class CablePathTests(CablePathTestCase): ) cable2.save() cable3 = Cable( - profile=CableProfileChoices.B_TO_MANY, + profile=CableProfileChoices.STRAIGHT_MULTI, a_terminations=[interfaces[1]], b_terminations=[front_ports[2], front_ports[3]] ) @@ -844,3 +710,65 @@ class CablePathTests(CablePathTestCase): # Test SVG generation CableTraceSVG(interfaces[0]).render() + + # TODO: Revisit this test under FR #20564 + @skipIf(True, "Waiting for FR #20564") + def test_223_single_path_via_multiple_pass_throughs_with_breakouts(self): + """ + [IF1] --C1-- [FP1] [RP1] --C2-- [IF3] + [IF2] [FP2] [RP2] [IF4] + """ + interfaces = [ + Interface.objects.create(device=self.device, name='Interface 1'), + Interface.objects.create(device=self.device, name='Interface 2'), + Interface.objects.create(device=self.device, name='Interface 3'), + 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( + profile=CableProfileChoices.STRAIGHT_MULTI, + a_terminations=[interfaces[0], interfaces[1]], + b_terminations=[frontport1, frontport2] + ) + cable1.save() + cable2 = Cable( + profile=CableProfileChoices.STRAIGHT_MULTI, + a_terminations=[rearport1, rearport2], + b_terminations=[interfaces[2], interfaces[3]] + ) + cable2.save() + + for path in CablePath.objects.all(): + print(f'{path}: {path.path_objects}') + + # Validate paths + self.assertPathExists( + (interfaces[0], cable1, [frontport1, frontport2], [rearport1, rearport2], cable2, interfaces[2]), + is_complete=True, + is_active=True + ) + self.assertPathExists( + (interfaces[1], cable1, [frontport1, frontport2], [rearport1, rearport2], cable2, interfaces[3]), + is_complete=True, + is_active=True + ) + self.assertPathExists( + (interfaces[2], cable2, [rearport1, rearport2], [frontport1, frontport2], cable1, interfaces[0]), + is_complete=True, + is_active=True + ) + self.assertPathExists( + (interfaces[3], cable2, [rearport1, rearport2], [frontport1, frontport2], cable1, interfaces[1]), + is_complete=True, + is_active=True + ) + self.assertEqual(CablePath.objects.count(), 4)