1
0
mirror of https://github.com/netbox-community/netbox.git synced 2024-05-10 07:54:54 +00:00
This commit is contained in:
jeremystretch
2022-07-07 12:48:44 -04:00
parent 1beb8522b9
commit 9a7f3f8c1a
17 changed files with 169 additions and 170 deletions

View File

@ -1,4 +1,5 @@
from circuits import filtersets, models
from dcim.graphql.mixins import CabledObjectMixin
from extras.graphql.mixins import CustomFieldsMixin, TagsMixin
from netbox.graphql.types import ObjectType, OrganizationalObjectType, NetBoxObjectType
@ -11,7 +12,7 @@ __all__ = (
)
class CircuitTerminationType(CustomFieldsMixin, TagsMixin, ObjectType):
class CircuitTerminationType(CustomFieldsMixin, TagsMixin, CabledObjectMixin, ObjectType):
class Meta:
model = models.CircuitTermination

View File

@ -52,17 +52,14 @@ class CabledObjectSerializer(serializers.ModelSerializer):
"""
Return the appropriate serializer for the link termination model.
"""
if not obj.cable:
if not obj.link_peers:
return []
# Return serialized peer termination objects
if obj.link_peers:
serializer = get_serializer_for_model(obj.link_peers[0], prefix='Nested')
context = {'request': self.context['request']}
return serializer(obj.link_peers, context=context, many=True).data
return []
@swagger_serializer_method(serializer_or_field=serializers.BooleanField)
def get__occupied(self, obj):
return obj._occupied
@ -77,8 +74,7 @@ class ConnectedEndpointsSerializer(serializers.ModelSerializer):
connected_endpoints_reachable = serializers.SerializerMethodField(read_only=True)
def get_connected_endpoints_type(self, obj):
endpoints = obj.connected_endpoints
if endpoints:
if endpoints := obj.connected_endpoints:
return f'{endpoints[0]._meta.app_label}.{endpoints[0]._meta.model_name}'
@swagger_serializer_method(serializer_or_field=serializers.ListField)
@ -86,8 +82,7 @@ class ConnectedEndpointsSerializer(serializers.ModelSerializer):
"""
Return the appropriate serializer for the type of connected object.
"""
endpoints = obj.connected_endpoints
if endpoints:
if endpoints := obj.connected_endpoints:
serializer = get_serializer_for_model(endpoints[0], prefix='Nested')
context = {'request': self.context['request']}
return serializer(endpoints, many=True, context=context).data
@ -1016,15 +1011,15 @@ class CableSerializer(NetBoxModelSerializer):
]
def _get_terminations_type(self, obj, side):
assert side.lower() in ('a', 'b')
terms = [t.termination for t in obj.terminations.all() if t.cable_end == side.upper()]
assert side in CableEndChoices.values()
terms = getattr(obj, f'get_{side.lower()}_terminations')()
if terms:
ct = ContentType.objects.get_for_model(terms[0])
return f"{ct.app_label}.{ct.model}"
def _get_terminations(self, obj, side):
assert side.lower() in ('a', 'b')
terms = [t.termination for t in obj.terminations.all() if t.cable_end == side.upper()]
assert side in CableEndChoices.values()
terms = getattr(obj, f'get_{side.lower()}_terminations')()
if not terms:
return []
@ -1037,19 +1032,19 @@ class CableSerializer(NetBoxModelSerializer):
@swagger_serializer_method(serializer_or_field=serializers.CharField)
def get_a_terminations_type(self, obj):
return self._get_terminations_type(obj, 'a')
return self._get_terminations_type(obj, CableEndChoices.SIDE_A)
@swagger_serializer_method(serializer_or_field=serializers.CharField)
def get_b_terminations_type(self, obj):
return self._get_terminations_type(obj, 'b')
return self._get_terminations_type(obj, CableEndChoices.SIDE_B)
@swagger_serializer_method(serializer_or_field=serializers.DictField)
def get_a_terminations(self, obj):
return self._get_terminations(obj, 'a')
return self._get_terminations(obj, CableEndChoices.SIDE_A)
@swagger_serializer_method(serializer_or_field=serializers.DictField)
def get_b_terminations(self, obj):
return self._get_terminations(obj, 'b')
return self._get_terminations(obj, CableEndChoices.SIDE_B)
class TracedCableSerializer(serializers.ModelSerializer):
@ -1066,7 +1061,7 @@ class TracedCableSerializer(serializers.ModelSerializer):
class CableTerminationSerializer(NetBoxModelSerializer):
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:cable-detail')
url = serializers.HyperlinkedIdentityField(view_name='dcim-api:cabletermination-detail')
termination_type = ContentTypeField(
queryset=ContentType.objects.filter(CABLE_TERMINATION_MODELS)
)

View File

@ -15,6 +15,7 @@ from circuits.models import Circuit
from dcim import filtersets
from dcim.constants import CABLE_TRACE_SVG_DEFAULT_WIDTH
from dcim.models import *
from dcim.svg import CableTraceSVG
from extras.api.views import ConfigContextQuerySetMixin
from ipam.models import Prefix, VLAN
from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired
@ -52,37 +53,30 @@ class PathEndpointMixin(object):
# Initialize the path array
path = []
# Render SVG image if requested
if request.GET.get('render', None) == 'svg':
# Render SVG
try:
width = int(request.GET.get('width', CABLE_TRACE_SVG_DEFAULT_WIDTH))
except (ValueError, TypeError):
width = CABLE_TRACE_SVG_DEFAULT_WIDTH
drawing = obj.get_trace_svg(
base_url=request.build_absolute_uri('/'),
width=width
)
return HttpResponse(drawing.tostring(), content_type='image/svg+xml')
drawing = CableTraceSVG(self, base_url=request.build_absolute_uri('/'), width=width)
return HttpResponse(drawing.render().tostring(), content_type='image/svg+xml')
# Serialize path objects, iterating over each three-tuple in the path
for near_end, cable, far_end in obj.trace():
if near_end is None:
# Split paths
break
# Serialize each object
if near_end is not None:
serializer_a = get_serializer_for_model(near_end[0], prefix='Nested')
x = serializer_a(near_end, many=True, context={'request': request}).data
if cable is not None:
y = serializers.TracedCableSerializer(cable[0], context={'request': request}).data
near_end = serializer_a(near_end, many=True, context={'request': request}).data
else:
y = None
# Path is split; stop here
break
if cable is not None:
cable = serializers.TracedCableSerializer(cable[0], context={'request': request}).data
if far_end is not None:
serializer_b = get_serializer_for_model(far_end[0], prefix='Nested')
z = serializer_b(far_end, many=True, context={'request': request}).data
else:
z = None
far_end = serializer_b(far_end, many=True, context={'request': request}).data
path.append((x, y, z))
path.append((near_end, cable, far_end))
return Response(path)

View File

@ -1294,7 +1294,7 @@ class CableEndChoices(ChoiceSet):
CHOICES = (
(SIDE_A, 'A'),
(SIDE_B, 'B'),
('', ''),
# ('', ''),
)

View File

@ -0,0 +1,5 @@
class CabledObjectMixin:
def resolve_cable_end(self, info):
# Handle empty values
return self.cable_end or None

View File

@ -7,6 +7,7 @@ from extras.graphql.mixins import (
from ipam.graphql.mixins import IPAddressesMixin, VLANGroupsMixin
from netbox.graphql.scalars import BigInt
from netbox.graphql.types import BaseObjectType, OrganizationalObjectType, NetBoxObjectType
from .mixins import CabledObjectMixin
__all__ = (
'CableType',
@ -107,7 +108,7 @@ class CableTerminationType(NetBoxObjectType):
filterset_class = filtersets.CableTerminationFilterSet
class ConsolePortType(ComponentObjectType):
class ConsolePortType(ComponentObjectType, CabledObjectMixin):
class Meta:
model = models.ConsolePort
@ -129,7 +130,7 @@ class ConsolePortTemplateType(ComponentTemplateObjectType):
return self.type or None
class ConsoleServerPortType(ComponentObjectType):
class ConsoleServerPortType(ComponentObjectType, CabledObjectMixin):
class Meta:
model = models.ConsoleServerPort
@ -211,7 +212,7 @@ class DeviceTypeType(NetBoxObjectType):
return self.airflow or None
class FrontPortType(ComponentObjectType):
class FrontPortType(ComponentObjectType, CabledObjectMixin):
class Meta:
model = models.FrontPort
@ -227,7 +228,7 @@ class FrontPortTemplateType(ComponentTemplateObjectType):
filterset_class = filtersets.FrontPortTemplateFilterSet
class InterfaceType(IPAddressesMixin, ComponentObjectType):
class InterfaceType(IPAddressesMixin, ComponentObjectType, CabledObjectMixin):
class Meta:
model = models.Interface
@ -330,7 +331,7 @@ class PlatformType(OrganizationalObjectType):
filterset_class = filtersets.PlatformFilterSet
class PowerFeedType(NetBoxObjectType):
class PowerFeedType(NetBoxObjectType, CabledObjectMixin):
class Meta:
model = models.PowerFeed
@ -338,7 +339,7 @@ class PowerFeedType(NetBoxObjectType):
filterset_class = filtersets.PowerFeedFilterSet
class PowerOutletType(ComponentObjectType):
class PowerOutletType(ComponentObjectType, CabledObjectMixin):
class Meta:
model = models.PowerOutlet
@ -374,7 +375,7 @@ class PowerPanelType(NetBoxObjectType):
filterset_class = filtersets.PowerPanelFilterSet
class PowerPortType(ComponentObjectType):
class PowerPortType(ComponentObjectType, CabledObjectMixin):
class Meta:
model = models.PowerPort
@ -426,7 +427,7 @@ class RackRoleType(OrganizationalObjectType):
filterset_class = filtersets.RackRoleFilterSet
class RearPortType(ComponentObjectType):
class RearPortType(ComponentObjectType, CabledObjectMixin):
class Meta:
model = models.RearPort

View File

@ -27,7 +27,7 @@ class Migration(migrations.Migration):
),
migrations.AddConstraint(
model_name='cabletermination',
constraint=models.UniqueConstraint(fields=('termination_type', 'termination_id'), name='unique_termination'),
constraint=models.UniqueConstraint(fields=('termination_type', 'termination_id'), name='dcim_cable_termination_unique_termination'),
),
# Update CablePath model

View File

@ -1,3 +1,4 @@
import itertools
from collections import defaultdict
from django.contrib.contenttypes.fields import GenericForeignKey
@ -11,15 +12,14 @@ from django.urls import reverse
from dcim.choices import *
from dcim.constants import *
from dcim.fields import PathField
from dcim.utils import decompile_path_node, flatten_path, object_to_path_node, path_node_to_object
from dcim.utils import decompile_path_node, object_to_path_node, path_node_to_object
from netbox.models import NetBoxModel
from utilities.fields import ColorField
from utilities.querysets import RestrictedQuerySet
from utilities.utils import to_meters
from wireless.models import WirelessLink
from .devices import Device
from .device_components import FrontPort, RearPort
from .devices import Device
__all__ = (
'Cable',
@ -110,7 +110,8 @@ class Cable(NetBoxModel):
# Cache the original status so we can check later if it's been changed
self._orig_status = self.status
# Assign associated CableTerminations (if any)
# Assign any *new* CableTerminations for the instance. These will replace any existing
# terminations on save().
if a_terminations is not None:
self.a_terminations = a_terminations
if b_terminations is not None:
@ -133,28 +134,25 @@ class Cable(NetBoxModel):
self.length_unit = ''
a_terminations = [
CableTermination(cable=self, cable_end='A', termination=t) for t in getattr(self, 'a_terminations', [])
CableTermination(cable=self, cable_end='A', termination=t)
for t in getattr(self, 'a_terminations', [])
]
b_terminations = [
CableTermination(cable=self, cable_end='B', termination=t) for t in getattr(self, 'b_terminations', [])
CableTermination(cable=self, cable_end='B', termination=t)
for t in getattr(self, 'b_terminations', [])
]
# Check that all termination objects for either end are of the same type
for terms in (a_terminations, b_terminations):
if terms and len(terms) > 1:
if not all(t.termination_type == terms[0].termination_type for t in terms[1:]):
raise ValidationError(
"Cannot connect different termination types to same end of cable."
)
if len(terms) > 1 and not all(t.termination_type == terms[0].termination_type for t in terms[1:]):
raise ValidationError("Cannot connect different termination types to same end of cable.")
# Check that termination types are compatible
if a_terminations and b_terminations:
a_type = a_terminations[0].termination_type.model
b_type = b_terminations[0].termination_type.model
if b_type not in COMPATIBLE_TERMINATION_TYPES.get(a_type):
raise ValidationError(
f"Incompatible termination types: {a_type} and {b_type}"
)
raise ValidationError(f"Incompatible termination types: {a_type} and {b_type}")
# Run clean() on any new CableTerminations
for cabletermination in [*a_terminations, *b_terminations]:
@ -169,6 +167,7 @@ class Cable(NetBoxModel):
else:
self._abs_length = None
# TODO: Need to come with a proper solution for filtering by termination parent
# Store the parent Device for the A and B terminations (if applicable) to enable filtering
if hasattr(self, 'a_terminations'):
self._termination_a_device = getattr(self.a_terminations[0], 'device', None)
@ -210,13 +209,15 @@ class Cable(NetBoxModel):
return LinkStatusChoices.colors.get(self.status)
def get_a_terminations(self):
# Query self.terminations.all() to leverage cached results
return [
term.termination for term in CableTermination.objects.filter(cable=self, cable_end='A')
ct.termination for ct in self.terminations.all() if ct.cable_end == CableEndChoices.SIDE_A
]
def get_b_terminations(self):
# Query self.terminations.all() to leverage cached results
return [
term.termination for term in CableTermination.objects.filter(cable=self, cable_end='B')
ct.termination for ct in self.terminations.all() if ct.cable_end == CableEndChoices.SIDE_B
]
@ -253,7 +254,7 @@ class CableTermination(models.Model):
constraints = (
models.UniqueConstraint(
fields=('termination_type', 'termination_id'),
name='unique_termination'
name='dcim_cable_termination_unique_termination'
),
)
@ -289,34 +290,48 @@ class CableTermination(models.Model):
# Delete the cable association on the terminating object
termination_model = self.termination._meta.model
termination_model.objects.filter(pk=self.termination_id).update(cable=None, cable_end='', _path=None)
termination_model.objects.filter(pk=self.termination_id).update(
cable=None,
cable_end=''
)
super().delete(*args, **kwargs)
class CablePath(models.Model):
"""
A CablePath instance represents the physical path from an origin to a destination, including all intermediate
elements in the path. Every instance must specify an `origin`, whereas `destination` may be null (for paths which do
not terminate on a PathEndpoint).
A CablePath instance represents the physical path from a set of origin nodes to a set of destination nodes,
including all intermediate elements.
`path` contains a list of nodes within the path, each represented by a tuple of (type, ID). The first element in the
path must be a Cable instance, followed by a pair of pass-through ports. For example, consider the following
`path` contains the ordered set of nodes, arranged in lists of (type, ID) tuples. (Each cable in the path can
terminate to one or more objects.) For example, consider the following
topology:
1 2 3
Interface A --- Front Port A | Rear Port A --- Rear Port B | Front Port B --- Interface B
A B C
Interface 1 --- Front Port 1 | Rear Port 1 --- Rear Port 2 | Front Port 3 --- Interface 2
Front Port 2 Front Port 4
This path would be expressed as:
CablePath(
origin = Interface A
destination = Interface B
path = [Cable 1, Front Port A, Rear Port A, Cable 2, Rear Port B, Front Port B, Cable 3]
path = [
[Interface 1],
[Cable A],
[Front Port 1, Front Port 2],
[Rear Port 1],
[Cable B],
[Rear Port 2],
[Front Port 3, Front Port 4],
[Cable C],
[Interface 2],
]
)
`is_active` is set to True only if 1) `destination` is not null, and 2) every Cable within the path has a status of
"connected".
`is_active` is set to True only if every Cable within the path has a status of "connected". `is_complete` is True
if the instance represents a complete end-to-end path from origin(s) to destination(s). `is_split` is True if the
path diverges across multiple cables.
`_nodes` retains a flattened list of all nodes within the path to enable simple filtering.
"""
path = models.JSONField(
default=list
@ -332,34 +347,30 @@ class CablePath(models.Model):
)
_nodes = PathField()
class Meta:
pass
def __str__(self):
status = ' (active)' if self.is_active else ' (split)' if self.is_split else ''
return f"Path #{self.pk}: {len(self.path)} nodes{status}"
return f"Path #{self.pk}: {len(self.path)} hops"
def save(self, *args, **kwargs):
# Save the flattened nodes list
self._nodes = flatten_path(self.path)
self._nodes = list(itertools.chain(*self.path))
super().save(*args, **kwargs)
# Record a direct reference to this CablePath on its originating object(s)
origin_model = self.origins[0]._meta.model
origin_ids = [o.id for o in self.origins]
origin_model = self.origin_type.model_class()
origin_ids = [decompile_path_node(node)[1] for node in self.path[0]]
origin_model.objects.filter(pk__in=origin_ids).update(_path=self.pk)
@property
def origin_type(self):
if self.path:
ct_id, _ = decompile_path_node(self.path[0][0])
return ContentType.objects.get_for_id(ct_id)
@property
def destination_type(self):
if not self.is_complete:
return None
if self.is_complete:
ct_id, _ = decompile_path_node(self.path[-1][0])
return ContentType.objects.get_for_id(ct_id)
@ -375,7 +386,7 @@ class CablePath(models.Model):
@property
def origins(self):
"""
Return the list of originating objects (from cache, if available).
Return the list of originating objects.
"""
if hasattr(self, '_path_objects'):
return self.path_objects[0]
@ -386,7 +397,7 @@ class CablePath(models.Model):
@property
def destinations(self):
"""
Return the list of destination objects (from cache, if available), if the path is complete.
Return the list of destination objects, if the path is complete.
"""
if not self.is_complete:
return []

View File

@ -10,7 +10,6 @@ from mptt.models import MPTTModel, TreeForeignKey
from dcim.choices import *
from dcim.constants import *
from dcim.fields import MACAddressField, WWNField
from dcim.svg import CableTraceSVG
from netbox.models import OrganizationalModel, NetBoxModel
from utilities.choices import ColorChoices
from utilities.fields import ColorField, NaturalOrderingField
@ -105,7 +104,8 @@ class ModularComponentModel(ComponentModel):
class CabledObjectModel(models.Model):
"""
An abstract model inherited by all models to which a Cable can terminate.
An abstract model inherited by all models to which a Cable can terminate. Provides the `cable` and `cable_end`
fields for caching cable associations, as well as `mark_connected` to designate "fake" connections.
"""
cable = models.ForeignKey(
to='dcim.Cable',
@ -134,8 +134,11 @@ class CabledObjectModel(models.Model):
raise ValidationError({
"cable_end": "Must specify cable end (A or B) when attaching a cable."
})
if self.mark_connected and self.cable_id:
if self.cable_end and not self.cable:
raise ValidationError({
"cable_end": "Cable end must not be set without a cable."
})
if self.mark_connected and self.cable:
raise ValidationError({
"mark_connected": "Cannot mark as connected with a cable attached."
})
@ -167,12 +170,13 @@ class CabledObjectModel(models.Model):
"""
Generic wrapper for a Cable, WirelessLink, or some other relation to a connected termination.
"""
# TODO: Support WirelessLinks
return self.cable
class PathEndpoint(models.Model):
"""
An abstract model inherited by any CableTermination subclass which represents the end of a CablePath; specifically,
An abstract model inherited by any CabledObjectModel subclass which represents the end of a CablePath; specifically,
these include ConsolePort, ConsoleServerPort, PowerPort, PowerOutlet, Interface, and PowerFeed.
`_path` references the CablePath originating from this instance, if any. It is set or cleared by the receivers in
@ -215,14 +219,6 @@ class PathEndpoint(models.Model):
# Return the path as a list of three-tuples (A termination(s), cable(s), B termination(s))
return list(zip(*[iter(path)] * 3))
def get_trace_svg(self, base_url=None, width=CABLE_TRACE_SVG_DEFAULT_WIDTH):
trace = CableTraceSVG(self, base_url=base_url, width=width)
return trace.render()
@property
def path(self):
return self._path
@property
def connected_endpoints(self):
"""
@ -338,7 +334,15 @@ class PowerPort(ModularComponentModel, CabledObjectModel, PathEndpoint):
def get_downstream_powerports(self, leg=None):
"""
Return a queryset of all PowerPorts connected via cable to a child PowerOutlet.
Return a queryset of all PowerPorts connected via cable to a child PowerOutlet. For example, in the topology
below, PP1.get_downstream_powerports() would return PP2-4.
---- PO1 <---> PP2
/
PP1 ------- PO2 <---> PP3
\
---- PO3 <---> PP4
"""
poweroutlets = self.poweroutlets.filter(cable__isnull=False)
if leg:

View File

@ -438,9 +438,9 @@ class Rack(NetBoxModel):
peer for peer in powerfeed.link_peers if isinstance(peer, PowerPort)
])
allocated_draw = 0
for powerport in powerports:
allocated_draw += powerport.get_power_draw()['allocated']
allocated_draw = sum([
powerport.get_power_draw()['allocated'] for powerport in powerports
])
return int(allocated_draw / available_power_total * 100)

View File

@ -3,7 +3,7 @@ import logging
from django.db.models.signals import post_save, post_delete, pre_delete
from django.dispatch import receiver
from .choices import LinkStatusChoices
from .choices import CableEndChoices, LinkStatusChoices
from .models import Cable, CablePath, CableTermination, Device, PathEndpoint, PowerPanel, Rack, Location, VirtualChassis
from .models.cables import trace_paths
from .utils import create_cablepath, rebuild_paths
@ -83,7 +83,7 @@ def update_connected_endpoints(instance, created, raw=False, **kwargs):
a_terminations = []
b_terminations = []
for t in instance.terminations.all():
if t.cable_end == 'A':
if t.cable_end == CableEndChoices.SIDE_A:
a_terminations.append(t.termination)
else:
b_terminations.append(t.termination)

View File

@ -13,21 +13,17 @@ CABLE_LENGTH = """
{% if record.length %}{{ record.length|simplify_decimal }} {{ record.length_unit }}{% endif %}
"""
CABLE_TERMINATION = """
{{ value|join:", " }}
"""
CABLE_TERMINATION_PARENT = """
{% with value.0 as termination %}
{% if termination.device %}
<a href="{{ termination.device.get_absolute_url }}">{{ termination.device }}</a>
{% elif termination.circuit %}
<a href="{{ termination.circuit.get_absolute_url }}">{{ termination.circuit }}</a>
{% elif termination.power_panel %}
<a href="{{ termination.power_panel.get_absolute_url }}">{{ termination.power_panel }}</a>
{% endif %}
{% endwith %}
"""
# CABLE_TERMINATION_PARENT = """
# {% with value.0 as termination %}
# {% if termination.device %}
# <a href="{{ termination.device.get_absolute_url }}">{{ termination.device }}</a>
# {% elif termination.circuit %}
# <a href="{{ termination.circuit.get_absolute_url }}">{{ termination.circuit }}</a>
# {% elif termination.power_panel %}
# <a href="{{ termination.power_panel.get_absolute_url }}">{{ termination.power_panel }}</a>
# {% endif %}
# {% endwith %}
# """
DEVICE_LINK = """
<a href="{% url 'dcim:device' pk=record.pk %}">

View File

@ -3,6 +3,7 @@ from django.test import TestCase
from circuits.models import *
from dcim.choices import LinkStatusChoices
from dcim.models import *
from dcim.svg import CableTraceSVG
from dcim.utils import object_to_path_node
@ -107,7 +108,7 @@ class CablePathTestCase(TestCase):
self.assertPathIsSet(interface2, path2)
# Test SVG generation
interface1.get_trace_svg()
CableTraceSVG(interface1).render()
# Delete cable 1
cable1.delete()
@ -146,7 +147,7 @@ class CablePathTestCase(TestCase):
self.assertPathIsSet(consoleserverport1, path2)
# Test SVG generation
consoleport1.get_trace_svg()
CableTraceSVG(consoleport1).render()
# Delete cable 1
cable1.delete()
@ -185,7 +186,7 @@ class CablePathTestCase(TestCase):
self.assertPathIsSet(poweroutlet1, path2)
# Test SVG generation
powerport1.get_trace_svg()
CableTraceSVG(powerport1).render()
# Delete cable 1
cable1.delete()
@ -224,7 +225,7 @@ class CablePathTestCase(TestCase):
self.assertPathIsSet(powerfeed1, path2)
# Test SVG generation
powerport1.get_trace_svg()
CableTraceSVG(powerport1).render()
# Delete cable 1
cable1.delete()
@ -267,7 +268,7 @@ class CablePathTestCase(TestCase):
self.assertPathIsSet(interface3, path2)
# Test SVG generation
interface1.get_trace_svg()
CableTraceSVG(interface1).render()
# Delete cable 1
cable1.delete()
@ -319,7 +320,7 @@ class CablePathTestCase(TestCase):
self.assertPathIsSet(interface4, path2)
# Test SVG generation
interface1.get_trace_svg()
CableTraceSVG(interface1).render()
# Delete cable 1
cable1.delete()

View File

@ -1,3 +1,5 @@
import itertools
from django.contrib.contenttypes.models import ContentType
from django.db import transaction
@ -29,16 +31,6 @@ def path_node_to_object(repr):
return ct.model_class().objects.get(pk=object_id)
def flatten_path(path):
"""
Flatten a two-dimensional array (list of lists) into a flat list.
"""
ret = []
for step in path:
ret.extend(step)
return ret
def create_cablepath(terminations):
"""
Create CablePaths for all paths originating from the specified set of nodes.
@ -54,7 +46,7 @@ def create_cablepath(terminations):
def rebuild_paths(terminations):
"""
Rebuild all CablePaths which traverse the specified node
Rebuild all CablePaths which traverse the specified nodes.
"""
from dcim.models import CablePath

View File

@ -28,6 +28,18 @@ from .choices import DeviceFaceChoices
from .constants import NONCONNECTABLE_IFACE_TYPES
from .models import *
CABLE_TERMINATION_TYPES = {
'dcim.consoleport': ConsolePort,
'dcim.consoleserverport': ConsoleServerPort,
'dcim.powerport': PowerPort,
'dcim.poweroutlet': PowerOutlet,
'dcim.interface': Interface,
'dcim.frontport': FrontPort,
'dcim.rearport': RearPort,
'dcim.powerfeed': PowerFeed,
'circuits.circuittermination': CircuitTermination,
}
class DeviceComponentsView(generic.ObjectChildrenView):
queryset = Device.objects.all()
@ -2818,22 +2830,10 @@ class CableEditView(generic.ObjectEditView):
# If creating a new Cable, initialize the form class using URL query params
if 'pk' not in kwargs:
termination_types = {
'dcim.consoleport': ConsolePort,
'dcim.consoleserverport': ConsoleServerPort,
'dcim.powerport': PowerPort,
'dcim.poweroutlet': PowerOutlet,
'dcim.interface': Interface,
'dcim.frontport': FrontPort,
'dcim.rearport': RearPort,
'dcim.powerfeed': PowerFeed,
'circuits.circuittermination': CircuitTermination,
}
a_type = termination_types.get(request.GET.get('a_terminations_type'))
b_type = termination_types.get(request.GET.get('b_terminations_type'))
self.form = forms.get_cable_form(a_type, b_type)
self.form = forms.get_cable_form(
a_type=CABLE_TERMINATION_TYPES.get(request.GET.get('a_terminations_type')),
b_type=CABLE_TERMINATION_TYPES.get(request.GET.get('b_terminations_type'))
)
return super().dispatch(request, *args, **kwargs)

View File

@ -179,8 +179,8 @@ class ExceptionHandlingMiddleware:
def process_exception(self, request, exception):
# Handle exceptions that occur from REST API requests
if is_api_request(request):
return rest_api_server_error(request)
# if is_api_request(request):
# return rest_api_server_error(request)
# Don't catch exceptions when in debug mode
if settings.DEBUG:

View File

@ -3,7 +3,6 @@ import sys
from django.conf import settings
from django.core.cache import cache
from django.db.models import F
from django.http import HttpResponseServerError
from django.shortcuts import redirect, render
from django.template import loader
@ -37,13 +36,13 @@ class HomeView(View):
return redirect("login")
connected_consoleports = ConsolePort.objects.restrict(request.user, 'view').prefetch_related('_path').filter(
_path__is_active=True
_path__is_complete=True
)
connected_powerports = PowerPort.objects.restrict(request.user, 'view').prefetch_related('_path').filter(
_path__is_active=True
_path__is_complete=True
)
connected_interfaces = Interface.objects.restrict(request.user, 'view').prefetch_related('_path').filter(
_path__is_active=True
_path__is_complete=True
)
def build_stats():