Coverage for dcim/models/modules.py: 26%
294 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from collections.abc import Iterable, Mapping
3import jsonschema
4import yaml
5from django.conf import settings
6from django.core.exceptions import ValidationError
7from django.db import OperationalError, models, router, transaction
8from django.db.models.signals import post_save
9from django.utils.translation import gettext_lazy as _
10from jsonschema.exceptions import ValidationError as JSONValidationError
12from dcim.choices import *
13from dcim.utils import create_port_mappings, update_interface_bridges, update_interface_parents
14from extras.models import CustomField
15from netbox.models import PrimaryModel
16from netbox.models.features import ImageAttachmentsMixin
17from netbox.models.mixins import WeightMixin
18from utilities.data import normalize_update_fields
19from utilities.exceptions import AbortRequest
20from utilities.fields import ColorField, CounterCacheField
21from utilities.jsonschema import validate_schema
22from utilities.string import title
23from utilities.tracking import TrackingModelMixin
25from .device_components import *
26from .module_moves import ModuleMovePlan
28__all__ = (
29 'Module',
30 'ModuleBayType',
31 'ModuleType',
32 'ModuleTypeProfile',
33)
36class ModuleBayType(PrimaryModel):
37 """
38 A type classification for module bays. When bay types are assigned to both a ModuleBay and a
39 ModuleType, module installation is permitted only if the two sets share at least one common
40 member (i.e. an empty set on either side means unconstrained).
41 """
42 name = models.CharField(
43 verbose_name=_('name'),
44 max_length=100,
45 )
46 slug = models.SlugField(
47 verbose_name=_('slug'),
48 max_length=100,
49 )
50 manufacturer = models.ForeignKey(
51 to='dcim.Manufacturer',
52 on_delete=models.PROTECT,
53 related_name='module_bay_types',
54 blank=True,
55 null=True,
56 )
57 color = ColorField(
58 verbose_name=_('color'),
59 blank=True,
60 )
62 clone_fields = ('manufacturer', 'color')
64 class Meta:
65 ordering = ('manufacturer', 'name')
66 constraints = (
67 models.UniqueConstraint(
68 fields=('manufacturer', 'name'),
69 name='%(app_label)s_%(class)s_unique_manufacturer_name',
70 nulls_distinct=False,
71 ),
72 models.UniqueConstraint(
73 fields=('manufacturer', 'slug'),
74 name='%(app_label)s_%(class)s_unique_manufacturer_slug',
75 nulls_distinct=False,
76 ),
77 )
78 verbose_name = _('module bay type')
79 verbose_name_plural = _('module bay types')
81 def __str__(self):
82 return self.name
84 @property
85 def full_name(self):
86 if self.manufacturer:
87 return f'{self.manufacturer} {self.name}'
88 return self.name
91class ModuleTypeProfile(PrimaryModel):
92 """
93 A profile which defines the attributes which can be set on one or more ModuleTypes.
94 """
95 name = models.CharField(
96 verbose_name=_('name'),
97 max_length=100,
98 unique=True
99 )
100 schema = models.JSONField(
101 blank=True,
102 null=True,
103 validators=[validate_schema],
104 verbose_name=_('schema'),
105 )
107 clone_fields = ('schema',)
109 class Meta:
110 ordering = ('name',)
111 verbose_name = _('module type profile')
112 verbose_name_plural = _('module type profiles')
114 def __str__(self):
115 return self.name
118class ModuleType(ImageAttachmentsMixin, PrimaryModel, WeightMixin):
119 """
120 A ModuleType represents a hardware element that can be installed within a device and which houses additional
121 components; for example, a line card within a chassis-based switch such as the Cisco Catalyst 6500. Like a
122 DeviceType, each ModuleType can have console, power, interface, pass-through port, and module bay templates assigned
123 to it. It cannot, however, house device bays.
124 """
125 profile = models.ForeignKey(
126 to='dcim.ModuleTypeProfile',
127 on_delete=models.PROTECT,
128 related_name='module_types',
129 blank=True,
130 null=True
131 )
132 manufacturer = models.ForeignKey(
133 to='dcim.Manufacturer',
134 on_delete=models.PROTECT,
135 related_name='module_types'
136 )
137 model = models.CharField(
138 verbose_name=_('model'),
139 max_length=100
140 )
141 part_number = models.CharField(
142 verbose_name=_('part number'),
143 max_length=50,
144 blank=True,
145 help_text=_('Discrete part number (optional)')
146 )
147 airflow = models.CharField(
148 verbose_name=_('airflow'),
149 max_length=50,
150 choices=ModuleAirflowChoices,
151 blank=True,
152 null=True
153 )
154 cooling_method = models.CharField(
155 verbose_name=_('cooling method'),
156 max_length=50,
157 choices=CoolingMethodChoices,
158 blank=True,
159 null=True
160 )
161 end_of_life = models.DateField(
162 verbose_name=_('end of life'),
163 blank=True,
164 null=True,
165 help_text=_('The date after which this module type is no longer supported by the manufacturer')
166 )
167 attribute_data = models.JSONField(
168 blank=True,
169 null=True,
170 verbose_name=_('attributes')
171 )
172 module_bay_types = models.ManyToManyField(
173 to='dcim.ModuleBayType',
174 related_name='module_types',
175 blank=True,
176 verbose_name=_('module bay types'),
177 help_text=_('Types of module bays this module type can be installed in (empty = unconstrained)'),
178 )
179 module_count = CounterCacheField(
180 to_model='dcim.Module',
181 to_field='module_type'
182 )
184 # Counter fields
185 console_port_template_count = CounterCacheField(
186 to_model='dcim.ConsolePortTemplate',
187 to_field='module_type'
188 )
189 console_server_port_template_count = CounterCacheField(
190 to_model='dcim.ConsoleServerPortTemplate',
191 to_field='module_type'
192 )
193 power_port_template_count = CounterCacheField(
194 to_model='dcim.PowerPortTemplate',
195 to_field='module_type'
196 )
197 power_outlet_template_count = CounterCacheField(
198 to_model='dcim.PowerOutletTemplate',
199 to_field='module_type'
200 )
201 cooling_intake_template_count = CounterCacheField(
202 to_model='dcim.CoolingIntakeTemplate',
203 to_field='module_type'
204 )
205 cooling_outflow_template_count = CounterCacheField(
206 to_model='dcim.CoolingOutflowTemplate',
207 to_field='module_type'
208 )
209 interface_template_count = CounterCacheField(
210 to_model='dcim.InterfaceTemplate',
211 to_field='module_type'
212 )
213 front_port_template_count = CounterCacheField(
214 to_model='dcim.FrontPortTemplate',
215 to_field='module_type'
216 )
217 rear_port_template_count = CounterCacheField(
218 to_model='dcim.RearPortTemplate',
219 to_field='module_type'
220 )
221 module_bay_template_count = CounterCacheField(
222 to_model='dcim.ModuleBayTemplate',
223 to_field='module_type'
224 )
226 clone_fields = ('profile', 'manufacturer', 'weight', 'weight_unit', 'airflow', 'cooling_method')
227 prerequisite_models = (
228 'dcim.Manufacturer',
229 )
231 class Meta:
232 ordering = ('profile', 'manufacturer', 'model')
233 constraints = (
234 models.UniqueConstraint(
235 fields=('manufacturer', 'model'),
236 name='%(app_label)s_%(class)s_unique_manufacturer_model'
237 ),
238 )
239 indexes = (
240 models.Index(fields=('profile', 'manufacturer', 'model')), # Default ordering
241 )
242 verbose_name = _('module type')
243 verbose_name_plural = _('module types')
245 def __str__(self):
246 return self.model
248 @property
249 def full_name(self):
250 return f"{self.manufacturer} {self.model}"
252 @property
253 def attributes(self):
254 """
255 Returns a human-friendly representation of the attributes defined for a ModuleType according to its profile.
256 """
257 if not self.attribute_data or self.profile is None or not self.profile.schema:
258 return {}
259 attrs = {}
260 for name, options in self.profile.schema.get('properties', {}).items():
261 key = options.get('title', title(name))
262 value = self.attribute_data.get(name)
263 if isinstance(value, Iterable) and not isinstance(value, (str, bytes, Mapping)):
264 value = ', '.join(str(v) for v in value)
265 attrs[key] = value
266 return dict(sorted(attrs.items()))
268 def get_incompatible_modules(self):
269 """
270 Return a queryset of Module instances of this type that are installed in bays whose
271 bay type sets are non-empty and share no members with this type's bay type set.
272 If this type has no bay type constraints, no installation can be incompatible.
273 """
274 type_type_pks = list(self.module_bay_types.values_list('pk', flat=True))
275 if not type_type_pks:
276 return Module.objects.none()
277 # Bays that have any compatible type (the intersection is non-empty)
278 compatible_bay_pks = ModuleBay.objects.filter(
279 module_bay_types__pk__in=type_type_pks
280 ).values_list('pk', flat=True)
281 # Bays with at least one type set (constrained bays); distinct() prevents M2M join duplicates
282 constrained_bay_pks = ModuleBay.objects.filter(
283 module_bay_types__isnull=False
284 ).distinct().values_list('pk', flat=True)
285 return Module.objects.filter(
286 module_type=self,
287 module_bay__pk__in=constrained_bay_pks,
288 ).exclude(
289 module_bay__pk__in=compatible_bay_pks,
290 )
292 def clean(self):
293 super().clean()
295 # Validate any attributes against the assigned profile's schema
296 if self.profile and self.profile.schema:
297 try:
298 jsonschema.validate(self.attribute_data, schema=self.profile.schema)
299 except JSONValidationError as e:
300 raise ValidationError(_("Invalid schema: {error}").format(error=e))
301 else:
302 self.attribute_data = None
304 def get_cooling_method_color(self):
305 return CoolingMethodChoices.colors.get(self.cooling_method)
307 def to_yaml(self):
308 data = {
309 'profile': self.profile.name if self.profile else None,
310 'manufacturer': self.manufacturer.name,
311 'model': self.model,
312 'part_number': self.part_number,
313 'description': self.description,
314 'weight': float(self.weight) if self.weight is not None else None,
315 'weight_unit': self.weight_unit,
316 'airflow': self.airflow,
317 'cooling_method': self.cooling_method,
318 'end_of_life': self.end_of_life.isoformat() if self.end_of_life else None,
319 'attribute_data': self.attribute_data,
320 'comments': self.comments,
321 'module_bay_types': [t.name for t in self.module_bay_types.all()],
322 }
324 # Component templates
325 if self.consoleporttemplates.exists():
326 data['console-ports'] = [
327 c.to_yaml() for c in self.consoleporttemplates.all()
328 ]
329 if self.consoleserverporttemplates.exists():
330 data['console-server-ports'] = [
331 c.to_yaml() for c in self.consoleserverporttemplates.all()
332 ]
333 if self.powerporttemplates.exists():
334 data['power-ports'] = [
335 c.to_yaml() for c in self.powerporttemplates.all()
336 ]
337 if self.poweroutlettemplates.exists():
338 data['power-outlets'] = [
339 c.to_yaml() for c in self.poweroutlettemplates.all()
340 ]
341 if self.coolingintaketemplates.exists():
342 data['cooling-intakes'] = [
343 c.to_yaml() for c in self.coolingintaketemplates.all()
344 ]
345 if self.coolingoutflowtemplates.exists():
346 data['cooling-outflows'] = [
347 c.to_yaml() for c in self.coolingoutflowtemplates.all()
348 ]
349 if self.interfacetemplates.exists():
350 data['interfaces'] = [
351 c.to_yaml() for c in self.interfacetemplates.all()
352 ]
353 if self.frontporttemplates.exists():
354 data['front-ports'] = [
355 c.to_yaml() for c in self.frontporttemplates.all()
356 ]
357 if self.rearporttemplates.exists():
358 data['rear-ports'] = [
359 c.to_yaml() for c in self.rearporttemplates.all()
360 ]
362 # Port mappings
363 port_mapping_data = [
364 c.to_yaml() for c in self.port_mappings.all()
365 ]
367 if port_mapping_data:
368 data['port-mappings'] = port_mapping_data
370 return yaml.dump(dict(data), sort_keys=False)
373class Module(TrackingModelMixin, PrimaryModel):
374 """
375 A Module represents a field-installable component within a Device which may itself hold multiple device components
376 (for example, a line card within a chassis switch). Modules are instantiated from ModuleTypes.
377 """
378 device = models.ForeignKey(
379 to='dcim.Device',
380 on_delete=models.CASCADE,
381 related_name='modules'
382 )
383 module_bay = models.OneToOneField(
384 to='dcim.ModuleBay',
385 on_delete=models.CASCADE,
386 related_name='installed_module'
387 )
388 module_type = models.ForeignKey(
389 to='dcim.ModuleType',
390 on_delete=models.PROTECT,
391 related_name='instances'
392 )
393 status = models.CharField(
394 verbose_name=_('status'),
395 max_length=50,
396 choices=ModuleStatusChoices,
397 default=ModuleStatusChoices.STATUS_ACTIVE
398 )
399 serial = models.CharField(
400 max_length=50,
401 blank=True,
402 verbose_name=_('serial number')
403 )
404 asset_tag = models.CharField(
405 max_length=50,
406 blank=True,
407 null=True,
408 unique=True,
409 verbose_name=_('asset tag'),
410 help_text=_('A unique tag used to identify this device')
411 )
413 clone_fields = ('device', 'module_type', 'status')
415 class Meta:
416 ordering = ('module_bay',)
417 verbose_name = _('module')
418 verbose_name_plural = _('modules')
420 def __str__(self):
421 return f'{self.module_bay.name}: {self.module_type} ({self.pk})'
423 def get_status_color(self):
424 return ModuleStatusChoices.colors.get(self.status)
426 @property
427 def is_bay_compatible(self):
428 """
429 Return True if this module's type is compatible with its installed bay, or if either has no type constraints.
430 Returns False when both the bay and module type have non-empty, disjoint bay type sets.
431 """
432 if not (self.module_bay_id and self.module_type_id):
433 return True
434 # Use .all() so a prefetch cache (populated by the API list queryset) is honoured; .values_list()
435 # creates a fresh queryset that bypasses the cache and causes N+1 hits in list views.
436 bay_types = {t.pk for t in self.module_bay.module_bay_types.all()}
437 type_types = {t.pk for t in self.module_type.module_bay_types.all()}
438 if bay_types and type_types and not (bay_types & type_types):
439 return False
440 return True
442 def clean(self):
443 super().clean()
445 if hasattr(self, "module_bay") and (self.module_bay.device != self.device):
446 raise ValidationError(
447 _("Module must be installed within a module bay belonging to the assigned device ({device}).").format(
448 device=self.device
449 )
450 )
452 if not self.is_bay_compatible:
453 raise ValidationError(
454 _('Module type {module_type} is not compatible with module bay {module_bay}: '
455 'their bay type sets have no common members.').format(
456 module_type=self.module_type,
457 module_bay=self.module_bay,
458 )
459 )
461 # Prevent module from being installed in a disabled bay
462 if hasattr(self, 'module_bay') and self.module_bay and not self.module_bay.enabled:
463 current_module_bay_id = Module.objects.filter(pk=self.pk).values_list('module_bay_id', flat=True).first()
464 if self.pk is None or current_module_bay_id != self.module_bay_id:
465 raise ValidationError({
466 'module_bay': _("Cannot install a module in a disabled module bay.")
467 })
469 # Prevent installation into an occupied module bay
470 if hasattr(self, 'module_bay') and self.module_bay_id and (
471 occupant := Module.objects.filter(module_bay_id=self.module_bay_id).exclude(pk=self.pk).first()
472 ):
473 raise ValidationError({
474 'module_bay': _(
475 "Module bay {module_bay} is already occupied by module {module}."
476 ).format(module_bay=self.module_bay, module=occupant)
477 })
479 # Validate a requested move (device and/or module bay change) of an existing module
480 if not self._state.adding and hasattr(self, 'module_bay') and self.module_bay_id and self.device_id:
481 old = Module.objects.filter(pk=self.pk).first()
482 if old and (old.device_id != self.device_id or old.module_bay_id != self.module_bay_id):
483 if old.module_type_id != self.module_type_id:
484 raise ValidationError({
485 'module_type': _(
486 "Changing a module's type while moving it is not supported. Change the module "
487 "type and move the module as separate operations."
488 )
489 })
490 try:
491 ModuleMovePlan.from_module(old_module=old, new_module=self).validate()
492 except ValueError as e:
493 raise ValidationError({'module_bay': str(e)}) from e
495 # Check for recursion
496 module = self
497 module_bays = []
498 modules = []
499 while module:
500 module_module_bay = getattr(module, "module_bay", None)
501 if module.pk in modules or (module_module_bay and module_module_bay.pk in module_bays):
502 raise ValidationError(_("A module bay cannot belong to a module installed within it."))
503 modules.append(module.pk)
504 if module_module_bay:
505 module_bays.append(module_module_bay.pk)
506 module = module_module_bay.module if module_module_bay else None
508 def save(self, *args, **kwargs):
509 # Normalize before the pk branch so _save_new() and _save_existing() forward a re-iterable value.
510 update_fields = normalize_update_fields(kwargs)
512 if self.pk is None:
513 self._save_new(*args, **kwargs)
514 return
516 placement_fields = {'device', 'device_id', 'module_bay', 'module_bay_id'}
517 if update_fields is not None and placement_fields.isdisjoint(update_fields):
518 # Placement columns cannot be written by this save, so no move can occur.
519 super().save(*args, **kwargs)
520 return
522 self._save_existing(*args, **kwargs)
524 def _save_new(self, *args, **kwargs):
525 super().save(*args, **kwargs)
527 using = self._state.db
529 adopt_components = getattr(self, '_adopt_components', False)
530 disable_replication = getattr(self, '_disable_replication', False)
532 # We skip adding components if both replication and component adoption is disabled
533 if disable_replication and not adopt_components:
534 return
536 # Iterate all component types
537 for templates, component_attribute, component_model in [
538 ("consoleporttemplates", "consoleports", ConsolePort),
539 ("consoleserverporttemplates", "consoleserverports", ConsoleServerPort),
540 ("interfacetemplates", "interfaces", Interface),
541 ("powerporttemplates", "powerports", PowerPort),
542 ("poweroutlettemplates", "poweroutlets", PowerOutlet),
543 ("coolingintaketemplates", "coolingintakes", CoolingIntake),
544 ("coolingoutflowtemplates", "coolingoutflows", CoolingOutflow),
545 ("rearporttemplates", "rearports", RearPort),
546 ("frontporttemplates", "frontports", FrontPort),
547 ("modulebaytemplates", "modulebays", ModuleBay),
548 ]:
549 create_instances = []
550 update_instances = []
552 # Prefetch installed components
553 installed_components = {
554 component.name: component
555 for component in getattr(self.device, component_attribute).db_manager(using).filter(
556 module__isnull=True
557 )
558 }
560 # Get the template for the module type.
561 for template in getattr(self.module_type, templates).all():
562 template_instance = template.instantiate(device=self.device, module=self)
563 template_instance._source_template = template
565 if adopt_components:
566 existing_item = installed_components.get(template_instance.name)
568 # Check if there's a component with the same name already
569 if existing_item:
570 # Assign it to the module
571 existing_item.module = self
572 update_instances.append(existing_item)
573 continue
575 # Only create new components if replication is enabled
576 if not disable_replication:
577 create_instances.append(template_instance)
579 # Set default values for any applicable custom fields
580 if cf_defaults := CustomField.objects.get_defaults_for_model(component_model):
581 for component in create_instances:
582 component.custom_field_data = cf_defaults
584 # Set denormalized references
585 for component in create_instances:
586 component._site = self.device.site
587 component._location = self.device.location
588 component._rack = self.device.rack
590 # Bulk-create new instances. ModuleBay is ltree-backed: its parent is set
591 # in ModuleBayTemplate.instantiate() (bulk_create bypasses ModuleBay.save()),
592 # and the BEFORE INSERT trigger derives path/sort_path from parent_id per row.
593 component_model.objects.using(using).bulk_create(create_instances)
595 # Copy M2M module_bay_types from template to new ModuleBay instances.
596 if component_model is ModuleBay:
597 for component in create_instances:
598 if src := getattr(component, '_source_template', None):
599 component.module_bay_types.set(src.module_bay_types.db_manager(using).all())
601 # Emit the post_save signal for each newly created object
602 for component in create_instances:
603 post_save.send(
604 sender=component_model,
605 instance=component,
606 created=True,
607 raw=False,
608 using=using,
609 update_fields=None
610 )
612 update_fields = ['module']
613 # ModuleBay.parent is derived from .module in ModuleBay.save(), and the
614 # path/sort_path trigger only fires on parent_id/name changes. A bare
615 # bulk_update(['module']) bypasses both, leaving the adopted bay rooted
616 # at its pre-adoption location. Set parent in lockstep so the BEFORE
617 # trigger recomputes path/sort_path.
618 if component_model is ModuleBay:
619 for instance in update_instances:
620 instance.parent = self.module_bay
621 update_fields = ['module', 'parent']
623 component_model.objects.using(using).bulk_update(
624 update_instances, update_fields, batch_size=settings.BULK_UPDATE_CHUNK_SIZE
625 )
626 # Emit the post_save signal for each updated object
627 for component in update_instances:
628 post_save.send(
629 sender=component_model,
630 instance=component,
631 created=False,
632 raw=False,
633 using=using,
634 update_fields=update_fields
635 )
637 # Replicate any front/rear port mappings from the ModuleType
638 create_port_mappings(self.device, self.module_type, self)
640 # Interface parents & bridges have to be set after interface instantiation. Parents are applied first so that
641 # channel subinterfaces validate against a populated parent.
642 update_interface_parents(self.device, self.module_type.interfacetemplates, self)
643 update_interface_bridges(self.device, self.module_type.interfacetemplates, self)
645 def _save_existing(self, *args, **kwargs):
646 try:
647 with transaction.atomic(using=router.db_for_write(Module)):
648 # Root row locks first (matches API ETag path); all routing below decides from this locked read
649 locked_old = Module.objects.select_for_update().only(
650 'device', 'module_bay', 'module_type'
651 ).filter(pk=self.pk).first()
652 if locked_old is None:
653 # A new pk, or a row concurrently deleted; create instead.
654 self._save_new(*args, **kwargs)
655 return
657 delta_fields = []
658 if locked_old.device_id != self.device_id:
659 delta_fields.append('device')
660 if locked_old.module_bay_id != self.module_bay_id:
661 delta_fields.append('module_bay')
663 if not delta_fields:
664 super().save(*args, **kwargs)
665 return
667 update_fields = kwargs.get('update_fields')
668 if update_fields is not None:
669 field_attnames = {'device': 'device_id', 'module_bay': 'module_bay_id'}
670 listed = {
671 field for field in delta_fields
672 if field in update_fields or field_attnames[field] in update_fields
673 }
674 if not listed:
675 # None of the changed placement fields are part of this write, so no move happens.
676 super().save(*args, **kwargs)
677 return
678 if listed != set(delta_fields):
679 raise AbortRequest(_(
680 "A module move must include every changed placement field in update_fields: "
681 "'device' (or 'device_id') and/or 'module_bay' (or 'module_bay_id')."
682 ))
684 if locked_old.module_type_id != self.module_type_id:
685 raise AbortRequest(_(
686 "Changing a module's type while moving it is not supported. Change the module type and "
687 "move the module as separate operations."
688 ))
690 try:
691 plan = ModuleMovePlan.from_module(old_module=locked_old, new_module=self)
692 plan.lock()
693 except ValueError as e:
694 raise AbortRequest(str(e)) from e
695 try:
696 plan.validate()
697 except ValidationError as e:
698 raise AbortRequest(' '.join(e.messages)) from e
700 super().save(*args, **kwargs)
701 plan.apply_after_root_save()
702 except OperationalError as e:
703 if getattr(e.__cause__, 'sqlstate', None) == '40P01':
704 raise AbortRequest(_(
705 "This module or its components are being modified by another request. Please try again."
706 )) from e
707 raise