Coverage for ipam/forms/bulk_import.py: 45%
262 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from django import forms
2from django.contrib.contenttypes.models import ContentType
3from django.contrib.postgres.forms import SimpleArrayField
4from django.core.exceptions import ValidationError as DjangoValidationError
5from django.utils.translation import gettext_lazy as _
7from dcim.forms.mixins import ScopedImportForm
8from dcim.models import Device, Interface, Site
9from ipam.choices import *
10from ipam.constants import *
11from ipam.models import *
12from ipam.utils import expand_port_mapping, split_port_mapping
13from ipam.validators import validate_port_mappings
14from netbox.forms import NetBoxModelImportForm, OrganizationalModelImportForm, PrimaryModelImportForm
15from tenancy.models import Tenant
16from utilities.forms.fields import (
17 CSVChoiceField,
18 CSVContentTypeField,
19 CSVModelChoiceField,
20 CSVModelMultipleChoiceField,
21 NumericRangeArrayField,
22 SlugField,
23)
24from virtualization.models import VirtualMachine, VMInterface
26__all__ = (
27 'ASNImportForm',
28 'ASNRangeImportForm',
29 'AggregateImportForm',
30 'FHRPGroupImportForm',
31 'IPAddressImportForm',
32 'IPRangeImportForm',
33 'PrefixImportForm',
34 'RIRImportForm',
35 'RoleImportForm',
36 'RouteTargetImportForm',
37 'ServiceImportForm',
38 'ServiceTemplateImportForm',
39 'VLANGroupImportForm',
40 'VLANImportForm',
41 'VLANTranslationPolicyImportForm',
42 'VLANTranslationRuleImportForm',
43 'VRFImportForm',
44)
47class VRFImportForm(PrimaryModelImportForm):
48 tenant = CSVModelChoiceField(
49 label=_('Tenant'),
50 queryset=Tenant.objects.all(),
51 required=False,
52 to_field_name='name',
53 help_text=_('Assigned tenant')
54 )
55 import_targets = CSVModelMultipleChoiceField(
56 queryset=RouteTarget.objects.all(),
57 required=False,
58 to_field_name='name',
59 help_text=_('Import route targets')
60 )
61 export_targets = CSVModelMultipleChoiceField(
62 queryset=RouteTarget.objects.all(),
63 required=False,
64 to_field_name='name',
65 help_text=_('Export route targets')
66 )
68 class Meta:
69 model = VRF
70 fields = (
71 'name', 'rd', 'tenant', 'enforce_unique', 'description', 'import_targets', 'export_targets', 'owner',
72 'comments', 'tags',
73 )
76class RouteTargetImportForm(PrimaryModelImportForm):
77 tenant = CSVModelChoiceField(
78 label=_('Tenant'),
79 queryset=Tenant.objects.all(),
80 required=False,
81 to_field_name='name',
82 help_text=_('Assigned tenant')
83 )
85 class Meta:
86 model = RouteTarget
87 fields = ('name', 'tenant', 'description', 'owner', 'comments', 'tags')
90class RIRImportForm(OrganizationalModelImportForm):
91 slug = SlugField()
93 class Meta:
94 model = RIR
95 fields = ('name', 'slug', 'is_private', 'description', 'owner', 'comments', 'tags')
98class AggregateImportForm(PrimaryModelImportForm):
99 rir = CSVModelChoiceField(
100 label=_('RIR'),
101 queryset=RIR.objects.all(),
102 to_field_name='name',
103 help_text=_('Assigned RIR')
104 )
105 tenant = CSVModelChoiceField(
106 label=_('Tenant'),
107 queryset=Tenant.objects.all(),
108 required=False,
109 to_field_name='name',
110 help_text=_('Assigned tenant')
111 )
113 class Meta:
114 model = Aggregate
115 fields = ('prefix', 'rir', 'tenant', 'date_added', 'description', 'owner', 'comments', 'tags')
118class ASNRangeImportForm(OrganizationalModelImportForm):
119 rir = CSVModelChoiceField(
120 label=_('RIR'),
121 queryset=RIR.objects.all(),
122 to_field_name='name',
123 help_text=_('Assigned RIR')
124 )
125 tenant = CSVModelChoiceField(
126 label=_('Tenant'),
127 queryset=Tenant.objects.all(),
128 required=False,
129 to_field_name='name',
130 help_text=_('Assigned tenant')
131 )
133 class Meta:
134 model = ASNRange
135 fields = ('name', 'slug', 'rir', 'start', 'end', 'tenant', 'description', 'owner', 'comments', 'tags')
138class ASNImportForm(PrimaryModelImportForm):
139 rir = CSVModelChoiceField(
140 label=_('RIR'),
141 queryset=RIR.objects.all(),
142 to_field_name='name',
143 help_text=_('Assigned RIR')
144 )
145 role = CSVModelChoiceField(
146 label=_('Role'),
147 queryset=Role.objects.all(),
148 required=False,
149 to_field_name='name',
150 help_text=_('Functional role')
151 )
152 tenant = CSVModelChoiceField(
153 label=_('Tenant'),
154 queryset=Tenant.objects.all(),
155 required=False,
156 to_field_name='name',
157 help_text=_('Assigned tenant')
158 )
160 class Meta:
161 model = ASN
162 fields = ('asn', 'rir', 'role', 'tenant', 'description', 'owner', 'comments', 'tags')
165class RoleImportForm(OrganizationalModelImportForm):
167 class Meta:
168 model = Role
169 fields = ('name', 'slug', 'weight', 'description', 'owner', 'comments', 'tags')
172class PrefixImportForm(ScopedImportForm, PrimaryModelImportForm):
173 vrf = CSVModelChoiceField(
174 label=_('VRF'),
175 queryset=VRF.objects.all(),
176 to_field_name='name',
177 required=False,
178 help_text=_('Assigned VRF')
179 )
180 tenant = CSVModelChoiceField(
181 label=_('Tenant'),
182 queryset=Tenant.objects.all(),
183 required=False,
184 to_field_name='name',
185 help_text=_('Assigned tenant')
186 )
187 vlan_group = CSVModelChoiceField(
188 label=_('VLAN group'),
189 queryset=VLANGroup.objects.all(),
190 required=False,
191 to_field_name='name',
192 help_text=_("VLAN's group (if any)")
193 )
194 vlan_site = CSVModelChoiceField(
195 label=_('VLAN Site'),
196 queryset=Site.objects.all(),
197 required=False,
198 to_field_name='name',
199 help_text=_("VLAN's site (if any)")
200 )
201 vlan = CSVModelChoiceField(
202 label=_('VLAN'),
203 queryset=VLAN.objects.all(),
204 required=False,
205 to_field_name='vid',
206 help_text=_("Assigned VLAN")
207 )
208 status = CSVChoiceField(
209 label=_('Status'),
210 choices=PrefixStatusChoices,
211 help_text=_('Operational status')
212 )
213 role = CSVModelChoiceField(
214 label=_('Role'),
215 queryset=Role.objects.all(),
216 required=False,
217 to_field_name='name',
218 help_text=_('Functional role')
219 )
221 class Meta:
222 model = Prefix
223 fields = (
224 'prefix', 'vrf', 'tenant', 'vlan_group', 'vlan_site', 'vlan', 'status', 'role', 'scope_type', 'scope_name',
225 'scope_id', 'is_pool', 'mark_utilized', 'description', 'owner', 'comments', 'tags',
226 )
227 labels = {
228 'scope_id': _('Scope ID'),
229 }
231 def __init__(self, data=None, *args, **kwargs):
232 super().__init__(data, *args, **kwargs)
234 if not data:
235 return
237 vlan_site = data.get('vlan_site')
238 vlan_group = data.get('vlan_group')
240 # Limit VLAN queryset by assigned site and/or group (if specified)
241 query = Q()
243 if vlan_site:
244 query |= Q(**{
245 f"site__{self.fields['vlan_site'].to_field_name}": vlan_site
246 })
248 if vlan_group:
249 query &= Q(**{
250 f"group__{self.fields['vlan_group'].to_field_name}": vlan_group
251 })
253 queryset = self.fields['vlan'].queryset.filter(query)
254 self.fields['vlan'].queryset = queryset
257class IPRangeImportForm(PrimaryModelImportForm):
258 vrf = CSVModelChoiceField(
259 label=_('VRF'),
260 queryset=VRF.objects.all(),
261 to_field_name='name',
262 required=False,
263 help_text=_('Assigned VRF')
264 )
265 tenant = CSVModelChoiceField(
266 label=_('Tenant'),
267 queryset=Tenant.objects.all(),
268 required=False,
269 to_field_name='name',
270 help_text=_('Assigned tenant')
271 )
272 status = CSVChoiceField(
273 label=_('Status'),
274 choices=IPRangeStatusChoices,
275 help_text=_('Operational status')
276 )
277 role = CSVModelChoiceField(
278 label=_('Role'),
279 queryset=Role.objects.all(),
280 required=False,
281 to_field_name='name',
282 help_text=_('Functional role')
283 )
285 class Meta:
286 model = IPRange
287 fields = (
288 'start_address', 'end_address', 'vrf', 'tenant', 'status', 'role', 'mark_populated', 'mark_utilized',
289 'description', 'owner', 'comments', 'tags',
290 )
293class IPAddressImportForm(PrimaryModelImportForm):
294 vrf = CSVModelChoiceField(
295 label=_('VRF'),
296 queryset=VRF.objects.all(),
297 to_field_name='name',
298 required=False,
299 help_text=_('Assigned VRF')
300 )
301 tenant = CSVModelChoiceField(
302 label=_('Tenant'),
303 queryset=Tenant.objects.all(),
304 to_field_name='name',
305 required=False,
306 help_text=_('Assigned tenant')
307 )
308 status = CSVChoiceField(
309 label=_('Status'),
310 choices=IPAddressStatusChoices,
311 help_text=_('Operational status')
312 )
313 role = CSVChoiceField(
314 label=_('Role'),
315 choices=IPAddressRoleChoices,
316 required=False,
317 help_text=_('Functional role')
318 )
319 device = CSVModelChoiceField(
320 label=_('Device'),
321 queryset=Device.objects.all(),
322 required=False,
323 to_field_name='name',
324 help_text=_('Parent device of assigned interface (if any)')
325 )
326 virtual_machine = CSVModelChoiceField(
327 label=_('Virtual machine'),
328 queryset=VirtualMachine.objects.all(),
329 required=False,
330 to_field_name='name',
331 help_text=_('Parent VM of assigned interface (if any)')
332 )
333 interface = CSVModelChoiceField(
334 label=_('Interface'),
335 queryset=Interface.objects.none(), # Can also refer to VMInterface
336 required=False,
337 to_field_name='name',
338 help_text=_('Assigned interface')
339 )
340 fhrp_group = CSVModelChoiceField(
341 label=_('FHRP Group'),
342 queryset=FHRPGroup.objects.all(),
343 required=False,
344 to_field_name='name',
345 help_text=_('Assigned FHRP Group name')
346 )
347 is_primary = forms.BooleanField(
348 label=_('Is primary'),
349 help_text=_('Make this the primary IP for the assigned device'),
350 required=False
351 )
352 is_oob = forms.BooleanField(
353 label=_('Is out-of-band'),
354 help_text=_('Designate this as the out-of-band IP address for the assigned device'),
355 required=False
356 )
358 class Meta:
359 model = IPAddress
360 fields = [
361 'address', 'vrf', 'tenant', 'status', 'role', 'device', 'virtual_machine', 'interface', 'fhrp_group',
362 'is_primary', 'is_oob', 'dns_name', 'description', 'owner', 'comments', 'tags',
363 ]
365 def __init__(self, data=None, *args, **kwargs):
366 super().__init__(data, *args, **kwargs)
368 if data:
370 # Limit interface queryset by assigned device
371 if data.get('device'):
372 self.fields['interface'].queryset = Interface.objects.filter(
373 **{f"device__{self.fields['device'].to_field_name}": data['device']}
374 )
376 # Limit interface queryset by assigned VM
377 elif data.get('virtual_machine'):
378 self.fields['interface'].queryset = VMInterface.objects.filter(
379 **{f"virtual_machine__{self.fields['virtual_machine'].to_field_name}": data['virtual_machine']}
380 )
382 def clean_is_primary(self):
383 # Make sure is_primary is None when it's not included in the uploaded data
384 if 'is_primary' not in self.data:
385 return None
386 return self.cleaned_data['is_primary']
388 def clean_is_oob(self):
389 # Make sure is_oob is None when it's not included in the uploaded data
390 if 'is_oob' not in self.data:
391 return None
392 return self.cleaned_data['is_oob']
394 def clean(self):
395 super().clean()
397 device = self.cleaned_data.get('device')
398 virtual_machine = self.cleaned_data.get('virtual_machine')
399 interface = self.cleaned_data.get('interface')
400 is_primary = self.cleaned_data.get('is_primary')
401 is_oob = self.cleaned_data.get('is_oob')
403 # Validate is_primary and is_oob
404 if is_primary and not device and not virtual_machine:
405 raise forms.ValidationError({
406 "is_primary": _("No device or virtual machine specified; cannot set as primary IP")
407 })
408 if is_oob and not device:
409 raise forms.ValidationError({
410 "is_oob": _("No device specified; cannot set as out-of-band IP")
411 })
412 if is_oob and virtual_machine:
413 raise forms.ValidationError({
414 "is_oob": _("Cannot set out-of-band IP for virtual machines")
415 })
416 if is_primary and not interface:
417 raise forms.ValidationError({
418 "is_primary": _("No interface specified; cannot set as primary IP")
419 })
420 if is_oob and not interface:
421 raise forms.ValidationError({
422 "is_oob": _("No interface specified; cannot set as out-of-band IP")
423 })
425 def save(self, *args, **kwargs):
427 # Set interface assignment
428 if self.cleaned_data.get('interface'):
429 self.instance.assigned_object = self.cleaned_data['interface']
430 if self.cleaned_data.get('fhrp_group'):
431 self.instance.assigned_object = self.cleaned_data['fhrp_group']
433 ipaddress = super().save(*args, **kwargs)
435 # Set as primary for device/VM
436 parent = self.cleaned_data.get('device') or self.cleaned_data.get('virtual_machine')
437 if parent and self.cleaned_data.get('is_primary') is not None:
438 if self.cleaned_data.get('is_primary'):
439 parent.snapshot()
440 if self.instance.address.version == 4:
441 parent.primary_ip4 = ipaddress
442 elif self.instance.address.version == 6:
443 parent.primary_ip6 = ipaddress
444 parent.save()
445 else:
446 # Only clear the primary IP if this IP is currently set as primary
447 if self.instance.address.version == 4 and parent.primary_ip4 == ipaddress:
448 parent.snapshot()
449 parent.primary_ip4 = None
450 parent.save()
451 elif self.instance.address.version == 6 and parent.primary_ip6 == ipaddress:
452 parent.snapshot()
453 parent.primary_ip6 = None
454 parent.save()
456 # Set as OOB for device
457 parent = self.cleaned_data.get('device')
458 if parent and self.cleaned_data.get('is_oob') is not None:
459 if self.cleaned_data.get('is_oob'):
460 parent.snapshot()
461 parent.oob_ip = ipaddress
462 parent.save()
463 elif parent.oob_ip == ipaddress:
464 # Only clear OOB if this IP is currently set as the OOB IP
465 parent.snapshot()
466 parent.oob_ip = None
467 parent.save()
469 return ipaddress
472class FHRPGroupImportForm(PrimaryModelImportForm):
473 protocol = CSVChoiceField(
474 label=_('Protocol'),
475 choices=FHRPGroupProtocolChoices
476 )
477 auth_type = CSVChoiceField(
478 label=_('Auth type'),
479 choices=FHRPGroupAuthTypeChoices,
480 required=False
481 )
483 class Meta:
484 model = FHRPGroup
485 fields = ('protocol', 'group_id', 'auth_type', 'auth_key', 'name', 'description', 'owner', 'comments', 'tags')
488class VLANGroupImportForm(ScopedImportForm, OrganizationalModelImportForm):
489 # Override ScopedImportForm.scope_type to set custom queryset
490 scope_type = CSVContentTypeField(
491 queryset=ContentType.objects.filter(model__in=VLANGROUP_SCOPE_TYPES),
492 required=False,
493 label=_('Scope type (app & model)')
494 )
495 vid_ranges = NumericRangeArrayField(
496 required=False
497 )
498 tenant = CSVModelChoiceField(
499 label=_('Tenant'),
500 queryset=Tenant.objects.all(),
501 required=False,
502 to_field_name='name',
503 help_text=_('Assigned tenant')
504 )
506 class Meta:
507 model = VLANGroup
508 fields = (
509 'name', 'slug', 'scope_type', 'scope_name', 'scope_id', 'vid_ranges', 'tenant', 'description', 'owner',
510 'comments', 'tags',
511 )
512 labels = {
513 'scope_id': _('Scope ID'),
514 }
517class VLANImportForm(PrimaryModelImportForm):
518 site = CSVModelChoiceField(
519 label=_('Site'),
520 queryset=Site.objects.all(),
521 required=False,
522 to_field_name='name',
523 help_text=_('Assigned site')
524 )
525 group = CSVModelChoiceField(
526 label=_('Group'),
527 queryset=VLANGroup.objects.all(),
528 required=False,
529 to_field_name='name',
530 help_text=_('Assigned VLAN group')
531 )
532 tenant = CSVModelChoiceField(
533 label=_('Tenant'),
534 queryset=Tenant.objects.all(),
535 to_field_name='name',
536 required=False,
537 help_text=_('Assigned tenant')
538 )
539 status = CSVChoiceField(
540 label=_('Status'),
541 choices=VLANStatusChoices,
542 help_text=_('Operational status')
543 )
544 role = CSVModelChoiceField(
545 label=_('Role'),
546 queryset=Role.objects.all(),
547 required=False,
548 to_field_name='name',
549 help_text=_('Functional role')
550 )
551 qinq_role = CSVChoiceField(
552 label=_('Q-in-Q role'),
553 choices=VLANQinQRoleChoices,
554 required=False,
555 help_text=_('Operational status')
556 )
557 qinq_svlan = CSVModelChoiceField(
558 label=_('Q-in-Q SVLAN'),
559 queryset=VLAN.objects.all(),
560 required=False,
561 to_field_name='vid',
562 help_text=_("Service VLAN (for Q-in-Q/802.1ad customer VLANs)")
563 )
565 class Meta:
566 model = VLAN
567 fields = (
568 'site', 'group', 'vid', 'name', 'tenant', 'status', 'role', 'description', 'qinq_role', 'qinq_svlan',
569 'owner', 'comments', 'tags',
570 )
573class VLANTranslationPolicyImportForm(PrimaryModelImportForm):
575 class Meta:
576 model = VLANTranslationPolicy
577 fields = ('name', 'description', 'owner', 'comments', 'tags')
580class VLANTranslationRuleImportForm(NetBoxModelImportForm):
581 policy = CSVModelChoiceField(
582 label=_('Policy'),
583 queryset=VLANTranslationPolicy.objects.all(),
584 to_field_name='name',
585 help_text=_('VLAN translation policy')
586 )
588 class Meta:
589 model = VLANTranslationRule
590 fields = ('policy', 'local_vid', 'remote_vid')
593class ServicePortMappingsImportMixin(forms.Form):
594 """
595 Adds a ``port_mappings`` CSV column parsed from a comma-separated list of ``protocol/port`` pairs
596 (e.g. "tcp/80,udp/53") into the model's flat ``['tcp/80', 'udp/53']`` list. A pair's port half may be
597 a hyphen range (e.g. "tcp/8000-8010"), matching the port syntax the edit form accepts.
598 """
599 port_mappings = SimpleArrayField(
600 base_field=forms.CharField(),
601 label=_('Port mappings'),
602 required=True,
603 help_text=_('Comma-separated list of protocol/port pairs in double quotes (e.g. "tcp/80,udp/53"). '
604 'A port range may be given with a hyphen (e.g. "tcp/8000-8010").')
605 )
607 def clean_port_mappings(self):
608 mappings = self.cleaned_data.get('port_mappings')
609 if not mappings:
610 return []
611 # Expand any hyphen range in a pair's port half (tcp/8000-8010 -> tcp/8000, tcp/8001, ...) so the
612 # CSV accepts the same port syntax as the edit form. validate_port_mappings then normalizes and
613 # checks each expanded pair, matching the protocol case-insensitively.
614 expanded = []
615 for mapping in mappings:
616 protocol, ports = split_port_mapping(mapping.strip())
617 try:
618 expanded.extend(expand_port_mapping(protocol, ports))
619 except DjangoValidationError as exc:
620 raise forms.ValidationError(exc.messages)
621 try:
622 expanded = validate_port_mappings(expanded)
623 except DjangoValidationError as exc:
624 raise forms.ValidationError(exc.messages)
625 return expanded
628class ServiceTemplateImportForm(ServicePortMappingsImportMixin, PrimaryModelImportForm):
630 class Meta:
631 model = ServiceTemplate
632 fields = ('name', 'port_mappings', 'description', 'owner', 'comments', 'tags')
635class ServiceImportForm(ServicePortMappingsImportMixin, PrimaryModelImportForm):
636 parent_object_type = CSVContentTypeField(
637 queryset=ContentType.objects.filter(SERVICE_ASSIGNMENT_MODELS),
638 required=True,
639 label=_('Parent type (app & model)')
640 )
641 parent = CSVModelChoiceField(
642 label=_('Parent'),
643 queryset=Device.objects.all(),
644 required=False,
645 to_field_name='name',
646 help_text=_('Parent object name')
647 )
648 parent_object_id = forms.IntegerField(
649 required=False,
650 help_text=_('Parent object ID'),
651 )
652 ipaddresses = CSVModelMultipleChoiceField(
653 queryset=IPAddress.objects.all(),
654 required=False,
655 to_field_name='address',
656 help_text=_('IP Address'),
657 )
659 class Meta:
660 model = Service
661 fields = (
662 'ipaddresses', 'name', 'port_mappings', 'description', 'owner', 'comments', 'tags',
663 )
665 def __init__(self, data=None, *args, **kwargs):
666 super().__init__(data, *args, **kwargs)
668 # Limit parent queryset by assigned parent object type
669 if data:
670 match data.get('parent_object_type'):
671 case 'dcim.device':
672 self.fields['parent'].queryset = Device.objects.all()
673 case 'ipam.fhrpgroup':
674 self.fields['parent'].queryset = FHRPGroup.objects.all()
675 case 'virtualization.virtualmachine':
676 self.fields['parent'].queryset = VirtualMachine.objects.all()
678 def save(self, *args, **kwargs):
679 if (parent := self.cleaned_data.get('parent')):
680 self.instance.parent = parent
682 return super().save(*args, **kwargs)
684 def clean(self):
685 super().clean()
687 if (parent_ct := self.cleaned_data.get('parent_object_type')):
688 if (parent := self.cleaned_data.get('parent')):
689 self.cleaned_data['parent_object_id'] = parent.pk
690 elif (parent_id := self.cleaned_data.get('parent_object_id')):
691 parent = parent_ct.model_class().objects.filter(id=parent_id).first()
692 self.cleaned_data['parent'] = parent
693 else:
694 # If a parent object type is passed and we've made it here, then raise a validation
695 # error since an associated parent object or parent object id has not been passed
696 raise forms.ValidationError(
697 _("One of parent or parent_object_id must be included with parent_object_type")
698 )
700 # making sure parent is defined. In cases where an import is resulting in an update, the
701 # import data might not include the parent object and so the above logic might not be
702 # triggered
703 parent = self.cleaned_data.get('parent')
704 for ip_address in self.cleaned_data.get('ipaddresses', []):
705 if not (assigned := ip_address.assigned_object) or ( # no assigned object
706 (isinstance(parent, FHRPGroup) and assigned != parent) # assigned to FHRPGroup
707 and getattr(assigned, 'parent_object') != parent # assigned to [VM]Interface
708 ):
709 raise forms.ValidationError(
710 _("{ip} is not assigned to this parent.").format(ip=ip_address)
711 )
713 return self.cleaned_data