Remove many-to-one profiles

This commit is contained in:
Jeremy Stretch
2025-11-17 12:14:06 -05:00
parent a75dee745e
commit aa7eedac42
5 changed files with 140 additions and 192 deletions
+12 -33
View File
@@ -12,10 +12,6 @@ class BaseCableProfile:
# Number of A & B terminations must match # Number of A & B terminations must match
symmetrical = True 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): def clean(self, cable):
if self.a_max_connections and len(cable.a_terminations) > self.a_max_connections: if self.a_max_connections and len(cable.a_terminations) > self.a_max_connections:
raise ValidationError({ raise ValidationError({
@@ -54,24 +50,21 @@ class BaseCableProfile:
def get_peer_terminations(self, terminations, position_stack): def get_peer_terminations(self, terminations, position_stack):
local_end = terminations[0].cable_end 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( qs = CableTermination.objects.filter(
cable=terminations[0].cable, cable=terminations[0].cable,
cable_end=terminations[0].opposite_cable_end cable_end=terminations[0].opposite_cable_end
) )
if position is not None:
qs = qs.filter(position=self.get_mapped_position(local_end, position)) # TODO: Optimize this to use a single query under any condition
return qs 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): class StraightSingleCableProfile(BaseCableProfile):
@@ -84,20 +77,6 @@ class StraightMultiCableProfile(BaseCableProfile):
b_max_connections = None 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): class Shuffle2x2MPO8CableProfile(BaseCableProfile):
a_max_connections = 8 a_max_connections = 8
b_max_connections = 8 b_max_connections = 8
@@ -129,7 +108,7 @@ class Shuffle4x4MPO8CableProfile(BaseCableProfile):
7: 6, 7: 6,
8: 8, 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()} _b_mapping = {v: k for k, v in _a_mapping.items()}
def get_mapped_position(self, side, position): def get_mapped_position(self, side, position):
-5
View File
@@ -1720,17 +1720,12 @@ class PortTypeChoices(ChoiceSet):
class CableProfileChoices(ChoiceSet): class CableProfileChoices(ChoiceSet):
STRAIGHT_SINGLE = 'straight-single' STRAIGHT_SINGLE = 'straight-single'
STRAIGHT_MULTI = 'straight-multi' STRAIGHT_MULTI = 'straight-multi'
A_TO_MANY = 'a-to-many'
B_TO_MANY = 'b-to-many'
SHUFFLE_2X2_MPO8 = 'shuffle-2x2-mpo8' SHUFFLE_2X2_MPO8 = 'shuffle-2x2-mpo8'
SHUFFLE_4X4_MPO8 = 'shuffle-4x4-mpo8' SHUFFLE_4X4_MPO8 = 'shuffle-4x4-mpo8'
CHOICES = ( CHOICES = (
(STRAIGHT_SINGLE, _('Straight (single position)')), (STRAIGHT_SINGLE, _('Straight (single position)')),
(STRAIGHT_MULTI, _('Straight (multi-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_2X2_MPO8, _('Shuffle (2x2 MPO8)')),
(SHUFFLE_4X4_MPO8, _('Shuffle (4x4 MPO8)')), (SHUFFLE_4X4_MPO8, _('Shuffle (4x4 MPO8)')),
) )
+3 -6
View File
@@ -21,7 +21,7 @@ from utilities.fields import ColorField, GenericArrayForeignKey
from utilities.querysets import RestrictedQuerySet from utilities.querysets import RestrictedQuerySet
from utilities.serialization import deserialize_object, serialize_object from utilities.serialization import deserialize_object, serialize_object
from wireless.models import WirelessLink from wireless.models import WirelessLink
from .device_components import FrontPort, RearPort, PathEndpoint from .device_components import FrontPort, PathEndpoint, RearPort
__all__ = ( __all__ = (
'Cable', 'Cable',
@@ -136,8 +136,6 @@ class Cable(PrimaryModel):
return { return {
CableProfileChoices.STRAIGHT_SINGLE: cable_profiles.StraightSingleCableProfile, CableProfileChoices.STRAIGHT_SINGLE: cable_profiles.StraightSingleCableProfile,
CableProfileChoices.STRAIGHT_MULTI: cable_profiles.StraightMultiCableProfile, 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_2X2_MPO8: cable_profiles.Shuffle2x2MPO8CableProfile,
CableProfileChoices.SHUFFLE_4X4_MPO8: cable_profiles.Shuffle4x4MPO8CableProfile, CableProfileChoices.SHUFFLE_4X4_MPO8: cable_profiles.Shuffle4x4MPO8CableProfile,
}.get(self.profile) }.get(self.profile)
@@ -328,7 +326,6 @@ class Cable(PrimaryModel):
Create/delete CableTerminations for this Cable to reflect its current state. Create/delete CableTerminations for this Cable to reflect its current state.
""" """
a_terminations, b_terminations = self.get_terminations() a_terminations, b_terminations = self.get_terminations()
profile = self.profile_class if self.profile else None
# Delete any stale CableTerminations # Delete any stale CableTerminations
for termination, ct in a_terminations.items(): for termination, ct in a_terminations.items():
@@ -341,11 +338,11 @@ class Cable(PrimaryModel):
# Save any new CableTerminations # Save any new CableTerminations
for i, termination in enumerate(self.a_terminations, start=1): for i, termination in enumerate(self.a_terminations, start=1):
if not termination.pk or termination not in a_terminations: 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() CableTermination(cable=self, cable_end='A', position=position, termination=termination).save()
for i, termination in enumerate(self.b_terminations, start=1): for i, termination in enumerate(self.b_terminations, start=1):
if not termination.pk or termination not in b_terminations: 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() CableTermination(cable=self, cable_end='B', position=position, termination=termination).save()
+49
View File
@@ -2191,6 +2191,55 @@ class LegacyCablePathTests(CablePathTestCase):
CableTraceSVG(interface1).render() CableTraceSVG(interface1).render()
CableTraceSVG(interface2).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): def test_301_create_path_via_existing_cable(self):
""" """
[IF1] --C1-- [FP1] [RP1] --C2-- [RP2] [FP2] --C3-- [IF2] [IF1] --C1-- [FP1] [RP1] --C2-- [RP2] [FP2] --C3-- [IF2]
+76 -148
View File
@@ -1,3 +1,5 @@
from unittest import skipIf
from circuits.models import CircuitTermination from circuits.models import CircuitTermination
from dcim.choices import CableProfileChoices from dcim.choices import CableProfileChoices
from dcim.models import * from dcim.models import *
@@ -126,143 +128,7 @@ class CablePathTests(CablePathTestCase):
# Check that all CablePaths have been deleted # Check that all CablePaths have been deleted
self.assertEqual(CablePath.objects.count(), 0) self.assertEqual(CablePath.objects.count(), 0)
def test_103_cable_profile_a_to_many(self): def test_103_cable_profile_2x2_mpo8(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):
""" """
[IF1:1] --C1-- [IF3:1] [IF1:1] --C1-- [IF3:1]
[IF1:2] [IF3:2] [IF1:2] [IF3:2]
@@ -374,7 +240,7 @@ class CablePathTests(CablePathTestCase):
# Check that all CablePaths have been deleted # Check that all CablePaths have been deleted
self.assertEqual(CablePath.objects.count(), 0) 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:1] --C1-- [IF3:1]
[IF1:2] [IF3:2] [IF1:2] [IF3:2]
@@ -507,13 +373,13 @@ class CablePathTests(CablePathTestCase):
# Create cables # Create cables
cable1 = Cable( cable1 = Cable(
profile=CableProfileChoices.B_TO_MANY, profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[interfaces[0], interfaces[1]], a_terminations=[interfaces[0], interfaces[1]],
b_terminations=[frontport1], b_terminations=[frontport1],
) )
cable1.save() cable1.save()
cable2 = Cable( cable2 = Cable(
profile=CableProfileChoices.A_TO_MANY, profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[rearport1], a_terminations=[rearport1],
b_terminations=[interfaces[2], interfaces[3]] b_terminations=[interfaces[2], interfaces[3]]
) )
@@ -586,13 +452,13 @@ class CablePathTests(CablePathTestCase):
# Create cables # Create cables
cable1 = Cable( cable1 = Cable(
profile=CableProfileChoices.B_TO_MANY, profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[interfaces[0], interfaces[1]], a_terminations=[interfaces[0], interfaces[1]],
b_terminations=[frontport1_1] b_terminations=[frontport1_1]
) )
cable1.save() cable1.save()
cable2 = Cable( cable2 = Cable(
profile=CableProfileChoices.B_TO_MANY, profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[interfaces[2], interfaces[3]], a_terminations=[interfaces[2], interfaces[3]],
b_terminations=[frontport1_2] b_terminations=[frontport1_2]
) )
@@ -604,13 +470,13 @@ class CablePathTests(CablePathTestCase):
) )
cable3.save() cable3.save()
cable4 = Cable( cable4 = Cable(
profile=CableProfileChoices.A_TO_MANY, profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[frontport2_1], a_terminations=[frontport2_1],
b_terminations=[interfaces[4], interfaces[5]] b_terminations=[interfaces[4], interfaces[5]]
) )
cable4.save() cable4.save()
cable5 = Cable( cable5 = Cable(
profile=CableProfileChoices.A_TO_MANY, profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[frontport2_2], a_terminations=[frontport2_2],
b_terminations=[interfaces[6], interfaces[7]] b_terminations=[interfaces[6], interfaces[7]]
) )
@@ -722,13 +588,13 @@ class CablePathTests(CablePathTestCase):
# Create cables # Create cables
cable1 = Cable( cable1 = Cable(
profile=CableProfileChoices.B_TO_MANY, profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[interfaces[0], interfaces[1]], a_terminations=[interfaces[0], interfaces[1]],
b_terminations=[circuittermination1] b_terminations=[circuittermination1]
) )
cable1.save() cable1.save()
cable2 = Cable( cable2 = Cable(
profile=CableProfileChoices.A_TO_MANY, profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[circuittermination2], a_terminations=[circuittermination2],
b_terminations=[interfaces[2], interfaces[3]] b_terminations=[interfaces[2], interfaces[3]]
) )
@@ -801,7 +667,7 @@ class CablePathTests(CablePathTestCase):
# Create cables # Create cables
cable1 = Cable( cable1 = Cable(
profile=CableProfileChoices.A_TO_MANY, profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[interfaces[0]], a_terminations=[interfaces[0]],
b_terminations=[front_ports[0], front_ports[1]] b_terminations=[front_ports[0], front_ports[1]]
) )
@@ -812,7 +678,7 @@ class CablePathTests(CablePathTestCase):
) )
cable2.save() cable2.save()
cable3 = Cable( cable3 = Cable(
profile=CableProfileChoices.B_TO_MANY, profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[interfaces[1]], a_terminations=[interfaces[1]],
b_terminations=[front_ports[2], front_ports[3]] b_terminations=[front_ports[2], front_ports[3]]
) )
@@ -844,3 +710,65 @@ class CablePathTests(CablePathTestCase):
# Test SVG generation # Test SVG generation
CableTraceSVG(interfaces[0]).render() 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)