mirror/netbox-community-netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2024-05-10 07:54:54 +00:00
Code Activity

Extend Cable model to support multiple A/B terminations

Browse Source
This commit is contained in:
jeremystretch
2022-04-25 17:10:15 -04:00
parent 6c290353c1
commit 4bb9b6ee26
16 changed files with 368 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -2,6 +2,7 @@ from collections import defaultdict
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 ObjectDoesNotExist, ValidationError
from django.db import models
from django.db.models import Sum
@@ -38,10 +39,9 @@ class Cable(NetBoxModel):
on_delete=models.PROTECT,
related_name='+'
)
termination_a_id = models.PositiveBigIntegerField()
termination_a = GenericForeignKey(
ct_field='termination_a_type',
fk_field='termination_a_id'
termination_a_ids = ArrayField(
base_field=models.PositiveBigIntegerField(),
null=True
)
termination_b_type = models.ForeignKey(
to=ContentType,
@@ -49,10 +49,9 @@ class Cable(NetBoxModel):
on_delete=models.PROTECT,
related_name='+'
)
termination_b_id = models.PositiveBigIntegerField()
termination_b = GenericForeignKey(
ct_field='termination_b_type',
fk_field='termination_b_id'
termination_b_ids = ArrayField(
base_field=models.PositiveBigIntegerField(),
null=True
)
type = models.CharField(
max_length=50,
@@ -115,10 +114,6 @@ class Cable(NetBoxModel):
class Meta:
ordering = ['pk']
unique_together = (
('termination_a_type', 'termination_a_id'),
('termination_b_type', 'termination_b_id'),
)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@@ -137,9 +132,9 @@ class Cable(NetBoxModel):
instance = super().from_db(db, field_names, values)
instance._orig_termination_a_type_id = instance.termination_a_type_id
instance._orig_termination_a_id = instance.termination_a_id
instance._orig_termination_a_ids = instance.termination_a_ids
instance._orig_termination_b_type_id = instance.termination_b_type_id
instance._orig_termination_b_id = instance.termination_b_id
instance._orig_termination_b_ids = instance.termination_b_ids
return instance
@@ -150,6 +145,18 @@ class Cable(NetBoxModel):
def get_absolute_url(self):
return reverse('dcim:cable', args=[self.pk])
@property
def termination_a(self):
if not hasattr(self, 'termination_a_type') or not self.termination_a_ids:
return []
return list(self.termination_a_type.model_class().objects.filter(pk__in=self.termination_a_ids))
@property
def termination_b(self):
if not hasattr(self, 'termination_b_type') or not self.termination_b_ids:
return []
return list(self.termination_b_type.model_class().objects.filter(pk__in=self.termination_b_ids))
def clean(self):
from circuits.models import CircuitTermination
@@ -158,9 +165,8 @@ class Cable(NetBoxModel):
# Validate that termination A exists
if not hasattr(self, 'termination_a_type'):
raise ValidationError('Termination A type has not been specified')
try:
self.termination_a_type.model_class().objects.get(pk=self.termination_a_id)
except ObjectDoesNotExist:
model = self.termination_a_type.model_class()
if model.objects.filter(pk__in=self.termination_a_ids).count() != len(self.termination_a_ids):
raise ValidationError({
'termination_a': 'Invalid ID for type {}'.format(self.termination_a_type)
})
@@ -168,9 +174,8 @@ class Cable(NetBoxModel):
# Validate that termination B exists
if not hasattr(self, 'termination_b_type'):
raise ValidationError('Termination B type has not been specified')
try:
self.termination_b_type.model_class().objects.get(pk=self.termination_b_id)
except ObjectDoesNotExist:
model = self.termination_a_type.model_class()
if model.objects.filter(pk__in=self.termination_b_ids).count() != len(self.termination_b_ids):
raise ValidationError({
'termination_b': 'Invalid ID for type {}'.format(self.termination_b_type)
})
@@ -180,14 +185,14 @@ class Cable(NetBoxModel):
err_msg = 'Cable termination points may not be modified. Delete and recreate the cable instead.'
if (
self.termination_a_type_id != self._orig_termination_a_type_id or
self.termination_a_id != self._orig_termination_a_id
set(self.termination_a_ids) != set(self._orig_termination_a_ids)
):
raise ValidationError({
'termination_a': err_msg
})
if (
self.termination_b_type_id != self._orig_termination_b_type_id or
self.termination_b_id != self._orig_termination_b_id
set(self.termination_b_ids) != set(self._orig_termination_b_ids)
):
raise ValidationError({
'termination_b': err_msg
@@ -197,18 +202,18 @@ class Cable(NetBoxModel):
type_b = self.termination_b_type.model
# Validate interface types
if type_a == 'interface' and self.termination_a.type in NONCONNECTABLE_IFACE_TYPES:
raise ValidationError({
'termination_a_id': 'Cables cannot be terminated to {} interfaces'.format(
self.termination_a.get_type_display()
)
})
if type_b == 'interface' and self.termination_b.type in NONCONNECTABLE_IFACE_TYPES:
raise ValidationError({
'termination_b_id': 'Cables cannot be terminated to {} interfaces'.format(
self.termination_b.get_type_display()
)
})
if type_a == 'interface':
for term in self.termination_a:
if term.type in NONCONNECTABLE_IFACE_TYPES:
raise ValidationError({
'termination_a_id': f'Cables cannot be terminated to {term.get_type_display()} interfaces'
})
if type_a == 'interface':
for term in self.termination_b:
if term.type in NONCONNECTABLE_IFACE_TYPES:
raise ValidationError({
'termination_b_id': f'Cables cannot be terminated to {term.get_type_display()} interfaces'
})
# Check that termination types are compatible
if type_b not in COMPATIBLE_TERMINATION_TYPES.get(type_a):
@@ -216,50 +221,48 @@ class Cable(NetBoxModel):
f"Incompatible termination types: {self.termination_a_type} and {self.termination_b_type}"
)
# Check that two connected RearPorts have the same number of positions (if both are >1)
if isinstance(self.termination_a, RearPort) and isinstance(self.termination_b, RearPort):
if self.termination_a.positions > 1 and self.termination_b.positions > 1:
if self.termination_a.positions != self.termination_b.positions:
raise ValidationError(
f"{self.termination_a} has {self.termination_a.positions} position(s) but "
f"{self.termination_b} has {self.termination_b.positions}. "
f"Both terminations must have the same number of positions (if greater than one)."
)
# TODO: Is this validation still necessary?
# # Check that two connected RearPorts have the same number of positions (if both are >1)
# if isinstance(self.termination_a, RearPort) and isinstance(self.termination_b, RearPort):
# if self.termination_a.positions > 1 and self.termination_b.positions > 1:
# if self.termination_a.positions != self.termination_b.positions:
# raise ValidationError(
# f"{self.termination_a} has {self.termination_a.positions} position(s) but "
# f"{self.termination_b} has {self.termination_b.positions}. "
# f"Both terminations must have the same number of positions (if greater than one)."
# )
# A termination point cannot be connected to itself
if self.termination_a == self.termination_b:
if set(self.termination_a).intersection(self.termination_b):
raise ValidationError(f"Cannot connect {self.termination_a_type} to itself")
# A front port cannot be connected to its corresponding rear port
if (
type_a in ['frontport', 'rearport'] and
type_b in ['frontport', 'rearport'] and
(
getattr(self.termination_a, 'rear_port', None) == self.termination_b or
getattr(self.termination_b, 'rear_port', None) == self.termination_a
)
):
raise ValidationError("A front port cannot be connected to it corresponding rear port")
# TODO
# # A front port cannot be connected to its corresponding rear port
# if (
# type_a in ['frontport', 'rearport'] and
# type_b in ['frontport', 'rearport'] and
# (
# getattr(self.termination_a, 'rear_port', None) == self.termination_b or
# getattr(self.termination_b, 'rear_port', None) == self.termination_a
# )
# ):
# raise ValidationError("A front port cannot be connected to it corresponding rear port")
# A CircuitTermination attached to a ProviderNetwork cannot have a Cable
if isinstance(self.termination_a, CircuitTermination) and self.termination_a.provider_network is not None:
raise ValidationError({
'termination_a_id': "Circuit terminations attached to a provider network may not be cabled."
})
if isinstance(self.termination_b, CircuitTermination) and self.termination_b.provider_network is not None:
raise ValidationError({
'termination_b_id': "Circuit terminations attached to a provider network may not be cabled."
})
# TODO
# # A CircuitTermination attached to a ProviderNetwork cannot have a Cable
# if isinstance(self.termination_a, CircuitTermination) and self.termination_a.provider_network is not None:
# raise ValidationError({
# 'termination_a_id': "Circuit terminations attached to a provider network may not be cabled."
# })
# if isinstance(self.termination_b, CircuitTermination) and self.termination_b.provider_network is not None:
# raise ValidationError({
# 'termination_b_id': "Circuit terminations attached to a provider network may not be cabled."
# })
# Check for an existing Cable connected to either termination object
if self.termination_a.cable not in (None, self):
raise ValidationError("{} already has a cable attached (#{})".format(
self.termination_a, self.termination_a.cable_id
))
if self.termination_b.cable not in (None, self):
raise ValidationError("{} already has a cable attached (#{})".format(
self.termination_b, self.termination_b.cable_id
))
for term in [*self.termination_a, *self.termination_b]:
if term.cable not in (None, self):
raise ValidationError(f'{term} already has a cable attached (#{term.cable_id})')
# Validate length and length_unit
if self.length is not None and not self.length_unit:
@@ -276,10 +279,10 @@ class Cable(NetBoxModel):
self._abs_length = None
# Store the parent Device for the A and B terminations (if applicable) to enable filtering
if hasattr(self.termination_a, 'device'):
self._termination_a_device = self.termination_a.device
if hasattr(self.termination_b, 'device'):
self._termination_b_device = self.termination_b.device
if hasattr(self.termination_a[0], 'device'):
self._termination_a_device = self.termination_a[0].device
if hasattr(self.termination_b[0], 'device'):
self._termination_b_device = self.termination_b[0].device
super().save(*args, **kwargs)
@@ -289,14 +292,6 @@ class Cable(NetBoxModel):
def get_status_color(self):
return LinkStatusChoices.colors.get(self.status)
def get_compatible_types(self):
"""
Return all termination types compatible with termination A.
"""
if self.termination_a is None:
return
return COMPATIBLE_TERMINATION_TYPES[self.termination_a._meta.model_name]
class CablePath(models.Model):
"""