From a324638f1f2be3f40ed98b57f8535075cd80aaf5 Mon Sep 17 00:00:00 2001 From: Jeremy Stretch Date: Mon, 19 Nov 2018 12:37:53 -0500 Subject: [PATCH] Improved logic for recording cable path connection status --- netbox/circuits/models.py | 2 +- netbox/dcim/models.py | 60 +++++++-------------- netbox/dcim/signals.py | 24 ++++++--- netbox/dcim/tests/test_models.py | 93 ++++++++++++++++++++++++++++++++ netbox/dcim/views.py | 2 +- 5 files changed, 129 insertions(+), 52 deletions(-) diff --git a/netbox/circuits/models.py b/netbox/circuits/models.py index e2f712479..776b24156 100644 --- a/netbox/circuits/models.py +++ b/netbox/circuits/models.py @@ -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)' diff --git a/netbox/dcim/models.py b/netbox/dcim/models.py index 501bf32cb..dcf5f93e4 100644 --- a/netbox/dcim/models.py +++ b/netbox/dcim/models.py @@ -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 # diff --git a/netbox/dcim/signals.py b/netbox/dcim/signals.py index 0bb6690b4..2ac3bee06 100644 --- a/netbox/dcim/signals.py +++ b/netbox/dcim/signals.py @@ -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 diff --git a/netbox/dcim/tests/test_models.py b/netbox/dcim/tests/test_models.py index 93336de24..757af61f4 100644 --- a/netbox/dcim/tests/test_models.py +++ b/netbox/dcim/tests/test_models.py @@ -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) diff --git a/netbox/dcim/views.py b/netbox/dcim/views.py index ceea73db3..e6fad5ee8 100644 --- a/netbox/dcim/views.py +++ b/netbox/dcim/views.py @@ -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), })