Coverage for netbox_data_import/forms.py: 97.88%
236 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-30 09:47 +0000
1# SPDX-License-Identifier: Apache-2.0
2# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com>
3import copy
5from django import forms
6from django.utils.html import format_html
7from dcim.models import Site, Location
8from tenancy.models import Tenant
9from netbox.forms import NetBoxModelBulkEditForm, NetBoxModelFilterSetForm, NetBoxModelForm, NetBoxModelImportForm
10from utilities.forms import BOOLEAN_WITH_BLANK_CHOICES, add_blank_choice
11from utilities.forms.fields import DynamicModelChoiceField, TagFilterField
12from utilities.forms.rendering import FieldSet
13from .adapters import adapter_choices, get_adapter, selectable_adapter_choices
14from .catalog import CATALOG
15from .models import (
16 CableClassMapping,
17 ImportProfile,
18 InferenceBackend,
19 ColumnMapping,
20 ClassRoleMapping,
21 DeviceTypeMapping,
22 ColumnTransformRule,
23 CableSegmentOverride,
24 PortClassMapping,
25 _require_adapter_config_mapping,
26 validate_registered_adapter,
27)
28from .cable_policy import cable_type_choices, compatible_cable_profile_choices
30_EXPLICIT_NONE = "__explicit_none__"
32_CREDENTIAL_REFERENCE_EXAMPLE = """{
33 "backend": "vault_kv_v2",
34 "mount": "secret",
35 "path": "inference/backend",
36 "field": "api_key"
37}"""
39_CREDENTIAL_REFERENCE_HELP = format_html(
40 "This JSON tells the plugin where to find the API key in Vault. It does not contain the API key. "
41 "Configure Vault first in the NetBox plugin settings at "
42 "<code>PLUGINS_CONFIG["netbox_data_import"]["vault"]</code>. "
43 '<details class="mt-1" data-credential-reference-help>'
44 "<summary>Show the required JSON shape</summary>"
45 '<pre class="mt-2 mb-0"><code>{}</code></pre>'
46 "</details>",
47 _CREDENTIAL_REFERENCE_EXAMPLE,
48)
51class _RuntimeCableChoiceField(forms.ChoiceField):
52 """Render only current choices while letting shared validation classify submitted values."""
54 def validate(self, value):
55 """Apply required-field validation without duplicating the runtime choice rule."""
56 forms.Field.validate(self, value)
59def _decision_choices(runtime_choices, *, unresolved=True):
60 """Add the explicit-none form state, and the unresolved one a complete decision does not offer."""
61 if _EXPLICIT_NONE in {value for value, _label in runtime_choices}:
62 raise RuntimeError("A NetBox Cable choice conflicts with the form's explicit-none control value.")
63 states = [("", "Unresolved")] if unresolved else []
64 return [*states, (_EXPLICIT_NONE, "Explicitly none"), *runtime_choices]
67def _with_stored_decision(choices, resolved, value):
68 """Keep a stored decision selectable after NetBox stops offering it.
70 A select cannot send back a value it does not list, so the browser would submit the first
71 option and `clean()` would record the loss as an unresolved decision without an error.
72 """
73 if not resolved or value is None or value in {key for key, _label in choices}:
74 return choices
75 return [*choices, (value, f"{value} (no longer offered)")]
78def _decision_initial(resolved, value):
79 """Return the form value for one stored tri-state decision."""
80 if not resolved:
81 return ""
82 return _EXPLICIT_NONE if value is None else value
85def _decode_decision(value):
86 """Return the resolved flag and nullable stored value for one form selection."""
87 if value in (None, ""):
88 return False, None
89 if value == _EXPLICIT_NONE:
90 return True, None
91 return True, value
94def cable_policy_form_initial(policy: dict) -> dict:
95 """Return the control values that offer one decided Cable policy as the default."""
96 if not policy:
97 return {}
98 return {
99 "cable_type": _decision_initial(True, policy["cable_type"]),
100 "cable_profile": _decision_initial(True, policy["cable_profile"]),
101 }
104def _profile_output_kinds(form):
105 """Return the output kinds of the profile this row belongs to, or None when unknown."""
106 profile = form.instance.profile_id or form.initial.get("profile")
107 if not profile:
108 return None
109 if not isinstance(profile, ImportProfile):
110 # NetBox seeds form initial from the query string, so a non-numeric value must not reach the query.
111 try:
112 profile = ImportProfile.objects.filter(pk=int(profile)).first()
113 except (TypeError, ValueError):
114 return None
115 if profile is None:
116 return None
117 return profile.output_kinds
120def _with_stored_target(choices, stored, output_kinds, *, allow_candidates=True):
121 """Keep a stored key-family target selectable, so an existing row can be re-saved.
123 CATALOG.choices lists fixed keys only, so a stored family key such as `extra_json:asset_id` is
124 never among them. Re-offer it only when the model would still accept it: the row's clean()
125 runs the same check, and offering more would put a choice in the list that saving rejects.
126 """
127 if not stored or stored in {key for key, _label in choices}:
128 return choices
129 if not CATALOG.is_valid(stored, output_kinds=output_kinds, allow_candidates=allow_candidates):
130 return choices
131 return [*choices, (stored, CATALOG.display(stored))]
134class ImportProfileForm(NetBoxModelForm):
135 """Form for creating and editing ImportProfile instances.
137 The Source Adapter is asked for first and is disabled after creation. The selected adapter
138 declares the remaining configuration fields.
139 """
141 class Meta:
142 model = ImportProfile
143 fields = ["name", "description", "source_adapter", "tags"]
145 def __init__(self, *args, **kwargs):
146 super().__init__(*args, **kwargs)
147 self._config_form_class = None
148 stored = _require_adapter_config_mapping(self.instance.adapter_config)
149 adapter = get_adapter(self._selected_adapter_key())
150 if adapter is None:
151 return
152 if self.instance.pk:
153 self.fields["source_adapter"].disabled = True
154 else:
155 self.fields["source_adapter"].choices = selectable_adapter_choices()
156 self._config_form_class = adapter.config_form_class()
157 for name, field in self._config_form_class.base_fields.items():
158 self.fields[name] = copy.deepcopy(field)
159 if name in stored:
160 self.initial.setdefault(name, stored[name])
162 def _selected_adapter_key(self):
163 """Return the adapter key this form edits: the stored one, or the submitted choice."""
164 if self.instance.pk:
165 return self.instance.source_adapter
166 if self.is_bound:
167 return self.data.get(self.add_prefix("source_adapter")) or self.instance.source_adapter
168 return self.initial.get("source_adapter") or self.instance.source_adapter
170 def clean(self):
171 """Collect the adapter-declared fields into ``adapter_config``."""
172 cleaned = super().clean()
173 if cleaned is None:
174 cleaned = self.cleaned_data
175 if self._config_form_class is not None:
176 self.instance.adapter_config = self._config_form_class.normalize(cleaned)
177 return cleaned
180class ImportProfileImportForm(NetBoxModelImportForm):
181 """CSV/YAML bulk-import form for ImportProfile objects (profile metadata only).
183 Adapter configuration is nested, so it is set through the edit UI, the REST API, or the
184 hierarchical profile YAML import.
185 """
187 class Meta:
188 model = ImportProfile
189 fields = ["name", "description", "source_adapter", "tags"]
192class ImportProfileBulkEditForm(NetBoxModelBulkEditForm):
193 """Bulk-edit fields that apply safely across import profiles.
195 The Source Adapter is immutable and ``adapter_config`` is adapter-scoped, so neither is
196 bulk-editable.
197 """
199 model = ImportProfile
201 description = forms.CharField(required=False, widget=forms.Textarea(attrs={"rows": 3}))
203 nullable_fields = ("description",)
206class ImportProfileFilterForm(NetBoxModelFilterSetForm):
207 """Filters tab of the Import Profile list."""
209 model = ImportProfile
210 fieldsets = (
211 FieldSet("q", "filter_id", "tag"),
212 FieldSet("source_adapter", name="Attributes"),
213 )
214 source_adapter = forms.ChoiceField(choices=lambda: add_blank_choice(adapter_choices()), required=False)
215 tag = TagFilterField(model)
218class ColumnMappingForm(forms.ModelForm):
219 """Form for creating and editing ColumnMapping instances."""
221 target_field = forms.ChoiceField(choices=CATALOG.choices)
223 class Meta:
224 model = ColumnMapping
225 fields = ["source_column", "target_field"]
227 def __init__(self, *args, **kwargs):
228 super().__init__(*args, **kwargs)
229 output_kinds = _profile_output_kinds(self)
230 choices = CATALOG.choices(output_kinds=output_kinds)
231 self.fields["target_field"].choices = _with_stored_target(choices, self.instance.target_field, output_kinds)
234class ClassRoleMappingForm(forms.ModelForm):
235 """Form for creating and editing ClassRoleMapping instances."""
237 class Meta:
238 model = ClassRoleMapping
239 fields = ["source_class", "creates_rack", "rack_type", "role_slug", "ignore"]
241 def clean(self):
242 """Require role_slug unless creates_rack or ignore is set."""
243 cleaned = super().clean()
244 creates_rack = cleaned.get("creates_rack")
245 ignore = cleaned.get("ignore")
246 role_slug = (cleaned.get("role_slug") or "").strip()
247 if not creates_rack and not ignore and not role_slug:
248 self.add_error(
249 "role_slug",
250 "A device role slug is required unless 'creates rack' or 'ignore' is checked.",
251 )
252 return cleaned
255class CableClassMappingForm(forms.ModelForm):
256 """Form for one source CableClass and its two independent target decisions."""
258 cable_type = _RuntimeCableChoiceField(required=False)
259 cable_profile = _RuntimeCableChoiceField(required=False)
261 class Meta:
262 model = CableClassMapping
263 fields = ["cable_class", "cable_type", "cable_profile"]
265 def __init__(self, *args, **kwargs):
266 super().__init__(*args, **kwargs)
267 self.fields["cable_type"].choices = _with_stored_decision(
268 _decision_choices(cable_type_choices()),
269 self.instance.cable_type_resolved,
270 self.instance.cable_type,
271 )
272 self.fields["cable_profile"].choices = _with_stored_decision(
273 _decision_choices(compatible_cable_profile_choices()),
274 self.instance.cable_profile_resolved,
275 self.instance.cable_profile,
276 )
277 self.initial["cable_type"] = _decision_initial(
278 self.instance.cable_type_resolved,
279 self.instance.cable_type,
280 )
281 self.initial["cable_profile"] = _decision_initial(
282 self.instance.cable_profile_resolved,
283 self.instance.cable_profile,
284 )
286 def clean(self):
287 """Decode each tri-state value and apply the shared runtime-choice validation."""
288 cleaned = super().clean()
289 type_resolved, cable_type = _decode_decision(cleaned.get("cable_type"))
290 profile_resolved, cable_profile = _decode_decision(cleaned.get("cable_profile"))
291 self.instance.cable_type_resolved = type_resolved
292 self.instance.cable_type = cable_type
293 self.instance.cable_profile_resolved = profile_resolved
294 self.instance.cable_profile = cable_profile
295 cleaned["cable_type"] = cable_type
296 cleaned["cable_profile"] = cable_profile
297 return cleaned
300class PortClassMappingForm(forms.ModelForm):
301 """Form for one source PortClass label and the fixed PortClass it maps to."""
303 class Meta:
304 model = PortClassMapping
305 fields = ["source_label", "port_class"]
308class CableSegmentOverrideForm(forms.ModelForm):
309 """Force both Cable dimensions on one planned segment, or only the Cable Type on a breakout leg."""
311 # One control per segment in a narrow panel, so the small variant keeps a long path readable.
312 cable_type = _RuntimeCableChoiceField(widget=forms.Select(attrs={"class": "form-select form-select-sm"}))
313 cable_profile = _RuntimeCableChoiceField(widget=forms.Select(attrs={"class": "form-select form-select-sm"}))
315 class Meta:
316 model = CableSegmentOverride
317 fields = ["cable_type", "cable_profile"]
319 def __init__(self, *args, breakout_leg=False, **kwargs):
320 super().__init__(*args, **kwargs)
321 # The empty state is a prompt the required field refuses, not a decision the override stores.
322 self.fields["cable_type"].choices = _with_stored_decision(
323 [("", "Choose a Cable Type"), *_decision_choices(cable_type_choices(), unresolved=False)],
324 True,
325 self.instance.cable_type,
326 )
327 if breakout_leg:
328 # The breakout decides a leg's profile, so the override keeps the profile it stores.
329 del self.fields["cable_profile"]
330 return
331 self.fields["cable_profile"].choices = _with_stored_decision(
332 [("", "Choose a Cable Profile"), *_decision_choices(compatible_cable_profile_choices(), unresolved=False)],
333 True,
334 self.instance.cable_profile,
335 )
337 def clean(self):
338 """Decode each explicit decision the form offers, which the model then validates against the instance."""
339 cleaned = super().clean()
340 for name in ("cable_type", "cable_profile"):
341 if name in self.fields:
342 _resolved, value = _decode_decision(cleaned.get(name))
343 setattr(self.instance, name, value)
344 cleaned[name] = value
345 return cleaned
348class DeviceTypeMappingForm(forms.ModelForm):
349 """Form for creating and editing DeviceTypeMapping instances."""
351 class Meta:
352 model = DeviceTypeMapping
353 fields = [
354 "source_make",
355 "source_model",
356 "netbox_manufacturer_slug",
357 "netbox_device_type_slug",
358 ]
361class ColumnTransformRuleForm(forms.ModelForm):
362 """Form for creating and editing ColumnTransformRule instances."""
364 group_1_target = forms.ChoiceField(required=False)
365 group_2_target = forms.ChoiceField(required=False)
367 class Meta:
368 model = ColumnTransformRule
369 fields = ["source_column", "pattern", "group_1_target", "group_2_target"]
371 def __init__(self, *args, **kwargs):
372 super().__init__(*args, **kwargs)
373 # A capture group yields text, so the candidate targets are not offered.
374 output_kinds = _profile_output_kinds(self)
375 choices = CATALOG.choices(output_kinds=output_kinds, allow_candidates=False)
376 for name in ("group_1_target", "group_2_target"):
377 stored = getattr(self.instance, name, "")
378 preserved = _with_stored_target(choices, stored, output_kinds, allow_candidates=False)
379 self.fields[name].choices = [("", "---------"), *preserved]
382class InferenceBackendForm(NetBoxModelForm):
383 """Create or edit one AI backend. Model validation applies the api_root trust boundary."""
385 class Meta:
386 model = InferenceBackend
387 fields = (
388 "backend_key",
389 "display_name",
390 "adapter_type",
391 "api_root",
392 "model",
393 "authentication",
394 "response_mode",
395 "credential_reference",
396 "connect_timeout",
397 "read_timeout",
398 "max_concurrent_requests",
399 "enabled",
400 "tags",
401 )
402 help_texts = {
403 "credential_reference": _CREDENTIAL_REFERENCE_HELP,
404 "model": (
405 "Enter the exact model id. After you save the backend, run the connection test to discover "
406 "model suggestions when the endpoint supports them."
407 ),
408 }
411class InferenceBackendFilterForm(NetBoxModelFilterSetForm):
412 """Filters tab of the AI backend list."""
414 model = InferenceBackend
415 fieldsets = (
416 FieldSet("q", "filter_id", "tag"),
417 FieldSet("adapter_type", "response_mode", "enabled", name="Attributes"),
418 )
419 adapter_type = forms.ChoiceField(choices=add_blank_choice(InferenceBackend.ADAPTER_TYPES), required=False)
420 response_mode = forms.ChoiceField(choices=add_blank_choice(InferenceBackend.RESPONSE_MODES), required=False)
421 enabled = forms.NullBooleanField(required=False, widget=forms.Select(choices=BOOLEAN_WITH_BLANK_CHOICES))
422 tag = TagFilterField(model)
425class ImportSetupForm(forms.Form):
426 """Form for the import wizard step 1: select profile, upload file, choose site/location/tenant."""
428 MAX_UPLOAD_SIZE = 10 * 1024 * 1024 # 10 MB
430 # ImportProfile has no REST API endpoint yet, so use a plain select
431 profile = forms.ModelChoiceField(
432 queryset=ImportProfile.objects.all(),
433 label="Import Profile",
434 empty_label="— Select a profile —",
435 )
436 excel_file = forms.FileField(
437 label="Excel File",
438 help_text="Upload the Excel file to import (.xlsx, max 10 MB)",
439 )
440 site = DynamicModelChoiceField(
441 queryset=Site.objects.all(),
442 label="Target Site",
443 )
444 location = DynamicModelChoiceField(
445 queryset=Location.objects.all(),
446 label="Location (optional)",
447 required=False,
448 query_params={"site_id": "$site"},
449 )
450 tenant = DynamicModelChoiceField(
451 queryset=Tenant.objects.all(),
452 label="Tenant (optional)",
453 required=False,
454 )
456 def __init__(self, *args, user=None, **kwargs):
457 super().__init__(*args, **kwargs)
458 if user is None:
459 return
460 self.fields["profile"].queryset = ImportProfile.objects.restrict(user, "change")
461 self.fields["site"].queryset = Site.objects.restrict(user, "view")
462 self.fields["location"].queryset = Location.objects.restrict(user, "view")
463 self.fields["tenant"].queryset = Tenant.objects.restrict(user, "view")
465 def clean_profile(self):
466 """Reject a profile whose stored Source Adapter this release no longer registers."""
467 profile = self.cleaned_data["profile"]
468 validate_registered_adapter(profile)
469 return profile
471 def clean(self):
472 """Reject a Location outside the selected Site, which no import can target or rank with."""
473 cleaned = super().clean()
474 site, location = cleaned.get("site"), cleaned.get("location")
475 if site is not None and location is not None and location.site_id != site.pk:
476 self.add_error("location", "The selected location does not belong to the selected site.")
477 return cleaned
479 def clean_excel_file(self):
480 """Reject files that exceed the maximum upload size."""
481 f = self.cleaned_data.get("excel_file")
482 if f and f.size > self.MAX_UPLOAD_SIZE:
483 limit_mb = self.MAX_UPLOAD_SIZE / (1024 * 1024)
484 raise forms.ValidationError(
485 f"File too large: {f.size / (1024 * 1024):.1f} MB. Maximum allowed is {limit_mb:.0f} MB."
486 )
487 return f