NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-24 09:28:38 -06:00
Code Issues Actions 9 Packages Projects Releases Wiki Activity

Refactor CablePath.from_origin()

Browse Source
This commit is contained in:
jeremystretch 2022-05-10 09:53:55 -04:00
parent b44bfa1aa6
commit 5667a9c456
3 changed files with 100 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
netbox/circuits/signals.py
View File

@ -23,5 +23,5 @@ def rebuild_cablepaths(instance, raw=False, **kwargs):
""" """
if not raw: if not raw:
peer_termination = instance.get_peer_termination() peer_termination = instance.get_peer_termination()
if peer_termination: # if peer_termination:
rebuild_paths(peer_termination) # rebuild_paths(peer_termination)

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

@ -14,6 +14,7 @@ from dcim.utils import decompile_path_node, flatten_path, object_to_path_node, p
from netbox.models import NetBoxModel from netbox.models import NetBoxModel
from utilities.fields import ColorField from utilities.fields import ColorField
from utilities.utils import to_meters from utilities.utils import to_meters
from wireless.models import WirelessLink
from .devices import Device from .devices import Device
from .device_components import FrontPort, RearPort from .device_components import FrontPort, RearPort
@ -329,124 +330,129 @@ class CablePath(models.Model):
@classmethod @classmethod
def from_origin(cls, terminations): def from_origin(cls, terminations):
""" """
Create a new CablePath instance as traced from the given path origin. Create a new CablePath instance as traced from the given termination objects. These can be any object to which a
Cable or WirelessLink connects (interfaces, console ports, circuit termination, etc.). All terminations must be
:param terminations: An iterable of one or more CableTermination objects. of the same type and must belong to the same parent object.
""" """
from circuits.models import CircuitTermination from circuits.models import CircuitTermination
if not terminations or terminations[0].termination.link is None:
return None
path = [] path = []
position_stack = [] position_stack = []
is_complete = False is_complete = False
is_active = True is_active = True
is_split = False is_split = False
# Start building the path from its originating CableTerminations while terminations:
path.append([
object_to_path_node(t.termination) for t in terminations
])
node = terminations[0].termination # Terminations must all be of the same type and belong to the same parent
while terminations and node.link is not None: assert all(isinstance(t, type(terminations[0])) for t in terminations[1:])
if hasattr(node.link, 'status') and node.link.status != LinkStatusChoices.STATUS_CONNECTED: assert all(t.parent is terminations[0].parent for t in terminations[1:])
# Step 1: Record the near-end termination object(s)
path.append([
object_to_path_node(t) for t in terminations
])
# Step 2: Determine the attached link (Cable or WirelessLink), if any
link = terminations[0].link
assert all(t.link is link for t in terminations[1:])
if link is None:
# No attached link; abort
break
assert type(link) in (Cable, WirelessLink)
# Step 3: Record the link and update path status if not "connected"
path.append([object_to_path_node(link)])
if hasattr(link, 'status') and link.status != LinkStatusChoices.STATUS_CONNECTED:
is_active = False is_active = False
# Append the cable # Step 4: Determine the far-end terminations
path.append([object_to_path_node(node.link)]) if isinstance(link, Cable):
termination_type = ContentType.objects.get_for_model(terminations[0])
# Follow the link to its far-end termination local_cable_terminations = CableTermination.objects.filter(
if terminations[0].cable_end == 'A': termination_type=termination_type,
peer_terminations = CableTermination.objects.filter(cable=terminations[0].cable, cable_end='B') termination_id__in=[t.pk for t in terminations]
else:
peer_terminations = CableTermination.objects.filter(cable=terminations[0].cable, cable_end='A')
# Follow FrontPorts to their corresponding RearPorts
if isinstance(peer_terminations[0].termination, FrontPort):
path.append([
object_to_path_node(t.termination) for t in peer_terminations
])
terminations = CableTermination.objects.filter(
termination_type=ContentType.objects.get_for_model(RearPort),
termination_id__in=[t.termination_id for t in peer_terminations]
) )
rear_ports = RearPort.objects.filter(pk__in=[t.termination.rear_port_id for t in peer_terminations]) # Terminations must all belong to same end of Cable
# TODO: We're assuming that each of the front-to-rear mapping use equivalent positions. local_cable_end = local_cable_terminations[0].cable_end
node = rear_ports[0] assert all(ct.cable_end == local_cable_end for ct in local_cable_terminations[1:])
if rear_ports[0].positions > 1: remote_cable_terminations = CableTermination.objects.filter(
position_stack.append(peer_terminations[0].termination.rear_port_position) cable=link,
path.append([ cable_end='A' if local_cable_end == 'B' else 'B'
object_to_path_node(rp) for rp in rear_ports )
]) remote_terminations = [ct.termination for ct in remote_cable_terminations]
else:
# WirelessLink
remote_terminations = [link.interface_b] if link.interface_a is terminations[0] else [link.interface_a]
# Follow RearPorts to their corresponding FrontPorts (if any) # Step 5: Record the far-end termination object(s)
elif isinstance(peer_terminations[0], RearPort): path.append([
path.append([ object_to_path_node(t) for t in remote_terminations
object_to_path_node(t.termination) for t in peer_terminations ])
])
# Determine the peer FrontPort's position # Step 6: Determine the "next hop" terminations, if applicable
if peer_terminations[0].termination.positions == 1: if isinstance(remote_terminations[0], FrontPort):
# Follow FrontPorts to their corresponding RearPorts
rear_ports = RearPort.objects.filter(
pk__in=[t.rear_port_id for t in remote_terminations]
)
# RearPorts must have the same number of positions
rp_position_count = rear_ports[0].positions
assert all(rp.positions == rp_position_count for rp in terminations[1:])
# Push position to stack if >1
if rp_position_count > 1:
position_stack.append(remote_terminations[0].rear_port_position)
terminations = rear_ports
elif isinstance(remote_terminations[0], RearPort):
# If the RearPort has multiple positions, pop the current position from the stack
rp_position_count = remote_terminations[0].positions
assert all(rp.positions == rp_position_count for rp in remote_terminations[1:])
if rp_position_count == 1:
position = 1 position = 1
elif position_stack: elif position_stack:
position = position_stack.pop() position = position_stack.pop()
else: else:
# No position indicated: path has split, so we stop at the RearPort # No position indicated: path has split, so we stop at the RearPorts
is_split = True is_split = True
break break
# Map FrontPorts to their corresponding RearPorts # Follow RearPorts to their corresponding FrontPorts (if any)
front_ports = FrontPort.objects.filter( front_ports = FrontPort.objects.filter(
rear_port_id__in=[t.rear_port_id for t in peer_terminations], rear_port_id__in=[t.pk for t in remote_terminations],
rear_port_position=position rear_port_position=position
) )
terminations = CableTermination.objects.filter(
termination_type=ContentType.objects.get_for_model(FrontPort),
termination_id__in=[fp.pk for fp in front_ports]
)
if terminations:
path.append([
object_to_path_node(t.termination) for t in terminations
])
# Follow a CircuitTermination to its corresponding CircuitTermination (A to Z or vice versa) terminations = front_ports
elif isinstance(peer_terminations[0], CircuitTermination):
path.append([
object_to_path_node(t.termination) for t in peer_terminations
])
# Get peer CircuitTerminations elif isinstance(remote_terminations[0], CircuitTermination):
term_side = 'Z' if peer_terminations[0].termination == 'A' else 'Z' # Follow a CircuitTermination to its corresponding CircuitTermination (A to Z or vice versa)
terminations = CircuitTermination.objects.filter( term_side = remote_terminations[0].term_side
circuit=peer_terminations[0].circuit, assert all(ct.term_side == term_side for ct in remote_terminations[1:])
term_side=term_side circuit_termination = CircuitTermination.objects.filter(
) circuit=remote_terminations[0].circuit,
# Tracing across multiple circuits not currently supported term_side='Z' if term_side == 'A' else 'Z'
if len(terminations) > 1: ).first()
is_split = True if circuit_termination is None:
break break
elif terminations: elif circuit_termination.provider_network:
# Circuit terminates to a ProviderNetwork
path.append([ path.append([
object_to_path_node(t.termination) for t in terminations object_to_path_node(circuit_termination.provider_network)
]) ])
# TODO
# if node.provider_network:
# destination = node.provider_network
# break
# elif node.site and not node.cable:
# destination = node.site
# break
else:
# No peer CircuitTermination exists; halt the trace
break break
elif circuit_termination.site and not circuit_termination.cable:
# Circuit terminates to a Site
path.append([
object_to_path_node(circuit_termination.site)
])
break
terminations = [circuit_termination]
# Anything else marks the end of the path # Anything else marks the end of the path
else: else:
path.append([
object_to_path_node(t.termination) for t in peer_terminations
])
is_complete = True is_complete = True
break break

19
netbox/dcim/signals.py
View File

@ -85,23 +85,20 @@ def update_connected_endpoints(instance, created, raw=False, **kwargs):
term for term in instance.terminations if not term.pk term for term in instance.terminations if not term.pk
]) ])
# Split terminations into A/B sets # Split terminations into A/B sets and save link assignments
_terms = defaultdict(list)
for term in instance.terminations:
_terms[term.cable_end].append(term)
# TODO: Update link peers # TODO: Update link peers
# Set cable on terminating endpoints _terms = defaultdict(list)
for term in instance.terminations: for t in instance.terminations:
if term.termination.cable != instance: if t.termination.cable != instance:
term.termination.cable = instance t.termination.cable = instance
term.termination.save() t.termination.save()
_terms[t.cable_end].append(t.termination)
# Create/update cable paths # Create/update cable paths
if created: if created:
for terms in _terms.values(): for terms in _terms.values():
# Examine type of first termination to determine object type (all must be the same) # Examine type of first termination to determine object type (all must be the same)
if isinstance(terms[0].termination, PathEndpoint): if isinstance(terms[0], PathEndpoint):
create_cablepath(terms) create_cablepath(terms)
# else: # else:
# rebuild_paths(terms) # rebuild_paths(terms)