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

Merge pull request #4521 from netbox-community/4388-cable-tracing

Fixes #4388: Improve connection endpoint detection
This commit is contained in:
Jeremy Stretch
2020-04-21 12:49:48 -04:00
committed by GitHub
8 changed files with 347 additions and 172 deletions

View File

@ -48,7 +48,7 @@ class CableTraceMixin(object):
# Initialize the path array # Initialize the path array
path = [] path = []
for near_end, cable, far_end in obj.trace(): for near_end, cable, far_end in obj.trace()[0]:
# Serialize each object # Serialize each object
serializer_a = get_serializer_for_model(near_end, prefix='Nested') serializer_a = get_serializer_for_model(near_end, prefix='Nested')

View File

@ -3,3 +3,12 @@ class LoopDetected(Exception):
A loop has been detected while tracing a cable path. A loop has been detected while tracing a cable path.
""" """
pass pass
class CableTraceSplit(Exception):
"""
A cable trace cannot be completed because a RearPort maps to multiple FrontPorts and
we don't know which one to follow.
"""
def __init__(self, termination, *args, **kwargs):
self.termination = termination

View File

@ -2205,26 +2205,3 @@ class Cable(ChangeLoggedModel):
if self.termination_a is None: if self.termination_a is None:
return return
return COMPATIBLE_TERMINATION_TYPES[self.termination_a._meta.model_name] return COMPATIBLE_TERMINATION_TYPES[self.termination_a._meta.model_name]
def get_path_endpoints(self):
"""
Traverse both ends of a cable path and return its connected endpoints. Note that one or both endpoints may be
None.
"""
a_path = self.termination_b.trace()
b_path = self.termination_a.trace()
# Determine overall path status (connected or planned)
if self.status == CableStatusChoices.STATUS_CONNECTED:
path_status = True
for segment in a_path[1:] + b_path[1:]:
if segment[1] is None or segment[1].status != CableStatusChoices.STATUS_CONNECTED:
path_status = False
break
else:
path_status = False
a_endpoint = a_path[-1][2]
b_endpoint = b_path[-1][2]
return a_endpoint, b_endpoint, path_status

View File

@ -10,6 +10,7 @@ from taggit.managers import TaggableManager
from dcim.choices import * from dcim.choices import *
from dcim.constants import * from dcim.constants import *
from dcim.exceptions import CableTraceSplit
from dcim.fields import MACAddressField from dcim.fields import MACAddressField
from extras.models import ObjectChange, TaggedItem from extras.models import ObjectChange, TaggedItem
from extras.utils import extras_features from extras.utils import extras_features
@ -91,7 +92,13 @@ class CableTermination(models.Model):
def trace(self): def trace(self):
""" """
Return a list representing a complete cable path, with each individual segment represented as a three-tuple: Return two items: the traceable portion of a cable path, and the termination points where it splits (if any).
This occurs when the trace is initiated from a midpoint along a path which traverses a RearPort. In cases where
the originating endpoint is unknown, it is not possible to know which corresponding FrontPort to follow.
The path is a list representing a complete cable path, with each individual segment represented as a
three-tuple:
[ [
(termination A, cable, termination B), (termination A, cable, termination B),
(termination C, cable, termination D), (termination C, cable, termination D),
@ -117,10 +124,7 @@ class CableTermination(models.Model):
# Can't map to a FrontPort without a position # Can't map to a FrontPort without a position
if not position_stack: if not position_stack:
# TODO: This behavior is broken. We need a mechanism by which to return all FrontPorts mapped raise CableTraceSplit(termination)
# to a given RearPort so that we can update end-to-end paths when a cable is created/deleted.
# For now, we're maintaining the current behavior of tracing only to the first FrontPort.
position_stack.append(1)
position = position_stack.pop() position = position_stack.pop()
@ -159,12 +163,12 @@ class CableTermination(models.Model):
if not endpoint.cable: if not endpoint.cable:
path.append((endpoint, None, None)) path.append((endpoint, None, None))
logger.debug("No cable connected") logger.debug("No cable connected")
return path return path, None
# Check for loops # Check for loops
if endpoint.cable in [segment[1] for segment in path]: if endpoint.cable in [segment[1] for segment in path]:
logger.debug("Loop detected!") logger.debug("Loop detected!")
return path return path, None
# Record the current segment in the path # Record the current segment in the path
far_end = endpoint.get_cable_peer() far_end = endpoint.get_cable_peer()
@ -174,9 +178,13 @@ class CableTermination(models.Model):
)) ))
# Get the peer port of the far end termination # Get the peer port of the far end termination
endpoint = get_peer_port(far_end) try:
endpoint = get_peer_port(far_end)
except CableTraceSplit as e:
return path, e.termination.frontports.all()
if endpoint is None: if endpoint is None:
return path return path, None
def get_cable_peer(self): def get_cable_peer(self):
if self.cable is None: if self.cable is None:
@ -186,6 +194,23 @@ class CableTermination(models.Model):
if self._cabled_as_b.exists(): if self._cabled_as_b.exists():
return self.cable.termination_a return self.cable.termination_a
def get_path_endpoints(self):
"""
Return all endpoints of paths which traverse this object.
"""
endpoints = []
# Get the far end of the last path segment
path, split_ends = self.trace()
endpoint = path[-1][2]
if split_ends is not None:
for termination in split_ends:
endpoints.extend(termination.get_path_endpoints())
elif endpoint is not None:
endpoints.append(endpoint)
return endpoints
# #
# Console ports # Console ports

View File

@ -3,6 +3,7 @@ import logging
from django.db.models.signals import post_save, pre_delete from django.db.models.signals import post_save, pre_delete
from django.dispatch import receiver from django.dispatch import receiver
from .choices import CableStatusChoices
from .models import Cable, Device, VirtualChassis from .models import Cable, Device, VirtualChassis
@ -48,16 +49,28 @@ def update_connected_endpoints(instance, **kwargs):
instance.termination_b.cable = instance instance.termination_b.cable = instance
instance.termination_b.save() instance.termination_b.save()
# Check if this Cable has formed a complete path. If so, update both endpoints. # Update any endpoints for this Cable.
endpoint_a, endpoint_b, path_status = instance.get_path_endpoints() endpoints = instance.termination_a.get_path_endpoints() + instance.termination_b.get_path_endpoints()
if getattr(endpoint_a, 'is_path_endpoint', False) and getattr(endpoint_b, 'is_path_endpoint', False): for endpoint in endpoints:
logger.debug("Updating path endpoints: {} <---> {}".format(endpoint_a, endpoint_b)) path, split_ends = endpoint.trace()
endpoint_a.connected_endpoint = endpoint_b # Determine overall path status (connected or planned)
endpoint_a.connection_status = path_status path_status = True
endpoint_a.save() for segment in path:
endpoint_b.connected_endpoint = endpoint_a if segment[1] is None or segment[1].status != CableStatusChoices.STATUS_CONNECTED:
endpoint_b.connection_status = path_status path_status = False
endpoint_b.save() break
endpoint_a = path[0][0]
endpoint_b = path[-1][2]
if getattr(endpoint_a, 'is_path_endpoint', False) and getattr(endpoint_b, 'is_path_endpoint', False):
logger.debug("Updating path endpoints: {} <---> {}".format(endpoint_a, endpoint_b))
endpoint_a.connected_endpoint = endpoint_b
endpoint_a.connection_status = path_status
endpoint_a.save()
endpoint_b.connected_endpoint = endpoint_a
endpoint_b.connection_status = path_status
endpoint_b.save()
@receiver(pre_delete, sender=Cable) @receiver(pre_delete, sender=Cable)
@ -67,7 +80,7 @@ def nullify_connected_endpoints(instance, **kwargs):
""" """
logger = logging.getLogger('netbox.dcim.cable') logger = logging.getLogger('netbox.dcim.cable')
endpoint_a, endpoint_b, _ = instance.get_path_endpoints() endpoints = instance.termination_a.get_path_endpoints() + instance.termination_b.get_path_endpoints()
# Disassociate the Cable from its termination points # Disassociate the Cable from its termination points
if instance.termination_a is not None: if instance.termination_a is not None:
@ -79,12 +92,10 @@ def nullify_connected_endpoints(instance, **kwargs):
instance.termination_b.cable = None instance.termination_b.cable = None
instance.termination_b.save() instance.termination_b.save()
# If this Cable was part of a complete path, tear it down # If this Cable was part of any complete end-to-end paths, tear them down.
if hasattr(endpoint_a, 'connected_endpoint') and hasattr(endpoint_b, 'connected_endpoint'): for endpoint in endpoints:
logger.debug("Tearing down path ({} <---> {})".format(endpoint_a, endpoint_b)) logger.debug(f"Removing path information for {endpoint}")
endpoint_a.connected_endpoint = None if hasattr(endpoint, 'connected_endpoint'):
endpoint_a.connection_status = None endpoint.connected_endpoint = None
endpoint_a.save() endpoint.connection_status = None
endpoint_b.connected_endpoint = None endpoint.save()
endpoint_b.connection_status = None
endpoint_b.save()

View File

@ -549,12 +549,21 @@ class CablePathTestCase(TestCase):
self.assertIsNone(endpoint_a.connection_status) self.assertIsNone(endpoint_a.connection_status)
self.assertIsNone(endpoint_b.connection_status) self.assertIsNone(endpoint_b.connection_status)
def test_connection_via_patch(self): def test_connections_via_patch(self):
""" """
1 2 3 Test two connections via patched rear ports:
[Device 1] ----- [Panel 1] ----- [Panel 2] ----- [Device 2] Device 1 <---> Device 2
Iface1 FP1 RP1 RP1 FP1 Iface1 Device 3 <---> Device 4
1 2
[Device 1] -----------+ +----------- [Device 2]
Iface1 | | Iface1
FP1 | 3 | FP1
[Panel 1] ----- [Panel 2]
FP2 | RP1 RP1 | FP2
Iface1 | | Iface1
[Device 3] -----------+ +----------- [Device 4]
4 5
""" """
# Create cables # Create cables
cable1 = Cable( cable1 = Cable(
@ -563,139 +572,43 @@ class CablePathTestCase(TestCase):
) )
cable1.save() cable1.save()
cable2 = Cable( cable2 = Cable(
termination_a=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'), termination_a=Interface.objects.get(device__name='Device 2', name='Interface 1'),
termination_b=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1')
)
cable2.save()
cable3 = Cable(
termination_a=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1'),
termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
)
cable3.save()
# Retrieve endpoints
endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1')
endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1')
# Validate connections
self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
self.assertTrue(endpoint_a.connection_status)
self.assertTrue(endpoint_b.connection_status)
# Delete cable 2
cable2.delete()
# Refresh endpoints
endpoint_a.refresh_from_db()
endpoint_b.refresh_from_db()
# Check that connections have been nullified
self.assertIsNone(endpoint_a.connected_endpoint)
self.assertIsNone(endpoint_b.connected_endpoint)
self.assertIsNone(endpoint_a.connection_status)
self.assertIsNone(endpoint_b.connection_status)
def test_connection_via_multiple_patches(self):
"""
1 2 3 4 5
[Device 1] ----- [Panel 1] ----- [Panel 2] ----- [Panel 3] ----- [Panel 4] ----- [Device 2]
Iface1 FP1 RP1 RP1 FP1 FP1 RP1 RP1 FP1 Iface1
"""
# Create cables
cable1 = Cable(
termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 1')
)
cable1.save()
cable2 = Cable(
termination_a=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
termination_b=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1')
)
cable2.save()
cable3 = Cable(
termination_a=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1'),
termination_b=FrontPort.objects.get(device__name='Panel 3', name='Front Port 1')
)
cable3.save()
cable4 = Cable(
termination_a=RearPort.objects.get(device__name='Panel 3', name='Rear Port 1'),
termination_b=RearPort.objects.get(device__name='Panel 4', name='Rear Port 1')
)
cable4.save()
cable5 = Cable(
termination_a=FrontPort.objects.get(device__name='Panel 4', name='Front Port 1'),
termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
)
cable5.save()
# Retrieve endpoints
endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1')
endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1')
# Validate connections
self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
self.assertTrue(endpoint_a.connection_status)
self.assertTrue(endpoint_b.connection_status)
# Delete cable 3
cable3.delete()
# Refresh endpoints
endpoint_a.refresh_from_db()
endpoint_b.refresh_from_db()
# Check that connections have been nullified
self.assertIsNone(endpoint_a.connected_endpoint)
self.assertIsNone(endpoint_b.connected_endpoint)
self.assertIsNone(endpoint_a.connection_status)
self.assertIsNone(endpoint_b.connection_status)
def test_connection_via_stacked_rear_ports(self):
"""
1 2 3 4 5
[Device 1] ----- [Panel 1] ----- [Panel 2] ----- [Panel 3] ----- [Panel 4] ----- [Device 2]
Iface1 FP1 RP1 FP1 RP1 RP1 FP1 RP1 FP1 Iface1
"""
# Create cables
cable1 = Cable(
termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 1')
)
cable1.save()
cable2 = Cable(
termination_a=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
termination_b=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1') termination_b=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1')
) )
cable2.save() cable2.save()
cable3 = Cable( cable3 = Cable(
termination_a=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1'), termination_a=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
termination_b=RearPort.objects.get(device__name='Panel 3', name='Rear Port 1') termination_b=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1')
) )
cable3.save() cable3.save()
cable4 = Cable( cable4 = Cable(
termination_a=FrontPort.objects.get(device__name='Panel 3', name='Front Port 1'), termination_a=Interface.objects.get(device__name='Device 3', name='Interface 1'),
termination_b=RearPort.objects.get(device__name='Panel 4', name='Rear Port 1') termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 2')
) )
cable4.save() cable4.save()
cable5 = Cable( cable5 = Cable(
termination_a=FrontPort.objects.get(device__name='Panel 4', name='Front Port 1'), termination_a=Interface.objects.get(device__name='Device 4', name='Interface 1'),
termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1') termination_b=FrontPort.objects.get(device__name='Panel 2', name='Front Port 2')
) )
cable5.save() cable5.save()
# Retrieve endpoints # Retrieve endpoints
endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1') endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1')
endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1') endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1')
endpoint_c = Interface.objects.get(device__name='Device 3', name='Interface 1')
endpoint_d = Interface.objects.get(device__name='Device 4', name='Interface 1')
# Validate connections # Validate connections
self.assertEqual(endpoint_a.connected_endpoint, endpoint_b) self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
self.assertEqual(endpoint_b.connected_endpoint, endpoint_a) self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
self.assertEqual(endpoint_c.connected_endpoint, endpoint_d)
self.assertEqual(endpoint_d.connected_endpoint, endpoint_c)
self.assertTrue(endpoint_a.connection_status) self.assertTrue(endpoint_a.connection_status)
self.assertTrue(endpoint_b.connection_status) self.assertTrue(endpoint_b.connection_status)
self.assertTrue(endpoint_c.connection_status)
self.assertTrue(endpoint_d.connection_status)
# Delete cable 3 # Delete cable 3
cable3.delete() cable3.delete()
@ -703,12 +616,204 @@ class CablePathTestCase(TestCase):
# Refresh endpoints # Refresh endpoints
endpoint_a.refresh_from_db() endpoint_a.refresh_from_db()
endpoint_b.refresh_from_db() endpoint_b.refresh_from_db()
endpoint_c.refresh_from_db()
endpoint_d.refresh_from_db()
# Check that connections have been nullified # Check that connections have been nullified
self.assertIsNone(endpoint_a.connected_endpoint) self.assertIsNone(endpoint_a.connected_endpoint)
self.assertIsNone(endpoint_b.connected_endpoint) self.assertIsNone(endpoint_b.connected_endpoint)
self.assertIsNone(endpoint_c.connected_endpoint)
self.assertIsNone(endpoint_d.connected_endpoint)
self.assertIsNone(endpoint_a.connection_status) self.assertIsNone(endpoint_a.connection_status)
self.assertIsNone(endpoint_b.connection_status) self.assertIsNone(endpoint_b.connection_status)
self.assertIsNone(endpoint_c.connection_status)
self.assertIsNone(endpoint_d.connection_status)
def test_connections_via_multiple_patches(self):
"""
Test two connections via patched rear ports:
Device 1 <---> Device 2
Device 3 <---> Device 4
1 2 3
[Device 1] -----------+ +---------------+ +----------- [Device 2]
Iface1 | | | | Iface1
FP1 | 4 | FP1 FP1 | 5 | FP1
[Panel 1] ----- [Panel 2] [Panel 3] ----- [Panel 4]
FP2 | RP1 RP1 | FP2 FP2 | RP1 RP1 | FP2
Iface1 | | | | Iface1
[Device 3] -----------+ +---------------+ +----------- [Device 4]
6 7 8
"""
# Create cables
cable1 = Cable(
termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 1')
)
cable1.save()
cable2 = Cable(
termination_a=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1'),
termination_b=FrontPort.objects.get(device__name='Panel 3', name='Front Port 1')
)
cable2.save()
cable3 = Cable(
termination_a=FrontPort.objects.get(device__name='Panel 4', name='Front Port 1'),
termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
)
cable3.save()
cable4 = Cable(
termination_a=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
termination_b=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1')
)
cable4.save()
cable5 = Cable(
termination_a=RearPort.objects.get(device__name='Panel 3', name='Rear Port 1'),
termination_b=RearPort.objects.get(device__name='Panel 4', name='Rear Port 1')
)
cable5.save()
cable6 = Cable(
termination_a=Interface.objects.get(device__name='Device 3', name='Interface 1'),
termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 2')
)
cable6.save()
cable7 = Cable(
termination_a=FrontPort.objects.get(device__name='Panel 2', name='Front Port 2'),
termination_b=FrontPort.objects.get(device__name='Panel 3', name='Front Port 2')
)
cable7.save()
cable8 = Cable(
termination_a=FrontPort.objects.get(device__name='Panel 4', name='Front Port 2'),
termination_b=Interface.objects.get(device__name='Device 4', name='Interface 1')
)
cable8.save()
# Retrieve endpoints
endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1')
endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1')
endpoint_c = Interface.objects.get(device__name='Device 3', name='Interface 1')
endpoint_d = Interface.objects.get(device__name='Device 4', name='Interface 1')
# Validate connections
self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
self.assertEqual(endpoint_c.connected_endpoint, endpoint_d)
self.assertEqual(endpoint_d.connected_endpoint, endpoint_c)
self.assertTrue(endpoint_a.connection_status)
self.assertTrue(endpoint_b.connection_status)
self.assertTrue(endpoint_c.connection_status)
self.assertTrue(endpoint_d.connection_status)
# Delete cables 4 and 5
cable4.delete()
cable5.delete()
# Refresh endpoints
endpoint_a.refresh_from_db()
endpoint_b.refresh_from_db()
endpoint_c.refresh_from_db()
endpoint_d.refresh_from_db()
# Check that connections have been nullified
self.assertIsNone(endpoint_a.connected_endpoint)
self.assertIsNone(endpoint_b.connected_endpoint)
self.assertIsNone(endpoint_c.connected_endpoint)
self.assertIsNone(endpoint_d.connected_endpoint)
self.assertIsNone(endpoint_a.connection_status)
self.assertIsNone(endpoint_b.connection_status)
self.assertIsNone(endpoint_c.connection_status)
self.assertIsNone(endpoint_d.connection_status)
def test_connections_via_nested_rear_ports(self):
"""
Test two connections via nested rear ports:
Device 1 <---> Device 2
Device 3 <---> Device 4
1 2
[Device 1] -----------+ +----------- [Device 2]
Iface1 | | Iface1
FP1 | 3 4 5 | FP1
[Panel 1] ----- [Panel 2] ----- [Panel 3] ----- [Panel 4]
FP2 | RP1 FP1 RP1 RP1 FP1 RP1 | FP2
Iface1 | | Iface1
[Device 3] -----------+ +----------- [Device 4]
6 7
"""
# Create cables
cable1 = Cable(
termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 1')
)
cable1.save()
cable2 = Cable(
termination_a=FrontPort.objects.get(device__name='Panel 4', name='Front Port 1'),
termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
)
cable2.save()
cable3 = Cable(
termination_a=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
termination_b=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1')
)
cable3.save()
cable4 = Cable(
termination_a=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1'),
termination_b=RearPort.objects.get(device__name='Panel 3', name='Rear Port 1')
)
cable4.save()
cable5 = Cable(
termination_a=FrontPort.objects.get(device__name='Panel 3', name='Front Port 1'),
termination_b=RearPort.objects.get(device__name='Panel 4', name='Rear Port 1')
)
cable5.save()
cable6 = Cable(
termination_a=Interface.objects.get(device__name='Device 3', name='Interface 1'),
termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 2')
)
cable6.save()
cable7 = Cable(
termination_a=FrontPort.objects.get(device__name='Panel 4', name='Front Port 2'),
termination_b=Interface.objects.get(device__name='Device 4', name='Interface 1')
)
cable7.save()
# Retrieve endpoints
endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1')
endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1')
endpoint_c = Interface.objects.get(device__name='Device 3', name='Interface 1')
endpoint_d = Interface.objects.get(device__name='Device 4', name='Interface 1')
# Validate connections
self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
self.assertEqual(endpoint_c.connected_endpoint, endpoint_d)
self.assertEqual(endpoint_d.connected_endpoint, endpoint_c)
self.assertTrue(endpoint_a.connection_status)
self.assertTrue(endpoint_b.connection_status)
self.assertTrue(endpoint_c.connection_status)
self.assertTrue(endpoint_d.connection_status)
# Delete cable 4
cable4.delete()
# Refresh endpoints
endpoint_a.refresh_from_db()
endpoint_b.refresh_from_db()
endpoint_c.refresh_from_db()
endpoint_d.refresh_from_db()
# Check that connections have been nullified
self.assertIsNone(endpoint_a.connected_endpoint)
self.assertIsNone(endpoint_b.connected_endpoint)
self.assertIsNone(endpoint_c.connected_endpoint)
self.assertIsNone(endpoint_d.connected_endpoint)
self.assertIsNone(endpoint_a.connection_status)
self.assertIsNone(endpoint_b.connection_status)
self.assertIsNone(endpoint_c.connection_status)
self.assertIsNone(endpoint_d.connection_status)
def test_connection_via_circuit(self): def test_connection_via_circuit(self):
""" """

View File

@ -32,6 +32,7 @@ from virtualization.models import VirtualMachine
from . import filters, forms, tables from . import filters, forms, tables
from .choices import DeviceFaceChoices from .choices import DeviceFaceChoices
from .constants import NONCONNECTABLE_IFACE_TYPES from .constants import NONCONNECTABLE_IFACE_TYPES
from .exceptions import CableTraceSplit
from .models import ( from .models import (
Cable, ConsolePort, ConsolePortTemplate, ConsoleServerPort, ConsoleServerPortTemplate, Device, DeviceBay, Cable, ConsolePort, ConsolePortTemplate, ConsoleServerPort, ConsoleServerPortTemplate, Device, DeviceBay,
DeviceBayTemplate, DeviceRole, DeviceType, FrontPort, FrontPortTemplate, Interface, InterfaceTemplate, DeviceBayTemplate, DeviceRole, DeviceType, FrontPort, FrontPortTemplate, Interface, InterfaceTemplate,
@ -2033,12 +2034,15 @@ class CableTraceView(PermissionRequiredMixin, View):
def get(self, request, model, pk): def get(self, request, model, pk):
obj = get_object_or_404(model, pk=pk) obj = get_object_or_404(model, pk=pk)
trace = obj.trace() path, split_ends = obj.trace()
total_length = sum([entry[1]._abs_length for entry in trace if entry[1] and entry[1]._abs_length]) total_length = sum(
[entry[1]._abs_length for entry in path if entry[1] and entry[1]._abs_length]
)
return render(request, 'dcim/cable_trace.html', { return render(request, 'dcim/cable_trace.html', {
'obj': obj, 'obj': obj,
'trace': trace, 'trace': path,
'split_ends': split_ends,
'total_length': total_length, 'total_length': total_length,
}) })

View File

@ -48,6 +48,50 @@
{% endif %} {% endif %}
</div> </div>
</div> </div>
{% if not forloop.last %}<hr />{% endif %} <hr />
{% endfor %} {% endfor %}
<div class="row">
{% if split_ends %}
<div class="col-md-7 col-md-offset-3">
<div class="panel panel-warning">
<div class="panel-heading">
<strong><i class="fa fa-warning"></i> Trace Split</strong>
</div>
<div class="panel-body">
There are multiple possible paths from this point. Select a port to continue.
</div>
</div>
<div class="panel panel-default">
<table class="panel-body table">
<thead>
<tr class="table-headings">
<th>Port</th>
<th>Connected</th>
<th>Type</th>
<th>Description</th>
</tr>
</thead>
{% for termination in split_ends %}
<tr>
<td><a href="{% url 'dcim:frontport_trace' pk=termination.pk %}">{{ termination }}</a></td>
<td>
{% if termination.cable %}
<i class="fa fa-check text-success" title="Yes"></i>
{% else %}
<i class="fa fa-times text-danger" title="No"></i>
{% endif %}
</td>
<td>{{ termination.get_type_display }}</td>
<td>{{ termination.description|placeholder }}</td>
</tr>
{% endfor %}
</table>
</div>
</div>
{% else %}
<div class="col-md-11 col-md-offset-1">
<h3 class="text-success text-center">Trace completed!</h3>
</div>
{% endif %}
</div>
{% endblock %} {% endblock %}