Coverage for netbox_data_import/forms.py: 97%

1# SPDX-License-Identifier: Apache-2.0

3import copy

5from django import forms

6from dcim.models import Site, Location

7from tenancy.models import Tenant

8from netbox.forms import NetBoxModelBulkEditForm, NetBoxModelForm, NetBoxModelImportForm

9from utilities.forms.fields import DynamicModelChoiceField

10from .adapters import get_adapter, selectable_adapter_choices

11from .catalog import CATALOG

12from .models import (

13 CableClassMapping,

14 ImportProfile,

15 InferenceBackend,

16 ColumnMapping,

17 ClassRoleMapping,

18 DeviceTypeMapping,

19 ColumnTransformRule,

20 _require_adapter_config_mapping,

21 validate_registered_adapter,

22 cable_type_choices,

23 compatible_cable_profile_choices,

24)

26_EXPLICIT_NONE = "__explicit_none__"

29class _RuntimeCableChoiceField(forms.ChoiceField):

30 """Render only current choices while letting shared validation classify submitted values."""

32 def validate(self, value):

33 """Apply required-field validation without duplicating the runtime choice rule."""

34 forms.Field.validate(self, value)

37def _decision_choices(runtime_choices):

38 """Add unresolved and explicit-none form states to current NetBox choices."""

39 if _EXPLICIT_NONE in {value for value, _label in runtime_choices}:

40 raise RuntimeError("A NetBox Cable choice conflicts with the form's explicit-none control value.")

41 return [("", "Unresolved"), (_EXPLICIT_NONE, "Explicitly none"), *runtime_choices]

44def _with_stored_decision(choices, resolved, value):

45 """Keep a stored decision selectable after NetBox stops offering it.

47 A select cannot send back a value it does not list, so the browser would submit the first

48 option and `clean()` would record the loss as an unresolved decision without an error.

49 """

50 if not resolved or value is None or value in {key for key, _label in choices}:

51 return choices

52 return [*choices, (value, f"{value} (no longer offered)")]

55def _decision_initial(resolved, value):

56 """Return the form value for one stored tri-state decision."""

57 if not resolved:

58 return ""

59 return _EXPLICIT_NONE if value is None else value

62def _decode_decision(value):

63 """Return the resolved flag and nullable stored value for one form selection."""

64 if value in (None, ""):

65 return False, None

66 if value == _EXPLICIT_NONE:

67 return True, None

68 return True, value

71def _profile_output_kinds(form):

72 """Return the output kinds of the profile this row belongs to, or None when unknown."""

73 profile = form.instance.profile_id or form.initial.get("profile")

74 if not profile:

75 return None

76 if not isinstance(profile, ImportProfile):

77 # NetBox seeds form initial from the query string, so a non-numeric value must not reach the query.

78 try:

79 profile = ImportProfile.objects.filter(pk=int(profile)).first()

80 except (TypeError, ValueError):

81 return None

82 if profile is None:

83 return None

84 return profile.output_kinds

87def _with_stored_target(choices, stored, output_kinds, *, allow_candidates=True):

88 """Keep a stored key-family target selectable, so an existing row can be re-saved.

90 CATALOG.choices lists fixed keys only, so a stored family key such as `extra_json:asset_id` is

91 never among them. Re-offer it only when the model would still accept it: the row's clean()

92 runs the same check, and offering more would put a choice in the list that saving rejects.

93 """

94 if not stored or stored in {key for key, _label in choices}:

95 return choices

96 if not CATALOG.is_valid(stored, output_kinds=output_kinds, allow_candidates=allow_candidates):

97 return choices

98 return [*choices, (stored, CATALOG.display(stored))]

100

101class ImportProfileForm(NetBoxModelForm):

102 """Form for creating and editing ImportProfile instances.

103

104 The Source Adapter is asked for first and is disabled after creation. The selected adapter

105 declares the remaining configuration fields.

106 """

107

108 class Meta:

109 model = ImportProfile

110 fields = ["name", "description", "source_adapter", "tags"]

111

112 def __init__(self, *args, **kwargs):

113 super().__init__(*args, **kwargs)

114 self._config_form_class = None

115 stored = _require_adapter_config_mapping(self.instance.adapter_config)

116 adapter = get_adapter(self._selected_adapter_key())

117 if adapter is None:

118 return

119 if self.instance.pk:

120 self.fields["source_adapter"].disabled = True

121 else:

122 self.fields["source_adapter"].choices = selectable_adapter_choices()

123 self._config_form_class = adapter.config_form_class()

124 for name, field in self._config_form_class.base_fields.items():

125 self.fields[name] = copy.deepcopy(field)

126 if name in stored:

127 self.initial.setdefault(name, stored[name])

128

129 def _selected_adapter_key(self):

130 """Return the adapter key this form edits: the stored one, or the submitted choice."""

131 if self.instance.pk:

132 return self.instance.source_adapter

133 if self.is_bound:

134 return self.data.get(self.add_prefix("source_adapter")) or self.instance.source_adapter

135 return self.instance.source_adapter

136

137 def clean(self):

138 """Collect the adapter-declared fields into ``adapter_config``."""

139 cleaned = super().clean()

140 if cleaned is None:

141 cleaned = self.cleaned_data

142 if self._config_form_class is not None:

143 self.instance.adapter_config = self._config_form_class.normalize(cleaned)

144 return cleaned

145

146

147class ImportProfileImportForm(NetBoxModelImportForm):

148 """CSV/YAML bulk-import form for ImportProfile objects (profile metadata only).

149

150 Adapter configuration is nested, so it is set through the edit UI, the REST API, or the

151 hierarchical profile YAML import.

152 """

153

154 class Meta:

155 model = ImportProfile

156 fields = ["name", "description", "source_adapter", "tags"]

157

158

159class ImportProfileBulkEditForm(NetBoxModelBulkEditForm):

160 """Bulk-edit fields that apply safely across import profiles.

161

162 The Source Adapter is immutable and ``adapter_config`` is adapter-scoped, so neither is

163 bulk-editable.

164 """

165

166 model = ImportProfile

167

168 description = forms.CharField(required=False, widget=forms.Textarea(attrs={"rows": 3}))

169

170 nullable_fields = ("description",)

171

172

173class ColumnMappingForm(forms.ModelForm):

174 """Form for creating and editing ColumnMapping instances."""

175

176 target_field = forms.ChoiceField(choices=CATALOG.choices)

177

178 class Meta:

179 model = ColumnMapping

180 fields = ["source_column", "target_field"]

181

182 def __init__(self, *args, **kwargs):

183 super().__init__(*args, **kwargs)

184 output_kinds = _profile_output_kinds(self)

185 choices = CATALOG.choices(output_kinds=output_kinds)

186 self.fields["target_field"].choices = _with_stored_target(choices, self.instance.target_field, output_kinds)

187

188

189class ClassRoleMappingForm(forms.ModelForm):

190 """Form for creating and editing ClassRoleMapping instances."""

191

192 class Meta:

193 model = ClassRoleMapping

194 fields = ["source_class", "creates_rack", "rack_type", "role_slug", "ignore"]

195

196 def clean(self):

197 """Require role_slug unless creates_rack or ignore is set."""

198 cleaned = super().clean()

199 creates_rack = cleaned.get("creates_rack")

200 ignore = cleaned.get("ignore")

201 role_slug = (cleaned.get("role_slug") or "").strip()

202 if not creates_rack and not ignore and not role_slug:

203 self.add_error(

204 "role_slug",

205 "A device role slug is required unless 'creates rack' or 'ignore' is checked.",

206 )

207 return cleaned

208

209

210class CableClassMappingForm(forms.ModelForm):

211 """Form for one source CableClass and its two independent target decisions."""

212

213 cable_type = _RuntimeCableChoiceField(required=False)

214 cable_profile = _RuntimeCableChoiceField(required=False)

215

216 class Meta:

217 model = CableClassMapping

218 fields = ["cable_class", "cable_type", "cable_profile"]

219

220 def __init__(self, *args, **kwargs):

221 super().__init__(*args, **kwargs)

222 self.fields["cable_type"].choices = _with_stored_decision(

223 _decision_choices(cable_type_choices()),

224 self.instance.cable_type_resolved,

225 self.instance.cable_type,

226 )

227 self.fields["cable_profile"].choices = _with_stored_decision(

228 _decision_choices(compatible_cable_profile_choices()),

229 self.instance.cable_profile_resolved,

230 self.instance.cable_profile,

231 )

232 self.initial["cable_type"] = _decision_initial(

233 self.instance.cable_type_resolved,

234 self.instance.cable_type,

235 )

236 self.initial["cable_profile"] = _decision_initial(

237 self.instance.cable_profile_resolved,

238 self.instance.cable_profile,

239 )

240

241 def clean(self):

242 """Decode each tri-state value and apply the shared runtime-choice validation."""

243 cleaned = super().clean()

244 type_resolved, cable_type = _decode_decision(cleaned.get("cable_type"))

245 profile_resolved, cable_profile = _decode_decision(cleaned.get("cable_profile"))

246 self.instance.cable_type_resolved = type_resolved

247 self.instance.cable_type = cable_type

248 self.instance.cable_profile_resolved = profile_resolved

249 self.instance.cable_profile = cable_profile

250 cleaned["cable_type"] = cable_type

251 cleaned["cable_profile"] = cable_profile

252 return cleaned

253

254

255class DeviceTypeMappingForm(forms.ModelForm):

256 """Form for creating and editing DeviceTypeMapping instances."""

257

258 class Meta:

259 model = DeviceTypeMapping

260 fields = [

261 "source_make",

262 "source_model",

263 "netbox_manufacturer_slug",

264 "netbox_device_type_slug",

265 ]

266

267

268class ColumnTransformRuleForm(forms.ModelForm):

269 """Form for creating and editing ColumnTransformRule instances."""

270

271 group_1_target = forms.ChoiceField(required=False)

272 group_2_target = forms.ChoiceField(required=False)

273

274 class Meta:

275 model = ColumnTransformRule

276 fields = ["source_column", "pattern", "group_1_target", "group_2_target"]

277

278 def __init__(self, *args, **kwargs):

279 super().__init__(*args, **kwargs)

280 # A capture group yields text, so the candidate targets are not offered.

281 output_kinds = _profile_output_kinds(self)

282 choices = CATALOG.choices(output_kinds=output_kinds, allow_candidates=False)

283 for name in ("group_1_target", "group_2_target"):

284 stored = getattr(self.instance, name, "")

285 preserved = _with_stored_target(choices, stored, output_kinds, allow_candidates=False)

286 self.fields[name].choices = [("", "---------"), *preserved]

287

288

289class InferenceBackendForm(NetBoxModelForm):

290 """Create or edit one AI backend. Model validation applies the api_root trust boundary."""

291

292 class Meta:

293 model = InferenceBackend

294 fields = (

295 "backend_key",

296 "display_name",

297 "adapter_type",

298 "api_root",

299 "model",

300 "authentication",

301 "response_mode",

302 "credential_reference",

303 "connect_timeout",

304 "read_timeout",

305 "enabled",

306 "tags",

307 )

308 help_texts = {

309 "credential_reference": (

310 "A typed Vault KV v2 reference: backend, mount, path and field. It never holds a key value."

311 ),

312 }

313

314

315class ImportSetupForm(forms.Form):

316 """Form for the import wizard step 1: select profile, upload file, choose site/location/tenant."""

317

318 MAX_UPLOAD_SIZE = 10 * 1024 * 1024 # 10 MB

319

320 # ImportProfile has no REST API endpoint yet, so use a plain select

321 profile = forms.ModelChoiceField(

322 queryset=ImportProfile.objects.all(),

323 label="Import Profile",

324 empty_label="— Select a profile —",

325 )

326 excel_file = forms.FileField(

327 label="Excel File",

328 help_text="Upload the Excel file to import (.xlsx, max 10 MB)",

329 )

330 site = DynamicModelChoiceField(

331 queryset=Site.objects.all(),

332 label="Target Site",

333 )

334 location = DynamicModelChoiceField(

335 queryset=Location.objects.all(),

336 label="Location (optional)",

337 required=False,

338 query_params={"site_id": "$site"},

339 )

340 tenant = DynamicModelChoiceField(

341 queryset=Tenant.objects.all(),

342 label="Tenant (optional)",

343 required=False,

344 )

345

346 def __init__(self, *args, user=None, **kwargs):

347 super().__init__(*args, **kwargs)

348 if user is None:

349 return

350 self.fields["profile"].queryset = ImportProfile.objects.restrict(user, "change")

351 self.fields["site"].queryset = Site.objects.restrict(user, "view")

352 self.fields["location"].queryset = Location.objects.restrict(user, "view")

353 self.fields["tenant"].queryset = Tenant.objects.restrict(user, "view")

354

355 def clean_profile(self):

356 """Reject a profile whose stored Source Adapter this release no longer registers."""

357 profile = self.cleaned_data["profile"]

358 validate_registered_adapter(profile)

359 return profile

360

361 def clean_excel_file(self):

362 """Reject files that exceed the maximum upload size."""

363 f = self.cleaned_data.get("excel_file")

364 if f and f.size > self.MAX_UPLOAD_SIZE:

365 limit_mb = self.MAX_UPLOAD_SIZE / (1024 * 1024)

366 raise forms.ValidationError(

367 f"File too large: {f.size / (1024 * 1024):.1f} MB. Maximum allowed is {limit_mb:.0f} MB."

368 )

369 return f