NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-12-26 23:27:46 -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 3140060f21
commit 15c32d44e7
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
fields = (
'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):

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

@@ -1,3 +1,4 @@
import django.contrib.postgres.fields
import django.core.validators
from django.db import migrations, models
@@ -10,14 +11,29 @@ class Migration(migrations.Migration):
operations = [
migrations.AddField(
model_name='circuittermination',
name='cable_position',
field=models.PositiveIntegerField(
name='cable_connector',
field=models.PositiveSmallIntegerField(
blank=True,
null=True,
validators=[
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
class BaseCableProfile:
# Maximum number of terminations allowed per side
a_max_connections = None
b_max_connections = None
def clean(self, cable):
# Enforce maximum connection limits
if self.a_max_connections and len(cable.a_terminations) > self.a_max_connections:
raise ValidationError({
'a_terminations': _(
'Maximum A side connections for profile {profile}: {max}'
).format(
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({
'b_terminations': _(
'Maximum B side connections for profile {profile}: {max}'
).format(
profile=cable.get_profile_display(),
max=self.b_max_connections,
)
})
pass
# # Enforce maximum connection limits
# if self.a_max_connections and len(cable.a_terminations) > self.a_max_connections:
# raise ValidationError({
# 'a_terminations': _(
# 'Maximum A side connections for profile {profile}: {max}'
# ).format(
# 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({
# 'b_terminations': _(
# 'Maximum B side connections for profile {profile}: {max}'
# ).format(
# 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.
By default, assume all positions are symmetrical.
Return the mapped far-end connector & position for a given cable end the local connector & position.
"""
return position
# By default, assume all positions are symmetrical.
return connector, position
def get_peer_terminations(self, terminations, position_stack):
local_end = terminations[0].cable_end
qs = CableTermination.objects.filter(
cable=terminations[0].cable,
cable_end=terminations[0].opposite_cable_end
def get_peer_termination(self, termination, position):
"""
Given a terminating object, return the peer terminating object (if any) on the opposite end of the cable.
"""
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
)
# 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)
print(f' Mapped to {connector}:{position}')
try:
ct = CableTermination.objects.get(
cable=termination.cable,
cable_end=termination.opposite_cable_end,
connector=connector,
positions__contains=[position],
)
print(f' Found termination {ct.termination}')
return ct.termination, position
except CableTermination.DoesNotExist:
print(f' Failed to resolve far end termination for {connector}:{position}')
return None, None
class StraightSingleCableProfile(BaseCableProfile):
a_max_connections = 1
b_max_connections = 1
class Straight1C1PCableProfile(BaseCableProfile):
a_connectors = {
1: [1],
}
b_connectors = a_connectors
class StraightMultiCableProfile(BaseCableProfile):
a_max_connections = None
b_max_connections = None
class Straight1C2PCableProfile(BaseCableProfile):
a_connectors = {
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):
a_max_connections = 8
b_max_connections = 8
a_connectors = {
1: [1, 2, 3, 4],
2: [1, 2, 3, 4],
}
b_connectors = a_connectors
_mapping = {
1: 1,
2: 2,
3: 5,
4: 6,
5: 3,
6: 4,
7: 7,
8: 8,
(1, 1): (1, 1),
(1, 2): (1, 2),
(1, 3): (2, 1),
(1, 4): (2, 2),
(2, 1): (1, 3),
(2, 2): (1, 4),
(2, 3): (2, 3),
(2, 4): (2, 4),
}
def get_mapped_position(self, side, position):
return self._mapping.get(position)
def get_mapped_position(self, side, connector, position):
return self._mapping.get((connector, position))
class Shuffle4x4MPO8CableProfile(BaseCableProfile):
a_max_connections = 8
b_max_connections = 8
# A side to B side position mapping
_a_mapping = {
1: 1,
2: 3,
3: 5,
4: 7,
5: 2,
6: 4,
7: 6,
8: 8,
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
_mapping = {
(1, 1): (1, 1),
(1, 2): (2, 1),
(1, 3): (3, 1),
(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):
if side.lower() == 'b':
return self._b_mapping.get(position)
return self._a_mapping.get(position)
def get_mapped_position(self, side, connector, position):
return self._mapping.get((connector, 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):
STRAIGHT_SINGLE = 'straight-single'
STRAIGHT_MULTI = 'straight-multi'
STRAIGHT_1C1P = 'straight-1c1p'
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_4X4_MPO8 = 'shuffle-4x4-mpo8'
SHUFFLE_BREAKOUT_2X8 = 'shuffle-breakout-2x8'
CHOICES = (
(STRAIGHT_SINGLE, _('Straight (single position)')),
(STRAIGHT_MULTI, _('Straight (multi-position)')),
(STRAIGHT_1C1P, _('Straight (1C1P)')),
(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_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
#
CABLE_CONNECTOR_MIN = 1
CABLE_CONNECTOR_MAX = 256
CABLE_POSITION_MIN = 1
CABLE_POSITION_MAX = 1024

14
netbox/dcim/filtersets.py
View File

@@ -1748,7 +1748,7 @@ class ConsolePortFilterSet(ModularDeviceComponentFilterSet, CabledObjectFilterSe
class Meta:
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
@@ -1760,7 +1760,7 @@ class ConsoleServerPortFilterSet(ModularDeviceComponentFilterSet, CabledObjectFi
class Meta:
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
@@ -1774,7 +1774,6 @@ class PowerPortFilterSet(ModularDeviceComponentFilterSet, CabledObjectFilterSet,
model = PowerPort
fields = (
'id', 'name', 'label', 'maximum_draw', 'allocated_draw', 'description', 'mark_connected', 'cable_end',
'cable_position',
)
@@ -1801,7 +1800,6 @@ class PowerOutletFilterSet(ModularDeviceComponentFilterSet, CabledObjectFilterSe
model = PowerOutlet
fields = (
'id', 'name', 'status', 'label', 'feed_leg', 'description', 'color', 'mark_connected', 'cable_end',
'cable_position',
)
@@ -2111,7 +2109,7 @@ class InterfaceFilterSet(
fields = (
'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',
'cable_id', 'cable_end', 'cable_position',
'cable_id', 'cable_end',
)
def filter_virtual_chassis_member_or_master(self, queryset, name, value):
@@ -2167,7 +2165,6 @@ class FrontPortFilterSet(ModularDeviceComponentFilterSet, CabledObjectFilterSet)
model = FrontPort
fields = (
'id', 'name', 'label', 'type', 'color', 'positions', 'description', 'mark_connected', 'cable_end',
'cable_position',
)
@@ -2188,7 +2185,6 @@ class RearPortFilterSet(ModularDeviceComponentFilterSet, CabledObjectFilterSet):
model = RearPort
fields = (
'id', 'name', 'label', 'type', 'color', 'positions', 'description', 'mark_connected', 'cable_end',
'cable_position',
)
@@ -2544,7 +2540,7 @@ class CableTerminationFilterSet(ChangeLoggedModelFilterSet):
class Meta:
model = CableTermination
fields = ('id', 'cable', 'cable_end', 'position', 'termination_type', 'termination_id')
fields = ('id', 'cable', 'cable_end', 'termination_type', 'termination_id')
@register_filterset
@@ -2663,7 +2659,7 @@ class PowerFeedFilterSet(PrimaryModelFilterSet, CabledObjectFilterSet, PathEndpo
model = PowerFeed
fields = (
'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):

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

@@ -1,3 +1,4 @@
import django.contrib.postgres.fields
import django.core.validators
from django.db import migrations, models
@@ -16,25 +17,40 @@ class Migration(migrations.Migration):
),
migrations.AddField(
model_name='cabletermination',
name='position',
field=models.PositiveIntegerField(
name='connector',
field=models.PositiveSmallIntegerField(
blank=True,
null=True,
validators=[
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(
name='cabletermination',
options={'ordering': ('cable', 'cable_end', 'position', 'pk')},
options={'ordering': ('cable', 'cable_end', 'connector', 'positions', 'pk')},
),
migrations.AddConstraint(
model_name='cabletermination',
constraint=models.UniqueConstraint(
fields=('cable', 'cable_end', 'position'),
name='dcim_cabletermination_unique_position'
fields=('cable', 'cable_end', 'connector'),
name='dcim_cabletermination_unique_connector'
),
),
]

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

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

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

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

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

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

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

@@ -2586,7 +2586,7 @@ class CableTest(APIViewTestCases.APIViewTestCase):
'object_id': interfaces[14].pk,
}],
'label': 'Cable 4',
'profile': CableProfileChoices.STRAIGHT_SINGLE,
'profile': CableProfileChoices.STRAIGHT_1C1P,
},
{
'a_terminations': [{
@@ -2598,7 +2598,7 @@ class CableTest(APIViewTestCases.APIViewTestCase):
'object_id': interfaces[15].pk,
}],
'label': 'Cable 5',
'profile': CableProfileChoices.STRAIGHT_SINGLE,
'profile': CableProfileChoices.STRAIGHT_1C1P,
},
{
'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 dcim.choices import CableProfileChoices
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
"""
def test_101_cable_profile_straight_single(self):
def test_101_cable_profile_straight_1c1p(self):
"""
[IF1] --C1-- [IF2]
Cable profile: Straight single
Cable profile: Straight 1C1P
"""
interfaces = [
Interface.objects.create(device=self.device, name='Interface 1'),
@@ -27,7 +29,7 @@ class CablePathTests(CablePathTestCase):
# Create cable 1
cable1 = Cable(
profile=CableProfileChoices.STRAIGHT_SINGLE,
profile=CableProfileChoices.STRAIGHT_1C1P,
a_terminations=[interfaces[0]],
b_terminations=[interfaces[1]],
)
@@ -49,8 +51,10 @@ class CablePathTests(CablePathTestCase):
interfaces[1].refresh_from_db()
self.assertPathIsSet(interfaces[0], path1)
self.assertPathIsSet(interfaces[1], path2)
self.assertEqual(interfaces[0].cable_position, 1)
self.assertEqual(interfaces[1].cable_position, 1)
self.assertEqual(interfaces[0].cable_connector, 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
CableTraceSVG(interfaces[0]).render()
@@ -61,12 +65,12 @@ class CablePathTests(CablePathTestCase):
# Check that all CablePaths have been deleted
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]
[IF2] [IF4]
Cable profile: Straight multi
Cable profile: Straight 2C1P
"""
interfaces = [
Interface.objects.create(device=self.device, name='Interface 1'),
@@ -77,7 +81,7 @@ class CablePathTests(CablePathTestCase):
# Create cable 1
cable1 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI,
profile=CableProfileChoices.STRAIGHT_2C1P,
a_terminations=[interfaces[0], interfaces[1]],
b_terminations=[interfaces[2], interfaces[3]],
)
@@ -112,10 +116,14 @@ class CablePathTests(CablePathTestCase):
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, 1)
self.assertEqual(interfaces[3].cable_position, 2)
self.assertEqual(interfaces[0].cable_connector, 1)
self.assertEqual(interfaces[0].cable_positions, [1])
self.assertEqual(interfaces[1].cable_connector, 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
CableTraceSVG(interfaces[0]).render()
@@ -126,7 +134,8 @@ class CablePathTests(CablePathTestCase):
# Check that all CablePaths have been deleted
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:2] [IF3:2]
@@ -227,7 +236,7 @@ class CablePathTests(CablePathTestCase):
interface.refresh_from_db()
self.assertPathIsSet(interface, path)
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
CableTraceSVG(interfaces[0]).render()
@@ -238,6 +247,7 @@ class CablePathTests(CablePathTestCase):
# Check that all CablePaths have been deleted
self.assertEqual(CablePath.objects.count(), 0)
@skip("Under development")
def test_104_cable_profile_4x4_mpo8(self):
"""
[IF1:1] --C1-- [IF3:1]
@@ -339,7 +349,7 @@ class CablePathTests(CablePathTestCase):
interface.refresh_from_db()
self.assertPathIsSet(interface, path)
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
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 4'),
]
rearport1 = RearPort.objects.create(device=self.device, name='Rear Port 1')
frontport1 = FrontPort.objects.create(device=self.device, name='Front 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', positions=4)
PortMapping.objects.bulk_create([
PortMapping(
device=self.device,
@@ -371,18 +381,39 @@ class CablePathTests(CablePathTestCase):
rear_port=rearport1,
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
cable1 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[interfaces[0], interfaces[1]],
b_terminations=[frontport1],
profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[frontport1],
b_terminations=[interfaces[0], interfaces[1]],
)
cable1.clean()
cable1.save()
cable2 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI,
profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[rearport1],
b_terminations=[interfaces[2], interfaces[3]]
)
@@ -424,10 +455,10 @@ class CablePathTests(CablePathTestCase):
def test_204_multiple_paths_via_pass_through_with_breakouts(self):
"""
[IF1] --C1-- [FP1:1] [RP1] --C3-- [RP2] [FP2:1] --C4-- [IF4]
[IF2] [IF5]
[IF3] --C2-- [FP1:2] [FP2:2] --C5-- [IF6]
[IF4] [IF7]
[IF1] --C1-- [FP1] [RP1] --C3-- [RP2] [FP3] --C4-- [IF5]
[IF2] [IF6]
[IF3] --C2-- [FP2] [FP4] --C5-- [IF7]
[IF4] [IF8]
"""
interfaces = [
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 8'),
]
rearport1 = RearPort.objects.create(device=self.device, name='Rear Port 1', positions=4)
rearport2 = RearPort.objects.create(device=self.device, name='Rear Port 2', positions=4)
frontport1_1 = FrontPort.objects.create(device=self.device, name='Front Port 1:1')
frontport1_2 = FrontPort.objects.create(device=self.device, name='Front Port 1:2')
frontport2_1 = FrontPort.objects.create(device=self.device, name='Front Port 2:1')
frontport2_2 = FrontPort.objects.create(device=self.device, name='Front Port 2:2')
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=8)
frontport1 = FrontPort.objects.create(device=self.device, name='Front Port 1:1', positions=4)
frontport2 = FrontPort.objects.create(device=self.device, name='Front Port 1:2', positions=4)
frontport3 = FrontPort.objects.create(device=self.device, name='Front Port 2:1', positions=4)
frontport4 = FrontPort.objects.create(device=self.device, name='Front Port 2:2', positions=4)
PortMapping.objects.bulk_create([
PortMapping(
device=self.device,
front_port=frontport1_1,
front_port=frontport1,
front_port_position=1,
rear_port=rearport1,
rear_port_position=1,
),
PortMapping(
device=self.device,
front_port=frontport1_2,
front_port_position=1,
front_port=frontport1,
front_port_position=2,
rear_port=rearport1,
rear_port_position=2,
),
PortMapping(
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,
rear_port=rearport2,
rear_port_position=1,
),
PortMapping(
device=self.device,
front_port=frontport2_2,
front_port_position=1,
front_port=frontport3,
front_port_position=2,
rear_port=rearport2,
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
cable1 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[interfaces[0], interfaces[1]],
b_terminations=[frontport1_1]
profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[frontport1],
b_terminations=[interfaces[0], interfaces[1]],
)
cable1.clean()
cable1.save()
cable2 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[interfaces[2], interfaces[3]],
b_terminations=[frontport1_2]
profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[frontport2],
b_terminations=[interfaces[2], interfaces[3]],
)
cable2.clean()
cable2.save()
cable3 = Cable(
profile=CableProfileChoices.STRAIGHT_SINGLE,
profile=CableProfileChoices.STRAIGHT_1C8P,
a_terminations=[rearport1],
b_terminations=[rearport2]
)
cable3.clean()
cable3.save()
cable4 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[frontport2_1],
b_terminations=[interfaces[4], interfaces[5]]
profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[frontport3],
b_terminations=[interfaces[4], interfaces[5]],
)
cable4.clean()
cable4.save()
cable5 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[frontport2_2],
b_terminations=[interfaces[6], interfaces[7]]
profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[frontport4],
b_terminations=[interfaces[6], interfaces[7]],
)
cable5.clean()
cable5.save()
@@ -516,7 +575,7 @@ class CablePathTests(CablePathTestCase):
paths = [
self.assertPathExists(
(
interfaces[0], cable1, frontport1_1, rearport1, cable3, rearport2, frontport2_1, cable4,
interfaces[0], cable1, frontport1, rearport1, cable3, rearport2, frontport3, cable4,
interfaces[4],
),
is_complete=True,
@@ -524,7 +583,7 @@ class CablePathTests(CablePathTestCase):
),
self.assertPathExists(
(
interfaces[1], cable1, frontport1_1, rearport1, cable3, rearport2, frontport2_1, cable4,
interfaces[1], cable1, frontport1, rearport1, cable3, rearport2, frontport3, cable4,
interfaces[5],
),
is_complete=True,
@@ -532,7 +591,7 @@ class CablePathTests(CablePathTestCase):
),
self.assertPathExists(
(
interfaces[2], cable2, frontport1_2, rearport1, cable3, rearport2, frontport2_2, cable5,
interfaces[2], cable2, frontport2, rearport1, cable3, rearport2, frontport4, cable5,
interfaces[6],
),
is_complete=True,
@@ -540,7 +599,7 @@ class CablePathTests(CablePathTestCase):
),
self.assertPathExists(
(
interfaces[3], cable2, frontport1_2, rearport1, cable3, rearport2, frontport2_2, cable5,
interfaces[3], cable2, frontport2, rearport1, cable3, rearport2, frontport4, cable5,
interfaces[7],
),
is_complete=True,
@@ -548,7 +607,7 @@ class CablePathTests(CablePathTestCase):
),
self.assertPathExists(
(
interfaces[4], cable4, frontport2_1, rearport2, cable3, rearport1, frontport1_1, cable1,
interfaces[4], cable4, frontport3, rearport2, cable3, rearport1, frontport1, cable1,
interfaces[0],
),
is_complete=True,
@@ -556,7 +615,7 @@ class CablePathTests(CablePathTestCase):
),
self.assertPathExists(
(
interfaces[5], cable4, frontport2_1, rearport2, cable3, rearport1, frontport1_1, cable1,
interfaces[5], cable4, frontport3, rearport2, cable3, rearport1, frontport1, cable1,
interfaces[1],
),
is_complete=True,
@@ -564,7 +623,7 @@ class CablePathTests(CablePathTestCase):
),
self.assertPathExists(
(
interfaces[6], cable5, frontport2_2, rearport2, cable3, rearport1, frontport1_2, cable2,
interfaces[6], cable5, frontport4, rearport2, cable3, rearport1, frontport2, cable2,
interfaces[2],
),
is_complete=True,
@@ -572,7 +631,7 @@ class CablePathTests(CablePathTestCase):
),
self.assertPathExists(
(
interfaces[7], cable5, frontport2_2, rearport2, cable3, rearport1, frontport1_2, cable2,
interfaces[7], cable5, frontport4, rearport2, cable3, rearport1, frontport2, cable2,
interfaces[3],
),
is_complete=True,
@@ -619,14 +678,14 @@ class CablePathTests(CablePathTestCase):
# Create cables
cable1 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI,
a_terminations=[interfaces[0], interfaces[1]],
b_terminations=[circuittermination1]
profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[circuittermination1],
b_terminations=[interfaces[0], interfaces[1]],
)
cable1.clean()
cable1.save()
cable2 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI,
profile=CableProfileChoices.BREAKOUT_1X4,
a_terminations=[circuittermination2],
b_terminations=[interfaces[2], interfaces[3]]
)
@@ -668,6 +727,8 @@ class CablePathTests(CablePathTestCase):
# Test SVG generation
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):
"""
[IF1] --C1-- [FP1] [RP1] --C2-- [RP3] [FP3] --C3-- [IF2]
@@ -722,7 +783,7 @@ class CablePathTests(CablePathTestCase):
# Create cables
cable1 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI,
profile=CableProfileChoices.STRAIGHT_2C1P,
a_terminations=[interfaces[0]],
b_terminations=[front_ports[0], front_ports[1]]
)
@@ -735,7 +796,7 @@ class CablePathTests(CablePathTestCase):
cable2.clean()
cable2.save()
cable3 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI,
profile=CableProfileChoices.STRAIGHT_2C1P,
a_terminations=[interfaces[1]],
b_terminations=[front_ports[2], front_ports[3]]
)
@@ -803,14 +864,14 @@ class CablePathTests(CablePathTestCase):
# Create cables
cable1 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI,
profile=CableProfileChoices.STRAIGHT_2C2P,
a_terminations=[interfaces[0], interfaces[1]],
b_terminations=[frontport1, frontport2]
)
cable1.clean()
cable1.save()
cable2 = Cable(
profile=CableProfileChoices.STRAIGHT_MULTI,
profile=CableProfileChoices.STRAIGHT_2C2P,
a_terminations=[rearport1, rearport2],
b_terminations=[interfaces[2], interfaces[3]]
)
@@ -819,22 +880,22 @@ class CablePathTests(CablePathTestCase):
# Validate paths
self.assertPathExists(
(interfaces[0], cable1, [frontport1, frontport2], [rearport1, rearport2], cable2, interfaces[2]),
(interfaces[0], cable1, frontport1, rearport1, cable2, interfaces[2]),
is_complete=True,
is_active=True
)
self.assertPathExists(
(interfaces[1], cable1, [frontport1, frontport2], [rearport1, rearport2], cable2, interfaces[3]),
(interfaces[1], cable1, frontport2, rearport2, cable2, interfaces[3]),
is_complete=True,
is_active=True
)
self.assertPathExists(
(interfaces[2], cable2, [rearport1, rearport2], [frontport1, frontport2], cable1, interfaces[0]),
(interfaces[2], cable2, rearport1, frontport1, cable1, interfaces[0]),
is_complete=True,
is_active=True
)
self.assertPathExists(
(interfaces[3], cable2, [rearport1, rearport2], [frontport1, frontport2], cable1, interfaces[1]),
(interfaces[3], cable2, rearport2, frontport2, cable1, interfaces[1]),
is_complete=True,
is_active=True
)

6
netbox/dcim/utils.py
View File

@@ -41,12 +41,12 @@ def create_cablepaths(objects):
"""
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)
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):
cp.save()