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

1import decimal 

2 

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 

7 

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 

19 

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 

27 

28__all__ = ( 

29 'DeviceSerializer', 

30 'MACAddressSerializer', 

31 'ModuleSerializer', 

32 'VirtualDeviceContextSerializer', 

33) 

34 

35 

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) 

91 

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) 

103 

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') 

118 

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 

129 

130 @extend_schema_field(serializers.JSONField(allow_null=True)) 

131 def get_config_context(self, obj): 

132 return obj.get_config_context() 

133 

134 

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) 

158 

159 # Related object counts 

160 interface_count = RelatedObjectCountField('interfaces') 

161 

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') 

170 

171 

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) 

192 

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') 

201 

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) 

207 

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) 

212 

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 }) 

228 

229 data = super().validate(data) 

230 

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 

235 

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 

239 

240 # Skip conflict checks when no component operations are requested. 

241 if not replicate_components and not adopt_components: 

242 return data 

243 

244 device = data.get('device') 

245 module_type = data.get('module_type') 

246 module_bay = data.get('module_bay') 

247 

248 # Required-field validation fires separately; skip here if any are missing. 

249 if not all([device, module_type, module_bay]): 

250 return data 

251 

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 

256 

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 } 

272 

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)) 

284 

285 existing_item = installed_components.get(resolved_name) 

286 

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 ) 

294 

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 ) 

302 

303 return data 

304 

305 def create(self, validated_data): 

306 replicate_components = validated_data.pop('replicate_components', True) 

307 adopt_components = validated_data.pop('adopt_components', False) 

308 

309 # Tags are handled after save; pop them here to pass to _save_tags() 

310 tags = validated_data.pop('tags', None) 

311 

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() 

320 

321 if tags is not None: 

322 self._save_tags(instance, tags) 

323 

324 return instance 

325 

326 

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) 

335 

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')