Coverage for netbox_data_import/profile_yaml.py: 97.99%
249 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# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com>
3"""Serialize and apply portable Import Profile YAML documents."""
5from dataclasses import dataclass
6from typing import Any
8import yaml
9from django.core.exceptions import ValidationError
10from django.db import transaction
11from yaml.constructor import ConstructorError
12from yaml.resolver import BaseResolver
14from . import branching
15from .catalog import POLICY_SECTIONS, policy_section
16from .models import (
17 CableClassMapping,
18 PortClassMapping,
19 ClassRoleMapping,
20 ColumnMapping,
21 ColumnTransformRule,
22 DeviceTypeMapping,
23 ImportProfile,
24 ManufacturerMapping,
25 locked_profile_policy,
26)
27from .object_permissions import (
28 ObjectPermissionDenied,
29 delete_permission_scoped_objects,
30 enforce_saved_object_permission,
31 save_permission_scoped_object,
32)
35@dataclass(frozen=True)
36class NaturalKeyField:
37 """Describe one related field represented by a stable lookup value."""
39 name: str
40 lookup: str
43@dataclass(frozen=True)
44class PolicyDocumentSchema:
45 """Describe one catalog policy section's portable document shape."""
47 model: Any
48 identity_fields: tuple[str, ...]
49 required_fields: tuple[str, ...]
50 fields: tuple[str, ...]
51 natural_keys: tuple[NaturalKeyField, ...] = ()
52 release_changed_before_write: bool = False
54 @property
55 def key(self) -> str:
56 """Return the catalog key owned by the model."""
57 return self.model.POLICY_SECTION
59 @property
60 def natural_key_map(self) -> dict[str, NaturalKeyField]:
61 """Return natural-key declarations by model field name."""
62 return {field.name: field for field in self.natural_keys}
65_POLICY_DOCUMENT_SCHEMAS = (
66 PolicyDocumentSchema(
67 model=ColumnMapping,
68 identity_fields=("source_column", "target_field"),
69 required_fields=("source_column", "target_field"),
70 fields=("source_column", "target_field"),
71 release_changed_before_write=True,
72 ),
73 PolicyDocumentSchema(
74 model=ClassRoleMapping,
75 identity_fields=("source_class",),
76 required_fields=("source_class",),
77 fields=("source_class", "creates_rack", "rack_type", "role_slug", "ignore"),
78 natural_keys=(NaturalKeyField("rack_type", "slug"),),
79 ),
80 PolicyDocumentSchema(
81 model=DeviceTypeMapping,
82 identity_fields=("source_make", "source_model"),
83 required_fields=(
84 "source_make",
85 "source_model",
86 "netbox_manufacturer_slug",
87 "netbox_device_type_slug",
88 ),
89 fields=(
90 "source_make",
91 "source_model",
92 "netbox_manufacturer_slug",
93 "netbox_device_type_slug",
94 ),
95 ),
96 PolicyDocumentSchema(
97 model=ManufacturerMapping,
98 identity_fields=("source_make",),
99 required_fields=("source_make", "netbox_manufacturer_slug"),
100 fields=("source_make", "netbox_manufacturer_slug"),
101 ),
102 PolicyDocumentSchema(
103 model=ColumnTransformRule,
104 identity_fields=("source_column",),
105 required_fields=("source_column", "pattern"),
106 fields=("source_column", "pattern", "group_1_target", "group_2_target"),
107 release_changed_before_write=True,
108 ),
109 PolicyDocumentSchema(
110 model=CableClassMapping,
111 identity_fields=("cable_class",),
112 required_fields=("cable_class",),
113 fields=(
114 "cable_class",
115 "cable_type_resolved",
116 "cable_type",
117 "cable_profile_resolved",
118 "cable_profile",
119 ),
120 ),
121 PolicyDocumentSchema(
122 model=PortClassMapping,
123 identity_fields=("source_label",),
124 required_fields=("source_label", "port_class"),
125 fields=("source_label", "port_class"),
126 ),
127)
129_SCHEMAS_BY_KEY = {schema.key: schema for schema in _POLICY_DOCUMENT_SCHEMAS}
130_PROFILE_FIELDS = ("description", "source_adapter")
133class ProfileDocumentInvalid(Exception):
134 """A profile YAML document breaks the document schema, and the message names the part that does."""
137class DuplicateYamlKeyError(ConstructorError):
138 """A YAML mapping repeats a key whose first value would otherwise be discarded."""
141class _UniqueKeySafeLoader(yaml.SafeLoader):
142 """Load the safe YAML subset and reject duplicate mapping keys at every depth."""
145def _construct_unique_mapping(loader, node, deep=False):
146 """Construct one mapping without PyYAML's last-value-wins behavior."""
147 loader.flatten_mapping(node)
148 mapping = {}
149 for key_node, value_node in node.value:
150 key = loader.construct_object(key_node, deep=deep)
151 try:
152 duplicate = key in mapping
153 except TypeError as exc:
154 raise ConstructorError(
155 "while constructing a mapping",
156 node.start_mark,
157 "found an unhashable mapping key",
158 key_node.start_mark,
159 ) from exc
160 if duplicate:
161 raise DuplicateYamlKeyError(
162 "while constructing a mapping",
163 node.start_mark,
164 f"found duplicate mapping key {key!r}",
165 key_node.start_mark,
166 )
167 mapping[key] = loader.construct_object(value_node, deep=deep)
168 return mapping
171_UniqueKeySafeLoader.add_constructor(BaseResolver.DEFAULT_MAPPING_TAG, _construct_unique_mapping)
174def load_yaml_document(stream) -> Any:
175 """Load untrusted YAML through the shared duplicate-key-rejecting safe loader."""
176 loader = _UniqueKeySafeLoader(stream)
177 try:
178 return loader.get_single_data()
179 finally:
180 loader.dispose()
183def serialize_profile(profile: ImportProfile, actor=None) -> dict[str, Any]:
184 """Return one portable profile document with only applicable policy sections."""
185 document: dict[str, Any] = {
186 "profile": {
187 "name": profile.name,
188 "description": profile.description,
189 "source_adapter": profile.source_adapter,
190 "adapter_config": profile.adapter_config,
191 }
192 }
193 for section in POLICY_SECTIONS:
194 schema = _SCHEMAS_BY_KEY.get(section.key)
195 if schema is None or not section.applies_to(profile.output_kinds):
196 continue
197 relation = schema.model._meta.get_field("profile").remote_field.get_accessor_name()
198 rows = getattr(profile, relation).all()
199 if actor is not None:
200 visible = schema.model.objects.restrict(actor, "view").filter(profile=profile)
201 if rows.exclude(pk__in=visible).exists():
202 raise ObjectPermissionDenied(f"{schema.model._meta.app_label}.view_{schema.model._meta.model_name}")
203 rows = visible
204 if schema.natural_keys:
205 rows = rows.select_related(*(field.name for field in schema.natural_keys))
206 if actor is not None:
207 _enforce_natural_key_view_permissions(schema, rows, actor)
208 document[section.key] = [_serialize_policy_row(schema, row) for row in rows]
209 return document
212def _enforce_natural_key_view_permissions(schema: PolicyDocumentSchema, rows, actor) -> None:
213 """Reject export when a natural-key reference names an object the actor cannot view."""
214 for natural_key in schema.natural_keys:
215 relation = schema.model._meta.get_field(natural_key.name)
216 related_model = relation.remote_field.model
217 referenced_ids = set(
218 rows.exclude(**{f"{natural_key.name}__isnull": True}).values_list(
219 f"{natural_key.name}__pk",
220 flat=True,
221 )
222 )
223 visible_ids = set(
224 related_model.objects.restrict(actor, "view").filter(pk__in=referenced_ids).values_list("pk", flat=True)
225 )
226 if referenced_ids - visible_ids:
227 raise ObjectPermissionDenied(f"{related_model._meta.app_label}.view_{related_model._meta.model_name}")
230def apply_profile_document(data: Any, actor=None) -> tuple[ImportProfile, dict[str, int]]:
231 """Create or update one profile and reconcile each supplied policy section."""
232 branching.refuse_branch()
233 profile_data, section_rows = _validate_document_shape(data)
234 with transaction.atomic():
235 profile_values = _profile_values(profile_data)
236 profile = ImportProfile.objects.filter(name=profile_data["name"]).first()
237 created = False
238 if profile is None:
239 profile = ImportProfile(name=profile_data["name"])
240 for field, value in profile_values.items():
241 setattr(profile, field, value)
242 _validate_instance(profile, "profile")
243 result = save_permission_scoped_object(
244 actor,
245 ImportProfile,
246 {"name": profile_data["name"]},
247 profile_values,
248 )
249 profile = result.instance
250 created = result.created
251 elif actor is not None and not ImportProfile.objects.restrict(actor, "change").filter(pk=profile.pk).exists():
252 raise ObjectPermissionDenied("netbox_data_import.change_importprofile")
254 with locked_profile_policy(profile.pk):
255 profile = ImportProfile.objects.get(pk=profile.pk)
256 if not created:
257 for field, value in profile_values.items():
258 setattr(profile, field, value)
259 _validate_instance(profile, "profile")
260 profile = save_permission_scoped_object(
261 actor,
262 ImportProfile,
263 {"pk": profile.pk},
264 profile_values,
265 ).instance
267 _validate_section_applicability(profile, section_rows)
268 prepared_rows = {}
269 for key, rows in section_rows.items():
270 schema = _SCHEMAS_BY_KEY[key]
271 prepared = [_prepare_policy_row(schema, row, index, actor) for index, row in enumerate(rows, 1)]
272 _validate_distinct_policy_identities(schema, prepared)
273 prepared_rows[key] = prepared
274 released_updates = {}
275 for key, rows in prepared_rows.items():
276 schema = _SCHEMAS_BY_KEY[key]
277 if schema.release_changed_before_write:
278 released_updates[key] = _release_changed_rows(profile, schema, rows, actor)
280 stats = {}
281 for section in POLICY_SECTIONS:
282 if section.key not in prepared_rows:
283 continue
284 schema = _SCHEMAS_BY_KEY[section.key]
285 stats[section.key] = _reconcile_policy_rows(
286 profile,
287 schema,
288 prepared_rows[section.key],
289 actor,
290 released_updates.get(section.key, {}),
291 )
292 # atomic-exit-safe: profile-import-committed
293 return profile, stats
296def _serialize_policy_row(schema: PolicyDocumentSchema, row) -> dict[str, Any]:
297 """Return one policy row using stable related-object references."""
298 natural_keys = schema.natural_key_map
299 serialized = {}
300 for name in schema.fields:
301 value = getattr(row, name)
302 if name in natural_keys:
303 value = None if value is None else getattr(value, natural_keys[name].lookup)
304 serialized[name] = value
305 return serialized
308def _validate_document_shape(data: Any) -> tuple[dict[str, Any], dict[str, list[dict[str, Any]]]]:
309 """Validate the document container and return its profile and section mappings."""
310 if not isinstance(data, dict) or "profile" not in data:
311 raise ProfileDocumentInvalid("YAML must contain a top-level 'profile' key.")
312 _require_string_keys(data, "Top-level")
313 profile_data = data["profile"]
314 if not isinstance(profile_data, dict):
315 raise ProfileDocumentInvalid("The 'profile' value must be a mapping (dict), not a scalar or list.")
316 _require_string_keys(profile_data, "'profile'")
317 if not profile_data.get("name"):
318 raise ProfileDocumentInvalid("Profile YAML must include a 'name' field.")
320 unknown = sorted(set(data) - {"profile", *_SCHEMAS_BY_KEY})
321 if unknown:
322 raise ProfileDocumentInvalid(f"Unknown profile YAML section(s): {', '.join(unknown)}")
324 sections = {}
325 for key in data:
326 if key == "profile":
327 continue
328 value = data[key]
329 if not isinstance(value, list):
330 raise ProfileDocumentInvalid(
331 f"'{key}' must be a list of mappings; use [] to explicitly remove all entries, "
332 f"got {type(value).__name__}."
333 )
334 for index, row in enumerate(value, 1):
335 if not isinstance(row, dict):
336 raise ProfileDocumentInvalid(f"'{key}[{index}]' must be a mapping, got {type(row).__name__}.")
337 _require_string_keys(row, f"'{key}[{index}]'")
338 sections[key] = value
339 return profile_data, sections
342def _require_string_keys(mapping: dict, label: str) -> None:
343 """Reject a key that is not a string: YAML also reads a number, a boolean or a date as a key."""
344 for key in mapping:
345 if not isinstance(key, str):
346 raise ProfileDocumentInvalid(f"{label} keys must be strings, got {key!r}.")
349def _profile_values(profile_data: dict[str, Any]) -> dict[str, Any]:
350 """Return validated scalar profile values and adapter configuration."""
351 accepted = {"name", "adapter_config", *_PROFILE_FIELDS}
352 unknown = sorted(set(profile_data) - accepted)
353 if unknown:
354 raise ProfileDocumentInvalid(f"Unknown profile key(s): {', '.join(unknown)}")
355 values = {field: profile_data[field] for field in _PROFILE_FIELDS if field in profile_data}
356 if "adapter_config" in profile_data:
357 values["adapter_config"] = profile_data["adapter_config"]
358 return values
361def _validate_section_applicability(profile: ImportProfile, sections: dict[str, list]) -> None:
362 """Reject each supplied section that the selected Source Adapter cannot use."""
363 for key in sections:
364 section = policy_section(key)
365 if section is None or not section.applies_to(profile.output_kinds):
366 raise ProfileDocumentInvalid(
367 f"Policy section '{key}' does not apply to source adapter '{profile.source_adapter}'."
368 )
371def _prepare_policy_row(schema: PolicyDocumentSchema, row: dict[str, Any], index: int, actor=None) -> dict[str, Any]:
372 """Validate one row's keys and resolve its stable related-object references."""
373 missing = [field for field in schema.required_fields if field not in row]
374 if missing:
375 raise ProfileDocumentInvalid(f"'{schema.key}[{index}]' missing required key(s): {', '.join(missing)}")
376 unknown = sorted(set(row) - set(schema.fields))
377 if unknown:
378 raise ProfileDocumentInvalid(f"'{schema.key}[{index}]' has unknown key(s): {', '.join(unknown)}")
379 # An identity or a lookup compares one value, so a list or a mapping there cannot name a row.
380 for name in (*schema.identity_fields, *(natural_key.name for natural_key in schema.natural_keys)):
381 if isinstance(row.get(name), (dict, list)):
382 raise ProfileDocumentInvalid(f"'{schema.key}[{index}]' key '{name}' must be a single value.")
384 prepared = dict(row)
385 for natural_key in schema.natural_keys:
386 if natural_key.name not in prepared or prepared[natural_key.name] is None:
387 continue
388 model_field = schema.model._meta.get_field(natural_key.name)
389 related_model = model_field.remote_field.model
390 visible = related_model.objects if actor is None else related_model.objects.restrict(actor, "view")
391 value = prepared[natural_key.name]
392 try:
393 prepared[natural_key.name] = visible.get(**{natural_key.lookup: value})
394 except related_model.DoesNotExist as exc:
395 raise ProfileDocumentInvalid(
396 f"{schema.key}[{index}]: {related_model._meta.verbose_name.title()} with "
397 f"{natural_key.lookup} '{value}' not found"
398 ) from exc
399 return prepared
402def _validate_distinct_policy_identities(schema: PolicyDocumentSchema, rows: list[dict[str, Any]]) -> None:
403 """Reject duplicate natural identities before any policy write."""
404 seen = set()
405 for row in rows:
406 identity = tuple(row[name] for name in schema.identity_fields)
407 if identity in seen:
408 display = "/".join(str(value) for value in identity)
409 raise ProfileDocumentInvalid(f"Duplicate {schema.key} identity: {display}")
410 seen.add(identity)
413def _release_changed_rows(
414 profile: ImportProfile,
415 schema: PolicyDocumentSchema,
416 rows: list[dict[str, Any]],
417 actor,
418) -> dict[tuple[Any, ...], Any]:
419 """Release target ownership while preserving logical update permissions."""
420 desired = {tuple(row[name] for name in schema.identity_fields): row for row in rows}
421 deleted_ids = []
422 released_updates = {}
423 for stored in schema.model.objects.filter(profile=profile):
424 key = tuple(getattr(stored, name) for name in schema.identity_fields)
425 row = desired.get(key)
426 if row is None:
427 deleted_ids.append(stored.pk)
428 elif any(getattr(stored, name) != value for name, value in row.items()):
429 enforce_saved_object_permission(stored, actor, "change")
430 released_updates[key] = stored
431 delete_permission_scoped_objects(actor, schema.model.objects.filter(pk__in=deleted_ids))
432 schema.model.objects.filter(pk__in=[row.pk for row in released_updates.values()]).delete()
433 return released_updates
436def _reconcile_policy_rows(
437 profile: ImportProfile,
438 schema: PolicyDocumentSchema,
439 rows: list[dict[str, Any]],
440 actor,
441 released_updates: dict[tuple[Any, ...], Any],
442) -> int:
443 """Create or update supplied rows and remove rows absent from the section."""
444 retained_ids = []
445 for row in rows:
446 identity = tuple(row[name] for name in schema.identity_fields)
447 lookup = {"profile": profile, **dict(zip(schema.identity_fields, identity, strict=True))}
448 instance = released_updates.get(identity) or schema.model.objects.filter(**lookup).first()
449 if instance is None:
450 instance = schema.model(**lookup)
451 for name, value in row.items():
452 setattr(instance, name, value)
453 _validate_instance(instance, _policy_row_label(schema, row))
454 values = {name: getattr(instance, name) for name in row if name not in schema.identity_fields}
455 if identity in released_updates:
456 instance.save(force_insert=True)
457 enforce_saved_object_permission(instance, actor, "change")
458 retained_ids.append(instance.pk)
459 else:
460 result = save_permission_scoped_object(actor, schema.model, lookup, values)
461 retained_ids.append(result.instance.pk)
462 delete_permission_scoped_objects(
463 actor,
464 schema.model.objects.filter(profile=profile).exclude(pk__in=retained_ids),
465 )
466 return len(rows)
469def _policy_row_label(schema: PolicyDocumentSchema, row: dict[str, Any]) -> str:
470 """Return one stable validation label from the row identity."""
471 identity = "/".join(str(row[field]) for field in schema.identity_fields)
472 return f"{schema.key}[{identity}]"
475def _validate_instance(instance, label: str) -> None:
476 """Validate a model and expose its field errors through the document interface."""
477 try:
478 instance.full_clean(validate_unique=False)
479 except ValidationError as exc:
480 if hasattr(exc, "message_dict"):
481 message = "; ".join(f"{field}: {', '.join(errors)}" for field, errors in exc.message_dict.items())
482 else:
483 message = "; ".join(exc.messages)
484 raise ProfileDocumentInvalid(f"Validation error in {label}: {message}") from exc
487__all__ = (
488 "DuplicateYamlKeyError",
489 "ProfileDocumentInvalid",
490 "apply_profile_document",
491 "load_yaml_document",
492 "serialize_profile",
493)