NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-22 20:12:00 -06:00
Code Issues Actions 8 Packages Projects Releases Wiki Activity

Improved logic for recording cable path connection status

Browse Source
This commit is contained in:
Jeremy Stretch 2018-11-19 12:37:53 -05:00
parent 7dde370ee1
commit a324638f1f
5 changed files with 129 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
netbox/circuits/models.py
View File

@ -237,7 +237,7 @@ class CircuitTermination(CableTermination):
)
connection_status = models.NullBooleanField(
choices=CONNECTION_STATUS_CHOICES,
default=CONNECTION_STATUS_CONNECTED
blank=True
)
port_speed = models.PositiveIntegerField(
verbose_name='Port speed (Kbps)'

60
netbox/dcim/models.py
View File

@ -88,7 +88,7 @@ class CableTermination(models.Model):
class Meta:
abstract = True
def trace(self, position=1):
def trace(self, position=1, follow_circuits=False):
"""
Return a list representing a complete cable path, with each individual segment represented as a three-tuple:
[
@ -97,7 +97,7 @@ class CableTermination(models.Model):
(termination E, cable, termination F)
]
"""
def get_peer_port(termination, position=1):
def get_peer_port(termination, position=1, follow_circuits=False):
from circuits.models import CircuitTermination
# Map a front port to its corresponding rear port
@ -117,7 +117,7 @@ class CableTermination(models.Model):
return peer_port, 1
# Follow a circuit to its other termination
elif isinstance(termination, CircuitTermination):
elif isinstance(termination, CircuitTermination) and follow_circuits:
peer_termination = termination.get_peer_termination()
if peer_termination is None:
return None, None
@ -133,7 +133,7 @@ class CableTermination(models.Model):
far_end = self.cable.termination_b if self.cable.termination_a == self else self.cable.termination_a
path = [(self, self.cable, far_end)]
peer_port, position = get_peer_port(far_end, position)
peer_port, position = get_peer_port(far_end, position, follow_circuits)
if peer_port is None:
return path
@ -1704,7 +1704,7 @@ class ConsolePort(CableTermination, ComponentModel):
)
connection_status = models.NullBooleanField(
choices=CONNECTION_STATUS_CHOICES,
default=CONNECTION_STATUS_CONNECTED
blank=True
)
objects = DeviceComponentManager()
@ -1792,7 +1792,7 @@ class PowerPort(CableTermination, ComponentModel):
)
connection_status = models.NullBooleanField(
choices=CONNECTION_STATUS_CHOICES,
default=CONNECTION_STATUS_CONNECTED
blank=True
)
objects = DeviceComponentManager()
@ -1897,7 +1897,7 @@ class Interface(CableTermination, ComponentModel):
)
connection_status = models.NullBooleanField(
choices=CONNECTION_STATUS_CHOICES,
default=CONNECTION_STATUS_CONNECTED
blank=True
)
lag = models.ForeignKey(
to='self',
@ -2554,7 +2554,7 @@ class Cable(ChangeLoggedModel):
))
# Virtual interfaces cannot be connected
endpoint_a, endpoint_b = self.get_path_endpoints()
endpoint_a, endpoint_b, _ = self.get_path_endpoints()
if (
(
isinstance(endpoint_a, Interface) and
@ -2600,42 +2600,18 @@ class Cable(ChangeLoggedModel):
Traverse both ends of a cable path and return its connected endpoints. Note that one or both endpoints may be
None.
"""
def trace_cable(termination, position=1):
a_path = self.termination_a.trace()
b_path = self.termination_b.trace()
# Given a front port, follow the cable connected to the corresponding rear port/position
if isinstance(termination, FrontPort):
peer_port = termination.rear_port
position = termination.rear_port_position
# Determine overall path status (connected or planned)
cables = [segment[1] for segment in a_path + b_path]
if all(cables) and all([c.status for c in cables]):
path_status = CONNECTION_STATUS_CONNECTED
else:
path_status = CONNECTION_STATUS_PLANNED
# Given a rear port/position, follow the cable connected to the corresponding front port
elif isinstance(termination, RearPort):
if position not in range(1, termination.positions + 1):
raise Exception("Invalid position for {} ({} positions): {})".format(
termination, termination.positions, position
))
peer_port = FrontPort.objects.get(
rear_port=termination,
rear_port_position=position,
)
position = 1
# Termination is not a pass-through port, so we've reached the end of the path
else:
return termination
# Find the cable (if any) attached to the peer port
next_cable = peer_port.cable
# If no cable exists, return None
if next_cable is None:
return None
far_end = next_cable.termination_b if next_cable.termination_a == peer_port else next_cable.termination_a
# Return the far side termination of the cable
return trace_cable(far_end, position)
return trace_cable(self.termination_a), trace_cable(self.termination_b)
# (A path end, B path end, connected/planned)
return a_path[-1][2], b_path[-1][2], path_status
#

24
netbox/dcim/signals.py
View File

@ -23,26 +23,35 @@ def clear_virtualchassis_members(instance, **kwargs):
@receiver(post_save, sender=Cable)
def update_connected_endpoints(instance, **kwargs):
"""
When a Cable is saved, check for and update its two connected endpoints
"""
# Cache the Cable on its two termination points
instance.termination_a.cable = instance
instance.termination_a.save()
instance.termination_b.cable = instance
instance.termination_b.save()
if instance.termination_a.cable != instance:
instance.termination_a.cable = instance
instance.termination_a.save()
if instance.termination_b.cable != instance:
instance.termination_b.cable = instance
instance.termination_b.save()
# Check if this Cable has formed a complete path. If so, update both endpoints.
endpoint_a, endpoint_b = instance.get_path_endpoints()
endpoint_a, endpoint_b, path_status = instance.get_path_endpoints()
if endpoint_a is not None and endpoint_b is not None:
endpoint_a.connected_endpoint = endpoint_b
endpoint_a.connection_status = True
endpoint_a.connection_status = path_status
endpoint_a.save()
endpoint_b.connected_endpoint = endpoint_a
endpoint_b.connection_status = True
endpoint_b.connection_status = path_status
endpoint_b.save()
@receiver(pre_delete, sender=Cable)
def nullify_connected_endpoints(instance, **kwargs):
"""
When a Cable is deleted, check for and update its two connected endpoints
"""
endpoint_a, endpoint_b, _ = instance.get_path_endpoints()
# Disassociate the Cable from its termination points
if instance.termination_a is not None:
@ -53,7 +62,6 @@ def nullify_connected_endpoints(instance, **kwargs):
instance.termination_b.save()
# If this Cable was part of a complete path, tear it down
endpoint_a, endpoint_b = instance.get_path_endpoints()
if endpoint_a is not None and endpoint_b is not None:
endpoint_a.connected_endpoint = None
endpoint_a.connection_status = None

93
netbox/dcim/tests/test_models.py
View File

@ -1,5 +1,6 @@
from django.test import TestCase
from dcim.constants import *
from dcim.models import *
@ -252,3 +253,95 @@ class CableTestCase(TestCase):
cable = Cable(termination_a=self.interface2, termination_b=virtual_interface)
with self.assertRaises(ValidationError):
cable.clean()
class CablePathTestCase(TestCase):
def setUp(self):
site = Site.objects.create(name='Test Site 1', slug='test-site-1')
manufacturer = Manufacturer.objects.create(name='Test Manufacturer 1', slug='test-manufacturer-1')
devicetype = DeviceType.objects.create(
manufacturer=manufacturer, model='Test Device Type 1', slug='test-device-type-1'
)
devicerole = DeviceRole.objects.create(
name='Test Device Role 1', slug='test-device-role-1', color='ff0000'
)
self.device1 = Device.objects.create(
device_type=devicetype, device_role=devicerole, name='Test Device 1', site=site
)
self.device2 = Device.objects.create(
device_type=devicetype, device_role=devicerole, name='Test Device 2', site=site
)
self.interface1 = Interface.objects.create(device=self.device1, name='eth0')
self.interface2 = Interface.objects.create(device=self.device2, name='eth0')
self.panel1 = Device.objects.create(
device_type=devicetype, device_role=devicerole, name='Test Panel 1', site=site
)
self.panel2 = Device.objects.create(
device_type=devicetype, device_role=devicerole, name='Test Panel 2', site=site
)
self.rear_port1 = RearPort.objects.create(
device=self.panel1, name='Rear Port 1', type=PORT_TYPE_8P8C
)
self.front_port1 = FrontPort.objects.create(
device=self.panel1, name='Front Port 1', type=PORT_TYPE_8P8C, rear_port=self.rear_port1
)
self.rear_port2 = RearPort.objects.create(
device=self.panel2, name='Rear Port 2', type=PORT_TYPE_8P8C
)
self.front_port2 = FrontPort.objects.create(
device=self.panel2, name='Front Port 2', type=PORT_TYPE_8P8C, rear_port=self.rear_port2
)
def test_path_completion(self):
# First segment
cable1 = Cable(termination_a=self.interface1, termination_b=self.front_port1)
cable1.save()
interface1 = Interface.objects.get(pk=self.interface1.pk)
self.assertIsNone(interface1.connected_endpoint)
self.assertIsNone(interface1.connection_status)
# Second segment
cable2 = Cable(termination_a=self.rear_port1, termination_b=self.rear_port2)
cable2.save()
interface1 = Interface.objects.get(pk=self.interface1.pk)
self.assertIsNone(interface1.connected_endpoint)
self.assertIsNone(interface1.connection_status)
# Third segment
cable3 = Cable(termination_a=self.front_port2, termination_b=self.interface2, status=CONNECTION_STATUS_PLANNED)
cable3.save()
interface1 = Interface.objects.get(pk=self.interface1.pk)
self.assertEqual(interface1.connected_endpoint, self.interface2)
self.assertEqual(interface1.connection_status, CONNECTION_STATUS_PLANNED)
# Switch third segment from planned to connected
cable3.status = CONNECTION_STATUS_CONNECTED
cable3.save()
interface1 = Interface.objects.get(pk=self.interface1.pk)
self.assertEqual(interface1.connected_endpoint, self.interface2)
self.assertEqual(interface1.connection_status, CONNECTION_STATUS_CONNECTED)
def test_path_teardown(self):
# Build the path
cable1 = Cable(termination_a=self.interface1, termination_b=self.front_port1)
cable1.save()
cable2 = Cable(termination_a=self.rear_port1, termination_b=self.rear_port2)
cable2.save()
cable3 = Cable(termination_a=self.front_port2, termination_b=self.interface2)
cable3.save()
interface1 = Interface.objects.get(pk=self.interface1.pk)
self.assertEqual(interface1.connected_endpoint, self.interface2)
self.assertEqual(interface1.connection_status, CONNECTION_STATUS_CONNECTED)
# Remove a cable
cable2.delete()
interface1 = Interface.objects.get(pk=self.interface1.pk)
self.assertIsNone(interface1.connected_endpoint)
self.assertIsNone(interface1.connection_status)
interface2 = Interface.objects.get(pk=self.interface2.pk)
self.assertIsNone(interface2.connected_endpoint)
self.assertIsNone(interface2.connection_status)

2
netbox/dcim/views.py
View File

@ -1625,7 +1625,7 @@ class CableTraceView(View):
return render(request, 'dcim/cable_trace.html', {
'obj': obj,
'trace': obj.trace(),
'trace': obj.trace(follow_circuits=True),
})