From ca13045515b8038b3d15054962c853c5cd268188 Mon Sep 17 00:00:00 2001 From: Jeremy Stretch Date: Fri, 10 Jan 2020 14:22:22 -0500 Subject: [PATCH] Closes #3092: Split DCIM models into separate files for easier management --- netbox/dcim/{models.py => models/__init__.py} | 1415 +---------------- .../dcim/models/device_component_templates.py | 400 +++++ netbox/dcim/models/device_components.py | 1019 ++++++++++++ netbox/dcim/tests/test_models.py | 5 +- 4 files changed, 1471 insertions(+), 1368 deletions(-) rename netbox/dcim/{models.py => models/__init__.py} (63%) create mode 100644 netbox/dcim/models/device_component_templates.py create mode 100644 netbox/dcim/models/device_components.py diff --git a/netbox/dcim/models.py b/netbox/dcim/models/__init__.py similarity index 63% rename from netbox/dcim/models.py rename to netbox/dcim/models/__init__.py index c8ee9fa8a..ab1cf1701 100644 --- a/netbox/dcim/models.py +++ b/netbox/dcim/models/__init__.py @@ -17,173 +17,59 @@ from mptt.models import MPTTModel, TreeForeignKey from taggit.managers import TaggableManager from timezone_field import TimeZoneField -from extras.models import ConfigContextModel, CustomFieldModel, ObjectChange, TaggedItem +from dcim.choices import * +from dcim.constants import * +from dcim.fields import ASNField +from extras.models import ConfigContextModel, CustomFieldModel, TaggedItem from utilities.fields import ColorField from utilities.managers import NaturalOrderingManager from utilities.models import ChangeLoggedModel -from utilities.utils import foreground_color, serialize_object, to_meters -from virtualization.choices import VMInterfaceTypeChoices - -from .choices import * -from .constants import * -from .exceptions import LoopDetected -from .fields import ASNField, MACAddressField -from .managers import InterfaceManager +from utilities.utils import foreground_color, to_meters +from .device_components import ( + CableTermination, ConsolePort, ConsoleServerPort, DeviceBay, FrontPort, Interface, InventoryItem, PowerOutlet, + PowerPort, RearPort, +) +from .device_component_templates import ( + ConsolePortTemplate, ConsoleServerPortTemplate, DeviceBayTemplate, FrontPortTemplate, InterfaceTemplate, + PowerOutletTemplate, PowerPortTemplate, RearPortTemplate, +) -class ComponentTemplateModel(models.Model): - - class Meta: - abstract = True - - def instantiate(self, device): - """ - Instantiate a new component on the specified Device. - """ - raise NotImplementedError() - - def to_objectchange(self, action): - return ObjectChange( - changed_object=self, - object_repr=str(self), - action=action, - related_object=self.device_type, - object_data=serialize_object(self) - ) - - -class ComponentModel(models.Model): - description = models.CharField( - max_length=100, - blank=True - ) - - class Meta: - abstract = True - - def to_objectchange(self, action): - # Annotate the parent Device/VM - try: - parent = getattr(self, 'device', None) or getattr(self, 'virtual_machine', None) - except ObjectDoesNotExist: - # The parent device/VM has already been deleted - parent = None - - return ObjectChange( - changed_object=self, - object_repr=str(self), - action=action, - related_object=parent, - object_data=serialize_object(self) - ) - - @property - def parent(self): - return getattr(self, 'device', None) - - -class CableTermination(models.Model): - cable = models.ForeignKey( - to='dcim.Cable', - on_delete=models.SET_NULL, - related_name='+', - blank=True, - null=True - ) - - # Generic relations to Cable. These ensure that an attached Cable is deleted if the terminated object is deleted. - _cabled_as_a = GenericRelation( - to='dcim.Cable', - content_type_field='termination_a_type', - object_id_field='termination_a_id' - ) - _cabled_as_b = GenericRelation( - to='dcim.Cable', - content_type_field='termination_b_type', - object_id_field='termination_b_id' - ) - - is_path_endpoint = True - - class Meta: - abstract = True - - def trace(self, position=1, follow_circuits=False, cable_history=None): - """ - Return a list representing a complete cable path, with each individual segment represented as a three-tuple: - [ - (termination A, cable, termination B), - (termination C, cable, termination D), - (termination E, cable, termination F) - ] - """ - def get_peer_port(termination, position=1, follow_circuits=False): - from circuits.models import CircuitTermination - - # Map a front port to its corresponding rear port - if isinstance(termination, FrontPort): - return termination.rear_port, termination.rear_port_position - - # Map a rear port/position to its 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 - )) - try: - peer_port = FrontPort.objects.get( - rear_port=termination, - rear_port_position=position, - ) - return peer_port, 1 - except ObjectDoesNotExist: - return None, None - - # Follow a circuit to its other termination - elif isinstance(termination, CircuitTermination) and follow_circuits: - peer_termination = termination.get_peer_termination() - if peer_termination is None: - return None, None - return peer_termination, position - - # Termination is not a pass-through port - else: - return None, None - - if not self.cable: - return [(self, None, None)] - - # Record cable history to detect loops - if cable_history is None: - cable_history = [] - elif self.cable in cable_history: - raise LoopDetected() - cable_history.append(self.cable) - - 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, follow_circuits) - if peer_port is None: - return path - - try: - next_segment = peer_port.trace(position, follow_circuits, cable_history) - except LoopDetected: - return path - - if next_segment is None: - return path + [(peer_port, None, None)] - - return path + next_segment - - def get_cable_peer(self): - if self.cable is None: - return None - if self._cabled_as_a.exists(): - return self.cable.termination_b - if self._cabled_as_b.exists(): - return self.cable.termination_a +__all__ = ( + 'Cable', + 'CableTermination', + 'ConsolePort', + 'ConsolePortTemplate', + 'ConsoleServerPort', + 'ConsoleServerPortTemplate', + 'Device', + 'DeviceBay', + 'DeviceBayTemplate', + 'DeviceRole', + 'DeviceType', + 'FrontPort', + 'FrontPortTemplate', + 'Interface', + 'InterfaceTemplate', + 'InventoryItem', + 'Manufacturer', + 'Platform', + 'PowerFeed', + 'PowerOutlet', + 'PowerOutletTemplate', + 'PowerPanel', + 'PowerPort', + 'PowerPortTemplate', + 'Rack', + 'RackGroup', + 'RackReservation', + 'RackRole', + 'RearPort', + 'RearPortTemplate', + 'Region', + 'Site', + 'VirtualChassis', +) # @@ -1171,360 +1057,6 @@ class DeviceType(ChangeLoggedModel, CustomFieldModel): return self.subdevice_role == SubdeviceRoleChoices.ROLE_CHILD -class ConsolePortTemplate(ComponentTemplateModel): - """ - A template for a ConsolePort to be created for a new Device. - """ - device_type = models.ForeignKey( - to='dcim.DeviceType', - on_delete=models.CASCADE, - related_name='consoleport_templates' - ) - name = models.CharField( - max_length=50 - ) - type = models.CharField( - max_length=50, - choices=ConsolePortTypeChoices, - blank=True - ) - - objects = NaturalOrderingManager() - - class Meta: - ordering = ['device_type', 'name'] - unique_together = ['device_type', 'name'] - - def __str__(self): - return self.name - - def instantiate(self, device): - return ConsolePort( - device=device, - name=self.name, - type=self.type - ) - - -class ConsoleServerPortTemplate(ComponentTemplateModel): - """ - A template for a ConsoleServerPort to be created for a new Device. - """ - device_type = models.ForeignKey( - to='dcim.DeviceType', - on_delete=models.CASCADE, - related_name='consoleserverport_templates' - ) - name = models.CharField( - max_length=50 - ) - type = models.CharField( - max_length=50, - choices=ConsolePortTypeChoices, - blank=True - ) - - objects = NaturalOrderingManager() - - class Meta: - ordering = ['device_type', 'name'] - unique_together = ['device_type', 'name'] - - def __str__(self): - return self.name - - def instantiate(self, device): - return ConsoleServerPort( - device=device, - name=self.name, - type=self.type - ) - - -class PowerPortTemplate(ComponentTemplateModel): - """ - A template for a PowerPort to be created for a new Device. - """ - device_type = models.ForeignKey( - to='dcim.DeviceType', - on_delete=models.CASCADE, - related_name='powerport_templates' - ) - name = models.CharField( - max_length=50 - ) - type = models.CharField( - max_length=50, - choices=PowerPortTypeChoices, - blank=True - ) - maximum_draw = models.PositiveSmallIntegerField( - blank=True, - null=True, - validators=[MinValueValidator(1)], - help_text="Maximum power draw (watts)" - ) - allocated_draw = models.PositiveSmallIntegerField( - blank=True, - null=True, - validators=[MinValueValidator(1)], - help_text="Allocated power draw (watts)" - ) - - objects = NaturalOrderingManager() - - class Meta: - ordering = ['device_type', 'name'] - unique_together = ['device_type', 'name'] - - def __str__(self): - return self.name - - def instantiate(self, device): - return PowerPort( - device=device, - name=self.name, - maximum_draw=self.maximum_draw, - allocated_draw=self.allocated_draw - ) - - -class PowerOutletTemplate(ComponentTemplateModel): - """ - A template for a PowerOutlet to be created for a new Device. - """ - device_type = models.ForeignKey( - to='dcim.DeviceType', - on_delete=models.CASCADE, - related_name='poweroutlet_templates' - ) - name = models.CharField( - max_length=50 - ) - type = models.CharField( - max_length=50, - choices=PowerOutletTypeChoices, - blank=True - ) - power_port = models.ForeignKey( - to='dcim.PowerPortTemplate', - on_delete=models.SET_NULL, - blank=True, - null=True, - related_name='poweroutlet_templates' - ) - feed_leg = models.CharField( - max_length=50, - choices=PowerOutletFeedLegChoices, - blank=True, - help_text="Phase (for three-phase feeds)" - ) - - objects = NaturalOrderingManager() - - class Meta: - ordering = ['device_type', 'name'] - unique_together = ['device_type', 'name'] - - def __str__(self): - return self.name - - def clean(self): - - # Validate power port assignment - if self.power_port and self.power_port.device_type != self.device_type: - raise ValidationError( - "Parent power port ({}) must belong to the same device type".format(self.power_port) - ) - - def instantiate(self, device): - if self.power_port: - power_port = PowerPort.objects.get(device=device, name=self.power_port.name) - else: - power_port = None - return PowerOutlet( - device=device, - name=self.name, - power_port=power_port, - feed_leg=self.feed_leg - ) - - -class InterfaceTemplate(ComponentTemplateModel): - """ - A template for a physical data interface on a new Device. - """ - device_type = models.ForeignKey( - to='dcim.DeviceType', - on_delete=models.CASCADE, - related_name='interface_templates' - ) - name = models.CharField( - max_length=64 - ) - type = models.CharField( - max_length=50, - choices=InterfaceTypeChoices - ) - mgmt_only = models.BooleanField( - default=False, - verbose_name='Management only' - ) - - objects = InterfaceManager() - - class Meta: - ordering = ['device_type', 'name'] - unique_together = ['device_type', 'name'] - - def __str__(self): - return self.name - - def instantiate(self, device): - return Interface( - device=device, - name=self.name, - type=self.type, - mgmt_only=self.mgmt_only - ) - - -class FrontPortTemplate(ComponentTemplateModel): - """ - Template for a pass-through port on the front of a new Device. - """ - device_type = models.ForeignKey( - to='dcim.DeviceType', - on_delete=models.CASCADE, - related_name='frontport_templates' - ) - name = models.CharField( - max_length=64 - ) - type = models.CharField( - max_length=50, - choices=PortTypeChoices - ) - rear_port = models.ForeignKey( - to='dcim.RearPortTemplate', - on_delete=models.CASCADE, - related_name='frontport_templates' - ) - rear_port_position = models.PositiveSmallIntegerField( - default=1, - validators=[MinValueValidator(1), MaxValueValidator(64)] - ) - - objects = NaturalOrderingManager() - - class Meta: - ordering = ['device_type', 'name'] - unique_together = [ - ['device_type', 'name'], - ['rear_port', 'rear_port_position'], - ] - - def __str__(self): - return self.name - - def clean(self): - - # Validate rear port assignment - if self.rear_port.device_type != self.device_type: - raise ValidationError( - "Rear port ({}) must belong to the same device type".format(self.rear_port) - ) - - # Validate rear port position assignment - if self.rear_port_position > self.rear_port.positions: - raise ValidationError( - "Invalid rear port position ({}); rear port {} has only {} positions".format( - self.rear_port_position, self.rear_port.name, self.rear_port.positions - ) - ) - - def instantiate(self, device): - if self.rear_port: - rear_port = RearPort.objects.get(device=device, name=self.rear_port.name) - else: - rear_port = None - return FrontPort( - device=device, - name=self.name, - type=self.type, - rear_port=rear_port, - rear_port_position=self.rear_port_position - ) - - -class RearPortTemplate(ComponentTemplateModel): - """ - Template for a pass-through port on the rear of a new Device. - """ - device_type = models.ForeignKey( - to='dcim.DeviceType', - on_delete=models.CASCADE, - related_name='rearport_templates' - ) - name = models.CharField( - max_length=64 - ) - type = models.CharField( - max_length=50, - choices=PortTypeChoices - ) - positions = models.PositiveSmallIntegerField( - default=1, - validators=[MinValueValidator(1), MaxValueValidator(64)] - ) - - objects = NaturalOrderingManager() - - class Meta: - ordering = ['device_type', 'name'] - unique_together = ['device_type', 'name'] - - def __str__(self): - return self.name - - def instantiate(self, device): - return RearPort( - device=device, - name=self.name, - type=self.type, - positions=self.positions - ) - - -class DeviceBayTemplate(ComponentTemplateModel): - """ - A template for a DeviceBay to be created for a new parent Device. - """ - device_type = models.ForeignKey( - to='dcim.DeviceType', - on_delete=models.CASCADE, - related_name='device_bay_templates' - ) - name = models.CharField( - max_length=50 - ) - - objects = NaturalOrderingManager() - - class Meta: - ordering = ['device_type', 'name'] - unique_together = ['device_type', 'name'] - - def __str__(self): - return self.name - - def instantiate(self, device): - return DeviceBay( - device=device, - name=self.name - ) - - # # Devices # @@ -2053,857 +1585,6 @@ class Device(ChangeLoggedModel, ConfigContextModel, CustomFieldModel): return self.STATUS_CLASS_MAP.get(self.status) -# -# Console ports -# - -class ConsolePort(CableTermination, ComponentModel): - """ - A physical console port within a Device. ConsolePorts connect to ConsoleServerPorts. - """ - device = models.ForeignKey( - to='dcim.Device', - on_delete=models.CASCADE, - related_name='consoleports' - ) - name = models.CharField( - max_length=50 - ) - type = models.CharField( - max_length=50, - choices=ConsolePortTypeChoices, - blank=True - ) - connected_endpoint = models.OneToOneField( - to='dcim.ConsoleServerPort', - on_delete=models.SET_NULL, - related_name='connected_endpoint', - blank=True, - null=True - ) - connection_status = models.NullBooleanField( - choices=CONNECTION_STATUS_CHOICES, - blank=True - ) - - objects = NaturalOrderingManager() - tags = TaggableManager(through=TaggedItem) - - csv_headers = ['device', 'name', 'type', 'description'] - - class Meta: - ordering = ['device', 'name'] - unique_together = ['device', 'name'] - - def __str__(self): - return self.name - - def get_absolute_url(self): - return self.device.get_absolute_url() - - def to_csv(self): - return ( - self.device.identifier, - self.name, - self.type, - self.description, - ) - - -# -# Console server ports -# - -class ConsoleServerPort(CableTermination, ComponentModel): - """ - A physical port within a Device (typically a designated console server) which provides access to ConsolePorts. - """ - device = models.ForeignKey( - to='dcim.Device', - on_delete=models.CASCADE, - related_name='consoleserverports' - ) - name = models.CharField( - max_length=50 - ) - type = models.CharField( - max_length=50, - choices=ConsolePortTypeChoices, - blank=True - ) - connection_status = models.NullBooleanField( - choices=CONNECTION_STATUS_CHOICES, - blank=True - ) - - objects = NaturalOrderingManager() - tags = TaggableManager(through=TaggedItem) - - csv_headers = ['device', 'name', 'type', 'description'] - - class Meta: - unique_together = ['device', 'name'] - - def __str__(self): - return self.name - - def get_absolute_url(self): - return self.device.get_absolute_url() - - def to_csv(self): - return ( - self.device.identifier, - self.name, - self.type, - self.description, - ) - - -# -# Power ports -# - -class PowerPort(CableTermination, ComponentModel): - """ - A physical power supply (intake) port within a Device. PowerPorts connect to PowerOutlets. - """ - device = models.ForeignKey( - to='dcim.Device', - on_delete=models.CASCADE, - related_name='powerports' - ) - name = models.CharField( - max_length=50 - ) - type = models.CharField( - max_length=50, - choices=PowerPortTypeChoices, - blank=True - ) - maximum_draw = models.PositiveSmallIntegerField( - blank=True, - null=True, - validators=[MinValueValidator(1)], - help_text="Maximum power draw (watts)" - ) - allocated_draw = models.PositiveSmallIntegerField( - blank=True, - null=True, - validators=[MinValueValidator(1)], - help_text="Allocated power draw (watts)" - ) - _connected_poweroutlet = models.OneToOneField( - to='dcim.PowerOutlet', - on_delete=models.SET_NULL, - related_name='connected_endpoint', - blank=True, - null=True - ) - _connected_powerfeed = models.OneToOneField( - to='dcim.PowerFeed', - on_delete=models.SET_NULL, - related_name='+', - blank=True, - null=True - ) - connection_status = models.NullBooleanField( - choices=CONNECTION_STATUS_CHOICES, - blank=True - ) - - objects = NaturalOrderingManager() - tags = TaggableManager(through=TaggedItem) - - csv_headers = ['device', 'name', 'type', 'maximum_draw', 'allocated_draw', 'description'] - - class Meta: - ordering = ['device', 'name'] - unique_together = ['device', 'name'] - - def __str__(self): - return self.name - - def get_absolute_url(self): - return self.device.get_absolute_url() - - def to_csv(self): - return ( - self.device.identifier, - self.name, - self.get_type_display(), - self.maximum_draw, - self.allocated_draw, - self.description, - ) - - @property - def connected_endpoint(self): - if self._connected_poweroutlet: - return self._connected_poweroutlet - return self._connected_powerfeed - - @connected_endpoint.setter - def connected_endpoint(self, value): - if value is None: - self._connected_poweroutlet = None - self._connected_powerfeed = None - elif isinstance(value, PowerOutlet): - self._connected_poweroutlet = value - self._connected_powerfeed = None - elif isinstance(value, PowerFeed): - self._connected_poweroutlet = None - self._connected_powerfeed = value - else: - raise ValueError( - "Connected endpoint must be a PowerOutlet or PowerFeed, not {}.".format(type(value)) - ) - - def get_power_draw(self): - """ - Return the allocated and maximum power draw (in VA) and child PowerOutlet count for this PowerPort. - """ - # Calculate aggregate draw of all child power outlets if no numbers have been defined manually - if self.allocated_draw is None and self.maximum_draw is None: - outlet_ids = PowerOutlet.objects.filter(power_port=self).values_list('pk', flat=True) - utilization = PowerPort.objects.filter(_connected_poweroutlet_id__in=outlet_ids).aggregate( - maximum_draw_total=Sum('maximum_draw'), - allocated_draw_total=Sum('allocated_draw'), - ) - ret = { - 'allocated': utilization['allocated_draw_total'] or 0, - 'maximum': utilization['maximum_draw_total'] or 0, - 'outlet_count': len(outlet_ids), - 'legs': [], - } - - # Calculate per-leg aggregates for three-phase feeds - if self._connected_powerfeed and self._connected_powerfeed.phase == PowerFeedPhaseChoices.PHASE_3PHASE: - for leg, leg_name in PowerOutletFeedLegChoices: - outlet_ids = PowerOutlet.objects.filter(power_port=self, feed_leg=leg).values_list('pk', flat=True) - utilization = PowerPort.objects.filter(_connected_poweroutlet_id__in=outlet_ids).aggregate( - maximum_draw_total=Sum('maximum_draw'), - allocated_draw_total=Sum('allocated_draw'), - ) - ret['legs'].append({ - 'name': leg_name, - 'allocated': utilization['allocated_draw_total'] or 0, - 'maximum': utilization['maximum_draw_total'] or 0, - 'outlet_count': len(outlet_ids), - }) - - return ret - - # Default to administratively defined values - return { - 'allocated': self.allocated_draw or 0, - 'maximum': self.maximum_draw or 0, - 'outlet_count': PowerOutlet.objects.filter(power_port=self).count(), - 'legs': [], - } - - -# -# Power outlets -# - -class PowerOutlet(CableTermination, ComponentModel): - """ - A physical power outlet (output) within a Device which provides power to a PowerPort. - """ - device = models.ForeignKey( - to='dcim.Device', - on_delete=models.CASCADE, - related_name='poweroutlets' - ) - name = models.CharField( - max_length=50 - ) - type = models.CharField( - max_length=50, - choices=PowerOutletTypeChoices, - blank=True - ) - power_port = models.ForeignKey( - to='dcim.PowerPort', - on_delete=models.SET_NULL, - blank=True, - null=True, - related_name='poweroutlets' - ) - feed_leg = models.CharField( - max_length=50, - choices=PowerOutletFeedLegChoices, - blank=True, - help_text="Phase (for three-phase feeds)" - ) - connection_status = models.NullBooleanField( - choices=CONNECTION_STATUS_CHOICES, - blank=True - ) - - objects = NaturalOrderingManager() - tags = TaggableManager(through=TaggedItem) - - csv_headers = ['device', 'name', 'type', 'power_port', 'feed_leg', 'description'] - - class Meta: - unique_together = ['device', 'name'] - - def __str__(self): - return self.name - - def get_absolute_url(self): - return self.device.get_absolute_url() - - def to_csv(self): - return ( - self.device.identifier, - self.name, - self.get_type_display(), - self.power_port.name if self.power_port else None, - self.get_feed_leg_display(), - self.description, - ) - - def clean(self): - - # Validate power port assignment - if self.power_port and self.power_port.device != self.device: - raise ValidationError( - "Parent power port ({}) must belong to the same device".format(self.power_port) - ) - - -# -# Interfaces -# - -class Interface(CableTermination, ComponentModel): - """ - A network interface within a Device or VirtualMachine. A physical Interface can connect to exactly one other - Interface. - """ - device = models.ForeignKey( - to='Device', - on_delete=models.CASCADE, - related_name='interfaces', - null=True, - blank=True - ) - virtual_machine = models.ForeignKey( - to='virtualization.VirtualMachine', - on_delete=models.CASCADE, - related_name='interfaces', - null=True, - blank=True - ) - name = models.CharField( - max_length=64 - ) - _connected_interface = models.OneToOneField( - to='self', - on_delete=models.SET_NULL, - related_name='+', - blank=True, - null=True - ) - _connected_circuittermination = models.OneToOneField( - to='circuits.CircuitTermination', - on_delete=models.SET_NULL, - related_name='+', - blank=True, - null=True - ) - connection_status = models.NullBooleanField( - choices=CONNECTION_STATUS_CHOICES, - blank=True - ) - lag = models.ForeignKey( - to='self', - on_delete=models.SET_NULL, - related_name='member_interfaces', - null=True, - blank=True, - verbose_name='Parent LAG' - ) - type = models.CharField( - max_length=50, - choices=InterfaceTypeChoices - ) - enabled = models.BooleanField( - default=True - ) - mac_address = MACAddressField( - null=True, - blank=True, - verbose_name='MAC Address' - ) - mtu = models.PositiveIntegerField( - blank=True, - null=True, - validators=[MinValueValidator(1), MaxValueValidator(65536)], - verbose_name='MTU' - ) - mgmt_only = models.BooleanField( - default=False, - verbose_name='OOB Management', - help_text='This interface is used only for out-of-band management' - ) - mode = models.CharField( - max_length=50, - choices=InterfaceModeChoices, - blank=True, - ) - untagged_vlan = models.ForeignKey( - to='ipam.VLAN', - on_delete=models.SET_NULL, - related_name='interfaces_as_untagged', - null=True, - blank=True, - verbose_name='Untagged VLAN' - ) - tagged_vlans = models.ManyToManyField( - to='ipam.VLAN', - related_name='interfaces_as_tagged', - blank=True, - verbose_name='Tagged VLANs' - ) - - objects = InterfaceManager() - tags = TaggableManager(through=TaggedItem) - - csv_headers = [ - 'device', 'virtual_machine', 'name', 'lag', 'type', 'enabled', 'mac_address', 'mtu', 'mgmt_only', - 'description', 'mode', - ] - - class Meta: - ordering = ['device', 'name'] - unique_together = ['device', 'name'] - - def __str__(self): - return self.name - - def get_absolute_url(self): - return reverse('dcim:interface', kwargs={'pk': self.pk}) - - def to_csv(self): - return ( - self.device.identifier if self.device else None, - self.virtual_machine.name if self.virtual_machine else None, - self.name, - self.lag.name if self.lag else None, - self.get_type_display(), - self.enabled, - self.mac_address, - self.mtu, - self.mgmt_only, - self.description, - self.get_mode_display(), - ) - - def clean(self): - - # An Interface must belong to a Device *or* to a VirtualMachine - if self.device and self.virtual_machine: - raise ValidationError("An interface cannot belong to both a device and a virtual machine.") - if not self.device and not self.virtual_machine: - raise ValidationError("An interface must belong to either a device or a virtual machine.") - - # VM interfaces must be virtual - if self.virtual_machine and self.type not in VMInterfaceTypeChoices.values(): - raise ValidationError({ - 'type': "Invalid interface type for a virtual machine: {}".format(self.type) - }) - - # Virtual interfaces cannot be connected - if self.type in NONCONNECTABLE_IFACE_TYPES and ( - self.cable or getattr(self, 'circuit_termination', False) - ): - raise ValidationError({ - 'type': "Virtual and wireless interfaces cannot be connected to another interface or circuit. " - "Disconnect the interface or choose a suitable type." - }) - - # An interface's LAG must belong to the same device (or VC master) - if self.lag and self.lag.device not in [self.device, self.device.get_vc_master()]: - raise ValidationError({ - 'lag': "The selected LAG interface ({}) belongs to a different device ({}).".format( - self.lag.name, self.lag.device.name - ) - }) - - # A virtual interface cannot have a parent LAG - if self.type in NONCONNECTABLE_IFACE_TYPES and self.lag is not None: - raise ValidationError({ - 'lag': "{} interfaces cannot have a parent LAG interface.".format(self.get_type_display()) - }) - - # Only a LAG can have LAG members - if self.type != InterfaceTypeChoices.TYPE_LAG and self.member_interfaces.exists(): - raise ValidationError({ - 'type': "Cannot change interface type; it has LAG members ({}).".format( - ", ".join([iface.name for iface in self.member_interfaces.all()]) - ) - }) - - # Validate untagged VLAN - if self.untagged_vlan and self.untagged_vlan.site not in [self.parent.site, None]: - raise ValidationError({ - 'untagged_vlan': "The untagged VLAN ({}) must belong to the same site as the interface's parent " - "device/VM, or it must be global".format(self.untagged_vlan) - }) - - def save(self, *args, **kwargs): - - # Remove untagged VLAN assignment for non-802.1Q interfaces - if self.mode is None: - self.untagged_vlan = None - - # Only "tagged" interfaces may have tagged VLANs assigned. ("tagged all" implies all VLANs are assigned.) - if self.pk and self.mode is not InterfaceModeChoices.MODE_TAGGED: - self.tagged_vlans.clear() - - return super().save(*args, **kwargs) - - def to_objectchange(self, action): - # Annotate the parent Device/VM - try: - parent_obj = self.device or self.virtual_machine - except ObjectDoesNotExist: - parent_obj = None - - return ObjectChange( - changed_object=self, - object_repr=str(self), - action=action, - related_object=parent_obj, - object_data=serialize_object(self) - ) - - @property - def connected_endpoint(self): - if self._connected_interface: - return self._connected_interface - return self._connected_circuittermination - - @connected_endpoint.setter - def connected_endpoint(self, value): - from circuits.models import CircuitTermination - - if value is None: - self._connected_interface = None - self._connected_circuittermination = None - elif isinstance(value, Interface): - self._connected_interface = value - self._connected_circuittermination = None - elif isinstance(value, CircuitTermination): - self._connected_interface = None - self._connected_circuittermination = value - else: - raise ValueError( - "Connected endpoint must be an Interface or CircuitTermination, not {}.".format(type(value)) - ) - - @property - def parent(self): - return self.device or self.virtual_machine - - @property - def is_connectable(self): - return self.type not in NONCONNECTABLE_IFACE_TYPES - - @property - def is_virtual(self): - return self.type in VIRTUAL_IFACE_TYPES - - @property - def is_wireless(self): - return self.type in WIRELESS_IFACE_TYPES - - @property - def is_lag(self): - return self.type == InterfaceTypeChoices.TYPE_LAG - - @property - def count_ipaddresses(self): - return self.ip_addresses.count() - - -# -# Pass-through ports -# - -class FrontPort(CableTermination, ComponentModel): - """ - A pass-through port on the front of a Device. - """ - device = models.ForeignKey( - to='dcim.Device', - on_delete=models.CASCADE, - related_name='frontports' - ) - name = models.CharField( - max_length=64 - ) - type = models.CharField( - max_length=50, - choices=PortTypeChoices - ) - rear_port = models.ForeignKey( - to='dcim.RearPort', - on_delete=models.CASCADE, - related_name='frontports' - ) - rear_port_position = models.PositiveSmallIntegerField( - default=1, - validators=[MinValueValidator(1), MaxValueValidator(64)] - ) - - is_path_endpoint = False - - objects = NaturalOrderingManager() - tags = TaggableManager(through=TaggedItem) - - csv_headers = ['device', 'name', 'type', 'rear_port', 'rear_port_position', 'description'] - - class Meta: - ordering = ['device', 'name'] - unique_together = [ - ['device', 'name'], - ['rear_port', 'rear_port_position'], - ] - - def __str__(self): - return self.name - - def to_csv(self): - return ( - self.device.identifier, - self.name, - self.get_type_display(), - self.rear_port.name, - self.rear_port_position, - self.description, - ) - - def clean(self): - - # Validate rear port assignment - if self.rear_port.device != self.device: - raise ValidationError( - "Rear port ({}) must belong to the same device".format(self.rear_port) - ) - - # Validate rear port position assignment - if self.rear_port_position > self.rear_port.positions: - raise ValidationError( - "Invalid rear port position ({}); rear port {} has only {} positions".format( - self.rear_port_position, self.rear_port.name, self.rear_port.positions - ) - ) - - -class RearPort(CableTermination, ComponentModel): - """ - A pass-through port on the rear of a Device. - """ - device = models.ForeignKey( - to='dcim.Device', - on_delete=models.CASCADE, - related_name='rearports' - ) - name = models.CharField( - max_length=64 - ) - type = models.CharField( - max_length=50, - choices=PortTypeChoices - ) - positions = models.PositiveSmallIntegerField( - default=1, - validators=[MinValueValidator(1), MaxValueValidator(64)] - ) - - is_path_endpoint = False - - objects = NaturalOrderingManager() - tags = TaggableManager(through=TaggedItem) - - csv_headers = ['device', 'name', 'type', 'positions', 'description'] - - class Meta: - ordering = ['device', 'name'] - unique_together = ['device', 'name'] - - def __str__(self): - return self.name - - def to_csv(self): - return ( - self.device.identifier, - self.name, - self.get_type_display(), - self.positions, - self.description, - ) - - -# -# Device bays -# - -class DeviceBay(ComponentModel): - """ - An empty space within a Device which can house a child device - """ - device = models.ForeignKey( - to='dcim.Device', - on_delete=models.CASCADE, - related_name='device_bays' - ) - name = models.CharField( - max_length=50, - verbose_name='Name' - ) - installed_device = models.OneToOneField( - to='dcim.Device', - on_delete=models.SET_NULL, - related_name='parent_bay', - blank=True, - null=True - ) - - objects = NaturalOrderingManager() - tags = TaggableManager(through=TaggedItem) - - csv_headers = ['device', 'name', 'installed_device', 'description'] - - class Meta: - ordering = ['device', 'name'] - unique_together = ['device', 'name'] - - def __str__(self): - return '{} - {}'.format(self.device.name, self.name) - - def get_absolute_url(self): - return self.device.get_absolute_url() - - def to_csv(self): - return ( - self.device.identifier, - self.name, - self.installed_device.identifier if self.installed_device else None, - self.description, - ) - - def clean(self): - - # Validate that the parent Device can have DeviceBays - if not self.device.device_type.is_parent_device: - raise ValidationError("This type of device ({}) does not support device bays.".format( - self.device.device_type - )) - - # Cannot install a device into itself, obviously - if self.device == self.installed_device: - raise ValidationError("Cannot install a device into itself.") - - # Check that the installed device is not already installed elsewhere - if self.installed_device: - current_bay = DeviceBay.objects.filter(installed_device=self.installed_device).first() - if current_bay and current_bay != self: - raise ValidationError({ - 'installed_device': "Cannot install the specified device; device is already installed in {}".format( - current_bay - ) - }) - - -# -# Inventory items -# - -class InventoryItem(ComponentModel): - """ - An InventoryItem represents a serialized piece of hardware within a Device, such as a line card or power supply. - InventoryItems are used only for inventory purposes. - """ - device = models.ForeignKey( - to='dcim.Device', - on_delete=models.CASCADE, - related_name='inventory_items' - ) - parent = models.ForeignKey( - to='self', - on_delete=models.CASCADE, - related_name='child_items', - blank=True, - null=True - ) - name = models.CharField( - max_length=50, - verbose_name='Name' - ) - manufacturer = models.ForeignKey( - to='dcim.Manufacturer', - on_delete=models.PROTECT, - related_name='inventory_items', - blank=True, - null=True - ) - part_id = models.CharField( - max_length=50, - verbose_name='Part ID', - blank=True - ) - serial = models.CharField( - max_length=50, - verbose_name='Serial number', - blank=True - ) - asset_tag = models.CharField( - max_length=50, - unique=True, - blank=True, - null=True, - verbose_name='Asset tag', - help_text='A unique tag used to identify this item' - ) - discovered = models.BooleanField( - default=False, - verbose_name='Discovered' - ) - - tags = TaggableManager(through=TaggedItem) - - csv_headers = [ - 'device', 'name', 'manufacturer', 'part_id', 'serial', 'asset_tag', 'discovered', 'description', - ] - - class Meta: - ordering = ['device__id', 'parent__id', 'name'] - unique_together = ['device', 'parent', 'name'] - - def __str__(self): - return self.name - - def get_absolute_url(self): - return self.device.get_absolute_url() - - def to_csv(self): - return ( - self.device.name or '{{{}}}'.format(self.device.pk), - self.name, - self.manufacturer.name if self.manufacturer else None, - self.part_id, - self.serial, - self.asset_tag, - self.discovered, - self.description, - ) - - # # Virtual chassis # diff --git a/netbox/dcim/models/device_component_templates.py b/netbox/dcim/models/device_component_templates.py new file mode 100644 index 000000000..2aa46d0ea --- /dev/null +++ b/netbox/dcim/models/device_component_templates.py @@ -0,0 +1,400 @@ +from django.core.exceptions import ValidationError +from django.core.validators import MaxValueValidator, MinValueValidator +from django.db import models + +from dcim.choices import * +from dcim.constants import * +from dcim.managers import InterfaceManager +from extras.models import ObjectChange +from utilities.managers import NaturalOrderingManager +from utilities.utils import serialize_object +from .device_components import ( + ConsolePort, ConsoleServerPort, DeviceBay, FrontPort, Interface, PowerOutlet, PowerPort, RearPort, +) + + +__all__ = ( + 'ConsolePortTemplate', + 'ConsoleServerPortTemplate', + 'DeviceBayTemplate', + 'FrontPortTemplate', + 'InterfaceTemplate', + 'PowerOutletTemplate', + 'PowerPortTemplate', + 'RearPortTemplate', +) + + +class ComponentTemplateModel(models.Model): + + class Meta: + abstract = True + + def instantiate(self, device): + """ + Instantiate a new component on the specified Device. + """ + raise NotImplementedError() + + def to_objectchange(self, action): + return ObjectChange( + changed_object=self, + object_repr=str(self), + action=action, + related_object=self.device_type, + object_data=serialize_object(self) + ) + + +class ConsolePortTemplate(ComponentTemplateModel): + """ + A template for a ConsolePort to be created for a new Device. + """ + device_type = models.ForeignKey( + to='dcim.DeviceType', + on_delete=models.CASCADE, + related_name='consoleport_templates' + ) + name = models.CharField( + max_length=50 + ) + type = models.CharField( + max_length=50, + choices=ConsolePortTypeChoices, + blank=True + ) + + objects = NaturalOrderingManager() + + class Meta: + ordering = ['device_type', 'name'] + unique_together = ['device_type', 'name'] + + def __str__(self): + return self.name + + def instantiate(self, device): + return ConsolePort( + device=device, + name=self.name, + type=self.type + ) + + +class ConsoleServerPortTemplate(ComponentTemplateModel): + """ + A template for a ConsoleServerPort to be created for a new Device. + """ + device_type = models.ForeignKey( + to='dcim.DeviceType', + on_delete=models.CASCADE, + related_name='consoleserverport_templates' + ) + name = models.CharField( + max_length=50 + ) + type = models.CharField( + max_length=50, + choices=ConsolePortTypeChoices, + blank=True + ) + + objects = NaturalOrderingManager() + + class Meta: + ordering = ['device_type', 'name'] + unique_together = ['device_type', 'name'] + + def __str__(self): + return self.name + + def instantiate(self, device): + return ConsoleServerPort( + device=device, + name=self.name, + type=self.type + ) + + +class PowerPortTemplate(ComponentTemplateModel): + """ + A template for a PowerPort to be created for a new Device. + """ + device_type = models.ForeignKey( + to='dcim.DeviceType', + on_delete=models.CASCADE, + related_name='powerport_templates' + ) + name = models.CharField( + max_length=50 + ) + type = models.CharField( + max_length=50, + choices=PowerPortTypeChoices, + blank=True + ) + maximum_draw = models.PositiveSmallIntegerField( + blank=True, + null=True, + validators=[MinValueValidator(1)], + help_text="Maximum power draw (watts)" + ) + allocated_draw = models.PositiveSmallIntegerField( + blank=True, + null=True, + validators=[MinValueValidator(1)], + help_text="Allocated power draw (watts)" + ) + + objects = NaturalOrderingManager() + + class Meta: + ordering = ['device_type', 'name'] + unique_together = ['device_type', 'name'] + + def __str__(self): + return self.name + + def instantiate(self, device): + return PowerPort( + device=device, + name=self.name, + maximum_draw=self.maximum_draw, + allocated_draw=self.allocated_draw + ) + + +class PowerOutletTemplate(ComponentTemplateModel): + """ + A template for a PowerOutlet to be created for a new Device. + """ + device_type = models.ForeignKey( + to='dcim.DeviceType', + on_delete=models.CASCADE, + related_name='poweroutlet_templates' + ) + name = models.CharField( + max_length=50 + ) + type = models.CharField( + max_length=50, + choices=PowerOutletTypeChoices, + blank=True + ) + power_port = models.ForeignKey( + to='dcim.PowerPortTemplate', + on_delete=models.SET_NULL, + blank=True, + null=True, + related_name='poweroutlet_templates' + ) + feed_leg = models.CharField( + max_length=50, + choices=PowerOutletFeedLegChoices, + blank=True, + help_text="Phase (for three-phase feeds)" + ) + + objects = NaturalOrderingManager() + + class Meta: + ordering = ['device_type', 'name'] + unique_together = ['device_type', 'name'] + + def __str__(self): + return self.name + + def clean(self): + + # Validate power port assignment + if self.power_port and self.power_port.device_type != self.device_type: + raise ValidationError( + "Parent power port ({}) must belong to the same device type".format(self.power_port) + ) + + def instantiate(self, device): + if self.power_port: + power_port = PowerPort.objects.get(device=device, name=self.power_port.name) + else: + power_port = None + return PowerOutlet( + device=device, + name=self.name, + power_port=power_port, + feed_leg=self.feed_leg + ) + + +class InterfaceTemplate(ComponentTemplateModel): + """ + A template for a physical data interface on a new Device. + """ + device_type = models.ForeignKey( + to='dcim.DeviceType', + on_delete=models.CASCADE, + related_name='interface_templates' + ) + name = models.CharField( + max_length=64 + ) + type = models.CharField( + max_length=50, + choices=InterfaceTypeChoices + ) + mgmt_only = models.BooleanField( + default=False, + verbose_name='Management only' + ) + + objects = InterfaceManager() + + class Meta: + ordering = ['device_type', 'name'] + unique_together = ['device_type', 'name'] + + def __str__(self): + return self.name + + def instantiate(self, device): + return Interface( + device=device, + name=self.name, + type=self.type, + mgmt_only=self.mgmt_only + ) + + +class FrontPortTemplate(ComponentTemplateModel): + """ + Template for a pass-through port on the front of a new Device. + """ + device_type = models.ForeignKey( + to='dcim.DeviceType', + on_delete=models.CASCADE, + related_name='frontport_templates' + ) + name = models.CharField( + max_length=64 + ) + type = models.CharField( + max_length=50, + choices=PortTypeChoices + ) + rear_port = models.ForeignKey( + to='dcim.RearPortTemplate', + on_delete=models.CASCADE, + related_name='frontport_templates' + ) + rear_port_position = models.PositiveSmallIntegerField( + default=1, + validators=[MinValueValidator(1), MaxValueValidator(64)] + ) + + objects = NaturalOrderingManager() + + class Meta: + ordering = ['device_type', 'name'] + unique_together = [ + ['device_type', 'name'], + ['rear_port', 'rear_port_position'], + ] + + def __str__(self): + return self.name + + def clean(self): + + # Validate rear port assignment + if self.rear_port.device_type != self.device_type: + raise ValidationError( + "Rear port ({}) must belong to the same device type".format(self.rear_port) + ) + + # Validate rear port position assignment + if self.rear_port_position > self.rear_port.positions: + raise ValidationError( + "Invalid rear port position ({}); rear port {} has only {} positions".format( + self.rear_port_position, self.rear_port.name, self.rear_port.positions + ) + ) + + def instantiate(self, device): + if self.rear_port: + rear_port = RearPort.objects.get(device=device, name=self.rear_port.name) + else: + rear_port = None + return FrontPort( + device=device, + name=self.name, + type=self.type, + rear_port=rear_port, + rear_port_position=self.rear_port_position + ) + + +class RearPortTemplate(ComponentTemplateModel): + """ + Template for a pass-through port on the rear of a new Device. + """ + device_type = models.ForeignKey( + to='dcim.DeviceType', + on_delete=models.CASCADE, + related_name='rearport_templates' + ) + name = models.CharField( + max_length=64 + ) + type = models.CharField( + max_length=50, + choices=PortTypeChoices + ) + positions = models.PositiveSmallIntegerField( + default=1, + validators=[MinValueValidator(1), MaxValueValidator(64)] + ) + + objects = NaturalOrderingManager() + + class Meta: + ordering = ['device_type', 'name'] + unique_together = ['device_type', 'name'] + + def __str__(self): + return self.name + + def instantiate(self, device): + return RearPort( + device=device, + name=self.name, + type=self.type, + positions=self.positions + ) + + +class DeviceBayTemplate(ComponentTemplateModel): + """ + A template for a DeviceBay to be created for a new parent Device. + """ + device_type = models.ForeignKey( + to='dcim.DeviceType', + on_delete=models.CASCADE, + related_name='device_bay_templates' + ) + name = models.CharField( + max_length=50 + ) + + objects = NaturalOrderingManager() + + class Meta: + ordering = ['device_type', 'name'] + unique_together = ['device_type', 'name'] + + def __str__(self): + return self.name + + def instantiate(self, device): + return DeviceBay( + device=device, + name=self.name + ) diff --git a/netbox/dcim/models/device_components.py b/netbox/dcim/models/device_components.py new file mode 100644 index 000000000..68bab8037 --- /dev/null +++ b/netbox/dcim/models/device_components.py @@ -0,0 +1,1019 @@ +from django.contrib.contenttypes.fields import GenericRelation +from django.core.exceptions import ObjectDoesNotExist, ValidationError +from django.core.validators import MaxValueValidator, MinValueValidator +from django.db import models +from django.db.models import Sum +from django.urls import reverse +from taggit.managers import TaggableManager + +from dcim.choices import * +from dcim.constants import * +from dcim.exceptions import LoopDetected +from dcim.fields import MACAddressField +from dcim.managers import InterfaceManager +from extras.models import ObjectChange, TaggedItem +from utilities.managers import NaturalOrderingManager +from utilities.utils import serialize_object +from virtualization.choices import VMInterfaceTypeChoices + + +__all__ = ( + 'CableTermination', + 'ConsolePort', + 'ConsoleServerPort', + 'DeviceBay', + 'FrontPort', + 'Interface', + 'InventoryItem', + 'PowerOutlet', + 'PowerPort', + 'RearPort', +) + + +class ComponentModel(models.Model): + description = models.CharField( + max_length=100, + blank=True + ) + + class Meta: + abstract = True + + def to_objectchange(self, action): + # Annotate the parent Device/VM + try: + parent = getattr(self, 'device', None) or getattr(self, 'virtual_machine', None) + except ObjectDoesNotExist: + # The parent device/VM has already been deleted + parent = None + + return ObjectChange( + changed_object=self, + object_repr=str(self), + action=action, + related_object=parent, + object_data=serialize_object(self) + ) + + @property + def parent(self): + return getattr(self, 'device', None) + + +class CableTermination(models.Model): + cable = models.ForeignKey( + to='dcim.Cable', + on_delete=models.SET_NULL, + related_name='+', + blank=True, + null=True + ) + + # Generic relations to Cable. These ensure that an attached Cable is deleted if the terminated object is deleted. + _cabled_as_a = GenericRelation( + to='dcim.Cable', + content_type_field='termination_a_type', + object_id_field='termination_a_id' + ) + _cabled_as_b = GenericRelation( + to='dcim.Cable', + content_type_field='termination_b_type', + object_id_field='termination_b_id' + ) + + is_path_endpoint = True + + class Meta: + abstract = True + + def trace(self, position=1, follow_circuits=False, cable_history=None): + """ + Return a list representing a complete cable path, with each individual segment represented as a three-tuple: + [ + (termination A, cable, termination B), + (termination C, cable, termination D), + (termination E, cable, termination F) + ] + """ + def get_peer_port(termination, position=1, follow_circuits=False): + from circuits.models import CircuitTermination + + # Map a front port to its corresponding rear port + if isinstance(termination, FrontPort): + return termination.rear_port, termination.rear_port_position + + # Map a rear port/position to its 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 + )) + try: + peer_port = FrontPort.objects.get( + rear_port=termination, + rear_port_position=position, + ) + return peer_port, 1 + except ObjectDoesNotExist: + return None, None + + # Follow a circuit to its other termination + elif isinstance(termination, CircuitTermination) and follow_circuits: + peer_termination = termination.get_peer_termination() + if peer_termination is None: + return None, None + return peer_termination, position + + # Termination is not a pass-through port + else: + return None, None + + if not self.cable: + return [(self, None, None)] + + # Record cable history to detect loops + if cable_history is None: + cable_history = [] + elif self.cable in cable_history: + raise LoopDetected() + cable_history.append(self.cable) + + 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, follow_circuits) + if peer_port is None: + return path + + try: + next_segment = peer_port.trace(position, follow_circuits, cable_history) + except LoopDetected: + return path + + if next_segment is None: + return path + [(peer_port, None, None)] + + return path + next_segment + + def get_cable_peer(self): + if self.cable is None: + return None + if self._cabled_as_a.exists(): + return self.cable.termination_b + if self._cabled_as_b.exists(): + return self.cable.termination_a + + +# +# Console ports +# + +class ConsolePort(CableTermination, ComponentModel): + """ + A physical console port within a Device. ConsolePorts connect to ConsoleServerPorts. + """ + device = models.ForeignKey( + to='dcim.Device', + on_delete=models.CASCADE, + related_name='consoleports' + ) + name = models.CharField( + max_length=50 + ) + type = models.CharField( + max_length=50, + choices=ConsolePortTypeChoices, + blank=True + ) + connected_endpoint = models.OneToOneField( + to='dcim.ConsoleServerPort', + on_delete=models.SET_NULL, + related_name='connected_endpoint', + blank=True, + null=True + ) + connection_status = models.NullBooleanField( + choices=CONNECTION_STATUS_CHOICES, + blank=True + ) + + objects = NaturalOrderingManager() + tags = TaggableManager(through=TaggedItem) + + csv_headers = ['device', 'name', 'type', 'description'] + + class Meta: + ordering = ['device', 'name'] + unique_together = ['device', 'name'] + + def __str__(self): + return self.name + + def get_absolute_url(self): + return self.device.get_absolute_url() + + def to_csv(self): + return ( + self.device.identifier, + self.name, + self.type, + self.description, + ) + + +# +# Console server ports +# + +class ConsoleServerPort(CableTermination, ComponentModel): + """ + A physical port within a Device (typically a designated console server) which provides access to ConsolePorts. + """ + device = models.ForeignKey( + to='dcim.Device', + on_delete=models.CASCADE, + related_name='consoleserverports' + ) + name = models.CharField( + max_length=50 + ) + type = models.CharField( + max_length=50, + choices=ConsolePortTypeChoices, + blank=True + ) + connection_status = models.NullBooleanField( + choices=CONNECTION_STATUS_CHOICES, + blank=True + ) + + objects = NaturalOrderingManager() + tags = TaggableManager(through=TaggedItem) + + csv_headers = ['device', 'name', 'type', 'description'] + + class Meta: + unique_together = ['device', 'name'] + + def __str__(self): + return self.name + + def get_absolute_url(self): + return self.device.get_absolute_url() + + def to_csv(self): + return ( + self.device.identifier, + self.name, + self.type, + self.description, + ) + + +# +# Power ports +# + +class PowerPort(CableTermination, ComponentModel): + """ + A physical power supply (intake) port within a Device. PowerPorts connect to PowerOutlets. + """ + device = models.ForeignKey( + to='dcim.Device', + on_delete=models.CASCADE, + related_name='powerports' + ) + name = models.CharField( + max_length=50 + ) + type = models.CharField( + max_length=50, + choices=PowerPortTypeChoices, + blank=True + ) + maximum_draw = models.PositiveSmallIntegerField( + blank=True, + null=True, + validators=[MinValueValidator(1)], + help_text="Maximum power draw (watts)" + ) + allocated_draw = models.PositiveSmallIntegerField( + blank=True, + null=True, + validators=[MinValueValidator(1)], + help_text="Allocated power draw (watts)" + ) + _connected_poweroutlet = models.OneToOneField( + to='dcim.PowerOutlet', + on_delete=models.SET_NULL, + related_name='connected_endpoint', + blank=True, + null=True + ) + _connected_powerfeed = models.OneToOneField( + to='dcim.PowerFeed', + on_delete=models.SET_NULL, + related_name='+', + blank=True, + null=True + ) + connection_status = models.NullBooleanField( + choices=CONNECTION_STATUS_CHOICES, + blank=True + ) + + objects = NaturalOrderingManager() + tags = TaggableManager(through=TaggedItem) + + csv_headers = ['device', 'name', 'type', 'maximum_draw', 'allocated_draw', 'description'] + + class Meta: + ordering = ['device', 'name'] + unique_together = ['device', 'name'] + + def __str__(self): + return self.name + + def get_absolute_url(self): + return self.device.get_absolute_url() + + def to_csv(self): + return ( + self.device.identifier, + self.name, + self.get_type_display(), + self.maximum_draw, + self.allocated_draw, + self.description, + ) + + @property + def connected_endpoint(self): + if self._connected_poweroutlet: + return self._connected_poweroutlet + return self._connected_powerfeed + + @connected_endpoint.setter + def connected_endpoint(self, value): + # TODO: Fix circular import + from . import PowerFeed + + if value is None: + self._connected_poweroutlet = None + self._connected_powerfeed = None + elif isinstance(value, PowerOutlet): + self._connected_poweroutlet = value + self._connected_powerfeed = None + elif isinstance(value, PowerFeed): + self._connected_poweroutlet = None + self._connected_powerfeed = value + else: + raise ValueError( + "Connected endpoint must be a PowerOutlet or PowerFeed, not {}.".format(type(value)) + ) + + def get_power_draw(self): + """ + Return the allocated and maximum power draw (in VA) and child PowerOutlet count for this PowerPort. + """ + # Calculate aggregate draw of all child power outlets if no numbers have been defined manually + if self.allocated_draw is None and self.maximum_draw is None: + outlet_ids = PowerOutlet.objects.filter(power_port=self).values_list('pk', flat=True) + utilization = PowerPort.objects.filter(_connected_poweroutlet_id__in=outlet_ids).aggregate( + maximum_draw_total=Sum('maximum_draw'), + allocated_draw_total=Sum('allocated_draw'), + ) + ret = { + 'allocated': utilization['allocated_draw_total'] or 0, + 'maximum': utilization['maximum_draw_total'] or 0, + 'outlet_count': len(outlet_ids), + 'legs': [], + } + + # Calculate per-leg aggregates for three-phase feeds + if self._connected_powerfeed and self._connected_powerfeed.phase == PowerFeedPhaseChoices.PHASE_3PHASE: + for leg, leg_name in PowerOutletFeedLegChoices: + outlet_ids = PowerOutlet.objects.filter(power_port=self, feed_leg=leg).values_list('pk', flat=True) + utilization = PowerPort.objects.filter(_connected_poweroutlet_id__in=outlet_ids).aggregate( + maximum_draw_total=Sum('maximum_draw'), + allocated_draw_total=Sum('allocated_draw'), + ) + ret['legs'].append({ + 'name': leg_name, + 'allocated': utilization['allocated_draw_total'] or 0, + 'maximum': utilization['maximum_draw_total'] or 0, + 'outlet_count': len(outlet_ids), + }) + + return ret + + # Default to administratively defined values + return { + 'allocated': self.allocated_draw or 0, + 'maximum': self.maximum_draw or 0, + 'outlet_count': PowerOutlet.objects.filter(power_port=self).count(), + 'legs': [], + } + + +# +# Power outlets +# + +class PowerOutlet(CableTermination, ComponentModel): + """ + A physical power outlet (output) within a Device which provides power to a PowerPort. + """ + device = models.ForeignKey( + to='dcim.Device', + on_delete=models.CASCADE, + related_name='poweroutlets' + ) + name = models.CharField( + max_length=50 + ) + type = models.CharField( + max_length=50, + choices=PowerOutletTypeChoices, + blank=True + ) + power_port = models.ForeignKey( + to='dcim.PowerPort', + on_delete=models.SET_NULL, + blank=True, + null=True, + related_name='poweroutlets' + ) + feed_leg = models.CharField( + max_length=50, + choices=PowerOutletFeedLegChoices, + blank=True, + help_text="Phase (for three-phase feeds)" + ) + connection_status = models.NullBooleanField( + choices=CONNECTION_STATUS_CHOICES, + blank=True + ) + + objects = NaturalOrderingManager() + tags = TaggableManager(through=TaggedItem) + + csv_headers = ['device', 'name', 'type', 'power_port', 'feed_leg', 'description'] + + class Meta: + unique_together = ['device', 'name'] + + def __str__(self): + return self.name + + def get_absolute_url(self): + return self.device.get_absolute_url() + + def to_csv(self): + return ( + self.device.identifier, + self.name, + self.get_type_display(), + self.power_port.name if self.power_port else None, + self.get_feed_leg_display(), + self.description, + ) + + def clean(self): + + # Validate power port assignment + if self.power_port and self.power_port.device != self.device: + raise ValidationError( + "Parent power port ({}) must belong to the same device".format(self.power_port) + ) + + +# +# Interfaces +# + +class Interface(CableTermination, ComponentModel): + """ + A network interface within a Device or VirtualMachine. A physical Interface can connect to exactly one other + Interface. + """ + device = models.ForeignKey( + to='Device', + on_delete=models.CASCADE, + related_name='interfaces', + null=True, + blank=True + ) + virtual_machine = models.ForeignKey( + to='virtualization.VirtualMachine', + on_delete=models.CASCADE, + related_name='interfaces', + null=True, + blank=True + ) + name = models.CharField( + max_length=64 + ) + _connected_interface = models.OneToOneField( + to='self', + on_delete=models.SET_NULL, + related_name='+', + blank=True, + null=True + ) + _connected_circuittermination = models.OneToOneField( + to='circuits.CircuitTermination', + on_delete=models.SET_NULL, + related_name='+', + blank=True, + null=True + ) + connection_status = models.NullBooleanField( + choices=CONNECTION_STATUS_CHOICES, + blank=True + ) + lag = models.ForeignKey( + to='self', + on_delete=models.SET_NULL, + related_name='member_interfaces', + null=True, + blank=True, + verbose_name='Parent LAG' + ) + type = models.CharField( + max_length=50, + choices=InterfaceTypeChoices + ) + enabled = models.BooleanField( + default=True + ) + mac_address = MACAddressField( + null=True, + blank=True, + verbose_name='MAC Address' + ) + mtu = models.PositiveIntegerField( + blank=True, + null=True, + validators=[MinValueValidator(1), MaxValueValidator(65536)], + verbose_name='MTU' + ) + mgmt_only = models.BooleanField( + default=False, + verbose_name='OOB Management', + help_text='This interface is used only for out-of-band management' + ) + mode = models.CharField( + max_length=50, + choices=InterfaceModeChoices, + blank=True, + ) + untagged_vlan = models.ForeignKey( + to='ipam.VLAN', + on_delete=models.SET_NULL, + related_name='interfaces_as_untagged', + null=True, + blank=True, + verbose_name='Untagged VLAN' + ) + tagged_vlans = models.ManyToManyField( + to='ipam.VLAN', + related_name='interfaces_as_tagged', + blank=True, + verbose_name='Tagged VLANs' + ) + + objects = InterfaceManager() + tags = TaggableManager(through=TaggedItem) + + csv_headers = [ + 'device', 'virtual_machine', 'name', 'lag', 'type', 'enabled', 'mac_address', 'mtu', 'mgmt_only', + 'description', 'mode', + ] + + class Meta: + ordering = ['device', 'name'] + unique_together = ['device', 'name'] + + def __str__(self): + return self.name + + def get_absolute_url(self): + return reverse('dcim:interface', kwargs={'pk': self.pk}) + + def to_csv(self): + return ( + self.device.identifier if self.device else None, + self.virtual_machine.name if self.virtual_machine else None, + self.name, + self.lag.name if self.lag else None, + self.get_type_display(), + self.enabled, + self.mac_address, + self.mtu, + self.mgmt_only, + self.description, + self.get_mode_display(), + ) + + def clean(self): + + # An Interface must belong to a Device *or* to a VirtualMachine + if self.device and self.virtual_machine: + raise ValidationError("An interface cannot belong to both a device and a virtual machine.") + if not self.device and not self.virtual_machine: + raise ValidationError("An interface must belong to either a device or a virtual machine.") + + # VM interfaces must be virtual + if self.virtual_machine and self.type not in VMInterfaceTypeChoices.values(): + raise ValidationError({ + 'type': "Invalid interface type for a virtual machine: {}".format(self.type) + }) + + # Virtual interfaces cannot be connected + if self.type in NONCONNECTABLE_IFACE_TYPES and ( + self.cable or getattr(self, 'circuit_termination', False) + ): + raise ValidationError({ + 'type': "Virtual and wireless interfaces cannot be connected to another interface or circuit. " + "Disconnect the interface or choose a suitable type." + }) + + # An interface's LAG must belong to the same device (or VC master) + if self.lag and self.lag.device not in [self.device, self.device.get_vc_master()]: + raise ValidationError({ + 'lag': "The selected LAG interface ({}) belongs to a different device ({}).".format( + self.lag.name, self.lag.device.name + ) + }) + + # A virtual interface cannot have a parent LAG + if self.type in NONCONNECTABLE_IFACE_TYPES and self.lag is not None: + raise ValidationError({ + 'lag': "{} interfaces cannot have a parent LAG interface.".format(self.get_type_display()) + }) + + # Only a LAG can have LAG members + if self.type != InterfaceTypeChoices.TYPE_LAG and self.member_interfaces.exists(): + raise ValidationError({ + 'type': "Cannot change interface type; it has LAG members ({}).".format( + ", ".join([iface.name for iface in self.member_interfaces.all()]) + ) + }) + + # Validate untagged VLAN + if self.untagged_vlan and self.untagged_vlan.site not in [self.parent.site, None]: + raise ValidationError({ + 'untagged_vlan': "The untagged VLAN ({}) must belong to the same site as the interface's parent " + "device/VM, or it must be global".format(self.untagged_vlan) + }) + + def save(self, *args, **kwargs): + + # Remove untagged VLAN assignment for non-802.1Q interfaces + if self.mode is None: + self.untagged_vlan = None + + # Only "tagged" interfaces may have tagged VLANs assigned. ("tagged all" implies all VLANs are assigned.) + if self.pk and self.mode is not InterfaceModeChoices.MODE_TAGGED: + self.tagged_vlans.clear() + + return super().save(*args, **kwargs) + + def to_objectchange(self, action): + # Annotate the parent Device/VM + try: + parent_obj = self.device or self.virtual_machine + except ObjectDoesNotExist: + parent_obj = None + + return ObjectChange( + changed_object=self, + object_repr=str(self), + action=action, + related_object=parent_obj, + object_data=serialize_object(self) + ) + + @property + def connected_endpoint(self): + if self._connected_interface: + return self._connected_interface + return self._connected_circuittermination + + @connected_endpoint.setter + def connected_endpoint(self, value): + from circuits.models import CircuitTermination + + if value is None: + self._connected_interface = None + self._connected_circuittermination = None + elif isinstance(value, Interface): + self._connected_interface = value + self._connected_circuittermination = None + elif isinstance(value, CircuitTermination): + self._connected_interface = None + self._connected_circuittermination = value + else: + raise ValueError( + "Connected endpoint must be an Interface or CircuitTermination, not {}.".format(type(value)) + ) + + @property + def parent(self): + return self.device or self.virtual_machine + + @property + def is_connectable(self): + return self.type not in NONCONNECTABLE_IFACE_TYPES + + @property + def is_virtual(self): + return self.type in VIRTUAL_IFACE_TYPES + + @property + def is_wireless(self): + return self.type in WIRELESS_IFACE_TYPES + + @property + def is_lag(self): + return self.type == InterfaceTypeChoices.TYPE_LAG + + @property + def count_ipaddresses(self): + return self.ip_addresses.count() + + +# +# Pass-through ports +# + +class FrontPort(CableTermination, ComponentModel): + """ + A pass-through port on the front of a Device. + """ + device = models.ForeignKey( + to='dcim.Device', + on_delete=models.CASCADE, + related_name='frontports' + ) + name = models.CharField( + max_length=64 + ) + type = models.CharField( + max_length=50, + choices=PortTypeChoices + ) + rear_port = models.ForeignKey( + to='dcim.RearPort', + on_delete=models.CASCADE, + related_name='frontports' + ) + rear_port_position = models.PositiveSmallIntegerField( + default=1, + validators=[MinValueValidator(1), MaxValueValidator(64)] + ) + + is_path_endpoint = False + + objects = NaturalOrderingManager() + tags = TaggableManager(through=TaggedItem) + + csv_headers = ['device', 'name', 'type', 'rear_port', 'rear_port_position', 'description'] + + class Meta: + ordering = ['device', 'name'] + unique_together = [ + ['device', 'name'], + ['rear_port', 'rear_port_position'], + ] + + def __str__(self): + return self.name + + def to_csv(self): + return ( + self.device.identifier, + self.name, + self.get_type_display(), + self.rear_port.name, + self.rear_port_position, + self.description, + ) + + def clean(self): + + # Validate rear port assignment + if self.rear_port.device != self.device: + raise ValidationError( + "Rear port ({}) must belong to the same device".format(self.rear_port) + ) + + # Validate rear port position assignment + if self.rear_port_position > self.rear_port.positions: + raise ValidationError( + "Invalid rear port position ({}); rear port {} has only {} positions".format( + self.rear_port_position, self.rear_port.name, self.rear_port.positions + ) + ) + + +class RearPort(CableTermination, ComponentModel): + """ + A pass-through port on the rear of a Device. + """ + device = models.ForeignKey( + to='dcim.Device', + on_delete=models.CASCADE, + related_name='rearports' + ) + name = models.CharField( + max_length=64 + ) + type = models.CharField( + max_length=50, + choices=PortTypeChoices + ) + positions = models.PositiveSmallIntegerField( + default=1, + validators=[MinValueValidator(1), MaxValueValidator(64)] + ) + + is_path_endpoint = False + + objects = NaturalOrderingManager() + tags = TaggableManager(through=TaggedItem) + + csv_headers = ['device', 'name', 'type', 'positions', 'description'] + + class Meta: + ordering = ['device', 'name'] + unique_together = ['device', 'name'] + + def __str__(self): + return self.name + + def to_csv(self): + return ( + self.device.identifier, + self.name, + self.get_type_display(), + self.positions, + self.description, + ) + + +# +# Device bays +# + +class DeviceBay(ComponentModel): + """ + An empty space within a Device which can house a child device + """ + device = models.ForeignKey( + to='dcim.Device', + on_delete=models.CASCADE, + related_name='device_bays' + ) + name = models.CharField( + max_length=50, + verbose_name='Name' + ) + installed_device = models.OneToOneField( + to='dcim.Device', + on_delete=models.SET_NULL, + related_name='parent_bay', + blank=True, + null=True + ) + + objects = NaturalOrderingManager() + tags = TaggableManager(through=TaggedItem) + + csv_headers = ['device', 'name', 'installed_device', 'description'] + + class Meta: + ordering = ['device', 'name'] + unique_together = ['device', 'name'] + + def __str__(self): + return '{} - {}'.format(self.device.name, self.name) + + def get_absolute_url(self): + return self.device.get_absolute_url() + + def to_csv(self): + return ( + self.device.identifier, + self.name, + self.installed_device.identifier if self.installed_device else None, + self.description, + ) + + def clean(self): + + # Validate that the parent Device can have DeviceBays + if not self.device.device_type.is_parent_device: + raise ValidationError("This type of device ({}) does not support device bays.".format( + self.device.device_type + )) + + # Cannot install a device into itself, obviously + if self.device == self.installed_device: + raise ValidationError("Cannot install a device into itself.") + + # Check that the installed device is not already installed elsewhere + if self.installed_device: + current_bay = DeviceBay.objects.filter(installed_device=self.installed_device).first() + if current_bay and current_bay != self: + raise ValidationError({ + 'installed_device': "Cannot install the specified device; device is already installed in {}".format( + current_bay + ) + }) + + +# +# Inventory items +# + +class InventoryItem(ComponentModel): + """ + An InventoryItem represents a serialized piece of hardware within a Device, such as a line card or power supply. + InventoryItems are used only for inventory purposes. + """ + device = models.ForeignKey( + to='dcim.Device', + on_delete=models.CASCADE, + related_name='inventory_items' + ) + parent = models.ForeignKey( + to='self', + on_delete=models.CASCADE, + related_name='child_items', + blank=True, + null=True + ) + name = models.CharField( + max_length=50, + verbose_name='Name' + ) + manufacturer = models.ForeignKey( + to='dcim.Manufacturer', + on_delete=models.PROTECT, + related_name='inventory_items', + blank=True, + null=True + ) + part_id = models.CharField( + max_length=50, + verbose_name='Part ID', + blank=True + ) + serial = models.CharField( + max_length=50, + verbose_name='Serial number', + blank=True + ) + asset_tag = models.CharField( + max_length=50, + unique=True, + blank=True, + null=True, + verbose_name='Asset tag', + help_text='A unique tag used to identify this item' + ) + discovered = models.BooleanField( + default=False, + verbose_name='Discovered' + ) + + tags = TaggableManager(through=TaggedItem) + + csv_headers = [ + 'device', 'name', 'manufacturer', 'part_id', 'serial', 'asset_tag', 'discovered', 'description', + ] + + class Meta: + ordering = ['device__id', 'parent__id', 'name'] + unique_together = ['device', 'parent', 'name'] + + def __str__(self): + return self.name + + def get_absolute_url(self): + return self.device.get_absolute_url() + + def to_csv(self): + return ( + self.device.name or '{{{}}}'.format(self.device.pk), + self.name, + self.manufacturer.name if self.manufacturer else None, + self.part_id, + self.serial, + self.asset_tag, + self.discovered, + self.description, + ) diff --git a/netbox/dcim/tests/test_models.py b/netbox/dcim/tests/test_models.py index 5cc3d0309..18d16150d 100644 --- a/netbox/dcim/tests/test_models.py +++ b/netbox/dcim/tests/test_models.py @@ -1,5 +1,8 @@ +from django.core.exceptions import ValidationError from django.test import TestCase +from dcim.choices import * +from dcim.constants import CONNECTION_STATUS_CONNECTED, CONNECTION_STATUS_PLANNED from dcim.models import * from tenancy.models import Tenant @@ -498,7 +501,7 @@ class CablePathTestCase(TestCase): self.assertEqual(interface1.connection_status, CONNECTION_STATUS_PLANNED) # Switch third segment from planned to connected - cable3.status = CONNECTION_STATUS_CONNECTED + cable3.status = CableStatusChoices.STATUS_CONNECTED cable3.save() interface1 = Interface.objects.get(pk=self.interface1.pk) self.assertEqual(interface1.connected_endpoint, self.interface2)