Coverage for dcim/api/serializers_/devices.py: 54%
164 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import decimal
3from django.contrib.contenttypes.models import ContentType
4from django.utils.translation import gettext as _
5from drf_spectacular.utils import extend_schema_field
6from rest_framework import serializers
8from dcim.choices import *
9from dcim.constants import MACADDRESS_ASSIGNMENT_MODELS, MODULE_TOKEN
10from dcim.models import Device, DeviceBay, MACAddress, Module, VirtualDeviceContext
11from dcim.utils import get_module_bay_positions, resolve_module_placeholder
12from extras.api.serializers_.configtemplates import ConfigTemplateSerializer
13from ipam.api.serializers_.ip import IPAddressSerializer
14from netbox.api.fields import ChoiceField, ContentTypeField, RelatedObjectCountField
15from netbox.api.gfk_fields import GFKSerializerField
16from netbox.api.serializers import PrimaryModelSerializer
17from tenancy.api.serializers_.tenants import TenantSerializer
18from virtualization.api.serializers_.clusters import ClusterSerializer
20from .devicetypes import *
21from .nested import NestedDeviceBaySerializer, NestedDeviceSerializer, NestedModuleBaySerializer
22from .platforms import PlatformSerializer
23from .racks import RackSerializer
24from .roles import DeviceRoleSerializer
25from .sites import LocationSerializer, SiteSerializer
26from .virtualchassis import VirtualChassisSerializer
28__all__ = (
29 'DeviceSerializer',
30 'MACAddressSerializer',
31 'ModuleSerializer',
32 'VirtualDeviceContextSerializer',
33)
36class DeviceSerializer(PrimaryModelSerializer):
37 device_type = DeviceTypeSerializer(nested=True)
38 role = DeviceRoleSerializer(nested=True)
39 tenant = TenantSerializer(
40 nested=True,
41 required=False,
42 allow_null=True,
43 default=None
44 )
45 platform = PlatformSerializer(nested=True, required=False, allow_null=True)
46 site = SiteSerializer(nested=True)
47 location = LocationSerializer(nested=True, required=False, allow_null=True, default=None)
48 rack = RackSerializer(nested=True, required=False, allow_null=True, default=None)
49 face = ChoiceField(choices=DeviceFaceChoices, allow_blank=True, default=lambda: '')
50 position = serializers.DecimalField(
51 max_digits=4,
52 decimal_places=1,
53 allow_null=True,
54 label=_('Position (U)'),
55 min_value=decimal.Decimal(0.5),
56 default=None
57 )
58 status = ChoiceField(choices=DeviceStatusChoices, required=False)
59 airflow = ChoiceField(choices=DeviceAirflowChoices, allow_blank=True, required=False)
60 cooling_method = ChoiceField(choices=CoolingMethodChoices, allow_blank=True, required=False, allow_null=True)
61 primary_ip = IPAddressSerializer(
62 nested=True,
63 read_only=True,
64 allow_null=True,
65 fields=[*IPAddressSerializer.Meta.brief_fields, 'dns_name', 'nat_inside', 'nat_outside'],
66 )
67 primary_ip4 = IPAddressSerializer(
68 nested=True,
69 required=False,
70 allow_null=True,
71 fields=[*IPAddressSerializer.Meta.brief_fields, 'dns_name', 'nat_inside', 'nat_outside'],
72 )
73 primary_ip6 = IPAddressSerializer(
74 nested=True,
75 required=False,
76 allow_null=True,
77 fields=[*IPAddressSerializer.Meta.brief_fields, 'dns_name', 'nat_inside', 'nat_outside'],
78 )
79 oob_ip = IPAddressSerializer(
80 nested=True,
81 required=False,
82 allow_null=True,
83 fields=[*IPAddressSerializer.Meta.brief_fields, 'dns_name', 'nat_inside', 'nat_outside'],
84 )
85 parent_device = serializers.SerializerMethodField()
86 cluster = ClusterSerializer(nested=True, required=False, allow_null=True)
87 virtual_chassis = VirtualChassisSerializer(nested=True, required=False, allow_null=True, default=None)
88 vc_position = serializers.IntegerField(allow_null=True, max_value=255, min_value=0, default=None)
89 config_template = ConfigTemplateSerializer(nested=True, required=False, allow_null=True, default=None)
90 config_context = serializers.SerializerMethodField(read_only=True)
92 # Counter fields
93 console_port_count = serializers.IntegerField(read_only=True)
94 console_server_port_count = serializers.IntegerField(read_only=True)
95 power_port_count = serializers.IntegerField(read_only=True)
96 power_outlet_count = serializers.IntegerField(read_only=True)
97 interface_count = serializers.IntegerField(read_only=True)
98 front_port_count = serializers.IntegerField(read_only=True)
99 rear_port_count = serializers.IntegerField(read_only=True)
100 device_bay_count = serializers.IntegerField(read_only=True)
101 module_bay_count = serializers.IntegerField(read_only=True)
102 inventory_item_count = serializers.IntegerField(read_only=True)
104 class Meta:
105 model = Device
106 fields = [
107 'id', 'url', 'display_url', 'display', 'name', 'device_type', 'role', 'tenant', 'platform', 'serial',
108 'asset_tag', 'site', 'location', 'rack', 'position', 'face', 'latitude', 'longitude', 'parent_device',
109 'status', 'airflow', 'cooling_method', 'primary_ip', 'primary_ip4', 'primary_ip6', 'oob_ip', 'cluster',
110 'virtual_chassis', 'vc_position', 'vc_priority', 'description', 'owner', 'comments', 'config_template',
111 'config_context',
112 'local_context_data', 'tags', 'custom_fields', 'created', 'last_updated', 'console_port_count',
113 'console_server_port_count', 'power_port_count', 'power_outlet_count', 'cooling_intake_count',
114 'cooling_outflow_count', 'interface_count', 'front_port_count', 'rear_port_count', 'device_bay_count',
115 'module_bay_count', 'inventory_item_count',
116 ]
117 brief_fields = ('id', 'url', 'display', 'name', 'description')
119 @extend_schema_field(NestedDeviceSerializer(allow_null=True))
120 def get_parent_device(self, obj):
121 try:
122 device_bay = obj.parent_bay
123 except DeviceBay.DoesNotExist:
124 return None
125 context = {'request': self.context['request']}
126 data = NestedDeviceSerializer(instance=device_bay.device, context=context).data
127 data['device_bay'] = NestedDeviceBaySerializer(instance=device_bay, context=context).data
128 return data
130 @extend_schema_field(serializers.JSONField(allow_null=True))
131 def get_config_context(self, obj):
132 return obj.get_config_context()
135class VirtualDeviceContextSerializer(PrimaryModelSerializer):
136 device = DeviceSerializer(nested=True)
137 identifier = serializers.IntegerField(allow_null=True, max_value=32767, min_value=0, required=False, default=None)
138 tenant = TenantSerializer(nested=True, required=False, allow_null=True, default=None)
139 primary_ip = IPAddressSerializer(
140 nested=True,
141 read_only=True,
142 allow_null=True,
143 fields=[*IPAddressSerializer.Meta.brief_fields, 'dns_name', 'nat_inside', 'nat_outside'],
144 )
145 primary_ip4 = IPAddressSerializer(
146 nested=True,
147 required=False,
148 allow_null=True,
149 fields=[*IPAddressSerializer.Meta.brief_fields, 'dns_name', 'nat_inside', 'nat_outside'],
150 )
151 primary_ip6 = IPAddressSerializer(
152 nested=True,
153 required=False,
154 allow_null=True,
155 fields=[*IPAddressSerializer.Meta.brief_fields, 'dns_name', 'nat_inside', 'nat_outside'],
156 )
157 status = ChoiceField(choices=VirtualDeviceContextStatusChoices)
159 # Related object counts
160 interface_count = RelatedObjectCountField('interfaces')
162 class Meta:
163 model = VirtualDeviceContext
164 fields = [
165 'id', 'url', 'display_url', 'display', 'name', 'device', 'identifier', 'tenant', 'primary_ip',
166 'primary_ip4', 'primary_ip6', 'status', 'description', 'owner', 'comments', 'tags', 'custom_fields',
167 'created', 'last_updated', 'interface_count',
168 ]
169 brief_fields = ('id', 'url', 'display', 'name', 'identifier', 'device', 'description')
172class ModuleSerializer(PrimaryModelSerializer):
173 device = DeviceSerializer(nested=True)
174 module_bay = NestedModuleBaySerializer()
175 module_type = ModuleTypeSerializer(nested=True)
176 status = ChoiceField(choices=ModuleStatusChoices, required=False)
177 replicate_components = serializers.BooleanField(
178 required=False,
179 default=True,
180 write_only=True,
181 label=_('Replicate components'),
182 help_text=_('Automatically populate components associated with this module type (default: true)')
183 )
184 adopt_components = serializers.BooleanField(
185 required=False,
186 default=False,
187 write_only=True,
188 label=_('Adopt components'),
189 help_text=_('Adopt already existing components')
190 )
191 is_bay_compatible = serializers.BooleanField(read_only=True)
193 class Meta:
194 model = Module
195 fields = [
196 'id', 'url', 'display_url', 'display', 'device', 'module_bay', 'module_type', 'status', 'serial',
197 'asset_tag', 'description', 'owner', 'comments', 'tags', 'custom_fields', 'created', 'last_updated',
198 'replicate_components', 'adopt_components', 'is_bay_compatible',
199 ]
200 brief_fields = ('id', 'url', 'display', 'device', 'module_bay', 'module_type', 'description')
202 def validate(self, data):
203 # When used as a nested serializer (e.g. as the `module` field on device component
204 # serializers), `data` is already a resolved Module instance — skip our custom logic.
205 if self.nested:
206 return super().validate(data)
208 # Pop write-only transient fields before ValidatedModelSerializer tries to
209 # construct a Module instance for full_clean(); restore them afterwards.
210 replicate_components = data.pop('replicate_components', True)
211 adopt_components = data.pop('adopt_components', False)
213 if self.instance is not None:
214 # Derive device from module_bay so full_clean() validates a consistent pair.
215 if 'module_bay' in data and 'device' not in data:
216 data['device'] = data['module_bay'].device
217 move_requested = (
218 ('module_bay' in data and data['module_bay'].pk != self.instance.module_bay_id) or
219 ('device' in data and data['device'].pk != self.instance.device_id)
220 )
221 if move_requested and 'module_type' in data and data['module_type'].pk != self.instance.module_type_id:
222 raise serializers.ValidationError({
223 'module_type': _(
224 "Changing a module's type while moving it is not supported. Change the module "
225 "type and move the module as separate operations."
226 )
227 })
229 data = super().validate(data)
231 # For updates these fields are not meaningful; omit them from validated_data so that
232 # ModelSerializer.update() does not set unexpected attributes on the instance.
233 if self.instance:
234 return data
236 # Always pass the flags to create() so it can set the correct private attributes.
237 data['replicate_components'] = replicate_components
238 data['adopt_components'] = adopt_components
240 # Skip conflict checks when no component operations are requested.
241 if not replicate_components and not adopt_components:
242 return data
244 device = data.get('device')
245 module_type = data.get('module_type')
246 module_bay = data.get('module_bay')
248 # Required-field validation fires separately; skip here if any are missing.
249 if not all([device, module_type, module_bay]):
250 return data
252 try:
253 positions = get_module_bay_positions(module_bay)
254 except ValueError as e:
255 raise serializers.ValidationError({'module_bay': str(e)}) from e
257 for templates_attr, component_attr in [
258 ('consoleporttemplates', 'consoleports'),
259 ('consoleserverporttemplates', 'consoleserverports'),
260 ('interfacetemplates', 'interfaces'),
261 ('powerporttemplates', 'powerports'),
262 ('poweroutlettemplates', 'poweroutlets'),
263 ('coolingintaketemplates', 'coolingintakes'),
264 ('coolingoutflowtemplates', 'coolingoutflows'),
265 ('rearporttemplates', 'rearports'),
266 ('frontporttemplates', 'frontports'),
267 ]:
268 installed_components = {
269 component.name: component
270 for component in getattr(device, component_attr).all()
271 }
273 for template in getattr(module_type, templates_attr).all():
274 resolved_name = template.name
275 if MODULE_TOKEN in template.name:
276 if not module_bay.position:
277 raise serializers.ValidationError(
278 _("Cannot install module with placeholder values in a module bay with no position defined.")
279 )
280 try:
281 resolved_name = resolve_module_placeholder(template.name, positions)
282 except ValueError as e:
283 raise serializers.ValidationError(str(e))
285 existing_item = installed_components.get(resolved_name)
287 if adopt_components and existing_item and existing_item.module:
288 raise serializers.ValidationError(
289 _("Cannot adopt {model} {name} as it already belongs to a module").format(
290 model=template.component_model.__name__,
291 name=resolved_name
292 )
293 )
295 if not adopt_components and resolved_name in installed_components:
296 raise serializers.ValidationError(
297 _("A {model} named {name} already exists").format(
298 model=template.component_model.__name__,
299 name=resolved_name
300 )
301 )
303 return data
305 def create(self, validated_data):
306 replicate_components = validated_data.pop('replicate_components', True)
307 adopt_components = validated_data.pop('adopt_components', False)
309 # Tags are handled after save; pop them here to pass to _save_tags()
310 tags = validated_data.pop('tags', None)
312 # _adopt_components and _disable_replication must be set on the instance before
313 # save() is called, so we cannot delegate to super().create() here.
314 instance = self.Meta.model(**validated_data)
315 if adopt_components:
316 instance._adopt_components = True
317 if not replicate_components:
318 instance._disable_replication = True
319 instance.save()
321 if tags is not None:
322 self._save_tags(instance, tags)
324 return instance
327class MACAddressSerializer(PrimaryModelSerializer):
328 assigned_object_type = ContentTypeField(
329 queryset=ContentType.objects.filter(MACADDRESS_ASSIGNMENT_MODELS),
330 required=False,
331 allow_null=True
332 )
333 assigned_object = GFKSerializerField(read_only=True)
334 is_primary = serializers.BooleanField(read_only=True)
336 class Meta:
337 model = MACAddress
338 fields = [
339 'id', 'url', 'display_url', 'display', 'mac_address', 'assigned_object_type', 'assigned_object_id',
340 'assigned_object', 'is_primary', 'description', 'owner', 'comments', 'tags', 'custom_fields', 'created',
341 'last_updated',
342 ]
343 brief_fields = ('id', 'url', 'display', 'mac_address', 'description')