NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2026-01-10 05:42:16 -06:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

#20788: Map positions by connector (WIP)

Browse Source
This commit is contained in:
Jeremy Stretch
2025-12-11 16:33:30 -05:00
parent f56015e03d
commit aaa05fe071
13 changed files with 650 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
netbox/circuits/filtersets.py
View File

@@ -353,7 +353,7 @@ class CircuitTerminationFilterSet(NetBoxModelFilterSet, CabledObjectFilterSet):
model = CircuitTermination model = CircuitTermination
fields = ( fields = (
'id', 'termination_id', 'term_side', 'port_speed', 'upstream_speed', 'xconnect_id', 'description', 'id', 'termination_id', 'term_side', 'port_speed', 'upstream_speed', 'xconnect_id', 'description',
'mark_connected', 'pp_info', 'cable_end', 'cable_position', 'mark_connected', 'pp_info', 'cable_end',
) )
def search(self, queryset, name, value): def search(self, queryset, name, value):

22
netbox/circuits/migrations/0054_cable_position.py
View File

@@ -1,3 +1,4 @@
import django.contrib.postgres.fields
import django.core.validators import django.core.validators
from django.db import migrations, models from django.db import migrations, models
@@ -10,14 +11,29 @@ class Migration(migrations.Migration):
operations = [ operations = [
migrations.AddField( migrations.AddField(
model_name='circuittermination', model_name='circuittermination',
name='cable_position', name='cable_connector',
field=models.PositiveIntegerField( field=models.PositiveSmallIntegerField(
blank=True, blank=True,
null=True, null=True,
validators=[ validators=[
django.core.validators.MinValueValidator(1), django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024), django.core.validators.MaxValueValidator(256)
], ],
), ),
), ),
migrations.AddField(
model_name='circuittermination',
name='cable_positions',
field=django.contrib.postgres.fields.ArrayField(
base_field=models.PositiveSmallIntegerField(
validators=[
django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024),
]
),
blank=True,
null=True,
size=None,
),
),
] ]

303
netbox/dcim/cable_profiles.py
View File

@@ -1,108 +1,247 @@
from django.core.exceptions import ValidationError
from django.utils.translation import gettext_lazy as _
from dcim.models import CableTermination from dcim.models import CableTermination
class BaseCableProfile: class BaseCableProfile:
# Maximum number of terminations allowed per side
a_max_connections = None
b_max_connections = None
def clean(self, cable): def clean(self, cable):
# Enforce maximum connection limits pass
if self.a_max_connections and len(cable.a_terminations) > self.a_max_connections: # # Enforce maximum connection limits
raise ValidationError({ # if self.a_max_connections and len(cable.a_terminations) > self.a_max_connections:
'a_terminations': _( # raise ValidationError({
'Maximum A side connections for profile {profile}: {max}' # 'a_terminations': _(
).format( # 'Maximum A side connections for profile {profile}: {max}'
profile=cable.get_profile_display(), # ).format(
max=self.a_max_connections, # profile=cable.get_profile_display(),
) # max=self.a_max_connections,
}) # )
if self.b_max_connections and len(cable.b_terminations) > self.b_max_connections: # })
raise ValidationError({ # if self.b_max_connections and len(cable.b_terminations) > self.b_max_connections:
'b_terminations': _( # raise ValidationError({
'Maximum B side connections for profile {profile}: {max}' # 'b_terminations': _(
).format( # 'Maximum B side connections for profile {profile}: {max}'
profile=cable.get_profile_display(), # ).format(
max=self.b_max_connections, # profile=cable.get_profile_display(),
) # max=self.b_max_connections,
}) # )
# })
def get_mapped_position(self, side, position): def get_mapped_position(self, side, connector, position):
""" """
Return the mapped position for a given cable end and position. Return the mapped far-end connector & position for a given cable end the local connector & position.
By default, assume all positions are symmetrical.
""" """
return position # By default, assume all positions are symmetrical.
return connector, position
def get_peer_terminations(self, terminations, position_stack): def get_peer_termination(self, termination, position):
local_end = terminations[0].cable_end """
qs = CableTermination.objects.filter( Given a terminating object, return the peer terminating object (if any) on the opposite end of the cable.
cable=terminations[0].cable, """
cable_end=terminations[0].opposite_cable_end print(f'get_peer_termination({termination}, {position})')
print(f' Mapping {termination.cable_end} {termination.cable_connector}:{position}...')
connector, position = self.get_mapped_position(
termination.cable_end,
termination.cable_connector,
position
) )
print(f' Mapped to {connector}:{position}')
# TODO: Optimize this to use a single query under any condition try:
if position_stack: ct = CableTermination.objects.get(
# Attempt to find a peer termination at the same position currently in the stack. Pop the stack only if cable=termination.cable,
# we find one. Otherwise, return any peer terminations with a null position. cable_end=termination.opposite_cable_end,
position = self.get_mapped_position(local_end, position_stack[-1][0]) connector=connector,
if peers := qs.filter(position=position): positions__contains=[position],
position_stack.pop() )
return peers print(f' Found termination {ct.termination}')
return ct.termination, position
return qs.filter(position=None) except CableTermination.DoesNotExist:
print(f' Failed to resolve far end termination for {connector}:{position}')
return None, None
class StraightSingleCableProfile(BaseCableProfile): class Straight1C1PCableProfile(BaseCableProfile):
a_max_connections = 1 a_connectors = {
b_max_connections = 1 1: [1],
}
b_connectors = a_connectors
class StraightMultiCableProfile(BaseCableProfile): class Straight1C2PCableProfile(BaseCableProfile):
a_max_connections = None a_connectors = {
b_max_connections = None 1: [1, 2],
}
b_connectors = a_connectors
class Straight1C4PCableProfile(BaseCableProfile):
a_connectors = {
1: [1, 2, 3, 4],
}
b_connectors = a_connectors
class Straight1C8PCableProfile(BaseCableProfile):
a_connectors = {
1: [1, 2, 3, 4, 5, 6, 7, 8],
}
b_connectors = a_connectors
class Straight2C1PCableProfile(BaseCableProfile):
a_connectors = {
1: [1],
2: [1],
}
b_connectors = a_connectors
class Straight2C2PCableProfile(BaseCableProfile):
a_connectors = {
1: [1, 2],
2: [1, 2],
}
b_connectors = a_connectors
class Breakout1x4CableProfile(BaseCableProfile):
a_connectors = {
1: [1, 2, 3, 4],
}
b_connectors = {
1: [1],
2: [1],
3: [1],
4: [1],
}
_mapping = {
(1, 1): (1, 1),
(1, 2): (2, 1),
(1, 3): (3, 1),
(1, 4): (4, 1),
(2, 1): (1, 2),
(3, 1): (1, 3),
(4, 1): (1, 4),
}
def get_mapped_position(self, side, connector, position):
return self._mapping.get((connector, position))
class MPOTrunk4x4CableProfile(BaseCableProfile):
a_connectors = {
1: [1, 2, 3, 4],
2: [1, 2, 3, 4],
3: [1, 2, 3, 4],
4: [1, 2, 3, 4],
}
b_connectors = a_connectors
class MPOTrunk8x8CableProfile(BaseCableProfile):
a_connectors = {
1: [1, 2, 3, 4],
2: [1, 2, 3, 4],
3: [1, 2, 3, 4],
4: [1, 2, 3, 4],
5: [1, 2, 3, 4],
6: [1, 2, 3, 4],
7: [1, 2, 3, 4],
8: [1, 2, 3, 4],
}
b_connectors = a_connectors
class Shuffle2x2MPO8CableProfile(BaseCableProfile): class Shuffle2x2MPO8CableProfile(BaseCableProfile):
a_max_connections = 8 a_connectors = {
b_max_connections = 8 1: [1, 2, 3, 4],
2: [1, 2, 3, 4],
}
b_connectors = a_connectors
_mapping = { _mapping = {
1: 1, (1, 1): (1, 1),
2: 2, (1, 2): (1, 2),
3: 5, (1, 3): (2, 1),
4: 6, (1, 4): (2, 2),
5: 3, (2, 1): (1, 3),
6: 4, (2, 2): (1, 4),
7: 7, (2, 3): (2, 3),
8: 8, (2, 4): (2, 4),
} }
def get_mapped_position(self, side, position): def get_mapped_position(self, side, connector, position):
return self._mapping.get(position) return self._mapping.get((connector, position))
class Shuffle4x4MPO8CableProfile(BaseCableProfile): class Shuffle4x4MPO8CableProfile(BaseCableProfile):
a_max_connections = 8 a_connectors = {
b_max_connections = 8 1: [1, 2, 3, 4],
# A side to B side position mapping 2: [1, 2, 3, 4],
_a_mapping = { 3: [1, 2, 3, 4],
1: 1, 4: [1, 2, 3, 4],
2: 3, }
3: 5, b_connectors = a_connectors
4: 7, _mapping = {
5: 2, (1, 1): (1, 1),
6: 4, (1, 2): (2, 1),
7: 6, (1, 3): (3, 1),
8: 8, (1, 4): (4, 1),
(2, 1): (1, 2),
(2, 2): (2, 2),
(2, 3): (3, 2),
(2, 4): (4, 2),
(3, 1): (1, 3),
(3, 2): (2, 3),
(3, 3): (3, 3),
(3, 4): (4, 3),
(4, 1): (1, 4),
(4, 2): (2, 4),
(4, 3): (3, 4),
(4, 4): (4, 4),
} }
# 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): def get_mapped_position(self, side, connector, position):
if side.lower() == 'b': return self._mapping.get((connector, position))
return self._b_mapping.get(position)
return self._a_mapping.get(position)
class ShuffleBreakout2x8CableProfile(BaseCableProfile):
"""
Temporary solution for mapping 2 front/rear ports to 8 discrete interfaces
"""
a_connectors = {
1: [1, 2, 3, 4],
2: [1, 2, 3, 4],
}
b_connectors = {
1: [1],
2: [1],
3: [1],
4: [1],
5: [1],
6: [1],
7: [1],
8: [1],
}
_a_mapping = {
(1, 1): (1, 1),
(1, 2): (2, 1),
(1, 3): (5, 1),
(1, 4): (6, 1),
(2, 1): (3, 2),
(2, 2): (4, 2),
(2, 3): (7, 2),
(2, 4): (8, 2),
}
_b_mapping = {
(1, 1): (1, 1),
(2, 1): (1, 2),
(3, 1): (2, 1),
(4, 1): (2, 2),
(5, 1): (1, 3),
(6, 1): (1, 4),
(7, 1): (2, 3),
(8, 1): (2, 4),
}
def get_mapped_position(self, side, connector, position):
if side.lower() == 'a':
return self._a_mapping.get((connector, position))
return self._b_mapping.get((connector, position))

24
netbox/dcim/choices.py
View File

@@ -1722,16 +1722,32 @@ class PortTypeChoices(ChoiceSet):
# #
class CableProfileChoices(ChoiceSet): class CableProfileChoices(ChoiceSet):
STRAIGHT_SINGLE = 'straight-single' STRAIGHT_1C1P = 'straight-1c1p'
STRAIGHT_MULTI = 'straight-multi' STRAIGHT_1C2P = 'straight-1c2p'
STRAIGHT_1C4P = 'straight-1c4p'
STRAIGHT_1C8P = 'straight-1c8p'
STRAIGHT_2C1P = 'straight-2c1p'
STRAIGHT_2C2P = 'straight-2c2p'
BREAKOUT_1X4 = 'breakout-1x4'
MPO_TRUNK_4X4 = 'mpo-trunk-4x4'
MPO_TRUNK_8X8 = 'mpo-trunk-8x8'
SHUFFLE_2X2_MPO8 = 'shuffle-2x2-mpo8' SHUFFLE_2X2_MPO8 = 'shuffle-2x2-mpo8'
SHUFFLE_4X4_MPO8 = 'shuffle-4x4-mpo8' SHUFFLE_4X4_MPO8 = 'shuffle-4x4-mpo8'
SHUFFLE_BREAKOUT_2X8 = 'shuffle-breakout-2x8'
CHOICES = ( CHOICES = (
(STRAIGHT_SINGLE, _('Straight (single position)')), (STRAIGHT_1C1P, _('Straight (1C1P)')),
(STRAIGHT_MULTI, _('Straight (multi-position)')), (STRAIGHT_1C2P, _('Straight (1C2P)')),
(STRAIGHT_1C4P, _('Straight (1C4P)')),
(STRAIGHT_1C8P, _('Straight (1C8P)')),
(STRAIGHT_2C1P, _('Straight (2C1P)')),
(STRAIGHT_2C2P, _('Straight (2C2P)')),
(BREAKOUT_1X4, _('Breakout (1:4)')),
(MPO_TRUNK_4X4, _('MPO Trunk (4x4)')),
(MPO_TRUNK_8X8, _('MPO Trunk (8x8)')),
(SHUFFLE_2X2_MPO8, _('Shuffle (2x2 MPO8)')), (SHUFFLE_2X2_MPO8, _('Shuffle (2x2 MPO8)')),
(SHUFFLE_4X4_MPO8, _('Shuffle (4x4 MPO8)')), (SHUFFLE_4X4_MPO8, _('Shuffle (4x4 MPO8)')),
(SHUFFLE_BREAKOUT_2X8, _('Shuffle breakout (2x8)')),
) )

3
netbox/dcim/constants.py
View File

@@ -24,6 +24,9 @@ RACK_STARTING_UNIT_DEFAULT = 1
# Cables # Cables
# #
CABLE_CONNECTOR_MIN = 1
CABLE_CONNECTOR_MAX = 256
CABLE_POSITION_MIN = 1 CABLE_POSITION_MIN = 1
CABLE_POSITION_MAX = 1024 CABLE_POSITION_MAX = 1024

14
netbox/dcim/filtersets.py
View File

@@ -1748,7 +1748,7 @@ class ConsolePortFilterSet(ModularDeviceComponentFilterSet, CabledObjectFilterSe
class Meta: class Meta:
model = ConsolePort model = ConsolePort
fields = ('id', 'name', 'label', 'speed', 'description', 'mark_connected', 'cable_end', 'cable_position') fields = ('id', 'name', 'label', 'speed', 'description', 'mark_connected', 'cable_end')
@register_filterset @register_filterset
@@ -1760,7 +1760,7 @@ class ConsoleServerPortFilterSet(ModularDeviceComponentFilterSet, CabledObjectFi
class Meta: class Meta:
model = ConsoleServerPort model = ConsoleServerPort
fields = ('id', 'name', 'label', 'speed', 'description', 'mark_connected', 'cable_end', 'cable_position') fields = ('id', 'name', 'label', 'speed', 'description', 'mark_connected', 'cable_end')
@register_filterset @register_filterset
@@ -1774,7 +1774,6 @@ class PowerPortFilterSet(ModularDeviceComponentFilterSet, CabledObjectFilterSet,
model = PowerPort model = PowerPort
fields = ( fields = (
'id', 'name', 'label', 'maximum_draw', 'allocated_draw', 'description', 'mark_connected', 'cable_end', 'id', 'name', 'label', 'maximum_draw', 'allocated_draw', 'description', 'mark_connected', 'cable_end',
'cable_position',
) )
@@ -1801,7 +1800,6 @@ class PowerOutletFilterSet(ModularDeviceComponentFilterSet, CabledObjectFilterSe
model = PowerOutlet model = PowerOutlet
fields = ( fields = (
'id', 'name', 'status', 'label', 'feed_leg', 'description', 'color', 'mark_connected', 'cable_end', 'id', 'name', 'status', 'label', 'feed_leg', 'description', 'color', 'mark_connected', 'cable_end',
'cable_position',
) )
@@ -2111,7 +2109,7 @@ class InterfaceFilterSet(
fields = ( fields = (
'id', 'name', 'label', 'type', 'enabled', 'mtu', 'mgmt_only', 'poe_mode', 'poe_type', 'mode', 'rf_role', 'id', 'name', 'label', 'type', 'enabled', 'mtu', 'mgmt_only', 'poe_mode', 'poe_type', 'mode', 'rf_role',
'rf_channel', 'rf_channel_frequency', 'rf_channel_width', 'tx_power', 'description', 'mark_connected', 'rf_channel', 'rf_channel_frequency', 'rf_channel_width', 'tx_power', 'description', 'mark_connected',
'cable_id', 'cable_end', 'cable_position', 'cable_id', 'cable_end',
) )
def filter_virtual_chassis_member_or_master(self, queryset, name, value): def filter_virtual_chassis_member_or_master(self, queryset, name, value):
@@ -2167,7 +2165,6 @@ class FrontPortFilterSet(ModularDeviceComponentFilterSet, CabledObjectFilterSet)
model = FrontPort model = FrontPort
fields = ( fields = (
'id', 'name', 'label', 'type', 'color', 'positions', 'description', 'mark_connected', 'cable_end', 'id', 'name', 'label', 'type', 'color', 'positions', 'description', 'mark_connected', 'cable_end',
'cable_position',
) )
@@ -2188,7 +2185,6 @@ class RearPortFilterSet(ModularDeviceComponentFilterSet, CabledObjectFilterSet):
model = RearPort model = RearPort
fields = ( fields = (
'id', 'name', 'label', 'type', 'color', 'positions', 'description', 'mark_connected', 'cable_end', 'id', 'name', 'label', 'type', 'color', 'positions', 'description', 'mark_connected', 'cable_end',
'cable_position',
) )
@@ -2544,7 +2540,7 @@ class CableTerminationFilterSet(ChangeLoggedModelFilterSet):
class Meta: class Meta:
model = CableTermination model = CableTermination
fields = ('id', 'cable', 'cable_end', 'position', 'termination_type', 'termination_id') fields = ('id', 'cable', 'cable_end', 'termination_type', 'termination_id')
@register_filterset @register_filterset
@@ -2663,7 +2659,7 @@ class PowerFeedFilterSet(PrimaryModelFilterSet, CabledObjectFilterSet, PathEndpo
model = PowerFeed model = PowerFeed
fields = ( fields = (
'id', 'name', 'status', 'type', 'supply', 'phase', 'voltage', 'amperage', 'max_utilization', 'id', 'name', 'status', 'type', 'supply', 'phase', 'voltage', 'amperage', 'max_utilization',
'available_power', 'mark_connected', 'cable_end', 'cable_position', 'description', 'available_power', 'mark_connected', 'cable_end', 'description',
) )
def search(self, queryset, name, value): def search(self, queryset, name, value):

30
netbox/dcim/migrations/0220_cable_profile.py
View File

@@ -1,3 +1,4 @@
import django.contrib.postgres.fields
import django.core.validators import django.core.validators
from django.db import migrations, models from django.db import migrations, models
@@ -16,25 +17,40 @@ class Migration(migrations.Migration):
), ),
migrations.AddField( migrations.AddField(
model_name='cabletermination', model_name='cabletermination',
name='position', name='connector',
field=models.PositiveIntegerField( field=models.PositiveSmallIntegerField(
blank=True, blank=True,
null=True, null=True,
validators=[ validators=[
django.core.validators.MinValueValidator(1), django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024), django.core.validators.MaxValueValidator(256)
], ]
),
),
migrations.AddField(
model_name='cabletermination',
name='positions',
field=django.contrib.postgres.fields.ArrayField(
base_field=models.PositiveSmallIntegerField(
validators=[
django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024)
]
),
blank=True,
null=True,
size=None
), ),
), ),
migrations.AlterModelOptions( migrations.AlterModelOptions(
name='cabletermination', name='cabletermination',
options={'ordering': ('cable', 'cable_end', 'position', 'pk')}, options={'ordering': ('cable', 'cable_end', 'connector', 'positions', 'pk')},
), ),
migrations.AddConstraint( migrations.AddConstraint(
model_name='cabletermination', model_name='cabletermination',
constraint=models.UniqueConstraint( constraint=models.UniqueConstraint(
fields=('cable', 'cable_end', 'position'), fields=('cable', 'cable_end', 'connector'),
name='dcim_cabletermination_unique_position' name='dcim_cabletermination_unique_connector'
), ),
), ),
] ]

171
netbox/dcim/migrations/0221_cable_position.py
View File

@@ -1,3 +1,4 @@
import django.contrib.postgres.fields
import django.core.validators import django.core.validators
from django.db import migrations, models from django.db import migrations, models
@@ -10,98 +11,218 @@ class Migration(migrations.Migration):
operations = [ operations = [
migrations.AddField( migrations.AddField(
model_name='consoleport', model_name='consoleport',
name='cable_position', name='cable_connector',
field=models.PositiveIntegerField( field=models.PositiveSmallIntegerField(
blank=True, blank=True,
null=True, null=True,
validators=[
django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(256)
],
),
),
migrations.AddField(
model_name='consoleport',
name='cable_positions',
field=django.contrib.postgres.fields.ArrayField(
base_field=models.PositiveSmallIntegerField(
validators=[ validators=[
django.core.validators.MinValueValidator(1), django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024), django.core.validators.MaxValueValidator(1024),
]
),
blank=True,
null=True,
size=None,
),
),
migrations.AddField(
model_name='consoleserverport',
name='cable_connector',
field=models.PositiveSmallIntegerField(
blank=True,
null=True,
validators=[
django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(256)
], ],
), ),
), ),
migrations.AddField( migrations.AddField(
model_name='consoleserverport', model_name='consoleserverport',
name='cable_position', name='cable_positions',
field=models.PositiveIntegerField( field=django.contrib.postgres.fields.ArrayField(
base_field=models.PositiveSmallIntegerField(
validators=[
django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024),
]
),
blank=True,
null=True,
size=None,
),
),
migrations.AddField(
model_name='frontport',
name='cable_connector',
field=models.PositiveSmallIntegerField(
blank=True, blank=True,
null=True, null=True,
validators=[ validators=[
django.core.validators.MinValueValidator(1), django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024), django.core.validators.MaxValueValidator(256)
], ],
), ),
), ),
migrations.AddField( migrations.AddField(
model_name='frontport', model_name='frontport',
name='cable_position', name='cable_positions',
field=models.PositiveIntegerField( field=django.contrib.postgres.fields.ArrayField(
base_field=models.PositiveSmallIntegerField(
validators=[
django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024),
]
),
blank=True,
null=True,
size=None,
),
),
migrations.AddField(
model_name='interface',
name='cable_connector',
field=models.PositiveSmallIntegerField(
blank=True, blank=True,
null=True, null=True,
validators=[ validators=[
django.core.validators.MinValueValidator(1), django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024), django.core.validators.MaxValueValidator(256)
], ],
), ),
), ),
migrations.AddField( migrations.AddField(
model_name='interface', model_name='interface',
name='cable_position', name='cable_positions',
field=models.PositiveIntegerField( field=django.contrib.postgres.fields.ArrayField(
base_field=models.PositiveSmallIntegerField(
validators=[
django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024),
]
),
blank=True,
null=True,
size=None,
),
),
migrations.AddField(
model_name='powerfeed',
name='cable_connector',
field=models.PositiveSmallIntegerField(
blank=True, blank=True,
null=True, null=True,
validators=[ validators=[
django.core.validators.MinValueValidator(1), django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024), django.core.validators.MaxValueValidator(256)
], ],
), ),
), ),
migrations.AddField( migrations.AddField(
model_name='powerfeed', model_name='powerfeed',
name='cable_position', name='cable_positions',
field=models.PositiveIntegerField( field=django.contrib.postgres.fields.ArrayField(
base_field=models.PositiveSmallIntegerField(
validators=[
django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024),
]
),
blank=True,
null=True,
size=None,
),
),
migrations.AddField(
model_name='poweroutlet',
name='cable_connector',
field=models.PositiveSmallIntegerField(
blank=True, blank=True,
null=True, null=True,
validators=[ validators=[
django.core.validators.MinValueValidator(1), django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024), django.core.validators.MaxValueValidator(256)
], ],
), ),
), ),
migrations.AddField( migrations.AddField(
model_name='poweroutlet', model_name='poweroutlet',
name='cable_position', name='cable_positions',
field=models.PositiveIntegerField( field=django.contrib.postgres.fields.ArrayField(
base_field=models.PositiveSmallIntegerField(
validators=[
django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024),
]
),
blank=True,
null=True,
size=None,
),
),
migrations.AddField(
model_name='powerport',
name='cable_connector',
field=models.PositiveSmallIntegerField(
blank=True, blank=True,
null=True, null=True,
validators=[ validators=[
django.core.validators.MinValueValidator(1), django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024), django.core.validators.MaxValueValidator(256)
], ],
), ),
), ),
migrations.AddField( migrations.AddField(
model_name='powerport', model_name='powerport',
name='cable_position', name='cable_positions',
field=models.PositiveIntegerField( field=django.contrib.postgres.fields.ArrayField(
base_field=models.PositiveSmallIntegerField(
validators=[
django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024),
]
),
blank=True,
null=True,
size=None,
),
),
migrations.AddField(
model_name='rearport',
name='cable_connector',
field=models.PositiveSmallIntegerField(
blank=True, blank=True,
null=True, null=True,
validators=[ validators=[
django.core.validators.MinValueValidator(1), django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024), django.core.validators.MaxValueValidator(256)
], ],
), ),
), ),
migrations.AddField( migrations.AddField(
model_name='rearport', model_name='rearport',
name='cable_position', name='cable_positions',
field=models.PositiveIntegerField( field=django.contrib.postgres.fields.ArrayField(
blank=True, base_field=models.PositiveSmallIntegerField(
null=True,
validators=[ validators=[
django.core.validators.MinValueValidator(1), django.core.validators.MinValueValidator(1),
django.core.validators.MaxValueValidator(1024), django.core.validators.MaxValueValidator(1024),
], ]
),
blank=True,
null=True,
size=None,
), ),
), ),
] ]

73
netbox/dcim/models/cables.py
View File

@@ -3,6 +3,7 @@ import logging
from django.contrib.contenttypes.fields import GenericForeignKey from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType from django.contrib.contenttypes.models import ContentType
from django.contrib.postgres.fields import ArrayField
from django.core.exceptions import ValidationError from django.core.exceptions import ValidationError
from django.core.validators import MaxValueValidator, MinValueValidator from django.core.validators import MaxValueValidator, MinValueValidator
from django.db import models from django.db import models
@@ -136,10 +137,18 @@ class Cable(PrimaryModel):
def profile_class(self): def profile_class(self):
from dcim import cable_profiles from dcim import cable_profiles
return { return {
CableProfileChoices.STRAIGHT_SINGLE: cable_profiles.StraightSingleCableProfile, CableProfileChoices.STRAIGHT_1C1P: cable_profiles.Straight1C1PCableProfile,
CableProfileChoices.STRAIGHT_MULTI: cable_profiles.StraightMultiCableProfile, CableProfileChoices.STRAIGHT_1C2P: cable_profiles.Straight1C2PCableProfile,
CableProfileChoices.STRAIGHT_1C4P: cable_profiles.Straight1C4PCableProfile,
CableProfileChoices.STRAIGHT_1C8P: cable_profiles.Straight1C8PCableProfile,
CableProfileChoices.STRAIGHT_2C1P: cable_profiles.Straight2C1PCableProfile,
CableProfileChoices.STRAIGHT_2C2P: cable_profiles.Straight2C2PCableProfile,
CableProfileChoices.BREAKOUT_1X4: cable_profiles.Breakout1x4CableProfile,
CableProfileChoices.MPO_TRUNK_4X4: cable_profiles.MPOTrunk4x4CableProfile,
CableProfileChoices.MPO_TRUNK_8X8: cable_profiles.MPOTrunk8x8CableProfile,
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,
CableProfileChoices.SHUFFLE_BREAKOUT_2X8: cable_profiles.ShuffleBreakout2x8CableProfile,
}.get(self.profile) }.get(self.profile)
def _get_x_terminations(self, side): def _get_x_terminations(self, side):
@@ -340,12 +349,30 @@ 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 self.profile and isinstance(termination, PathEndpoint) else None connector = positions = None
CableTermination(cable=self, cable_end='A', position=position, termination=termination).save() if self.profile:
connector = i
positions = self.profile_class().a_connectors[i]
CableTermination(
cable=self,
cable_end='A',
connector=connector,
positions=positions,
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 self.profile and isinstance(termination, PathEndpoint) else None connector = positions = None
CableTermination(cable=self, cable_end='B', position=position, termination=termination).save() if self.profile:
connector = i
positions = self.profile_class().b_connectors[i]
CableTermination(
cable=self,
cable_end='B',
connector=connector,
positions=positions,
termination=termination
).save()
class CableTermination(ChangeLoggedModel): class CableTermination(ChangeLoggedModel):
@@ -372,13 +399,23 @@ class CableTermination(ChangeLoggedModel):
ct_field='termination_type', ct_field='termination_type',
fk_field='termination_id' fk_field='termination_id'
) )
position = models.PositiveIntegerField( connector = models.PositiveSmallIntegerField(
blank=True, blank=True,
null=True, null=True,
validators=(
MinValueValidator(CABLE_CONNECTOR_MIN),
MaxValueValidator(CABLE_CONNECTOR_MAX)
),
)
positions = ArrayField(
base_field=models.PositiveSmallIntegerField(
validators=( validators=(
MinValueValidator(CABLE_POSITION_MIN), MinValueValidator(CABLE_POSITION_MIN),
MaxValueValidator(CABLE_POSITION_MAX) MaxValueValidator(CABLE_POSITION_MAX)
) )
),
blank=True,
null=True,
) )
# Cached associations to enable efficient filtering # Cached associations to enable efficient filtering
@@ -410,15 +447,15 @@ class CableTermination(ChangeLoggedModel):
objects = RestrictedQuerySet.as_manager() objects = RestrictedQuerySet.as_manager()
class Meta: class Meta:
ordering = ('cable', 'cable_end', 'position', 'pk') ordering = ('cable', 'cable_end', 'connector', 'positions', 'pk')
constraints = ( constraints = (
models.UniqueConstraint( models.UniqueConstraint(
fields=('termination_type', 'termination_id'), fields=('termination_type', 'termination_id'),
name='%(app_label)s_%(class)s_unique_termination' name='%(app_label)s_%(class)s_unique_termination'
), ),
models.UniqueConstraint( models.UniqueConstraint(
fields=('cable', 'cable_end', 'position'), fields=('cable', 'cable_end', 'connector'),
name='%(app_label)s_%(class)s_unique_position' name='%(app_label)s_%(class)s_unique_connector'
), ),
) )
verbose_name = _('cable termination') verbose_name = _('cable termination')
@@ -483,7 +520,8 @@ class CableTermination(ChangeLoggedModel):
termination.snapshot() termination.snapshot()
termination.cable = self.cable termination.cable = self.cable
termination.cable_end = self.cable_end termination.cable_end = self.cable_end
termination.cable_position = self.position termination.cable_connector = self.connector
termination.cable_positions = self.positions
termination.save() termination.save()
def delete(self, *args, **kwargs): def delete(self, *args, **kwargs):
@@ -493,6 +531,7 @@ class CableTermination(ChangeLoggedModel):
termination.snapshot() termination.snapshot()
termination.cable = None termination.cable = None
termination.cable_end = None termination.cable_end = None
termination.cable_connector = None
termination.cable_position = None termination.cable_position = None
termination.save() termination.save()
@@ -701,9 +740,10 @@ class CablePath(models.Model):
path.append([ path.append([
object_to_path_node(t) for t in terminations object_to_path_node(t) for t in terminations
]) ])
# If not null, push cable_position onto the stack # If not null, push cable position onto the stack
if terminations[0].cable_position is not None: # TODO: Handle multiple positions?
position_stack.append([terminations[0].cable_position]) if isinstance(terminations[0], PathEndpoint) and terminations[0].cable_positions:
position_stack.append([terminations[0].cable_positions[0]])
# Step 2: Determine the attached links (Cable or WirelessLink), if any # Step 2: Determine the attached links (Cable or WirelessLink), if any
links = list(dict.fromkeys( links = list(dict.fromkeys(
@@ -744,8 +784,9 @@ class CablePath(models.Model):
# Profile-based tracing # Profile-based tracing
if links[0].profile: if links[0].profile:
cable_profile = links[0].profile_class() cable_profile = links[0].profile_class()
peer_cable_terminations = cable_profile.get_peer_terminations(terminations, position_stack) term, position = cable_profile.get_peer_termination(terminations[0], position_stack.pop()[0])
remote_terminations = [ct.termination for ct in peer_cable_terminations] remote_terminations = [term]
position_stack.append([position])
# Legacy (positionless) behavior # Legacy (positionless) behavior
else: else:

14
netbox/dcim/models/device_components.py
View File

@@ -1,6 +1,7 @@
from functools import cached_property from functools import cached_property
from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
from django.contrib.postgres.fields import ArrayField
from django.core.exceptions import ValidationError from django.core.exceptions import ValidationError
from django.core.validators import MaxValueValidator, MinValueValidator from django.core.validators import MaxValueValidator, MinValueValidator
from django.db import models from django.db import models
@@ -177,14 +178,23 @@ class CabledObjectModel(models.Model):
blank=True, blank=True,
null=True null=True
) )
cable_position = models.PositiveIntegerField( cable_connector = models.PositiveSmallIntegerField(
verbose_name=_('cable position'),
blank=True, blank=True,
null=True, null=True,
validators=(
MinValueValidator(CABLE_CONNECTOR_MIN),
MaxValueValidator(CABLE_CONNECTOR_MAX)
),
)
cable_positions = ArrayField(
base_field=models.PositiveSmallIntegerField(
validators=( validators=(
MinValueValidator(CABLE_POSITION_MIN), MinValueValidator(CABLE_POSITION_MIN),
MaxValueValidator(CABLE_POSITION_MAX) MaxValueValidator(CABLE_POSITION_MAX)
)
), ),
blank=True,
null=True,
) )
mark_connected = models.BooleanField( mark_connected = models.BooleanField(
verbose_name=_('mark connected'), verbose_name=_('mark connected'),

4
netbox/dcim/tests/test_api.py
View File

@@ -2586,7 +2586,7 @@ class CableTest(APIViewTestCases.APIViewTestCase):
'object_id': interfaces[14].pk, 'object_id': interfaces[14].pk,
}], }],
'label': 'Cable 4', 'label': 'Cable 4',
'profile': CableProfileChoices.STRAIGHT_SINGLE, 'profile': CableProfileChoices.STRAIGHT_1C1P,
}, },
{ {
'a_terminations': [{ 'a_terminations': [{
@@ -2598,7 +2598,7 @@ class CableTest(APIViewTestCases.APIViewTestCase):
'object_id': interfaces[15].pk, 'object_id': interfaces[15].pk,
}], }],
'label': 'Cable 5', 'label': 'Cable 5',
'profile': CableProfileChoices.STRAIGHT_SINGLE, 'profile': CableProfileChoices.STRAIGHT_1C1P,
}, },
{ {
'a_terminations': [{ 'a_terminations': [{

201
netbox/dcim/tests/test_cablepaths2.py
View File

@@ -1,3 +1,5 @@
from unittest import skip
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 *
@@ -14,11 +16,11 @@ class CablePathTests(CablePathTestCase):
2XX: Topology tests replicated from the legacy test case and adapted to use profiles 2XX: Topology tests replicated from the legacy test case and adapted to use profiles
""" """
def test_101_cable_profile_straight_single(self): def test_101_cable_profile_straight_1c1p(self):
""" """
[IF1] --C1-- [IF2] [IF1] --C1-- [IF2]
Cable profile: Straight single Cable profile: Straight 1C1P
""" """
interfaces = [ interfaces = [
Interface.objects.create(device=self.device, name='Interface 1'), Interface.objects.create(device=self.device, name='Interface 1'),
@@ -27,7 +29,7 @@ class CablePathTests(CablePathTestCase):
# Create cable 1 # Create cable 1
cable1 = Cable( cable1 = Cable(
profile=CableProfileChoices.STRAIGHT_SINGLE, profile=CableProfileChoices.STRAIGHT_1C1P,
a_terminations=[interfaces[0]], a_terminations=[interfaces[0]],
b_terminations=[interfaces[1]], b_terminations=[interfaces[1]],
) )
@@ -49,8 +51,10 @@ class CablePathTests(CablePathTestCase):
interfaces[1].refresh_from_db() interfaces[1].refresh_from_db()
self.assertPathIsSet(interfaces[0], path1) self.assertPathIsSet(interfaces[0], path1)
self.assertPathIsSet(interfaces[1], path2) self.assertPathIsSet(interfaces[1], path2)
self.assertEqual(interfaces[0].cable_position, 1) self.assertEqual(interfaces[0].cable_connector, 1)
self.assertEqual(interfaces[1].cable_position, 1) self.assertEqual(interfaces[0].cable_positions, [1])
self.assertEqual(interfaces[1].cable_connector, 1)
self.assertEqual(interfaces[1].cable_positions, [1])
# Test SVG generation # Test SVG generation
CableTraceSVG(interfaces[0]).render() CableTraceSVG(interfaces[0]).render()
@@ -61,12 +65,12 @@ 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_102_cable_profile_straight_multi(self): def test_102_cable_profile_straight_2c1p(self):
""" """
[IF1] --C1-- [IF3] [IF1] --C1-- [IF3]
[IF2] [IF4] [IF2] [IF4]
Cable profile: Straight multi Cable profile: Straight 2C1P
""" """
interfaces = [ interfaces = [
Interface.objects.create(device=self.device, name='Interface 1'), Interface.objects.create(device=self.device, name='Interface 1'),
@@ -77,7 +81,7 @@ class CablePathTests(CablePathTestCase):
# Create cable 1 # Create cable 1
cable1 = Cable( cable1 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI, profile=CableProfileChoices.STRAIGHT_2C1P,
a_terminations=[interfaces[0], interfaces[1]], a_terminations=[interfaces[0], interfaces[1]],
b_terminations=[interfaces[2], interfaces[3]], b_terminations=[interfaces[2], interfaces[3]],
) )
@@ -112,10 +116,14 @@ class CablePathTests(CablePathTestCase):
self.assertPathIsSet(interfaces[1], path2) self.assertPathIsSet(interfaces[1], path2)
self.assertPathIsSet(interfaces[2], path3) self.assertPathIsSet(interfaces[2], path3)
self.assertPathIsSet(interfaces[3], path4) self.assertPathIsSet(interfaces[3], path4)
self.assertEqual(interfaces[0].cable_position, 1) self.assertEqual(interfaces[0].cable_connector, 1)
self.assertEqual(interfaces[1].cable_position, 2) self.assertEqual(interfaces[0].cable_positions, [1])
self.assertEqual(interfaces[2].cable_position, 1) self.assertEqual(interfaces[1].cable_connector, 2)
self.assertEqual(interfaces[3].cable_position, 2) self.assertEqual(interfaces[1].cable_positions, [1])
self.assertEqual(interfaces[2].cable_connector, 1)
self.assertEqual(interfaces[2].cable_positions, [1])
self.assertEqual(interfaces[3].cable_connector, 2)
self.assertEqual(interfaces[3].cable_positions, [1])
# Test SVG generation # Test SVG generation
CableTraceSVG(interfaces[0]).render() CableTraceSVG(interfaces[0]).render()
@@ -126,7 +134,8 @@ 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_2x2_mpo8(self): @skip("Under development")
def test_103_cable_profile_shuffle_2x2_mpo8(self):
""" """
[IF1:1] --C1-- [IF3:1] [IF1:1] --C1-- [IF3:1]
[IF1:2] [IF3:2] [IF1:2] [IF3:2]
@@ -227,7 +236,7 @@ class CablePathTests(CablePathTestCase):
interface.refresh_from_db() interface.refresh_from_db()
self.assertPathIsSet(interface, path) self.assertPathIsSet(interface, path)
self.assertEqual(interface.cable_end, 'A' if i < 8 else 'B') self.assertEqual(interface.cable_end, 'A' if i < 8 else 'B')
self.assertEqual(interface.cable_position, (i % 8) + 1) self.assertEqual(interface.cable_positions, [(i % 8) + 1])
# Test SVG generation # Test SVG generation
CableTraceSVG(interfaces[0]).render() CableTraceSVG(interfaces[0]).render()
@@ -238,6 +247,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)
@skip("Under development")
def test_104_cable_profile_4x4_mpo8(self): def test_104_cable_profile_4x4_mpo8(self):
""" """
[IF1:1] --C1-- [IF3:1] [IF1:1] --C1-- [IF3:1]
@@ -339,7 +349,7 @@ class CablePathTests(CablePathTestCase):
interface.refresh_from_db() interface.refresh_from_db()
self.assertPathIsSet(interface, path) self.assertPathIsSet(interface, path)
self.assertEqual(interface.cable_end, 'A' if i < 8 else 'B') self.assertEqual(interface.cable_end, 'A' if i < 8 else 'B')
self.assertEqual(interface.cable_position, (i % 8) + 1) self.assertEqual(interface.cable_positions, [(i % 8) + 1])
# Test SVG generation # Test SVG generation
CableTraceSVG(interfaces[0]).render() CableTraceSVG(interfaces[0]).render()
@@ -361,8 +371,8 @@ class CablePathTests(CablePathTestCase):
Interface.objects.create(device=self.device, name='Interface 3'), Interface.objects.create(device=self.device, name='Interface 3'),
Interface.objects.create(device=self.device, name='Interface 4'), Interface.objects.create(device=self.device, name='Interface 4'),
] ]
rearport1 = RearPort.objects.create(device=self.device, name='Rear Port 1') rearport1 = RearPort.objects.create(device=self.device, name='Rear Port 1', positions=4)
frontport1 = FrontPort.objects.create(device=self.device, name='Front Port 1') frontport1 = FrontPort.objects.create(device=self.device, name='Front Port 1', positions=4)
PortMapping.objects.bulk_create([ PortMapping.objects.bulk_create([
PortMapping( PortMapping(
device=self.device, device=self.device,
@@ -371,18 +381,39 @@ class CablePathTests(CablePathTestCase):
rear_port=rearport1, rear_port=rearport1,
rear_port_position=1, rear_port_position=1,
), ),
PortMapping(
device=self.device,
front_port=frontport1,
front_port_position=2,
rear_port=rearport1,
rear_port_position=2,
),
PortMapping(
device=self.device,
front_port=frontport1,
front_port_position=3,
rear_port=rearport1,
rear_port_position=3,
),
PortMapping(
device=self.device,
front_port=frontport1,
front_port_position=4,
rear_port=rearport1,
rear_port_position=4,
),
]) ])
# Create cables # Create cables
cable1 = Cable( cable1 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI, profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[interfaces[0], interfaces[1]], a_terminations=[frontport1],
b_terminations=[frontport1], b_terminations=[interfaces[0], interfaces[1]],
) )
cable1.clean() cable1.clean()
cable1.save() cable1.save()
cable2 = Cable( cable2 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI, profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[rearport1], a_terminations=[rearport1],
b_terminations=[interfaces[2], interfaces[3]] b_terminations=[interfaces[2], interfaces[3]]
) )
@@ -424,10 +455,10 @@ class CablePathTests(CablePathTestCase):
def test_204_multiple_paths_via_pass_through_with_breakouts(self): def test_204_multiple_paths_via_pass_through_with_breakouts(self):
""" """
[IF1] --C1-- [FP1:1] [RP1] --C3-- [RP2] [FP2:1] --C4-- [IF4] [IF1] --C1-- [FP1] [RP1] --C3-- [RP2] [FP3] --C4-- [IF5]
[IF2] [IF5] [IF2] [IF6]
[IF3] --C2-- [FP1:2] [FP2:2] --C5-- [IF6] [IF3] --C2-- [FP2] [FP4] --C5-- [IF7]
[IF4] [IF7] [IF4] [IF8]
""" """
interfaces = [ interfaces = [
Interface.objects.create(device=self.device, name='Interface 1'), Interface.objects.create(device=self.device, name='Interface 1'),
@@ -439,76 +470,104 @@ class CablePathTests(CablePathTestCase):
Interface.objects.create(device=self.device, name='Interface 7'), Interface.objects.create(device=self.device, name='Interface 7'),
Interface.objects.create(device=self.device, name='Interface 8'), Interface.objects.create(device=self.device, name='Interface 8'),
] ]
rearport1 = RearPort.objects.create(device=self.device, name='Rear Port 1', positions=4) rearport1 = RearPort.objects.create(device=self.device, name='Rear Port 1', positions=8)
rearport2 = RearPort.objects.create(device=self.device, name='Rear Port 2', positions=4) rearport2 = RearPort.objects.create(device=self.device, name='Rear Port 2', positions=8)
frontport1_1 = FrontPort.objects.create(device=self.device, name='Front Port 1:1') frontport1 = FrontPort.objects.create(device=self.device, name='Front Port 1:1', positions=4)
frontport1_2 = FrontPort.objects.create(device=self.device, name='Front Port 1:2') frontport2 = FrontPort.objects.create(device=self.device, name='Front Port 1:2', positions=4)
frontport2_1 = FrontPort.objects.create(device=self.device, name='Front Port 2:1') frontport3 = FrontPort.objects.create(device=self.device, name='Front Port 2:1', positions=4)
frontport2_2 = FrontPort.objects.create(device=self.device, name='Front Port 2:2') frontport4 = FrontPort.objects.create(device=self.device, name='Front Port 2:2', positions=4)
PortMapping.objects.bulk_create([ PortMapping.objects.bulk_create([
PortMapping( PortMapping(
device=self.device, device=self.device,
front_port=frontport1_1, front_port=frontport1,
front_port_position=1, front_port_position=1,
rear_port=rearport1, rear_port=rearport1,
rear_port_position=1, rear_port_position=1,
), ),
PortMapping( PortMapping(
device=self.device, device=self.device,
front_port=frontport1_2, front_port=frontport1,
front_port_position=1, front_port_position=2,
rear_port=rearport1, rear_port=rearport1,
rear_port_position=2, rear_port_position=2,
), ),
PortMapping( PortMapping(
device=self.device, device=self.device,
front_port=frontport2_1, front_port=frontport2,
front_port_position=1,
rear_port=rearport1,
rear_port_position=5,
),
PortMapping(
device=self.device,
front_port=frontport2,
front_port_position=2,
rear_port=rearport1,
rear_port_position=6,
),
PortMapping(
device=self.device,
front_port=frontport3,
front_port_position=1, front_port_position=1,
rear_port=rearport2, rear_port=rearport2,
rear_port_position=1, rear_port_position=1,
), ),
PortMapping( PortMapping(
device=self.device, device=self.device,
front_port=frontport2_2, front_port=frontport3,
front_port_position=1, front_port_position=2,
rear_port=rearport2, rear_port=rearport2,
rear_port_position=2, rear_port_position=2,
), ),
PortMapping(
device=self.device,
front_port=frontport4,
front_port_position=1,
rear_port=rearport2,
rear_port_position=5,
),
PortMapping(
device=self.device,
front_port=frontport4,
front_port_position=2,
rear_port=rearport2,
rear_port_position=6,
),
]) ])
# Create cables # Create cables
cable1 = Cable( cable1 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI, profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[interfaces[0], interfaces[1]], a_terminations=[frontport1],
b_terminations=[frontport1_1] b_terminations=[interfaces[0], interfaces[1]],
) )
cable1.clean() cable1.clean()
cable1.save() cable1.save()
cable2 = Cable( cable2 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI, profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[interfaces[2], interfaces[3]], a_terminations=[frontport2],
b_terminations=[frontport1_2] b_terminations=[interfaces[2], interfaces[3]],
) )
cable2.clean() cable2.clean()
cable2.save() cable2.save()
cable3 = Cable( cable3 = Cable(
profile=CableProfileChoices.STRAIGHT_SINGLE, profile=CableProfileChoices.STRAIGHT_1C8P,
a_terminations=[rearport1], a_terminations=[rearport1],
b_terminations=[rearport2] b_terminations=[rearport2]
) )
cable3.clean() cable3.clean()
cable3.save() cable3.save()
cable4 = Cable( cable4 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI, profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[frontport2_1], a_terminations=[frontport3],
b_terminations=[interfaces[4], interfaces[5]] b_terminations=[interfaces[4], interfaces[5]],
) )
cable4.clean() cable4.clean()
cable4.save() cable4.save()
cable5 = Cable( cable5 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI, profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[frontport2_2], a_terminations=[frontport4],
b_terminations=[interfaces[6], interfaces[7]] b_terminations=[interfaces[6], interfaces[7]],
) )
cable5.clean() cable5.clean()
cable5.save() cable5.save()
@@ -516,7 +575,7 @@ class CablePathTests(CablePathTestCase):
paths = [ paths = [
self.assertPathExists( self.assertPathExists(
( (
interfaces[0], cable1, frontport1_1, rearport1, cable3, rearport2, frontport2_1, cable4, interfaces[0], cable1, frontport1, rearport1, cable3, rearport2, frontport3, cable4,
interfaces[4], interfaces[4],
), ),
is_complete=True, is_complete=True,
@@ -524,7 +583,7 @@ class CablePathTests(CablePathTestCase):
), ),
self.assertPathExists( self.assertPathExists(
( (
interfaces[1], cable1, frontport1_1, rearport1, cable3, rearport2, frontport2_1, cable4, interfaces[1], cable1, frontport1, rearport1, cable3, rearport2, frontport3, cable4,
interfaces[5], interfaces[5],
), ),
is_complete=True, is_complete=True,
@@ -532,7 +591,7 @@ class CablePathTests(CablePathTestCase):
), ),
self.assertPathExists( self.assertPathExists(
( (
interfaces[2], cable2, frontport1_2, rearport1, cable3, rearport2, frontport2_2, cable5, interfaces[2], cable2, frontport2, rearport1, cable3, rearport2, frontport4, cable5,
interfaces[6], interfaces[6],
), ),
is_complete=True, is_complete=True,
@@ -540,7 +599,7 @@ class CablePathTests(CablePathTestCase):
), ),
self.assertPathExists( self.assertPathExists(
( (
interfaces[3], cable2, frontport1_2, rearport1, cable3, rearport2, frontport2_2, cable5, interfaces[3], cable2, frontport2, rearport1, cable3, rearport2, frontport4, cable5,
interfaces[7], interfaces[7],
), ),
is_complete=True, is_complete=True,
@@ -548,7 +607,7 @@ class CablePathTests(CablePathTestCase):
), ),
self.assertPathExists( self.assertPathExists(
( (
interfaces[4], cable4, frontport2_1, rearport2, cable3, rearport1, frontport1_1, cable1, interfaces[4], cable4, frontport3, rearport2, cable3, rearport1, frontport1, cable1,
interfaces[0], interfaces[0],
), ),
is_complete=True, is_complete=True,
@@ -556,7 +615,7 @@ class CablePathTests(CablePathTestCase):
), ),
self.assertPathExists( self.assertPathExists(
( (
interfaces[5], cable4, frontport2_1, rearport2, cable3, rearport1, frontport1_1, cable1, interfaces[5], cable4, frontport3, rearport2, cable3, rearport1, frontport1, cable1,
interfaces[1], interfaces[1],
), ),
is_complete=True, is_complete=True,
@@ -564,7 +623,7 @@ class CablePathTests(CablePathTestCase):
), ),
self.assertPathExists( self.assertPathExists(
( (
interfaces[6], cable5, frontport2_2, rearport2, cable3, rearport1, frontport1_2, cable2, interfaces[6], cable5, frontport4, rearport2, cable3, rearport1, frontport2, cable2,
interfaces[2], interfaces[2],
), ),
is_complete=True, is_complete=True,
@@ -572,7 +631,7 @@ class CablePathTests(CablePathTestCase):
), ),
self.assertPathExists( self.assertPathExists(
( (
interfaces[7], cable5, frontport2_2, rearport2, cable3, rearport1, frontport1_2, cable2, interfaces[7], cable5, frontport4, rearport2, cable3, rearport1, frontport2, cable2,
interfaces[3], interfaces[3],
), ),
is_complete=True, is_complete=True,
@@ -619,14 +678,14 @@ class CablePathTests(CablePathTestCase):
# Create cables # Create cables
cable1 = Cable( cable1 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI, profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[interfaces[0], interfaces[1]], a_terminations=[circuittermination1],
b_terminations=[circuittermination1] b_terminations=[interfaces[0], interfaces[1]],
) )
cable1.clean() cable1.clean()
cable1.save() cable1.save()
cable2 = Cable( cable2 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI, profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[circuittermination2], a_terminations=[circuittermination2],
b_terminations=[interfaces[2], interfaces[3]] b_terminations=[interfaces[2], interfaces[3]]
) )
@@ -668,6 +727,8 @@ class CablePathTests(CablePathTestCase):
# Test SVG generation # Test SVG generation
CableTraceSVG(interfaces[0]).render() CableTraceSVG(interfaces[0]).render()
# TBD: Is this a topology we want to support?
@skip("Test applicability TBD")
def test_217_interface_to_interface_via_rear_ports(self): def test_217_interface_to_interface_via_rear_ports(self):
""" """
[IF1] --C1-- [FP1] [RP1] --C2-- [RP3] [FP3] --C3-- [IF2] [IF1] --C1-- [FP1] [RP1] --C2-- [RP3] [FP3] --C3-- [IF2]
@@ -722,7 +783,7 @@ class CablePathTests(CablePathTestCase):
# Create cables # Create cables
cable1 = Cable( cable1 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI, profile=CableProfileChoices.STRAIGHT_2C1P,
a_terminations=[interfaces[0]], a_terminations=[interfaces[0]],
b_terminations=[front_ports[0], front_ports[1]] b_terminations=[front_ports[0], front_ports[1]]
) )
@@ -735,7 +796,7 @@ class CablePathTests(CablePathTestCase):
cable2.clean() cable2.clean()
cable2.save() cable2.save()
cable3 = Cable( cable3 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI, profile=CableProfileChoices.STRAIGHT_2C1P,
a_terminations=[interfaces[1]], a_terminations=[interfaces[1]],
b_terminations=[front_ports[2], front_ports[3]] b_terminations=[front_ports[2], front_ports[3]]
) )
@@ -803,14 +864,14 @@ class CablePathTests(CablePathTestCase):
# Create cables # Create cables
cable1 = Cable( cable1 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI, profile=CableProfileChoices.STRAIGHT_2C2P,
a_terminations=[interfaces[0], interfaces[1]], a_terminations=[interfaces[0], interfaces[1]],
b_terminations=[frontport1, frontport2] b_terminations=[frontport1, frontport2]
) )
cable1.clean() cable1.clean()
cable1.save() cable1.save()
cable2 = Cable( cable2 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI, profile=CableProfileChoices.STRAIGHT_2C2P,
a_terminations=[rearport1, rearport2], a_terminations=[rearport1, rearport2],
b_terminations=[interfaces[2], interfaces[3]] b_terminations=[interfaces[2], interfaces[3]]
) )
@@ -819,22 +880,22 @@ class CablePathTests(CablePathTestCase):
# Validate paths # Validate paths
self.assertPathExists( self.assertPathExists(
(interfaces[0], cable1, [frontport1, frontport2], [rearport1, rearport2], cable2, interfaces[2]), (interfaces[0], cable1, frontport1, rearport1, cable2, interfaces[2]),
is_complete=True, is_complete=True,
is_active=True is_active=True
) )
self.assertPathExists( self.assertPathExists(
(interfaces[1], cable1, [frontport1, frontport2], [rearport1, rearport2], cable2, interfaces[3]), (interfaces[1], cable1, frontport2, rearport2, cable2, interfaces[3]),
is_complete=True, is_complete=True,
is_active=True is_active=True
) )
self.assertPathExists( self.assertPathExists(
(interfaces[2], cable2, [rearport1, rearport2], [frontport1, frontport2], cable1, interfaces[0]), (interfaces[2], cable2, rearport1, frontport1, cable1, interfaces[0]),
is_complete=True, is_complete=True,
is_active=True is_active=True
) )
self.assertPathExists( self.assertPathExists(
(interfaces[3], cable2, [rearport1, rearport2], [frontport1, frontport2], cable1, interfaces[1]), (interfaces[3], cable2, rearport2, frontport2, cable1, interfaces[1]),
is_complete=True, is_complete=True,
is_active=True is_active=True
) )

6
netbox/dcim/utils.py
View File

@@ -41,12 +41,12 @@ def create_cablepaths(objects):
""" """
from dcim.models import CablePath from dcim.models import CablePath
# Arrange objects by cable position. All objects with a null position are grouped together. # Arrange objects by cable connector. All objects with a null connector are grouped together.
origins = defaultdict(list) origins = defaultdict(list)
for obj in objects: for obj in objects:
origins[obj.cable_position].append(obj) origins[obj.cable_connector].append(obj)
for position, objects in origins.items(): for connector, objects in origins.items():
if cp := CablePath.from_origin(objects): if cp := CablePath.from_origin(objects):
cp.save() cp.save()