Coverage for dcim/models/device_components.py: 50%
650 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from functools import cached_property
3from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
4from django.contrib.postgres.fields import ArrayField
5from django.contrib.postgres.indexes import GistIndex
6from django.core.exceptions import ObjectDoesNotExist, ValidationError
7from django.core.validators import MaxValueValidator, MinValueValidator
8from django.db import models, router, transaction
9from django.utils.translation import gettext_lazy as _
11from dcim.choices import *
12from dcim.constants import *
13from dcim.fields import WWNField
14from dcim.models.base import PortMappingBase
15from dcim.models.mixins import (
16 CoolingLoopValidationMixin,
17 DiameterMixin,
18 InterfaceChannelRenameMixin,
19 InterfaceValidationMixin,
20 MaxFlowMixin,
21)
22from netbox.choices import ColorChoices
23from netbox.models import NetBoxModel, OrganizationalModel
24from netbox.models.features import ChangeLoggingMixin
25from netbox.models.ltree import LtreeManager, LtreeModel, SortPathField
26from netbox.models.mixins import OwnerMixin
27from utilities.fields import ColorField, NaturalOrderingField
28from utilities.ordering import naturalize_interface
29from utilities.query_functions import CollateAsChar
30from utilities.tracking import TrackingModelMixin
31from wireless.choices import *
32from wireless.utils import get_channel_attr
34__all__ = (
35 'BaseInterface',
36 'CabledObjectModel',
37 'ConsolePort',
38 'ConsoleServerPort',
39 'CoolingIntake',
40 'CoolingOutflow',
41 'DeviceBay',
42 'FrontPort',
43 'Interface',
44 'InventoryItem',
45 'InventoryItemRole',
46 'ModuleBay',
47 'PathEndpoint',
48 'PortMapping',
49 'PowerOutlet',
50 'PowerPort',
51 'RearPort',
52)
55class ComponentModel(OwnerMixin, NetBoxModel):
56 """
57 An abstract model inherited by any model which has a parent Device.
58 """
59 device = models.ForeignKey(
60 to='dcim.Device',
61 on_delete=models.CASCADE,
62 related_name='%(class)ss'
63 )
64 name = models.CharField(
65 verbose_name=_('name'),
66 max_length=64,
67 db_collation="natural_sort"
68 )
69 label = models.CharField(
70 verbose_name=_('label'),
71 max_length=64,
72 blank=True,
73 help_text=_('Physical label')
74 )
75 description = models.CharField(
76 verbose_name=_('description'),
77 max_length=200,
78 blank=True
79 )
81 # Denormalized references replicated from the parent Device
82 _site = models.ForeignKey(
83 to='dcim.Site',
84 on_delete=models.SET_NULL,
85 related_name='+',
86 blank=True,
87 null=True,
88 )
89 _location = models.ForeignKey(
90 to='dcim.Location',
91 on_delete=models.SET_NULL,
92 related_name='+',
93 blank=True,
94 null=True,
95 )
96 _rack = models.ForeignKey(
97 to='dcim.Rack',
98 on_delete=models.SET_NULL,
99 related_name='+',
100 blank=True,
101 null=True,
102 )
104 class Meta:
105 abstract = True
106 ordering = ('device', 'name')
107 constraints = (
108 models.UniqueConstraint(
109 fields=('device', 'name'),
110 name='%(app_label)s_%(class)s_unique_device_name'
111 ),
112 )
114 def __init__(self, *args, **kwargs):
115 super().__init__(*args, **kwargs)
117 # Cache the original Device ID for reference under clean()
118 self._original_device = self.__dict__.get('device_id')
120 def __str__(self):
121 if self.label: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true
122 return f"{self.name} ({self.label})"
123 return self.name
125 def to_objectchange(self, action):
126 objectchange = super().to_objectchange(action)
127 objectchange.related_object = self.device
128 return objectchange
130 def clean(self):
131 super().clean()
133 # Check list of Modules that allow device field to be changed
134 if (type(self) not in [InventoryItem]) and (self.pk is not None) and (self._original_device != self.device_id): 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true
135 raise ValidationError({
136 "device": _("Components cannot be moved to a different device.")
137 })
139 def save(self, *args, **kwargs):
140 # Save denormalized references
141 self._site = self.device.site
142 self._location = self.device.location
143 self._rack = self.device.rack
145 super().save(*args, **kwargs)
147 @property
148 def parent_object(self):
149 return self.device
152class ModularComponentModel(ComponentModel):
153 module = models.ForeignKey(
154 to='dcim.Module',
155 on_delete=models.CASCADE,
156 related_name='%(class)ss',
157 blank=True,
158 null=True
159 )
160 inventory_items = GenericRelation(
161 to='dcim.InventoryItem',
162 content_type_field='component_type',
163 object_id_field='component_id'
164 )
166 class Meta(ComponentModel.Meta):
167 abstract = True
170class CabledObjectModel(models.Model):
171 """
172 An abstract model inherited by all models to which a Cable can terminate. Provides the `cable` and `cable_end`
173 fields for caching cable associations, as well as `mark_connected` to designate "fake" connections.
174 """
175 cable = models.ForeignKey(
176 to='dcim.Cable',
177 on_delete=models.SET_NULL,
178 related_name='+',
179 blank=True,
180 null=True
181 )
182 cable_end = models.CharField(
183 verbose_name=_('cable end'),
184 max_length=1,
185 choices=CableEndChoices,
186 blank=True,
187 null=True
188 )
189 cable_connector = models.PositiveSmallIntegerField(
190 blank=True,
191 null=True,
192 validators=(
193 MinValueValidator(CABLE_CONNECTOR_MIN),
194 MaxValueValidator(CABLE_CONNECTOR_MAX)
195 ),
196 )
197 cable_positions = ArrayField(
198 base_field=models.PositiveSmallIntegerField(
199 validators=(
200 MinValueValidator(CABLE_POSITION_MIN),
201 MaxValueValidator(CABLE_POSITION_MAX)
202 )
203 ),
204 blank=True,
205 null=True,
206 )
207 mark_connected = models.BooleanField(
208 verbose_name=_('mark connected'),
209 default=False,
210 help_text=_('Treat as if a cable is connected')
211 )
213 cable_terminations = GenericRelation(
214 to='dcim.CableTermination',
215 content_type_field='termination_type',
216 object_id_field='termination_id',
217 related_query_name='%(class)s',
218 )
220 class Meta:
221 abstract = True
223 def clean(self):
224 super().clean()
226 if self.cable: 226 ↛ 227line 226 didn't jump to line 227 because the condition on line 226 was never true
227 if not self.cable_end:
228 raise ValidationError({
229 "cable_end": _("Must specify cable end (A or B) when attaching a cable.")
230 })
231 if self.cable_connector and not self.cable_positions:
232 raise ValidationError({
233 "cable_positions": _("Must specify position(s) when specifying a cable connector.")
234 })
235 if self.cable_positions and not self.cable_connector:
236 raise ValidationError({
237 "cable_positions": _("Cable positions cannot be set without a cable connector.")
238 })
239 if self.mark_connected:
240 raise ValidationError({
241 "mark_connected": _("Cannot mark as connected with a cable attached.")
242 })
243 else:
244 if self.cable_end: 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true
245 raise ValidationError({
246 "cable_end": _("Cable end must not be set without a cable.")
247 })
248 if self.cable_connector: 248 ↛ 249line 248 didn't jump to line 249 because the condition on line 248 was never true
249 raise ValidationError({
250 "cable_connector": _("Cable connector must not be set without a cable.")
251 })
252 if self.cable_positions: 252 ↛ 253line 252 didn't jump to line 253 because the condition on line 252 was never true
253 raise ValidationError({
254 "cable_positions": _("Cable termination positions must not be set without a cable.")
255 })
257 @property
258 def link(self):
259 """
260 Generic wrapper for a Cable, WirelessLink, or some other relation to a connected termination.
261 """
262 return self.cable
264 @cached_property
265 def link_peers(self):
266 if not self.cable: 266 ↛ 269line 266 didn't jump to line 269 because the condition on line 266 was always true
267 return []
269 if self.cable.profile:
270 return self._get_profile_link_peers()
272 return [peer.termination for peer in self.cable.terminations.all() if peer.cable_end != self.cable_end]
274 def _get_profile_link_peers(self):
275 if self.cable_end is None or self.cable_connector is None or not self.cable_positions:
276 return []
278 profile = self.cable.profile_class()
279 peer_terminations = {
280 (peer.connector, position): peer.termination
281 for peer in self.cable.terminations.all()
282 if peer.cable_end == self.opposite_cable_end and peer.connector is not None
283 for position in peer.positions or []
284 }
285 link_peers = []
287 for position in self.cable_positions:
288 mapped_position = profile.get_mapped_position(self.cable_end, self.cable_connector, position)
289 if mapped_position is None:
290 continue
292 peer = peer_terminations.get(mapped_position)
293 if peer is not None and peer not in link_peers:
294 link_peers.append(peer)
296 return link_peers
298 @property
299 def _occupied(self):
300 return bool(self.mark_connected or self.cable_id)
302 @property
303 def parent_object(self):
304 raise NotImplementedError(
305 _("{class_name} models must declare a parent_object property").format(class_name=self.__class__.__name__)
306 )
308 @property
309 def opposite_cable_end(self):
310 if not self.cable_end:
311 return None
312 return CableEndChoices.SIDE_A if self.cable_end == CableEndChoices.SIDE_B else CableEndChoices.SIDE_B
314 def set_cable_termination(self, termination):
315 """Save attributes from the given CableTermination on the terminating object."""
316 self.cable = termination.cable
317 self.cable_end = termination.cable_end
318 self.cable_connector = termination.connector
319 self.cable_positions = termination.positions
320 set_cable_termination.alters_data = True
322 def clear_cable_termination(self, termination):
323 """Clear all cable termination attributes from the terminating object."""
324 self.cable = None
325 self.cable_end = None
326 self.cable_connector = None
327 self.cable_positions = None
328 clear_cable_termination.alters_data = True
331class PathEndpoint(models.Model):
332 """
333 An abstract model inherited by any CabledObjectModel subclass which represents the end of a CablePath; specifically,
334 these include ConsolePort, ConsoleServerPort, PowerPort, PowerOutlet, Interface, and PowerFeed.
336 `_path` references the CablePath originating from this instance, if any. It is set or cleared by the receivers in
337 dcim.signals in response to changes in the cable path, and complements the `origin` GenericForeignKey field on the
338 CablePath model. `_path` should not be accessed directly; rather, use the `path` property.
340 `connected_endpoints()` is a convenience method for returning the destination of the associated CablePath, if any.
341 """
343 _path = models.ForeignKey(
344 to='dcim.CablePath',
345 on_delete=models.SET_NULL,
346 null=True,
347 blank=True,
348 )
350 class Meta:
351 abstract = True
353 def trace(self):
354 origin = self
355 path = []
357 # Construct the complete path (including e.g. bridged interfaces)
358 while origin is not None: 358 ↛ 383line 358 didn't jump to line 383 because the condition on line 358 was always true
359 # Go through the public accessor rather than dereferencing `_path`
360 # directly. During cable edits, CablePath rows can be deleted and
361 # recreated while this endpoint instance is still in memory.
362 cable_path = origin.path
363 if cable_path is None: 363 ↛ 366line 363 didn't jump to line 366 because the condition on line 363 was always true
364 break
366 path.extend(cable_path.path_objects)
368 # If the path ends at a non-connected pass-through port, pad out the link and far-end terminations
369 if len(path) % 3 == 1:
370 path.extend(([], []))
371 # If the path ends at a site or provider network, inject a null "link" to render an attachment
372 elif len(path) % 3 == 2:
373 path.insert(-1, [])
375 # Check for a bridged relationship to continue the trace.
376 destinations = cable_path.destinations
377 if len(destinations) == 1:
378 origin = getattr(destinations[0], 'bridge', None)
379 else:
380 origin = None
382 # Return the path as a list of three-tuples (A termination(s), cable(s), B termination(s))
383 return list(zip(*[iter(path)] * 3))
385 @property
386 def path(self):
387 """
388 Return this endpoint's current CablePath, if any.
390 `_path` is a denormalized reference that is updated from CablePath
391 save/delete handlers, including queryset.update() calls on origin
392 endpoints. That means an already-instantiated endpoint can briefly hold
393 a stale in-memory `_path` relation while the database already points to
394 a different CablePath (or to no path at all).
396 Two stale cases are repaired by refreshing only the `_path` field
397 from the database:
399 1. The endpoint is linked (by cable or wireless link) but `_path` is
400 unset, because the instance was loaded before its path was traced
401 (e.g. while queued for event serialization during link creation).
402 2. The cached relation points to a CablePath row that has just been
403 deleted.
405 Repairing case 1 costs one query per access for a linked endpoint
406 whose path is genuinely absent in the database. That state is
407 transient outside of tracing failures, so no result caching is
408 attempted here.
409 """
410 if self._path_id is None: 410 ↛ 418line 410 didn't jump to line 418 because the condition on line 410 was always true
411 has_link = self.cable_id is not None or getattr(self, 'wireless_link_id', None) is not None
412 if self.pk and has_link: 412 ↛ 413line 412 didn't jump to line 413 because the condition on line 412 was never true
413 self.refresh_from_db(fields=['_path'])
415 if self._path_id is None: 415 ↛ 418line 415 didn't jump to line 418 because the condition on line 415 was always true
416 return None
418 try:
419 return self._path
420 except ObjectDoesNotExist:
421 # Refresh only the denormalized FK instead of the whole model.
422 # The expected problem here is in-memory staleness during path
423 # rebuilds, not persistent database corruption.
424 self.refresh_from_db(fields=['_path'])
425 return self._path if self._path_id else None
427 @cached_property
428 def connected_endpoints(self):
429 """
430 Caching accessor for the attached CablePath's destinations (if any).
432 Always route through `path` so stale in-memory `_path` references are
433 repaired before we cache the result for the lifetime of this instance.
434 """
435 if cable_path := self.path: 435 ↛ 436line 435 didn't jump to line 436 because the condition on line 435 was never true
436 return cable_path.destinations
437 return []
440#
441# Console components
442#
444class ConsolePort(ModularComponentModel, CabledObjectModel, PathEndpoint, TrackingModelMixin):
445 """
446 A physical console port within a Device. ConsolePorts connect to ConsoleServerPorts.
447 """
448 type = models.CharField(
449 verbose_name=_('type'),
450 max_length=50,
451 choices=ConsolePortTypeChoices,
452 blank=True,
453 null=True,
454 help_text=_('Physical port type')
455 )
456 speed = models.PositiveIntegerField(
457 verbose_name=_('speed'),
458 choices=ConsolePortSpeedChoices,
459 blank=True,
460 null=True,
461 help_text=_('Port speed in bits per second')
462 )
464 clone_fields = ('device', 'module', 'type', 'speed')
466 class Meta(ModularComponentModel.Meta):
467 verbose_name = _('console port')
468 verbose_name_plural = _('console ports')
471class ConsoleServerPort(ModularComponentModel, CabledObjectModel, PathEndpoint, TrackingModelMixin):
472 """
473 A physical port within a Device (typically a designated console server) which provides access to ConsolePorts.
474 """
475 type = models.CharField(
476 verbose_name=_('type'),
477 max_length=50,
478 choices=ConsolePortTypeChoices,
479 blank=True,
480 null=True,
481 help_text=_('Physical port type')
482 )
483 speed = models.PositiveIntegerField(
484 verbose_name=_('speed'),
485 choices=ConsolePortSpeedChoices,
486 blank=True,
487 null=True,
488 help_text=_('Port speed in bits per second')
489 )
491 clone_fields = ('device', 'module', 'type', 'speed')
493 class Meta(ModularComponentModel.Meta):
494 verbose_name = _('console server port')
495 verbose_name_plural = _('console server ports')
498#
499# Power components
500#
502class PowerPort(ModularComponentModel, CabledObjectModel, PathEndpoint, TrackingModelMixin):
503 """
504 A physical power supply (intake) port within a Device. PowerPorts connect to PowerOutlets.
505 """
506 type = models.CharField(
507 verbose_name=_('type'),
508 max_length=50,
509 choices=PowerPortTypeChoices,
510 blank=True,
511 null=True,
512 help_text=_('Physical port type')
513 )
514 maximum_draw = models.PositiveIntegerField(
515 verbose_name=_('maximum draw'),
516 blank=True,
517 null=True,
518 validators=[MinValueValidator(1)],
519 help_text=_("Maximum power draw (watts)")
520 )
521 allocated_draw = models.PositiveIntegerField(
522 verbose_name=_('allocated draw'),
523 blank=True,
524 null=True,
525 validators=[MinValueValidator(1)],
526 help_text=_('Allocated power draw (watts)')
527 )
529 clone_fields = ('device', 'module', 'maximum_draw', 'allocated_draw')
531 class Meta(ModularComponentModel.Meta):
532 verbose_name = _('power port')
533 verbose_name_plural = _('power ports')
535 def clean(self):
536 super().clean()
538 if self.maximum_draw is not None and self.allocated_draw is not None:
539 if self.allocated_draw > self.maximum_draw:
540 raise ValidationError({
541 'allocated_draw': _(
542 "Allocated draw cannot exceed the maximum draw ({maximum_draw}W)."
543 ).format(maximum_draw=self.maximum_draw)
544 })
546 def get_downstream_powerports(self, leg=None):
547 """
548 Return a queryset of all PowerPorts connected via cable to a child PowerOutlet. For example, in the topology
549 below, PP1.get_downstream_powerports() would return PP2-4.
551 ---- PO1 <---> PP2
552 /
553 PP1 ------- PO2 <---> PP3
554 \
555 ---- PO3 <---> PP4
557 """
558 poweroutlets = self.poweroutlets.filter(cable__isnull=False)
559 if leg:
560 poweroutlets = poweroutlets.filter(feed_leg=leg)
561 if not poweroutlets:
562 return PowerPort.objects.none()
564 q = Q()
565 for poweroutlet in poweroutlets:
566 q |= Q(
567 cable=poweroutlet.cable,
568 cable_end=poweroutlet.opposite_cable_end
569 )
571 return PowerPort.objects.filter(q)
573 def get_power_draw(self, _seen=None):
574 """
575 Return the allocated and maximum power draw (in VA) and child PowerOutlet count for this PowerPort.
576 """
577 from dcim.models import PowerFeed
579 # Calculate aggregate draw of all child power outlets if no numbers have been defined manually
580 if self.allocated_draw is None and self.maximum_draw is None:
582 def _aggregate(powerports, seen):
583 # Recursively resolve the draw for each downstream PowerPort. Using the per-port value
584 # (rather than a SQL aggregate over allocated_draw/maximum_draw) allows the draw to
585 # propagate through intermediate auto-mode PowerPorts, e.g. PDU-internal fuse chains.
586 # `seen` tracks visited PowerPorts to prevent infinite recursion if the topology
587 # happens to form a cycle.
588 allocated_total = 0
589 maximum_total = 0
590 for powerport in powerports:
591 if powerport.pk in seen:
592 continue
593 seen.add(powerport.pk)
594 draw = powerport.get_power_draw(_seen=seen)
595 allocated_total += draw['allocated']
596 maximum_total += draw['maximum']
597 return allocated_total, maximum_total
599 # Seed each _aggregate() call with a fresh copy of the inherited visited set so the full
600 # and per-leg aggregations are independent. Otherwise, ports visited during the full
601 # aggregation would be skipped during the per-leg passes.
602 base_seen = set(_seen) if _seen else set()
603 base_seen.add(self.pk)
605 allocated, maximum = _aggregate(self.get_downstream_powerports(), set(base_seen))
606 ret = {
607 'allocated': allocated,
608 'maximum': maximum,
609 'outlet_count': self.poweroutlets.count(),
610 'legs': [],
611 }
613 # Calculate per-leg aggregates for three-phase power feeds
614 if len(self.link_peers) == 1 and isinstance(self.link_peers[0], PowerFeed) and \
615 self.link_peers[0].phase == PowerFeedPhaseChoices.PHASE_3PHASE:
616 for leg, leg_name in PowerOutletFeedLegChoices:
617 leg_allocated, leg_maximum = _aggregate(
618 self.get_downstream_powerports(leg=leg), set(base_seen)
619 )
620 ret['legs'].append({
621 'name': leg_name,
622 'allocated': leg_allocated,
623 'maximum': leg_maximum,
624 'outlet_count': self.poweroutlets.filter(feed_leg=leg).count(),
625 })
627 return ret
629 # Default to administratively defined values
630 return {
631 'allocated': self.allocated_draw or 0,
632 'maximum': self.maximum_draw or 0,
633 'outlet_count': self.poweroutlets.count(),
634 'legs': [],
635 }
638class PowerOutlet(ModularComponentModel, CabledObjectModel, PathEndpoint, TrackingModelMixin):
639 """
640 A physical power outlet (output) within a Device which provides power to a PowerPort.
641 """
642 status = models.CharField(
643 verbose_name=_('status'),
644 max_length=50,
645 choices=PowerOutletStatusChoices,
646 default=PowerOutletStatusChoices.STATUS_ENABLED
647 )
648 type = models.CharField(
649 verbose_name=_('type'),
650 max_length=50,
651 choices=PowerOutletTypeChoices,
652 blank=True,
653 null=True,
654 help_text=_('Physical port type')
655 )
656 power_port = models.ForeignKey(
657 to='dcim.PowerPort',
658 on_delete=models.SET_NULL,
659 blank=True,
660 null=True,
661 related_name='poweroutlets'
662 )
663 feed_leg = models.CharField(
664 verbose_name=_('feed leg'),
665 max_length=50,
666 choices=PowerOutletFeedLegChoices,
667 blank=True,
668 null=True,
669 help_text=_('Phase (for three-phase feeds)')
670 )
671 color = ColorField(
672 verbose_name=_('color'),
673 blank=True
674 )
676 clone_fields = ('device', 'module', 'type', 'power_port', 'feed_leg')
678 class Meta(ModularComponentModel.Meta):
679 verbose_name = _('power outlet')
680 verbose_name_plural = _('power outlets')
682 def clean(self):
683 super().clean()
685 # Validate power port assignment
686 if self.power_port and self.power_port.device != self.device:
687 raise ValidationError(
688 _("Parent power port ({power_port}) must belong to the same device").format(power_port=self.power_port)
689 )
691 def get_status_color(self):
692 return PowerOutletStatusChoices.colors.get(self.status)
695#
696# Cooling components
697#
699class CoolingIntake(
700 CoolingLoopValidationMixin, DiameterMixin, MaxFlowMixin, ModularComponentModel, TrackingModelMixin
701):
702 """
703 A coolant intake port within a Device (e.g. a server cold-plate inlet or CDU intake). A
704 CoolingIntake is supplied by an upstream CoolingOutflow. The serving CoolingFeed is
705 derived from the Device's Rack rather than referenced directly.
707 Unlike CoolingOutflow (whose parent intake is on the same Device and can therefore be
708 templated), an intake's upstream outflow typically lives on a different Device (e.g. a
709 CDU), so there is deliberately no upstream-outflow field on CoolingIntakeTemplate.
710 """
711 type = models.CharField(
712 verbose_name=_('type'),
713 max_length=50,
714 choices=CoolingConnectorTypeChoices,
715 blank=True,
716 null=True,
717 help_text=_('Physical connector type')
718 )
719 # diameter, diameter_unit, _abs_diameter provided by DiameterMixin
720 # max_flow, max_flow_unit, _abs_max_flow provided by MaxFlowMixin
721 cooling_outflow = models.ForeignKey(
722 to='dcim.CoolingOutflow',
723 on_delete=models.SET_NULL,
724 blank=True,
725 null=True,
726 related_name='coolingintakes',
727 help_text=_('The upstream cooling outflow supplying this intake')
728 )
730 clone_fields = (
731 'device', 'module', 'type', 'diameter', 'diameter_unit', 'max_flow',
732 'max_flow_unit',
733 )
734 upstream_field = 'cooling_outflow'
736 class Meta(ModularComponentModel.Meta):
737 verbose_name = _('cooling intake')
738 verbose_name_plural = _('cooling intakes')
740 def clean(self):
741 super().clean()
743 # Prevent the intake/outflow chain from forming a loop
744 self.validate_cooling_loop()
747class CoolingOutflow(CoolingLoopValidationMixin, DiameterMixin, ModularComponentModel, TrackingModelMixin):
748 """
749 A coolant outlet within a Device (e.g. a CDU or manifold outlet) which supplies one or more
750 CoolingIntakes (referenced via CoolingIntake.cooling_outflow).
751 """
752 type = models.CharField(
753 verbose_name=_('type'),
754 max_length=50,
755 choices=CoolingConnectorTypeChoices,
756 blank=True,
757 null=True,
758 help_text=_('Physical connector type')
759 )
760 # diameter, diameter_unit, _abs_diameter provided by DiameterMixin
761 cooling_intake = models.ForeignKey(
762 to='dcim.CoolingIntake',
763 on_delete=models.SET_NULL,
764 blank=True,
765 null=True,
766 related_name='coolingoutflows'
767 )
769 clone_fields = ('device', 'module', 'type', 'diameter', 'diameter_unit', 'cooling_intake')
770 upstream_field = 'cooling_intake'
772 class Meta(ModularComponentModel.Meta):
773 verbose_name = _('cooling outflow')
774 verbose_name_plural = _('cooling outflows')
776 def clean(self):
777 super().clean()
779 # Validate cooling intake assignment
780 if self.cooling_intake and self.cooling_intake.device != self.device:
781 raise ValidationError(
782 _("Parent cooling intake ({cooling_intake}) must belong to the same device").format(
783 cooling_intake=self.cooling_intake)
784 )
786 # Prevent the intake/outflow chain from forming a loop
787 self.validate_cooling_loop()
790#
791# Interfaces
792#
794class BaseInterface(models.Model):
795 """
796 Abstract base class for fields shared by dcim.Interface and virtualization.VMInterface.
797 """
798 enabled = models.BooleanField(
799 verbose_name=_('enabled'),
800 default=True
801 )
802 mtu = models.PositiveIntegerField(
803 blank=True,
804 null=True,
805 validators=[
806 MinValueValidator(INTERFACE_MTU_MIN),
807 MaxValueValidator(INTERFACE_MTU_MAX)
808 ],
809 verbose_name=_('MTU')
810 )
811 mode = models.CharField(
812 verbose_name=_('mode'),
813 max_length=50,
814 choices=InterfaceModeChoices,
815 blank=True,
816 null=True,
817 help_text=_('IEEE 802.1Q tagging strategy')
818 )
819 parent = models.ForeignKey(
820 to='self',
821 on_delete=models.RESTRICT,
822 related_name='child_interfaces',
823 null=True,
824 blank=True,
825 verbose_name=_('parent interface')
826 )
827 bridge = models.ForeignKey(
828 to='self',
829 on_delete=models.SET_NULL,
830 related_name='bridge_interfaces',
831 null=True,
832 blank=True,
833 verbose_name=_('bridge interface')
834 )
835 untagged_vlan = models.ForeignKey(
836 to='ipam.VLAN',
837 on_delete=models.SET_NULL,
838 related_name='%(class)ss_as_untagged',
839 null=True,
840 blank=True,
841 verbose_name=_('untagged VLAN')
842 )
843 tagged_vlans = models.ManyToManyField(
844 to='ipam.VLAN',
845 related_name='%(class)ss_as_tagged',
846 blank=True,
847 verbose_name=_('tagged VLANs')
848 )
849 qinq_svlan = models.ForeignKey(
850 to='ipam.VLAN',
851 on_delete=models.SET_NULL,
852 related_name='%(class)ss_svlan',
853 null=True,
854 blank=True,
855 verbose_name=_('Q-in-Q SVLAN')
856 )
857 vlan_translation_policy = models.ForeignKey(
858 to='ipam.VLANTranslationPolicy',
859 on_delete=models.PROTECT,
860 null=True,
861 blank=True,
862 verbose_name=_('VLAN Translation Policy')
863 )
864 primary_mac_address = models.OneToOneField(
865 to='dcim.MACAddress',
866 on_delete=models.SET_NULL,
867 related_name='+',
868 blank=True,
869 null=True,
870 verbose_name=_('primary MAC address')
871 )
873 class Meta:
874 abstract = True
876 def clean(self):
877 super().clean()
879 # SVLAN can be defined only for Q-in-Q interfaces
880 if self.qinq_svlan and self.mode != InterfaceModeChoices.MODE_Q_IN_Q: 880 ↛ 881line 880 didn't jump to line 881 because the condition on line 880 was never true
881 raise ValidationError({
882 'qinq_svlan': _("Only Q-in-Q interfaces may specify a service VLAN.")
883 })
885 # A primary MAC address must belong to this interface. On create the MAC is assigned by a
886 # post_save signal after this runs, so an as-yet-unassigned MAC is only rejected on update
887 # (self._state.adding is False), where no such signal fires. These are raised as non-field
888 # errors: primary_mac_address is not an InterfaceForm field (it's edited via the mac_address
889 # shortcut), so a field-keyed error would raise in the form's add_error() rather than render.
890 if self.primary_mac_address: 890 ↛ 891line 890 didn't jump to line 891 because the condition on line 890 was never true
891 if self.primary_mac_address.assigned_object is None:
892 if not self._state.adding:
893 raise ValidationError(
894 _("Only a MAC address assigned to this interface can be its primary MAC address.")
895 )
896 elif self.primary_mac_address.assigned_object != self:
897 raise ValidationError(
898 _("MAC address {mac_address} is assigned to a different interface ({interface}).").format(
899 mac_address=self.primary_mac_address,
900 interface=self.primary_mac_address.assigned_object,
901 )
902 )
904 def save(self, *args, **kwargs):
906 # Remove untagged VLAN assignment for non-802.1Q interfaces
907 if not self.mode: 907 ↛ 911line 907 didn't jump to line 911 because the condition on line 907 was always true
908 self.untagged_vlan = None
910 # Only "tagged" interfaces may have tagged VLANs assigned. ("tagged all" implies all VLANs are assigned.)
911 if not self._state.adding and self.mode != InterfaceModeChoices.MODE_TAGGED:
912 self.tagged_vlans.clear()
914 return super().save(*args, **kwargs)
916 def set_primary_mac_address(self, mac):
917 """
918 Set (or clear) this interface's primary MAC address as a single atomic, validated operation.
919 Pass a MACAddress instance to designate it primary, or None to clear the primary MAC. The
920 callers own permission checks; this method owns the validated write. To set from a submitted
921 address string (find-or-create on this interface) use set_primary_mac_address_from_value().
922 """
923 self._set_primary_mac_address(mac=mac)
924 set_primary_mac_address.alters_data = True
926 def set_primary_mac_address_from_value(self, mac_address):
927 """
928 Set this interface's primary MAC address from a submitted address string, finding an existing
929 MAC on the interface or creating one, all within the operation's locked transaction. An empty
930 value clears the primary MAC. For the form and API adapters, which receive a string.
931 """
932 self._set_primary_mac_address(mac_value=mac_address or None)
933 set_primary_mac_address_from_value.alters_data = True
935 def _set_primary_mac_address(self, mac=None, mac_value=None):
936 """
937 Shared implementation of the two public setters. Locks this interface's row, resolves a
938 submitted string to a MACAddress (find-or-create, inside the lock so concurrent requests can't
939 both create the same one), validates, and saves. Callers pass either a resolved MACAddress
940 (`mac`) or an address string (`mac_value`), never both.
941 """
942 with transaction.atomic(using=router.db_for_write(type(self))):
943 # Lock and re-fetch this interface so concurrent set-primary/find-or-create requests
944 # serialize, and mutate the freshly-loaded row rather than the caller's in-memory instance.
945 # The re-fetch resets change-tracking state (e.g. _original_device) to the persisted values,
946 # so full_clean() validates the persisted object plus this one change, not unrelated edits the
947 # adapter already validated and saved.
948 locked = type(self).objects.select_for_update().get(pk=self.pk)
950 # Resolve a submitted string to a MAC inside the lock, so two concurrent requests setting the
951 # same new value can't both miss the lookup and both create a duplicate.
952 if mac_value is not None:
953 mac = locked.mac_addresses.filter(mac_address=mac_value).first()
954 if mac is None:
955 mac = locked.mac_addresses.model(mac_address=mac_value, assigned_object=locked)
956 mac.full_clean()
957 mac.save()
959 target_id = mac.pk if mac is not None else None
960 if locked.primary_mac_address_id == target_id:
961 self.primary_mac_address = mac
962 self.__dict__.pop('mac_address', None)
963 return
965 # Snapshot the locked row (refetched after any adapter save this request) so the changelog
966 # records the correct pre-change state for this MAC change, not an earlier field edit.
967 locked.snapshot()
968 locked.primary_mac_address = mac
969 locked.full_clean(validate_unique=False)
970 locked.save()
972 # Reflect the change on the caller's instance (for success messages and API responses) and
973 # invalidate the cached read-side mac_address property.
974 self.primary_mac_address = mac
975 self.__dict__.pop('mac_address', None)
977 @property
978 def tunnel_termination(self):
979 return self.tunnel_terminations.first()
981 @property
982 def count_ipaddresses(self):
983 return self.ip_addresses.count()
985 @property
986 def count_fhrp_groups(self):
987 return self.fhrp_group_assignments.count()
989 @cached_property
990 def mac_address(self):
991 if self.primary_mac_address: 991 ↛ 992line 991 didn't jump to line 992 because the condition on line 991 was never true
992 return self.primary_mac_address.mac_address
993 return None
996class Interface(
997 InterfaceChannelRenameMixin,
998 InterfaceValidationMixin,
999 ModularComponentModel,
1000 BaseInterface,
1001 CabledObjectModel,
1002 PathEndpoint,
1003 TrackingModelMixin,
1004):
1005 """
1006 A network interface within a Device. A physical Interface can connect to exactly one other Interface.
1007 """
1008 # Override ComponentModel._name to specify naturalize_interface function
1009 _name = NaturalOrderingField(
1010 target_field='name',
1011 naturalize_function=naturalize_interface,
1012 max_length=100,
1013 blank=True
1014 )
1015 vdcs = models.ManyToManyField(
1016 to='dcim.VirtualDeviceContext',
1017 related_name='interfaces'
1018 )
1019 lag = models.ForeignKey(
1020 to='self',
1021 on_delete=models.SET_NULL,
1022 related_name='member_interfaces',
1023 null=True,
1024 blank=True,
1025 verbose_name=_('parent LAG')
1026 )
1027 type = models.CharField(
1028 verbose_name=_('type'),
1029 max_length=50,
1030 choices=InterfaceTypeChoices
1031 )
1032 channels = models.PositiveSmallIntegerField(
1033 verbose_name=_('channels'),
1034 blank=True,
1035 null=True,
1036 validators=(
1037 MinValueValidator(INTERFACE_CHANNELS_MIN),
1038 MaxValueValidator(INTERFACE_CHANNELS_MAX)
1039 ),
1040 help_text=_('The number of channels into which this interface is channelized')
1041 )
1042 channel_id = models.PositiveSmallIntegerField(
1043 verbose_name=_('channel ID'),
1044 blank=True,
1045 null=True,
1046 validators=(
1047 MinValueValidator(INTERFACE_CHANNELS_MIN),
1048 MaxValueValidator(INTERFACE_CHANNELS_MAX)
1049 ),
1050 help_text=_('The channel on the parent interface to which this subinterface is bound')
1051 )
1052 mgmt_only = models.BooleanField(
1053 default=False,
1054 verbose_name=_('management only'),
1055 help_text=_('This interface is used only for out-of-band management')
1056 )
1057 speed = models.PositiveBigIntegerField(
1058 blank=True,
1059 null=True,
1060 verbose_name=_('speed (Kbps)')
1061 )
1062 duplex = models.CharField(
1063 verbose_name=_('duplex'),
1064 max_length=50,
1065 blank=True,
1066 null=True,
1067 choices=InterfaceDuplexChoices
1068 )
1069 wwn = WWNField(
1070 null=True,
1071 blank=True,
1072 verbose_name=_('WWN'),
1073 help_text=_('64-bit World Wide Name')
1074 )
1075 rf_role = models.CharField(
1076 max_length=30,
1077 choices=WirelessRoleChoices,
1078 blank=True,
1079 null=True,
1080 verbose_name=_('wireless role')
1081 )
1082 rf_channel = models.CharField(
1083 max_length=50,
1084 choices=WirelessChannelChoices,
1085 blank=True,
1086 null=True,
1087 verbose_name=_('wireless channel')
1088 )
1089 rf_channel_frequency = models.DecimalField(
1090 max_digits=8,
1091 decimal_places=3,
1092 blank=True,
1093 null=True,
1094 verbose_name=_('channel frequency (MHz)'),
1095 help_text=_("Populated by selected channel (if set)")
1096 )
1097 rf_channel_width = models.DecimalField(
1098 max_digits=7,
1099 decimal_places=3,
1100 blank=True,
1101 null=True,
1102 verbose_name=('channel width (MHz)'),
1103 help_text=_("Populated by selected channel (if set)")
1104 )
1105 tx_power = models.SmallIntegerField(
1106 blank=True,
1107 null=True,
1108 validators=(
1109 MinValueValidator(-40),
1110 MaxValueValidator(127),
1111 ),
1112 verbose_name=_('transmit power (dBm)')
1113 )
1114 poe_mode = models.CharField(
1115 max_length=50,
1116 choices=InterfacePoEModeChoices,
1117 blank=True,
1118 null=True,
1119 verbose_name=_('PoE mode')
1120 )
1121 poe_type = models.CharField(
1122 max_length=50,
1123 choices=InterfacePoETypeChoices,
1124 blank=True,
1125 null=True,
1126 verbose_name=_('PoE type')
1127 )
1128 wireless_link = models.ForeignKey(
1129 to='wireless.WirelessLink',
1130 on_delete=models.SET_NULL,
1131 related_name='+',
1132 blank=True,
1133 null=True
1134 )
1135 wireless_lans = models.ManyToManyField(
1136 to='wireless.WirelessLAN',
1137 related_name='interfaces',
1138 blank=True,
1139 verbose_name=_('wireless LANs')
1140 )
1141 vrf = models.ForeignKey(
1142 to='ipam.VRF',
1143 on_delete=models.SET_NULL,
1144 related_name='interfaces',
1145 null=True,
1146 blank=True,
1147 verbose_name=_('VRF')
1148 )
1149 ip_addresses = GenericRelation(
1150 to='ipam.IPAddress',
1151 content_type_field='assigned_object_type',
1152 object_id_field='assigned_object_id',
1153 related_query_name='interface'
1154 )
1155 mac_addresses = GenericRelation(
1156 to='dcim.MACAddress',
1157 content_type_field='assigned_object_type',
1158 object_id_field='assigned_object_id',
1159 related_query_name='interface'
1160 )
1161 fhrp_group_assignments = GenericRelation(
1162 to='ipam.FHRPGroupAssignment',
1163 content_type_field='interface_type',
1164 object_id_field='interface_id',
1165 related_query_name='+'
1166 )
1167 tunnel_terminations = GenericRelation(
1168 to='vpn.TunnelTermination',
1169 content_type_field='termination_type',
1170 object_id_field='termination_id',
1171 related_query_name='interface'
1172 )
1173 l2vpn_terminations = GenericRelation(
1174 to='vpn.L2VPNTermination',
1175 content_type_field='assigned_object_type',
1176 object_id_field='assigned_object_id',
1177 related_query_name='interface',
1178 )
1180 clone_fields = (
1181 'device', 'module', 'parent', 'bridge', 'lag', 'type', 'channels', 'mgmt_only', 'mtu', 'mode', 'speed',
1182 'duplex', 'rf_role', 'rf_channel', 'rf_channel_frequency', 'rf_channel_width', 'tx_power', 'poe_mode',
1183 'poe_type', 'vrf',
1184 )
1186 class Meta(ModularComponentModel.Meta):
1187 ordering = ('device', CollateAsChar('_name'))
1188 verbose_name = _('interface')
1189 verbose_name_plural = _('interfaces')
1190 constraints = (
1191 *ModularComponentModel.Meta.constraints,
1192 models.UniqueConstraint(
1193 fields=('parent', 'channel_id'),
1194 name='%(app_label)s_%(class)s_unique_parent_channel_id'
1195 ),
1196 )
1198 def __init__(self, *args, **kwargs):
1199 # InterfaceChannelRenameMixin.__init__() (reached via super(), first in the MRO) sets _original_channels, used
1200 # below by InterfaceValidationMixin.clean() and by post_save signal handlers.
1201 super().__init__(*args, **kwargs)
1203 # Cache channelization-related fields so post-save signal handlers can detect changes which require rebuilding
1204 # cable paths (channelization does not involve modifying the Cable itself, so the cable signals do not fire).
1205 self._original_channel_id = self.__dict__.get('channel_id')
1206 self._original_parent_id = self.__dict__.get('parent_id')
1208 def clean(self):
1209 super().clean()
1211 # Virtual Interfaces cannot have a Cable attached
1212 if self.is_virtual and self.cable: 1212 ↛ 1213line 1212 didn't jump to line 1213 because the condition on line 1212 was never true
1213 raise ValidationError({
1214 'type': _("{display_type} interfaces cannot have a cable attached.").format(
1215 display_type=self.get_type_display()
1216 )
1217 })
1219 # Virtual Interfaces cannot be marked as connected
1220 if self.is_virtual and self.mark_connected: 1220 ↛ 1221line 1220 didn't jump to line 1221 because the condition on line 1220 was never true
1221 raise ValidationError({
1222 'mark_connected': _("{display_type} interfaces cannot be marked as connected.".format(
1223 display_type=self.get_type_display())
1224 )
1225 })
1227 # A channel subinterface's cable state is mirrored from its channelized parent (see
1228 # update_channelized_cable_paths()), so it cannot also carry its own CableTermination -- checking
1229 # cable_terminations rather than self.cable, since a valid channel child's self.cable is expected to
1230 # already reflect the parent's mirrored cable.
1231 if self.channel_id is not None and self.cable_terminations.exists(): 1231 ↛ 1232line 1231 didn't jump to line 1232 because the condition on line 1231 was never true
1232 raise ValidationError({
1233 'channel_id': _(
1234 "A channel ID cannot be assigned to an interface with an existing cable connection. Remove "
1235 "the cable first."
1236 )
1237 })
1239 # Parent validation (self-reference and interface-type restrictions are enforced by InterfaceValidationMixin)
1241 # An interface's parent must belong to the same device or virtual chassis
1242 if self.parent and self.parent.device != self.device: 1242 ↛ 1243line 1242 didn't jump to line 1243 because the condition on line 1242 was never true
1243 if self.device.virtual_chassis is None:
1244 raise ValidationError({
1245 'parent': _(
1246 "The selected parent interface ({interface}) belongs to a different device ({device})"
1247 ).format(interface=self.parent, device=self.parent.device)
1248 })
1249 if self.parent.device.virtual_chassis != self.device.virtual_chassis:
1250 raise ValidationError({
1251 'parent': _(
1252 "The selected parent interface ({interface}) belongs to {device}, which is not part of "
1253 "virtual chassis {virtual_chassis}."
1254 ).format(
1255 interface=self.parent,
1256 device=self.parent.device,
1257 virtual_chassis=self.device.virtual_chassis
1258 )
1259 })
1261 # Bridge validation
1263 # A bridged interface belongs to the same device or virtual chassis
1264 if self.bridge and self.bridge.device != self.device: 1264 ↛ 1265line 1264 didn't jump to line 1265 because the condition on line 1264 was never true
1265 if self.device.virtual_chassis is None:
1266 raise ValidationError({
1267 'bridge': _(
1268 "The selected bridge interface ({bridge}) belongs to a different device ({device})."
1269 ).format(bridge=self.bridge, device=self.bridge.device)
1270 })
1271 if self.bridge.device.virtual_chassis != self.device.virtual_chassis:
1272 raise ValidationError({
1273 'bridge': _(
1274 "The selected bridge interface ({interface}) belongs to {device}, which is not part of virtual "
1275 "chassis {virtual_chassis}."
1276 ).format(
1277 interface=self.bridge, device=self.bridge.device, virtual_chassis=self.device.virtual_chassis
1278 )
1279 })
1281 # LAG validation
1283 # A virtual interface cannot have a parent LAG
1284 if self.type == InterfaceTypeChoices.TYPE_VIRTUAL and self.lag is not None: 1284 ↛ 1285line 1284 didn't jump to line 1285 because the condition on line 1284 was never true
1285 raise ValidationError({'lag': _("Virtual interfaces cannot have a parent LAG interface.")})
1287 # A LAG interface cannot be its own parent
1288 if self.pk and self.lag_id == self.pk: 1288 ↛ 1289line 1288 didn't jump to line 1289 because the condition on line 1288 was never true
1289 raise ValidationError({'lag': _("A LAG interface cannot be its own parent.")})
1291 # An interface's LAG must belong to the same device or virtual chassis
1292 if self.lag and self.lag.device != self.device: 1292 ↛ 1293line 1292 didn't jump to line 1293 because the condition on line 1292 was never true
1293 if self.device.virtual_chassis is None:
1294 raise ValidationError({
1295 'lag': _(
1296 "The selected LAG interface ({lag}) belongs to a different device ({device})."
1297 ).format(lag=self.lag, device=self.lag.device)
1298 })
1299 if self.lag.device.virtual_chassis != self.device.virtual_chassis:
1300 raise ValidationError({
1301 'lag': _(
1302 "The selected LAG interface ({lag}) belongs to {device}, which is not part of virtual chassis "
1303 "{virtual_chassis}.".format(
1304 lag=self.lag, device=self.lag.device, virtual_chassis=self.device.virtual_chassis)
1305 )
1306 })
1308 # Wireless validation
1310 # RF channel may only be set for wireless interfaces
1311 if self.rf_channel and not self.is_wireless: 1311 ↛ 1312line 1311 didn't jump to line 1312 because the condition on line 1311 was never true
1312 raise ValidationError({'rf_channel': _("Channel may be set only on wireless interfaces.")})
1314 # Validate channel frequency against interface type and selected channel (if any)
1315 if self.rf_channel_frequency: 1315 ↛ 1316line 1315 didn't jump to line 1316 because the condition on line 1315 was never true
1316 if not self.is_wireless:
1317 raise ValidationError({
1318 'rf_channel_frequency': _("Channel frequency may be set only on wireless interfaces."),
1319 })
1320 if self.rf_channel and self.rf_channel_frequency != get_channel_attr(self.rf_channel, 'frequency'):
1321 raise ValidationError({
1322 'rf_channel_frequency': _("Cannot specify custom frequency with channel selected."),
1323 })
1325 # Validate channel width against interface type and selected channel (if any)
1326 if self.rf_channel_width: 1326 ↛ 1327line 1326 didn't jump to line 1327 because the condition on line 1326 was never true
1327 if not self.is_wireless:
1328 raise ValidationError({'rf_channel_width': _("Channel width may be set only on wireless interfaces.")})
1329 if self.rf_channel and self.rf_channel_width != get_channel_attr(self.rf_channel, 'width'):
1330 raise ValidationError({'rf_channel_width': _("Cannot specify custom width with channel selected.")})
1332 # VLAN validation
1333 if not self.mode and self.untagged_vlan: 1333 ↛ 1334line 1333 didn't jump to line 1334 because the condition on line 1333 was never true
1334 raise ValidationError({'untagged_vlan': _("Interface mode does not support an untagged vlan.")})
1336 # Validate untagged VLAN
1337 if self.untagged_vlan and self.untagged_vlan.site not in [self.device.site, None]: 1337 ↛ 1338line 1337 didn't jump to line 1338 because the condition on line 1337 was never true
1338 raise ValidationError({
1339 'untagged_vlan': _(
1340 "The untagged VLAN ({untagged_vlan}) must belong to the same site as the interface's parent "
1341 "device, or it must be global."
1342 ).format(untagged_vlan=self.untagged_vlan)
1343 })
1345 def save(self, *args, **kwargs):
1347 # Set absolute channel attributes from selected options
1348 if self.rf_channel and not self.rf_channel_frequency: 1348 ↛ 1349line 1348 didn't jump to line 1349 because the condition on line 1348 was never true
1349 self.rf_channel_frequency = get_channel_attr(self.rf_channel, 'frequency')
1350 if self.rf_channel and not self.rf_channel_width: 1350 ↛ 1351line 1350 didn't jump to line 1351 because the condition on line 1350 was never true
1351 self.rf_channel_width = get_channel_attr(self.rf_channel, 'width')
1353 # InterfaceChannelRenameMixin.save() (reached via super(), first in the MRO) detects and cascades a channelized
1354 # parent rename around this call.
1355 super().save(*args, **kwargs)
1357 @property
1358 def _occupied(self):
1359 return super()._occupied or bool(self.wireless_link_id)
1361 @property
1362 def is_wired(self):
1363 # Also excludes any channel subinterface, which derives its cable from the channelized parent.
1364 return self.type not in NONCONNECTABLE_IFACE_TYPES and self.channel_id is None
1366 @property
1367 def is_virtual(self):
1368 return self.type in VIRTUAL_IFACE_TYPES
1370 @property
1371 def is_wireless(self):
1372 return self.type in WIRELESS_IFACE_TYPES
1374 @property
1375 def is_lag(self):
1376 return self.type == InterfaceTypeChoices.TYPE_LAG
1378 @property
1379 def is_bridge(self):
1380 return self.type == InterfaceTypeChoices.TYPE_BRIDGE
1382 @property
1383 def is_channel(self):
1384 # Identified by channel_id, not type — it may keep its own specific physical type instead of "channel".
1385 return self.channel_id is not None
1387 @property
1388 def link(self):
1389 return self.cable or self.wireless_link
1391 @cached_property
1392 def link_peers(self):
1393 if self.cable: 1393 ↛ 1394line 1393 didn't jump to line 1394 because the condition on line 1393 was never true
1394 return super().link_peers
1395 if self.wireless_link: 1395 ↛ 1397line 1395 didn't jump to line 1397 because the condition on line 1395 was never true
1396 # Return the opposite side of the attached wireless link
1397 if self.wireless_link.interface_a == self:
1398 return [self.wireless_link.interface_b]
1399 return [self.wireless_link.interface_a]
1400 return []
1402 @property
1403 def l2vpn_termination(self):
1404 return self.l2vpn_terminations.first()
1406 @cached_property
1407 def connected_endpoints(self):
1408 # If this is a virtual interface, return the remote endpoint of the connected
1409 # virtual circuit, if any.
1410 if self.is_virtual and hasattr(self, 'virtual_circuit_termination'): 1410 ↛ 1411line 1410 didn't jump to line 1411 because the condition on line 1410 was never true
1411 return self.virtual_circuit_termination.peer_terminations
1412 return super().connected_endpoints
1414 def set_cable_termination(self, termination):
1415 super().set_cable_termination(termination)
1417 # A channelized interface carries no path of its own; instead, its cable is mirrored onto each channel
1418 # subinterface (occupying a single position of the shared connector) so that each channel traces independently.
1419 if self.channels:
1420 self.propagate_channel_cables()
1422 def clear_cable_termination(self, termination):
1423 super().clear_cable_termination(termination)
1425 if self.channels:
1426 self.clear_channel_cables()
1428 def propagate_channel_cables(self):
1429 """
1430 Mirror this channelized interface's cable attributes onto each of its channel subinterfaces, restricting each
1431 child to the single connector position identified by its channel_id. Only profiled cables map connector
1432 positions to channels; a positionless (unprofiled) cable carries no per-channel path, so nothing is mirrored.
1433 """
1434 # Only a profiled cable defines the connector positions that channels map onto; without one, clear any
1435 # previously-mirrored attributes rather than propagate an unusable cable reference.
1436 if not (self.cable and self.cable.profile):
1437 self.clear_channel_cables()
1438 return
1440 # Mirror via bulk_update() to issue a single UPDATE and, crucially, to bypass the post_save signal — a
1441 # per-child save() would re-trigger update_channelized_cable_paths() and recurse indefinitely.
1442 children = list(self.child_interfaces.filter(channel_id__isnull=False))
1443 for child in children:
1444 child.cable = self.cable
1445 child.cable_end = self.cable_end
1446 child.cable_connector = self.cable_connector
1447 child.cable_positions = [child.channel_id]
1448 type(self).objects.bulk_update(
1449 children, ['cable', 'cable_end', 'cable_connector', 'cable_positions']
1450 )
1452 def clear_channel_cables(self):
1453 """
1454 Clear the mirrored cable attributes from this channelized interface's channel subinterfaces.
1455 """
1456 # A queryset update() clears every child in a single query and bypasses the post_save signal (see above).
1457 # cable_end is cleared to '' to match the convention used elsewhere when nullifying a termination (see
1458 # nullify_connected_endpoints() and update_channelized_cable_paths() in dcim.signals).
1459 self.child_interfaces.filter(channel_id__isnull=False).update(
1460 cable=None,
1461 cable_end='',
1462 cable_connector=None,
1463 cable_positions=None,
1464 )
1467#
1468# Pass-through ports
1469#
1471class PortMapping(ChangeLoggingMixin, PortMappingBase):
1472 """
1473 Maps a FrontPort & position to a RearPort & position.
1474 """
1475 device = models.ForeignKey(
1476 to='dcim.Device',
1477 on_delete=models.CASCADE,
1478 related_name='port_mappings',
1479 )
1480 front_port = models.ForeignKey(
1481 to='dcim.FrontPort',
1482 on_delete=models.CASCADE,
1483 related_name='mappings',
1484 )
1485 rear_port = models.ForeignKey(
1486 to='dcim.RearPort',
1487 on_delete=models.CASCADE,
1488 related_name='mappings',
1489 )
1491 class Meta(PortMappingBase.Meta):
1492 # Inherit the unique constraints from PortMappingBase.Meta.
1493 pass
1495 def clean(self):
1496 super().clean()
1498 # Both ports must belong to the same device
1499 if self.front_port.device_id != self.rear_port.device_id:
1500 raise ValidationError({
1501 "rear_port": _("Rear port ({rear_port}) must belong to the same device").format(
1502 rear_port=self.rear_port
1503 )
1504 })
1506 def save(self, *args, **kwargs):
1507 # Associate the mapping with the parent Device
1508 self.device = self.front_port.device
1509 super().save(*args, **kwargs)
1512class FrontPort(ModularComponentModel, CabledObjectModel, TrackingModelMixin):
1513 """
1514 A pass-through port on the front of a Device.
1515 """
1516 type = models.CharField(
1517 verbose_name=_('type'),
1518 max_length=50,
1519 choices=PortTypeChoices
1520 )
1521 color = ColorField(
1522 verbose_name=_('color'),
1523 blank=True
1524 )
1525 positions = models.PositiveSmallIntegerField(
1526 verbose_name=_('positions'),
1527 default=1,
1528 validators=[
1529 MinValueValidator(PORT_POSITION_MIN),
1530 MaxValueValidator(PORT_POSITION_MAX)
1531 ],
1532 )
1534 clone_fields = ('device', 'type', 'color', 'positions')
1536 class Meta(ModularComponentModel.Meta):
1537 constraints = (
1538 models.UniqueConstraint(
1539 fields=('device', 'name'),
1540 name='%(app_label)s_%(class)s_unique_device_name'
1541 ),
1542 )
1543 verbose_name = _('front port')
1544 verbose_name_plural = _('front ports')
1546 def clean(self):
1547 super().clean()
1549 # Check that positions is greater than or equal to the number of associated RearPorts
1550 if not self._state.adding:
1551 mapping_count = self.mappings.count()
1552 if self.positions < mapping_count:
1553 raise ValidationError({
1554 "positions": _(
1555 "The number of positions cannot be less than the number of mapped rear ports ({count})"
1556 ).format(count=mapping_count)
1557 })
1560class RearPort(ModularComponentModel, CabledObjectModel, TrackingModelMixin):
1561 """
1562 A pass-through port on the rear of a Device.
1563 """
1564 type = models.CharField(
1565 verbose_name=_('type'),
1566 max_length=50,
1567 choices=PortTypeChoices
1568 )
1569 color = ColorField(
1570 verbose_name=_('color'),
1571 blank=True
1572 )
1573 positions = models.PositiveSmallIntegerField(
1574 verbose_name=_('positions'),
1575 default=1,
1576 validators=[
1577 MinValueValidator(PORT_POSITION_MIN),
1578 MaxValueValidator(PORT_POSITION_MAX)
1579 ],
1580 )
1582 clone_fields = ('device', 'type', 'color', 'positions')
1584 class Meta(ModularComponentModel.Meta):
1585 verbose_name = _('rear port')
1586 verbose_name_plural = _('rear ports')
1588 def clean(self):
1589 super().clean()
1591 # Check that positions count is greater than or equal to the number of associated FrontPorts
1592 if not self._state.adding:
1593 mapping_count = self.mappings.count()
1594 if self.positions < mapping_count:
1595 raise ValidationError({
1596 "positions": _(
1597 "The number of positions cannot be less than the number of mapped front ports "
1598 "({count})"
1599 ).format(count=mapping_count)
1600 })
1603#
1604# Bays
1605#
1607class ModuleBay(ModularComponentModel, TrackingModelMixin, LtreeModel):
1608 """
1609 An empty space within a Device which can house a child device
1610 """
1611 parent = models.ForeignKey(
1612 to='self',
1613 on_delete=models.CASCADE,
1614 related_name='children',
1615 blank=True,
1616 null=True,
1617 editable=False,
1618 db_index=True
1619 )
1620 position = models.CharField(
1621 verbose_name=_('position'),
1622 max_length=30,
1623 blank=True,
1624 help_text=_('Identifier to reference when renaming installed components')
1625 )
1626 enabled = models.BooleanField(
1627 verbose_name=_('enabled'),
1628 default=True,
1629 )
1630 module_bay_types = models.ManyToManyField(
1631 to='dcim.ModuleBayType',
1632 related_name='module_bays',
1633 blank=True,
1634 verbose_name=_('module bay types'),
1635 help_text=_('Types of modules that can be installed in this bay (empty = unconstrained)'),
1636 )
1637 # sort_path inherits `name`'s natural_sort collation automatically (LtreeModelBase),
1638 # so ORDER BY sort_path sorts siblings naturally (Slot 0..Slot 13) — as MPTT's
1639 # order_insertion_by=('name',) did — rather than lexicographically.
1640 sort_path = SortPathField(
1641 editable=False,
1642 blank=True,
1643 default='',
1644 )
1646 clone_fields = ('device', 'enabled')
1648 objects = LtreeManager()
1650 class Meta(ModularComponentModel.Meta):
1651 # Order by sort_path alone (not device-first), reproducing the MPTT
1652 # ModuleBayManager's ('_root_name', 'lft'): sort_path begins with the tree's
1653 # root-bay name (natural_sort collation), so the global list groups by
1654 # root-bay name across devices, descendants following their root. `pk`
1655 # gives a deterministic tie-break among same-named roots on different devices
1656 # (MPTT's lft=1 left this order arbitrary).
1657 ordering = ('sort_path', 'pk')
1658 indexes = (
1659 GistIndex(fields=['path'], name='dcim_modulebay_path_gist'),
1660 models.Index(fields=['sort_path'], name='dcim_modulebay_sort_path_idx'),
1661 )
1662 constraints = (
1663 models.UniqueConstraint(
1664 fields=('device', 'module', 'name'),
1665 name='%(app_label)s_%(class)s_unique_device_module_name'
1666 ),
1667 )
1668 verbose_name = _('module bay')
1669 verbose_name_plural = _('module bays')
1671 def clean(self):
1672 super().clean()
1674 # Check for recursion
1675 if module := self.module:
1676 module_bays = [self.pk]
1677 modules = []
1678 while module:
1679 if module.pk in modules or module.module_bay.pk in module_bays:
1680 raise ValidationError(_("A module bay cannot belong to a module installed within it."))
1681 modules.append(module.pk)
1682 module_bays.append(module.module_bay.pk)
1683 module = module.module_bay.module if module.module_bay else None
1685 def save(self, *args, **kwargs):
1686 if self.module:
1687 self.parent = self.module.module_bay
1688 else:
1689 self.parent = None
1690 super().save(*args, **kwargs)
1692 def _parent_creates_cycle(self):
1693 # A ModuleBay's parent is system-derived from its module (see save()), not
1694 # user-assigned, and module/bay recursion is validated in clean(); skip the
1695 # generic ltree cycle guard.
1696 return False
1698 @property
1699 def _occupied(self):
1700 """
1701 Indicates whether the module bay is occupied by a module.
1702 """
1703 return bool(not self.enabled or hasattr(self, 'installed_module'))
1705 @property
1706 def is_module_compatible(self):
1707 """
1708 Return True if the installed module (if any) is compatible with this bay's type constraints,
1709 or if this bay has no type constraints, or if no module is installed.
1710 Returns False when this bay and the installed module's type have non-empty, disjoint bay type sets.
1711 """
1712 module = getattr(self, 'installed_module', None)
1713 if module is None:
1714 return True
1715 # Use .all() so a prefetch cache is honoured; see Module.is_bay_compatible for details.
1716 bay_types = {t.pk for t in self.module_bay_types.all()}
1717 if not bay_types:
1718 return True
1719 type_types = {t.pk for t in module.module_type.module_bay_types.all()}
1720 if type_types and not (bay_types & type_types):
1721 return False
1722 return True
1724 def get_incompatible_module(self):
1725 """
1726 Return the installed Module if it is incompatible with this bay's type constraints, else None.
1727 """
1728 module = getattr(self, 'installed_module', None)
1729 return module if module and not self.is_module_compatible else None
1732class DeviceBay(ComponentModel, TrackingModelMixin):
1733 """
1734 An empty space within a Device which can house a child device
1735 """
1736 installed_device = models.OneToOneField(
1737 to='dcim.Device',
1738 on_delete=models.SET_NULL,
1739 related_name='parent_bay',
1740 blank=True,
1741 null=True
1742 )
1743 enabled = models.BooleanField(
1744 verbose_name=_('enabled'),
1745 default=True,
1746 )
1748 clone_fields = ('device', 'enabled')
1750 class Meta(ComponentModel.Meta):
1751 verbose_name = _('device bay')
1752 verbose_name_plural = _('device bays')
1754 def clean(self):
1755 super().clean()
1757 # Validate that the parent Device can have DeviceBays
1758 if hasattr(self, 'device') and not self.device.device_type.is_parent_device:
1759 raise ValidationError(_("This type of device ({device_type}) does not support device bays.").format(
1760 device_type=self.device.device_type
1761 ))
1763 # Prevent installing a device into a disabled bay
1764 if self.installed_device and not self.enabled:
1765 current_installed_device_id = (
1766 DeviceBay.objects.filter(pk=self.pk).values_list('installed_device_id', flat=True).first()
1767 )
1768 if self.pk is None or current_installed_device_id != self.installed_device_id:
1769 raise ValidationError({
1770 'installed_device': _("Cannot install a device in a disabled device bay.")
1771 })
1773 # Cannot install a device into itself, obviously
1774 if self.installed_device and getattr(self, 'device', None) == self.installed_device:
1775 raise ValidationError(_("Cannot install a device into itself."))
1777 # Check that the installed device is not already installed elsewhere
1778 if self.installed_device:
1779 current_bay = DeviceBay.objects.filter(installed_device=self.installed_device).first()
1780 if current_bay and current_bay != self:
1781 raise ValidationError({
1782 'installed_device': _(
1783 "Cannot install the specified device; device is already installed in {bay}."
1784 ).format(bay=current_bay)
1785 })
1787 @property
1788 def _occupied(self):
1789 """
1790 Indicates whether the device bay is occupied by a child device.
1791 """
1792 return bool(not self.enabled or self.installed_device_id)
1795#
1796# Inventory items
1797#
1800class InventoryItemRole(OrganizationalModel):
1801 """
1802 Inventory items may optionally be assigned a functional role.
1803 """
1804 color = ColorField(
1805 verbose_name=_('color'),
1806 default=ColorChoices.COLOR_GREY
1807 )
1809 class Meta:
1810 ordering = ('name',)
1811 verbose_name = _('inventory item role')
1812 verbose_name_plural = _('inventory item roles')
1815class InventoryItem(LtreeModel, ComponentModel, TrackingModelMixin):
1816 """
1817 An InventoryItem represents a serialized piece of hardware within a Device, such as a line card or power supply.
1818 InventoryItems are used only for inventory purposes.
1819 """
1820 parent = models.ForeignKey(
1821 to='self',
1822 on_delete=models.CASCADE,
1823 related_name='child_items',
1824 blank=True,
1825 null=True,
1826 db_index=True
1827 )
1828 component_type = models.ForeignKey(
1829 to='contenttypes.ContentType',
1830 on_delete=models.PROTECT,
1831 related_name='+',
1832 blank=True,
1833 null=True
1834 )
1835 component_id = models.PositiveBigIntegerField(
1836 blank=True,
1837 null=True
1838 )
1839 component = GenericForeignKey(
1840 ct_field='component_type',
1841 fk_field='component_id'
1842 )
1843 status = models.CharField(
1844 verbose_name=_('status'),
1845 max_length=50,
1846 choices=InventoryItemStatusChoices,
1847 default=InventoryItemStatusChoices.STATUS_ACTIVE
1848 )
1849 role = models.ForeignKey(
1850 to='dcim.InventoryItemRole',
1851 on_delete=models.PROTECT,
1852 related_name='inventory_items',
1853 blank=True,
1854 null=True
1855 )
1856 manufacturer = models.ForeignKey(
1857 to='dcim.Manufacturer',
1858 on_delete=models.PROTECT,
1859 related_name='inventory_items',
1860 blank=True,
1861 null=True
1862 )
1863 part_id = models.CharField(
1864 max_length=50,
1865 verbose_name=_('part ID'),
1866 blank=True,
1867 help_text=_('Manufacturer-assigned part identifier')
1868 )
1869 serial = models.CharField(
1870 max_length=50,
1871 verbose_name=_('serial number'),
1872 blank=True
1873 )
1874 asset_tag = models.CharField(
1875 max_length=50,
1876 unique=True,
1877 blank=True,
1878 null=True,
1879 verbose_name=_('asset tag'),
1880 help_text=_('A unique tag used to identify this item')
1881 )
1882 discovered = models.BooleanField(
1883 verbose_name=_('discovered'),
1884 default=False,
1885 help_text=_('This item was automatically discovered')
1886 )
1888 clone_fields = ('device', 'parent', 'role', 'manufacturer', 'status', 'part_id')
1890 objects = LtreeManager()
1892 class Meta:
1893 # Global list is flat + alphabetical by name (natural_sort collation). The
1894 # per-device Inventory tab renders the hierarchy instead — DeviceInventoryView
1895 # .get_children() orders that by `path`. `pk` is a deterministic tie-break for
1896 # same-named items on different devices.
1897 ordering = ('name', 'pk')
1898 indexes = (
1899 models.Index(fields=('component_type', 'component_id')),
1900 GistIndex(fields=['path'], name='dcim_inventoryitem_path_gist'),
1901 )
1902 constraints = (
1903 models.UniqueConstraint(
1904 fields=('device', 'parent', 'name'),
1905 name='%(app_label)s_%(class)s_unique_device_parent_name'
1906 ),
1907 )
1908 verbose_name = _('inventory item')
1909 verbose_name_plural = _('inventory items')
1911 def clean(self):
1912 super().clean()
1914 # Validation for moving InventoryItems
1915 if not self._state.adding:
1916 # Cannot move an InventoryItem to another device if it has a parent
1917 if self.parent and self.parent.device != self.device:
1918 raise ValidationError({
1919 "parent": _("Parent inventory item does not belong to the same device.")
1920 })
1922 # Prevent moving InventoryItems with children
1923 first_child = self.get_children().first()
1924 if first_child and first_child.device != self.device:
1925 raise ValidationError(_("Cannot move an inventory item with dependent children"))
1927 # When moving an InventoryItem to another device, remove any associated component
1928 if self.component and self.component.device != self.device:
1929 self.component = None
1930 else:
1931 if self.component and self.component.device != self.device:
1932 raise ValidationError({
1933 "device": _("Cannot assign inventory item to component on another device")
1934 })
1936 def get_status_color(self):
1937 return InventoryItemStatusChoices.colors.get(self.status)