Coverage for extras/forms/bulk_import.py: 47%
181 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import re
3from django import forms
4from django.contrib.postgres.forms import SimpleArrayField
5from django.core.exceptions import NON_FIELD_ERRORS, ObjectDoesNotExist, ValidationError
6from django.utils.translation import gettext_lazy as _
8from core.models import DataFile, DataSource, ObjectType
9from extras.choices import *
10from extras.models import *
11from netbox.event_rules import get_event_rule_action
12from netbox.events import get_event_type_choices
13from netbox.forms import NetBoxModelImportForm, OwnerCSVMixin, PrimaryModelImportForm
14from users.models import Group, User
15from utilities.forms import CSVModelForm
16from utilities.forms.fields import (
17 CSVChoiceField,
18 CSVContentTypeField,
19 CSVModelChoiceField,
20 CSVModelMultipleChoiceField,
21 CSVMultipleChoiceField,
22 CSVMultipleContentTypeField,
23 SlugField,
24)
26__all__ = (
27 'ConfigContextProfileImportForm',
28 'ConfigTemplateImportForm',
29 'CustomFieldChoiceSetImportForm',
30 'CustomFieldImportForm',
31 'CustomLinkImportForm',
32 'EventRuleImportForm',
33 'ExportTemplateImportForm',
34 'JournalEntryImportForm',
35 'NotificationGroupImportForm',
36 'SavedFilterImportForm',
37 'TagImportForm',
38 'WebhookImportForm',
39)
42class CustomFieldImportForm(OwnerCSVMixin, CSVModelForm):
43 object_types = CSVMultipleContentTypeField(
44 label=_('Object types'),
45 queryset=ObjectType.objects.with_feature('custom_fields'),
46 help_text=_("One or more assigned object types")
47 )
48 type = CSVChoiceField(
49 label=_('Type'),
50 choices=CustomFieldTypeChoices,
51 help_text=_('Field data type (e.g. text, integer, etc.)')
52 )
53 related_object_type = CSVContentTypeField(
54 label=_('Object type'),
55 queryset=ObjectType.objects.public(),
56 required=False,
57 help_text=_("Object type (for object or multi-object fields)")
58 )
59 choice_set = CSVModelChoiceField(
60 label=_('Choice set'),
61 queryset=CustomFieldChoiceSet.objects.all(),
62 to_field_name='name',
63 required=False,
64 help_text=_('Choice set (for selection fields)')
65 )
66 ui_visible = CSVChoiceField(
67 label=_('UI visible'),
68 choices=CustomFieldUIVisibleChoices,
69 required=False,
70 help_text=_('Whether the custom field is displayed in the UI')
71 )
72 ui_editable = CSVChoiceField(
73 label=_('UI editable'),
74 choices=CustomFieldUIEditableChoices,
75 required=False,
76 help_text=_('Whether the custom field is editable in the UI')
77 )
79 class Meta:
80 model = CustomField
81 fields = (
82 'name', 'label', 'group_name', 'type', 'object_types', 'related_object_type', 'required', 'unique',
83 'description', 'search_weight', 'filter_logic', 'default', 'choice_set', 'weight', 'validation_minimum',
84 'validation_maximum', 'validation_regex', 'validation_schema', 'ui_visible', 'ui_editable',
85 'is_cloneable', 'nulls_first', 'owner', 'comments',
86 )
89class CustomFieldChoiceSetImportForm(OwnerCSVMixin, CSVModelForm):
90 base_choices = CSVChoiceField(
91 choices=CustomFieldChoiceSetBaseChoices,
92 required=False,
93 help_text=_('The base set of predefined choices to use (if any)')
94 )
95 extra_choices = SimpleArrayField(
96 base_field=forms.CharField(),
97 required=False,
98 help_text=_(
99 'Quoted string of comma-separated field choices with optional labels separated by colon: '
100 '"choice1:First Choice,choice2:Second Choice"'
101 )
102 )
103 choice_colors = SimpleArrayField(
104 base_field=forms.CharField(),
105 required=False,
106 help_text=_(
107 'Quoted string of comma-separated color mappings in the format '
108 '"choice1:red,choice2:green". Supported colors: {colors}'
109 ).format(colors=', '.join(CustomFieldChoiceColorChoices.values())),
110 )
112 class Meta:
113 model = CustomFieldChoiceSet
114 fields = (
115 'name', 'description', 'base_choices', 'extra_choices', 'choice_colors', 'order_alphabetically', 'owner',
116 )
118 def clean_extra_choices(self):
119 if isinstance(self.cleaned_data['extra_choices'], list):
120 data = []
121 for line in self.cleaned_data['extra_choices']:
122 try:
123 value, label = re.split(r'(?<!\\):', line, maxsplit=1)
124 value = value.replace('\\:', ':')
125 label = label.replace('\\:', ':')
126 except ValueError:
127 value, label = line, line
128 data.append((value, label))
129 return data
130 return None
132 def clean_choice_colors(self):
133 if isinstance(self.cleaned_data['choice_colors'], list):
134 data = {}
135 for line in self.cleaned_data['choice_colors']:
136 try:
137 value, color = re.split(r'(?<!\\):', line, maxsplit=1)
138 value = value.replace('\\:', ':')
139 except ValueError as e:
140 raise forms.ValidationError(
141 _("Invalid color mapping '{line}'. Use the format value:color.").format(line=line)
142 ) from e
144 value = value.strip()
145 color = color.strip()
146 if value in data:
147 raise forms.ValidationError(
148 _("Duplicate color mapping defined for choice '{value}'.").format(value=value)
149 )
150 data[value] = color
151 return data
152 return {}
155class CustomLinkImportForm(OwnerCSVMixin, CSVModelForm):
156 object_types = CSVMultipleContentTypeField(
157 label=_('Object types'),
158 queryset=ObjectType.objects.with_feature('custom_links'),
159 help_text=_("One or more assigned object types")
160 )
161 button_class = CSVChoiceField(
162 label=_('button class'),
163 required=False,
164 choices=CustomLinkButtonClassChoices,
165 help_text=_('The class of the first link in a group will be used for the dropdown button')
166 )
168 class Meta:
169 model = CustomLink
170 fields = (
171 'name', 'object_types', 'enabled', 'weight', 'group_name', 'button_class', 'new_window', 'link_text',
172 'link_url', 'owner',
173 )
176class ExportTemplateImportForm(OwnerCSVMixin, CSVModelForm):
177 object_types = CSVMultipleContentTypeField(
178 label=_('Object types'),
179 queryset=ObjectType.objects.with_feature('export_templates'),
180 help_text=_("One or more assigned object types")
181 )
182 template_code = forms.CharField(
183 label=_('Template code'),
184 required=False,
185 help_text=_('Jinja2 template code, if not sourced from a data file')
186 )
187 data_source = CSVModelChoiceField(
188 label=_('Data source'),
189 queryset=DataSource.objects.all(),
190 required=False,
191 to_field_name='name',
192 help_text=_('Data source which provides the data file')
193 )
194 data_file = CSVModelChoiceField(
195 label=_('Data file'),
196 queryset=DataFile.objects.all(),
197 required=False,
198 to_field_name='path',
199 help_text=_('Data file containing the template code')
200 )
201 auto_sync_enabled = forms.BooleanField(
202 required=False,
203 label=_('Auto sync enabled'),
204 help_text=_("Enable automatic synchronization of template content when the data file is updated")
205 )
207 class Meta:
208 model = ExportTemplate
209 fields = (
210 'name', 'object_types', 'description', 'environment_params', 'mime_type', 'file_name', 'file_extension',
211 'as_attachment', 'template_code', 'data_source', 'data_file', 'auto_sync_enabled', 'owner',
212 )
214 def clean_data_source(self):
215 # Paths are unique only within a source, and Meta.fields cleans data_source before data_file
216 data_source = self.cleaned_data.get('data_source')
217 if data_source and 'data_file' in self.fields:
218 self.fields['data_file'].queryset = self.fields['data_file'].queryset.filter(source=data_source)
220 return data_source
222 def clean(self):
223 super().clean()
225 # An update record carries only the fields being changed, so fall back to the stored values
226 template_code = self.cleaned_data.get('template_code', self.instance.template_code)
227 data_file = self.cleaned_data.get('data_file', self.instance.data_file)
228 if not template_code and not data_file:
229 raise forms.ValidationError(_("Must specify either local content or a data file"))
231 return self.cleaned_data
234class ConfigContextProfileImportForm(PrimaryModelImportForm):
236 class Meta:
237 model = ConfigContextProfile
238 fields = [
239 'name', 'description', 'schema', 'owner', 'comments', 'tags',
240 ]
243class ConfigTemplateImportForm(OwnerCSVMixin, CSVModelForm):
244 data_source = CSVModelChoiceField(
245 label=_('Data source'),
246 queryset=DataSource.objects.all(),
247 required=False,
248 to_field_name='name',
249 help_text=_('Data source which provides the data file')
250 )
251 data_file = CSVModelChoiceField(
252 label=_('Data file'),
253 queryset=DataFile.objects.all(),
254 required=False,
255 to_field_name='path',
256 help_text=_('Data file containing the template code')
257 )
258 auto_sync_enabled = forms.BooleanField(
259 required=False,
260 label=_('Auto sync enabled'),
261 help_text=_("Enable automatic synchronization of template content when the data file is updated")
262 )
264 class Meta:
265 model = ConfigTemplate
266 fields = (
267 'name', 'description', 'template_code', 'data_source', 'data_file', 'auto_sync_enabled',
268 'environment_params', 'mime_type', 'file_name', 'file_extension', 'as_attachment', 'debug', 'owner',
269 'tags',
270 )
272 def clean_data_source(self):
273 # Paths are unique only within a source, and Meta.fields cleans data_source before data_file
274 data_source = self.cleaned_data.get('data_source')
275 if data_source and 'data_file' in self.fields:
276 self.fields['data_file'].queryset = self.fields['data_file'].queryset.filter(source=data_source)
278 return data_source
280 def clean(self):
281 super().clean()
283 # Make sure template_code is None when it's not included in the uploaded data
284 if not self.data.get('template_code') and not self.data.get('data_file'):
285 raise forms.ValidationError(_("Must specify either local content or a data file"))
286 return self.cleaned_data
289class SavedFilterImportForm(OwnerCSVMixin, CSVModelForm):
290 object_types = CSVMultipleContentTypeField(
291 label=_('Object types'),
292 queryset=ObjectType.objects.all(),
293 help_text=_("One or more assigned object types")
294 )
296 class Meta:
297 model = SavedFilter
298 fields = (
299 'name', 'slug', 'object_types', 'description', 'weight', 'enabled', 'shared', 'parameters', 'owner',
300 )
303class WebhookImportForm(OwnerCSVMixin, NetBoxModelImportForm):
305 class Meta:
306 model = Webhook
307 fields = (
308 'name', 'payload_url', 'http_method', 'http_content_type', 'additional_headers', 'body_template',
309 'secret', 'ssl_verification', 'ca_file_path', 'timeout', 'description', 'owner', 'tags'
310 )
313class EventRuleImportForm(OwnerCSVMixin, NetBoxModelImportForm):
314 object_types = CSVMultipleContentTypeField(
315 label=_('Object types'),
316 queryset=ObjectType.objects.with_feature('event_rules'),
317 help_text=_("One or more assigned object types")
318 )
319 event_types = CSVMultipleChoiceField(
320 choices=get_event_type_choices(),
321 label=_('Event types'),
322 help_text=_('The event type(s) which will trigger this rule')
323 )
324 action_object = forms.CharField(
325 label=_('Action object'),
326 required=False,
327 help_text=_(
328 'The target object for the action, if it requires one. The expected format depends on the action type '
329 '(e.g. a webhook or notification group name, or a script as dotted path module.Class).'
330 )
331 )
333 class Meta:
334 model = EventRule
335 fields = (
336 'name', 'description', 'enabled', 'conditions', 'object_types', 'event_types', 'action_type',
337 'owner', 'comments', 'tags'
338 )
340 def clean(self):
341 super().clean()
343 action_object = self.cleaned_data.get('action_object')
344 action_type = self.cleaned_data.get('action_type')
345 if not action_type:
346 return
348 action = get_event_rule_action(action_type)
349 if action is None:
350 raise forms.ValidationError({
351 'action_type': _('"{action_type}" is not a registered action type.').format(action_type=action_type)
352 })
354 if not action_object:
355 if action.object_required:
356 raise forms.ValidationError({
357 'action_object': _("This action type requires a target object."),
358 })
359 # Clear any action_object this instance previously had (relevant for a CSV row that
360 # updates an existing rule, matched by id, to a now-object-less action_type).
361 self.instance.action_object_type = None
362 self.instance.action_object_id = None
363 return
365 if action.object_model is None:
366 raise forms.ValidationError({
367 'action_object': _("This action type does not operate against a target object."),
368 })
370 try:
371 obj = action.resolve_import_object(action_object)
372 except ObjectDoesNotExist:
373 raise forms.ValidationError({
374 'action_object': _("{name} not found").format(name=action_object)
375 })
376 if obj is None:
377 raise forms.ValidationError({
378 'action_object': _("This action type does not support bulk import.")
379 })
381 # Assign the GFK itself (not just action_object_type/id) so EventRule.clean()'s later
382 # access to self.action_object hits the descriptor cache instead of a fresh SELECT --
383 # for a non-proxy object_model, where the concrete and non-concrete content types match.
384 self.instance.action_object = obj
385 self.instance.action_object_type = ObjectType.objects.get_for_model(obj, for_concrete_model=False)
387 def _update_errors(self, errors):
388 # Remap errors keyed by fields this form doesn't expose (e.g. action_object_id) to
389 # NON_FIELD_ERRORS; otherwise Django's add_error() raises ValueError instead of failing validation normally.
390 if hasattr(errors, 'error_dict'):
391 remapped = {}
392 for field, messages in errors.error_dict.items():
393 key = field if field == NON_FIELD_ERRORS or field in self.fields else NON_FIELD_ERRORS
394 remapped.setdefault(key, []).extend(messages)
395 errors = ValidationError(remapped)
396 super()._update_errors(errors)
399class TagImportForm(OwnerCSVMixin, CSVModelForm):
400 slug = SlugField()
401 object_types = CSVMultipleContentTypeField(
402 label=_('Object types'),
403 queryset=ObjectType.objects.with_feature('tags'),
404 help_text=_("One or more assigned object types"),
405 required=False,
406 )
408 class Meta:
409 model = Tag
410 fields = (
411 'name', 'slug', 'color', 'weight', 'description', 'object_types', 'owner',
412 )
415class JournalEntryImportForm(NetBoxModelImportForm):
416 assigned_object_type = CSVContentTypeField(
417 queryset=ObjectType.objects.all(),
418 label=_('Assigned object type'),
419 )
420 kind = CSVChoiceField(
421 label=_('Kind'),
422 choices=JournalEntryKindChoices,
423 help_text=_('The classification of entry')
424 )
425 comments = forms.CharField(
426 label=_('Comments'),
427 required=True
428 )
430 class Meta:
431 model = JournalEntry
432 fields = (
433 'assigned_object_type', 'assigned_object_id', 'created_by', 'kind', 'comments', 'tags'
434 )
436 def __init__(self, *args, **kwargs):
437 super().__init__(*args, **kwargs)
438 # If not creating a new JournalEntry, disable the created_by field
439 if self.instance and not self.instance._state.adding:
440 self.fields['created_by'].disabled = True
443class NotificationGroupImportForm(CSVModelForm):
444 users = CSVModelMultipleChoiceField(
445 label=_('Users'),
446 queryset=User.objects.all(),
447 required=False,
448 to_field_name='username',
449 help_text=_('User names separated by commas, encased with double quotes')
450 )
451 groups = CSVModelMultipleChoiceField(
452 label=_('Groups'),
453 queryset=Group.objects.all(),
454 required=False,
455 to_field_name='name',
456 help_text=_('Group names separated by commas, encased with double quotes')
457 )
459 class Meta:
460 model = NotificationGroup
461 fields = ('name', 'description', 'users', 'groups')