Coverage for dcim/models/mixins.py: 50%
207 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from decimal import Decimal
3from django.apps import apps
4from django.contrib.contenttypes.fields import GenericForeignKey
5from django.core.exceptions import ValidationError
6from django.core.validators import MinValueValidator
7from django.db import IntegrityError, models, transaction
8from django.utils.translation import gettext_lazy as _
10from dcim.choices import InterfaceTypeChoices
11from dcim.constants import NONCONNECTABLE_IFACE_TYPES, VIRTUAL_IFACE_TYPES, WIRELESS_IFACE_TYPES
12from netbox.choices import *
13from utilities.conversion import (
14 to_liters_per_minute,
15 to_millimeters,
16)
17from utilities.data import normalize_update_fields
19__all__ = (
20 'CachedScopeMixin',
21 'CoolingLoopValidationMixin',
22 'DiameterMixin',
23 'InterfaceChannelRenameMixin',
24 'InterfaceValidationMixin',
25 'MaxFlowMixin',
26 'RenderConfigMixin',
27)
30class RenderConfigMixin(models.Model):
31 config_template = models.ForeignKey(
32 to='extras.ConfigTemplate',
33 on_delete=models.PROTECT,
34 related_name='%(class)ss',
35 blank=True,
36 null=True
37 )
39 class Meta:
40 abstract = True
42 def get_config_template(self):
43 """
44 Return the appropriate ConfigTemplate (if any) for this Device.
45 """
46 if self.config_template: 46 ↛ 47line 46 didn't jump to line 47 because the condition on line 46 was never true
47 return self.config_template
48 if self.role and self.role.config_template: 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true
49 return self.role.config_template
50 if self.platform and self.platform.config_template: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true
51 return self.platform.config_template
52 return None
55class CachedScopeMixin(models.Model):
56 """
57 Mixin for adding a GenericForeignKey scope to a model that can point to a Region, SiteGroup, Site, or Location.
58 Includes cached fields for each to allow efficient filtering. Appropriate validation must be done in the clean()
59 method as this does not have any as validation is generally model-specific.
60 """
61 scope_type = models.ForeignKey(
62 to='contenttypes.ContentType',
63 on_delete=models.PROTECT,
64 related_name='+',
65 blank=True,
66 null=True
67 )
68 scope_id = models.PositiveBigIntegerField(
69 blank=True,
70 null=True
71 )
72 scope = GenericForeignKey(
73 ct_field='scope_type',
74 fk_field='scope_id'
75 )
77 _location = models.ForeignKey(
78 to='dcim.Location',
79 on_delete=models.CASCADE,
80 blank=True,
81 null=True
82 )
83 _site = models.ForeignKey(
84 to='dcim.Site',
85 on_delete=models.CASCADE,
86 blank=True,
87 null=True
88 )
89 # SET_NULL, not CASCADE: these cache an ancestor of the actual scope, so deleting that
90 # ancestor must not delete this object. Deletion of a Region/SiteGroup that *is* the
91 # actual scope is handled independently via its GenericRelation to this model.
92 _region = models.ForeignKey(
93 to='dcim.Region',
94 on_delete=models.SET_NULL,
95 blank=True,
96 null=True
97 )
98 _site_group = models.ForeignKey(
99 to='dcim.SiteGroup',
100 on_delete=models.SET_NULL,
101 blank=True,
102 null=True
103 )
105 class Meta:
106 abstract = True
108 def clean(self):
109 if self.scope_type and not (self.scope or self.scope_id):
110 scope_type = self.scope_type.model_class()
111 raise ValidationError(
112 _("Please select a {scope_type}.").format(scope_type=scope_type._meta.model_name)
113 )
114 super().clean()
116 def save(self, *args, **kwargs):
117 # Cache objects associated with the terminating object (for filtering)
118 self.cache_related_objects()
120 super().save(*args, **kwargs)
122 def cache_related_objects(self):
123 self._region = self._site_group = self._site = self._location = None
124 if self.scope_type:
125 scope_type = self.scope_type.model_class()
126 if scope_type == apps.get_model('dcim', 'region'): 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true
127 self._region = self.scope
128 elif scope_type == apps.get_model('dcim', 'sitegroup'): 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true
129 self._site_group = self.scope
130 elif scope_type == apps.get_model('dcim', 'site'): 130 ↛ 134line 130 didn't jump to line 134 because the condition on line 130 was always true
131 self._region = self.scope.region
132 self._site_group = self.scope.group
133 self._site = self.scope
134 elif scope_type == apps.get_model('dcim', 'location'):
135 self._region = self.scope.site.region
136 self._site_group = self.scope.site.group
137 self._site = self.scope.site
138 self._location = self.scope
139 cache_related_objects.alters_data = True
142class InterfaceValidationMixin:
144 def clean(self):
145 super().clean()
147 # An interface cannot be its own parent
148 if self.pk and self.parent_id == self.pk: 148 ↛ 149line 148 didn't jump to line 149 because the condition on line 148 was never true
149 raise ValidationError({'parent': _("An interface cannot be its own parent.")})
151 # A channel subinterface may keep its own specific physical type (e.g. 10GBASE-SR) instead of the
152 # generic "channel" type, but never a virtual or wireless type.
153 can_bind_to_channel = (
154 self.type == InterfaceTypeChoices.TYPE_CHANNEL or self.type not in NONCONNECTABLE_IFACE_TYPES
155 )
156 # During bulk-creation pattern validation (a replication base), channel_id is not yet assigned — it is
157 # supplied per-instance during expansion — so the parent/channel_id presence checks below are relaxed.
158 is_replicated_base = getattr(self, '_replicated_base', False)
160 # An interface may have a parent only if virtual, or bound to a channel on that parent.
161 if self.parent_id and self.type != InterfaceTypeChoices.TYPE_VIRTUAL: 161 ↛ 162line 161 didn't jump to line 162 because the condition on line 161 was never true
162 if self.channel_id is None and not (is_replicated_base and can_bind_to_channel):
163 raise ValidationError({
164 'parent': _(
165 "Only virtual interfaces, or a channel subinterface with a channel ID assigned, may be "
166 "assigned to a parent interface."
167 )
168 })
170 # Only one layer of channelization is permitted: an interface cannot be both channelized and a channel
171 if self.channels and self.channel_id: 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true
172 raise ValidationError(
173 _("An interface cannot be both channelized and bound to a channel on a parent interface.")
174 )
176 # Only physical interfaces may be channelized
177 if self.channels and self.type in NONCONNECTABLE_IFACE_TYPES: 177 ↛ 178line 177 didn't jump to line 178 because the condition on line 177 was never true
178 raise ValidationError({
179 'channels': _("{display_type} interfaces cannot be channelized.").format(
180 display_type=self.get_type_display()
181 )
182 })
184 # The channel type and channel_id are mutually dependent. The channel_id requirement is relaxed for a
185 # replication base (bulk creation), where each channel_id is supplied per-instance during expansion.
186 if self.type == InterfaceTypeChoices.TYPE_CHANNEL and self.channel_id is None and not is_replicated_base: 186 ↛ 187line 186 didn't jump to line 187 because the condition on line 186 was never true
187 raise ValidationError({
188 'channel_id': _("Channel interfaces must have a channel ID assigned.")
189 })
190 if self.channel_id is not None and not can_bind_to_channel: 190 ↛ 191line 190 didn't jump to line 191 because the condition on line 190 was never true
191 raise ValidationError({
192 'channel_id': _(
193 "A channel ID cannot be assigned to a virtual, LAG, bridge, or wireless interface."
194 )
195 })
197 # A channel subinterface must be bound to a channelized parent. A replication base is checked too, so an
198 # invalid parent selection is caught before pattern expansion rather than per-instance.
199 if self.channel_id is not None or (is_replicated_base and can_bind_to_channel and self.parent_id): 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 if self.parent is None:
201 raise ValidationError({
202 'parent': _("A channel subinterface must be assigned to a parent interface.")
203 })
204 if not self.parent.channels:
205 raise ValidationError({
206 'parent': _("The parent interface ({interface}) is not channelized.").format(
207 interface=self.parent
208 )
209 })
210 if self.channel_id and self.channel_id > self.parent.channels:
211 raise ValidationError({
212 'channel_id': _(
213 "Invalid channel ID ({channel_id}): the parent interface provides only {channels} channels."
214 ).format(channel_id=self.channel_id, channels=self.parent.channels)
215 })
217 # Reducing or clearing the channel count cannot orphan an existing child bound to a higher channel. Gated
218 # on channels/_original_channels so this stays off the hot path for never-channelized interfaces.
219 if self.pk and (self.channels or self._original_channels): 219 ↛ 220line 219 didn't jump to line 220 because the condition on line 219 was never true
220 max_child_channel_id = self.child_interfaces.filter(
221 channel_id__gt=self.channels or 0
222 ).aggregate(models.Max('channel_id'))['channel_id__max']
223 if max_child_channel_id is not None:
224 if self.channels:
225 message = _(
226 "Cannot set channels to {channels}: a channel subinterface is bound to channel "
227 "{channel_id}. Delete or reassign the affected subinterface(s) first."
228 ).format(channels=self.channels, channel_id=max_child_channel_id)
229 else:
230 message = _(
231 "Cannot remove channelization: a channel subinterface is bound to channel {channel_id}. "
232 "Delete or reassign the affected subinterface(s) first."
233 ).format(channel_id=max_child_channel_id)
234 raise ValidationError({'channels': message})
236 # An interface cannot be bridged to itself
237 if self.pk and self.bridge_id == self.pk: 237 ↛ 238line 237 didn't jump to line 238 because the condition on line 237 was never true
238 raise ValidationError({'bridge': _("An interface cannot be bridged to itself.")})
240 # Only physical interfaces may have a PoE mode/type assigned
241 if self.poe_mode and self.type in VIRTUAL_IFACE_TYPES: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true
242 raise ValidationError({
243 'poe_mode': _("Virtual interfaces cannot have a PoE mode.")
244 })
245 if self.poe_type and self.type in VIRTUAL_IFACE_TYPES: 245 ↛ 246line 245 didn't jump to line 246 because the condition on line 245 was never true
246 raise ValidationError({
247 'poe_type': _("Virtual interfaces cannot have a PoE type.")
248 })
250 # An interface with a PoE type set must also specify a mode
251 if self.poe_type and not self.poe_mode: 251 ↛ 252line 251 didn't jump to line 252 because the condition on line 251 was never true
252 raise ValidationError({
253 'poe_type': _("Must specify PoE mode when designating a PoE type.")
254 })
256 # RF role may be set only for wireless interfaces
257 if self.rf_role and self.type not in WIRELESS_IFACE_TYPES: 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 raise ValidationError({'rf_role': _("Wireless role may be set only on wireless interfaces.")})
261class InterfaceChannelRenameMixin:
262 """
263 Cooperative __init__()/save() mixin for Interface and InterfaceTemplate: detects a rename of a channelized
264 parent and cascades it to any channel subinterface which follows the "<parent name>:<channel ID>" naming
265 convention.
267 Must precede the model's other bases so its __init__()/save() sit ahead of them in the MRO; both delegate
268 onward via super(), so a consuming model only needs to list this mixin first among its bases and call
269 super().__init__()/super().save() as usual -- no extra wiring required.
270 """
272 def __init__(self, *args, **kwargs):
273 super().__init__(*args, **kwargs)
274 self._original_name = self.__dict__.get('name')
275 # Also relied on by InterfaceValidationMixin.clean() (a channel-count reduction that would orphan an
276 # existing child). Tracked here rather than per-model so both concerns share one source of truth.
277 self._original_channels = self.__dict__.get('channels')
279 def save(self, *args, **kwargs):
280 update_fields = normalize_update_fields(kwargs)
281 # A save() whose update_fields excludes 'name'/'channels' won't actually persist that attribute, so the
282 # cascade decision below can't treat self.name/self.channels as current in that case -- fall back to the
283 # last known persisted value instead. Without this, e.g. clearing self.channels in memory and saving
284 # with update_fields=['name'] would see a falsy self.channels and skip a cascade the DB still requires.
285 name_persisted = update_fields is None or 'name' in update_fields
286 channels_persisted = update_fields is None or 'channels' in update_fields
287 is_channelized = self.channels if channels_persisted else self._original_channels
288 old_name, new_name = self._original_name, self.name
289 renamed = bool(self.pk and is_channelized and name_persisted and new_name != old_name)
291 super().save(*args, **kwargs)
292 # Captured after super().save() so it reflects the DB actually used -- which, when this save() was
293 # called with an explicit using=, is not necessarily what router.db_for_write() would return if
294 # re-run here.
295 db_alias = self._state.db
297 if name_persisted: 297 ↛ 299line 297 didn't jump to line 299 because the condition on line 297 was always true
298 self._original_name = new_name
299 if channels_persisted: 299 ↛ 302line 299 didn't jump to line 302 because the condition on line 299 was always true
300 self._original_channels = self.channels
302 if renamed: 302 ↛ 304line 302 didn't jump to line 304 because the condition on line 302 was never true
303 # Defer until commit so a later save in the same transaction cannot overwrite the cascade.
304 transaction.on_commit(
305 lambda: self._rename_channel_subinterfaces(old_name, new_name, db_alias),
306 using=db_alias,
307 )
309 def _rename_channel_subinterfaces(self, old_name, new_name, db_alias):
310 """
311 Rename each channel subinterface following the "<parent name>:<channel ID>" convention to match this
312 interface's new name. A subinterface named otherwise is left untouched, as is one whose renamed form
313 would exceed the name field's max length or collide with an existing sibling.
314 """
315 max_name_length = self._meta.get_field('name').max_length
316 # This runs from an on_commit callback, after the triggering save()'s own transaction has already
317 # committed -- so without this outer atomic(), each child below would run in its own independent,
318 # auto-committing transaction rather than a savepoint, and an unexpected failure partway through
319 # could leave only some of the child set renamed.
320 with transaction.atomic(using=db_alias):
321 for child in self.child_interfaces.using(db_alias).filter(channel_id__isnull=False):
322 if child.name != f'{old_name}:{child.channel_id}':
323 continue
324 candidate_name = f'{new_name}:{child.channel_id}'
325 if len(candidate_name) > max_name_length:
326 continue
327 # A full save() (not a queryset update()) so _name, last_updated, and the changelog get updated
328 # too; a channel subinterface can never itself be channelized, so this can't recurse into the
329 # cascade. update_fields is restricted to what actually changed so unrelated receivers (e.g.
330 # Interface's own cable-path rebuild) can skip redundant work.
331 child.snapshot()
332 child.name = candidate_name
333 # Renamed in its own savepoint: the DB's unique constraint is the sole arbiter of a collision,
334 # and a collision on one child can't abort the rename of the others.
335 try:
336 with transaction.atomic(using=db_alias):
337 child.save(using=db_alias, update_fields=['name', '_name', 'last_updated'])
338 except IntegrityError:
339 # Confirm this was really the expected name collision (not some other constraint) before
340 # treating it as safe to skip. The (device, name) constraint is declared via
341 # Meta.constraints, which validate_unique() does not check -- only validate_constraints()
342 # does.
343 try:
344 child.validate_constraints()
345 except ValidationError:
346 continue
347 raise
350class CoolingLoopValidationMixin:
351 """
352 Adds loop detection to the coolant chain formed by cooling intakes and outflows. A CoolingIntake is supplied
353 by an upstream CoolingOutflow (via `cooling_outflow`), which may in turn be supplied by an upstream
354 CoolingIntake on the same device (via `cooling_intake`), and so on; this chain must remain acyclic.
356 Each concrete model declares `upstream_field`, the name of its foreign key to the next component upstream.
357 The chain alternates between the two models, so the walk simply follows each visited component's own
358 upstream field in turn. (The field is resolved by name rather than referencing the models directly, as this
359 module is imported by the ones defining them.)
360 """
361 upstream_field = None
363 @classmethod
364 def _get_upstream_field(cls):
365 return cls._meta.get_field(cls.upstream_field)
367 def validate_cooling_loop(self):
368 """
369 Raise a ValidationError if this component's upstream assignment forms a loop.
371 Each hop resolves only the next foreign key ID (a single indexed column lookup) rather than loading
372 full related objects, and the `seen` set of (model, pk) pairs guarantees termination.
373 """
374 seen = set()
375 if self.pk:
376 seen.add((type(self), self.pk))
378 # Seed the walk from this (possibly unsaved) component's in-memory foreign key
379 field = self._get_upstream_field()
380 model, pk = field.related_model, getattr(self, field.attname)
382 while pk is not None:
383 if (model, pk) in seen:
384 raise ValidationError(_("Cooling intake and outflow assignments cannot form a loop."))
385 seen.add((model, pk))
387 # Advance to the component upstream of the one just visited
388 field = model._get_upstream_field()
389 pk = model.objects.filter(pk=pk).values_list(field.attname, flat=True).first()
390 model = field.related_model
393class DiameterMixin(models.Model):
394 diameter = models.DecimalField(
395 verbose_name=_('diameter'),
396 max_digits=8,
397 decimal_places=2,
398 blank=True,
399 null=True,
400 validators=[MinValueValidator(Decimal('0.01'))],
401 )
402 diameter_unit = models.CharField(
403 verbose_name=_('diameter unit'),
404 max_length=50,
405 choices=DiameterUnitChoices,
406 blank=True,
407 null=True,
408 )
409 # Stores the normalized diameter (in millimeters) for database ordering
410 _abs_diameter = models.DecimalField(
411 max_digits=13,
412 decimal_places=4,
413 blank=True,
414 null=True
415 )
417 class Meta:
418 abstract = True
420 @property
421 def abs_diameter(self):
422 # Public alias for _abs_diameter; Django templates cannot access underscore-prefixed attributes.
423 return self._abs_diameter
425 def normalize_diameter(self):
426 """
427 Store the given diameter (if any) in millimeters for use in database ordering. Called by save(), and
428 directly by component instantiation, which bypasses save() via bulk_create().
429 """
430 if self.diameter is not None and self.diameter_unit:
431 self._abs_diameter = to_millimeters(self.diameter, self.diameter_unit)
432 else:
433 self._abs_diameter = None
434 if self.diameter is None:
435 self.diameter_unit = None
436 normalize_diameter.alters_data = True
438 def save(self, *args, **kwargs):
439 self.normalize_diameter()
441 super().save(*args, **kwargs)
443 def clean(self):
444 super().clean()
446 # Validate diameter and diameter_unit
447 if self.diameter is not None and not self.diameter_unit:
448 raise ValidationError(_("Must specify a unit when setting a diameter"))
451class MaxFlowMixin(models.Model):
452 """
453 Adds the maximum rate of coolant flow supported by an object, held as a value plus its unit alongside a
454 normalized column (in liters per minute) so that ordering and filtering work across mixed units.
455 """
456 max_flow = models.DecimalField(
457 verbose_name=_('max flow'),
458 max_digits=8,
459 decimal_places=2,
460 blank=True,
461 null=True,
462 validators=[MinValueValidator(Decimal('0.01'))],
463 )
464 max_flow_unit = models.CharField(
465 verbose_name=_('max flow unit'),
466 max_length=50,
467 choices=FlowRateUnitChoices,
468 blank=True,
469 null=True,
470 )
471 # Stores the normalized max flow (in liters per minute) for database ordering
472 _abs_max_flow = models.DecimalField(
473 max_digits=13,
474 decimal_places=4,
475 blank=True,
476 null=True
477 )
479 class Meta:
480 abstract = True
482 @property
483 def abs_max_flow(self):
484 # Public alias for _abs_max_flow; Django templates cannot access underscore-prefixed attributes.
485 return self._abs_max_flow
487 def normalize_max_flow(self):
488 """
489 Store the given max flow (if any) in liters per minute for use in database ordering. Called by save(),
490 and directly by component instantiation, which bypasses save() via bulk_create().
491 """
492 if self.max_flow is not None and self.max_flow_unit:
493 self._abs_max_flow = to_liters_per_minute(self.max_flow, self.max_flow_unit)
494 else:
495 self._abs_max_flow = None
496 if self.max_flow is None:
497 self.max_flow_unit = None
498 normalize_max_flow.alters_data = True
500 def save(self, *args, **kwargs):
501 self.normalize_max_flow()
503 super().save(*args, **kwargs)
505 def clean(self):
506 super().clean()
508 # Validate max_flow and max_flow_unit
509 if self.max_flow is not None and not self.max_flow_unit:
510 raise ValidationError(_("Must specify a unit when setting a maximum flow"))