mirror of
https://github.com/netbox-community/netbox.git
synced 2025-07-13 16:47:34 -06:00
363 lines
13 KiB
Python
363 lines
13 KiB
Python
import jsonschema
|
|
import yaml
|
|
from django.core.exceptions import ValidationError
|
|
from django.db import models
|
|
from django.db.models.signals import post_save
|
|
from django.utils.translation import gettext_lazy as _
|
|
from jsonschema.exceptions import ValidationError as JSONValidationError
|
|
|
|
from dcim.choices import *
|
|
from dcim.constants import MODULE_TOKEN
|
|
from dcim.utils import update_interface_bridges
|
|
from extras.models import ConfigContextModel, CustomField
|
|
from netbox.models import PrimaryModel
|
|
from netbox.models.features import ImageAttachmentsMixin
|
|
from netbox.models.mixins import WeightMixin
|
|
from utilities.jsonschema import validate_schema
|
|
from utilities.string import title
|
|
from .device_components import *
|
|
|
|
__all__ = (
|
|
'Module',
|
|
'ModuleType',
|
|
'ModuleTypeProfile',
|
|
)
|
|
|
|
|
|
class ModuleTypeProfile(PrimaryModel):
|
|
"""
|
|
A profile which defines the attributes which can be set on one or more ModuleTypes.
|
|
"""
|
|
name = models.CharField(
|
|
verbose_name=_('name'),
|
|
max_length=100,
|
|
unique=True
|
|
)
|
|
schema = models.JSONField(
|
|
blank=True,
|
|
null=True,
|
|
verbose_name=_('schema')
|
|
)
|
|
|
|
clone_fields = ('schema',)
|
|
|
|
class Meta:
|
|
ordering = ('name',)
|
|
verbose_name = _('module type profile')
|
|
verbose_name_plural = _('module type profiles')
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
def clean(self):
|
|
super().clean()
|
|
|
|
# Validate the schema definition
|
|
if self.schema is not None:
|
|
try:
|
|
validate_schema(self.schema)
|
|
except ValidationError as e:
|
|
raise ValidationError({
|
|
'schema': e.message,
|
|
})
|
|
|
|
|
|
class ModuleType(ImageAttachmentsMixin, PrimaryModel, WeightMixin):
|
|
"""
|
|
A ModuleType represents a hardware element that can be installed within a device and which houses additional
|
|
components; for example, a line card within a chassis-based switch such as the Cisco Catalyst 6500. Like a
|
|
DeviceType, each ModuleType can have console, power, interface, and pass-through port templates assigned to it. It
|
|
cannot, however house device bays or module bays.
|
|
"""
|
|
profile = models.ForeignKey(
|
|
to='dcim.ModuleTypeProfile',
|
|
on_delete=models.PROTECT,
|
|
related_name='module_types',
|
|
blank=True,
|
|
null=True
|
|
)
|
|
manufacturer = models.ForeignKey(
|
|
to='dcim.Manufacturer',
|
|
on_delete=models.PROTECT,
|
|
related_name='module_types'
|
|
)
|
|
model = models.CharField(
|
|
verbose_name=_('model'),
|
|
max_length=100
|
|
)
|
|
part_number = models.CharField(
|
|
verbose_name=_('part number'),
|
|
max_length=50,
|
|
blank=True,
|
|
help_text=_('Discrete part number (optional)')
|
|
)
|
|
airflow = models.CharField(
|
|
verbose_name=_('airflow'),
|
|
max_length=50,
|
|
choices=ModuleAirflowChoices,
|
|
blank=True,
|
|
null=True
|
|
)
|
|
attribute_data = models.JSONField(
|
|
blank=True,
|
|
null=True,
|
|
verbose_name=_('attributes')
|
|
)
|
|
|
|
clone_fields = ('profile', 'manufacturer', 'weight', 'weight_unit', 'airflow')
|
|
prerequisite_models = (
|
|
'dcim.Manufacturer',
|
|
)
|
|
|
|
class Meta:
|
|
ordering = ('profile', 'manufacturer', 'model')
|
|
constraints = (
|
|
models.UniqueConstraint(
|
|
fields=('manufacturer', 'model'),
|
|
name='%(app_label)s_%(class)s_unique_manufacturer_model'
|
|
),
|
|
)
|
|
verbose_name = _('module type')
|
|
verbose_name_plural = _('module types')
|
|
|
|
def __str__(self):
|
|
return self.model
|
|
|
|
@property
|
|
def full_name(self):
|
|
return f"{self.manufacturer} {self.model}"
|
|
|
|
@property
|
|
def attributes(self):
|
|
"""
|
|
Returns a human-friendly representation of the attributes defined for a ModuleType according to its profile.
|
|
"""
|
|
if not self.attribute_data or self.profile is None or not self.profile.schema:
|
|
return {}
|
|
attrs = {}
|
|
for name, options in self.profile.schema.get('properties', {}).items():
|
|
key = options.get('title', title(name))
|
|
attrs[key] = self.attribute_data.get(name)
|
|
return dict(sorted(attrs.items()))
|
|
|
|
def clean(self):
|
|
super().clean()
|
|
|
|
# Validate any attributes against the assigned profile's schema
|
|
if self.profile and self.profile.schema:
|
|
try:
|
|
jsonschema.validate(self.attribute_data, schema=self.profile.schema)
|
|
except JSONValidationError as e:
|
|
raise ValidationError(_("Invalid schema: {error}").format(error=e))
|
|
else:
|
|
self.attribute_data = None
|
|
|
|
def to_yaml(self):
|
|
data = {
|
|
'profile': self.profile.name if self.profile else None,
|
|
'manufacturer': self.manufacturer.name,
|
|
'model': self.model,
|
|
'part_number': self.part_number,
|
|
'description': self.description,
|
|
'weight': float(self.weight) if self.weight is not None else None,
|
|
'weight_unit': self.weight_unit,
|
|
'comments': self.comments,
|
|
}
|
|
|
|
# Component templates
|
|
if self.consoleporttemplates.exists():
|
|
data['console-ports'] = [
|
|
c.to_yaml() for c in self.consoleporttemplates.all()
|
|
]
|
|
if self.consoleserverporttemplates.exists():
|
|
data['console-server-ports'] = [
|
|
c.to_yaml() for c in self.consoleserverporttemplates.all()
|
|
]
|
|
if self.powerporttemplates.exists():
|
|
data['power-ports'] = [
|
|
c.to_yaml() for c in self.powerporttemplates.all()
|
|
]
|
|
if self.poweroutlettemplates.exists():
|
|
data['power-outlets'] = [
|
|
c.to_yaml() for c in self.poweroutlettemplates.all()
|
|
]
|
|
if self.interfacetemplates.exists():
|
|
data['interfaces'] = [
|
|
c.to_yaml() for c in self.interfacetemplates.all()
|
|
]
|
|
if self.frontporttemplates.exists():
|
|
data['front-ports'] = [
|
|
c.to_yaml() for c in self.frontporttemplates.all()
|
|
]
|
|
if self.rearporttemplates.exists():
|
|
data['rear-ports'] = [
|
|
c.to_yaml() for c in self.rearporttemplates.all()
|
|
]
|
|
|
|
return yaml.dump(dict(data), sort_keys=False)
|
|
|
|
|
|
class Module(PrimaryModel, ConfigContextModel):
|
|
"""
|
|
A Module represents a field-installable component within a Device which may itself hold multiple device components
|
|
(for example, a line card within a chassis switch). Modules are instantiated from ModuleTypes.
|
|
"""
|
|
device = models.ForeignKey(
|
|
to='dcim.Device',
|
|
on_delete=models.CASCADE,
|
|
related_name='modules'
|
|
)
|
|
module_bay = models.OneToOneField(
|
|
to='dcim.ModuleBay',
|
|
on_delete=models.CASCADE,
|
|
related_name='installed_module'
|
|
)
|
|
module_type = models.ForeignKey(
|
|
to='dcim.ModuleType',
|
|
on_delete=models.PROTECT,
|
|
related_name='instances'
|
|
)
|
|
status = models.CharField(
|
|
verbose_name=_('status'),
|
|
max_length=50,
|
|
choices=ModuleStatusChoices,
|
|
default=ModuleStatusChoices.STATUS_ACTIVE
|
|
)
|
|
serial = models.CharField(
|
|
max_length=50,
|
|
blank=True,
|
|
verbose_name=_('serial number')
|
|
)
|
|
asset_tag = models.CharField(
|
|
max_length=50,
|
|
blank=True,
|
|
null=True,
|
|
unique=True,
|
|
verbose_name=_('asset tag'),
|
|
help_text=_('A unique tag used to identify this device')
|
|
)
|
|
|
|
clone_fields = ('device', 'module_type', 'status')
|
|
|
|
class Meta:
|
|
ordering = ('module_bay',)
|
|
verbose_name = _('module')
|
|
verbose_name_plural = _('modules')
|
|
|
|
def __str__(self):
|
|
return f'{self.module_bay.name}: {self.module_type} ({self.pk})'
|
|
|
|
def get_status_color(self):
|
|
return ModuleStatusChoices.colors.get(self.status)
|
|
|
|
def clean(self):
|
|
super().clean()
|
|
|
|
if hasattr(self, "module_bay") and (self.module_bay.device != self.device):
|
|
raise ValidationError(
|
|
_("Module must be installed within a module bay belonging to the assigned device ({device}).").format(
|
|
device=self.device
|
|
)
|
|
)
|
|
|
|
# Check for recursion
|
|
module = self
|
|
module_bays = []
|
|
modules = []
|
|
while module:
|
|
if module.pk in modules or module.module_bay.pk in module_bays:
|
|
raise ValidationError(_("A module bay cannot belong to a module installed within it."))
|
|
modules.append(module.pk)
|
|
module_bays.append(module.module_bay.pk)
|
|
module = module.module_bay.module if module.module_bay else None
|
|
|
|
def save(self, *args, **kwargs):
|
|
is_new = self.pk is None
|
|
|
|
super().save(*args, **kwargs)
|
|
|
|
adopt_components = getattr(self, '_adopt_components', False)
|
|
disable_replication = getattr(self, '_disable_replication', False)
|
|
|
|
# We skip adding components if the module is being edited or
|
|
# both replication and component adoption is disabled
|
|
if not is_new or (disable_replication and not adopt_components):
|
|
return
|
|
|
|
# Iterate all component types
|
|
for templates, component_attribute, component_model in [
|
|
("consoleporttemplates", "consoleports", ConsolePort),
|
|
("consoleserverporttemplates", "consoleserverports", ConsoleServerPort),
|
|
("interfacetemplates", "interfaces", Interface),
|
|
("powerporttemplates", "powerports", PowerPort),
|
|
("poweroutlettemplates", "poweroutlets", PowerOutlet),
|
|
("rearporttemplates", "rearports", RearPort),
|
|
("frontporttemplates", "frontports", FrontPort),
|
|
("modulebaytemplates", "modulebays", ModuleBay),
|
|
]:
|
|
create_instances = []
|
|
update_instances = []
|
|
|
|
# Prefetch installed components
|
|
installed_components = {
|
|
component.name: component
|
|
for component in getattr(self.device, component_attribute).filter(module__isnull=True)
|
|
}
|
|
|
|
# Get the template for the module type.
|
|
for template in getattr(self.module_type, templates).all():
|
|
template_instance = template.instantiate(device=self.device, module=self)
|
|
|
|
if adopt_components:
|
|
existing_item = installed_components.get(template_instance.name)
|
|
|
|
# Check if there's a component with the same name already
|
|
if existing_item:
|
|
# Assign it to the module
|
|
existing_item.module = self
|
|
update_instances.append(existing_item)
|
|
continue
|
|
|
|
# Only create new components if replication is enabled
|
|
if not disable_replication:
|
|
create_instances.append(template_instance)
|
|
|
|
# Set default values for any applicable custom fields
|
|
if cf_defaults := CustomField.objects.get_defaults_for_model(component_model):
|
|
for component in create_instances:
|
|
component.custom_field_data = cf_defaults
|
|
|
|
if component_model is not ModuleBay:
|
|
component_model.objects.bulk_create(create_instances)
|
|
# Emit the post_save signal for each newly created object
|
|
for component in create_instances:
|
|
post_save.send(
|
|
sender=component_model,
|
|
instance=component,
|
|
created=True,
|
|
raw=False,
|
|
using='default',
|
|
update_fields=None
|
|
)
|
|
else:
|
|
# ModuleBays must be saved individually for MPTT
|
|
for instance in create_instances:
|
|
instance.name = instance.name.replace(MODULE_TOKEN, str(self.module_bay.position))
|
|
instance.save()
|
|
|
|
update_fields = ['module']
|
|
component_model.objects.bulk_update(update_instances, update_fields)
|
|
# Emit the post_save signal for each updated object
|
|
for component in update_instances:
|
|
post_save.send(
|
|
sender=component_model,
|
|
instance=component,
|
|
created=False,
|
|
raw=False,
|
|
using='default',
|
|
update_fields=update_fields
|
|
)
|
|
|
|
# Interface bridges have to be set after interface instantiation
|
|
update_interface_bridges(self.device, self.module_type.interfacetemplates, self)
|