Coverage for netbox_data_import/plan.py: 100%
256 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-09 20:50 +0000
1# SPDX-License-Identifier: Apache-2.0
2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com>
3"""The target-neutral Import Plan.
5An Import Plan is a serializable derived artifact: Synchronization Units holding typed Planned
6Changes, their dispositions, and structured diagnostics. It is the contract between the Import
7Engine, the Review Workspace, the session, and a background job payload.
9This module holds no NetBox import. Every value that enters a plan passes through a canonical JSON
10round trip, so a live ORM object, a queryset, a callable, or a template fragment cannot reach one.
11"""
13from __future__ import annotations
15import hashlib
16import json
17import re
18from collections import Counter
19from collections.abc import Mapping
20from dataclasses import dataclass, field
21from typing import Any
23SCHEMA_VERSION = 2
25_DIAGNOSTIC_CODE = re.compile(r"^[a-z0-9]+(?:_[a-z0-9]+)*\.[a-z0-9]+(?:_[a-z0-9]+)*$")
28class PlanError(Exception):
29 """Base class for the Import Plan failures the coordinator reports."""
32class PlanInvalid(PlanError):
33 """A plan violates a structural invariant and cannot execute."""
36class PlanSchemaMismatch(PlanError):
37 """A serialized plan states a schema version this release does not execute."""
40class Disposition:
41 """The exactly-one state a Synchronization Unit carries (section 4.2)."""
43 ACTIONABLE = "actionable"
44 NO_OP = "no-op"
45 BLOCKED = "blocked"
46 INVALID = "invalid"
47 EXCLUDED = "excluded"
49 ALL = frozenset({ACTIONABLE, NO_OP, BLOCKED, INVALID, EXCLUDED})
52class Severity:
53 """Diagnostic severity (section 4.2)."""
55 INFO = "info"
56 WARNING = "warning"
57 ERROR = "error"
59 ALL = frozenset({INFO, WARNING, ERROR})
62def canonical_json(value: Any) -> str:
63 """Return the canonical serialization used for every fingerprint.
65 ``allow_nan`` stays off: NaN and Infinity are not JSON, PostgreSQL rejects them in a JSONField,
66 and NaN never equals itself, which would make two identical shared changes look conflicting.
67 """
68 return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False)
71def fingerprint_of(value: Any) -> str:
72 """Return the SHA-256 digest of the canonical serialization of *value*."""
73 return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()
76class FrozenDict(dict):
77 """The JSON object a plan hands out: JSON data, closed to mutation.
79 A frozen plan value stays JSON data, because a Target Module writes one into a JSONField.
80 """
82 __slots__ = ()
84 def _immutable(self, *args, **kwargs):
85 """Refuse one mutation, so a shared plan value cannot change under its other readers."""
86 raise TypeError("An Import Plan value cannot be changed.")
88 __setitem__ = _immutable
89 __delitem__ = _immutable
90 __ior__ = _immutable
91 clear = _immutable
92 pop = _immutable
93 popitem = _immutable
94 setdefault = _immutable
95 update = _immutable
98def _freeze_json(value: Any) -> Any:
99 """Return recursively immutable JSON data."""
100 if isinstance(value, dict):
101 return FrozenDict({key: _freeze_json(item) for key, item in value.items()})
102 if isinstance(value, list):
103 return tuple(_freeze_json(item) for item in value)
104 return value
107def _thaw_json(value: Any) -> Any:
108 """Return detached plain JSON data from an immutable plan value."""
109 if isinstance(value, Mapping):
110 return {key: _thaw_json(item) for key, item in value.items()}
111 if isinstance(value, tuple):
112 return [_thaw_json(item) for item in value]
113 return value
116def _frozen_json(value: Any, label: str) -> Any:
117 """Validate, detach, and recursively freeze JSON data entering a plan."""
118 try:
119 return _freeze_json(json.loads(canonical_json(value)))
120 except (TypeError, ValueError) as exc:
121 raise PlanInvalid(f"{label} must be JSON-serializable plan data: {exc}") from exc
124def _plan_mapping(value: Any, label: str) -> Mapping[str, Any]:
125 """Return *value* as the immutable JSON object the field declares."""
126 frozen = _frozen_json(value, label)
127 if not isinstance(frozen, Mapping):
128 raise PlanInvalid(f"{label} must be a JSON object, not {type(value).__name__}.")
129 return frozen
132def _plan_text(value, label: str) -> str:
133 """Return *value* as the string the field declares."""
134 if not isinstance(value, str):
135 raise PlanInvalid(f"{label} must be a string, not {type(value).__name__}.")
136 return value
139def _plan_int(value, label: str) -> int:
140 """Return *value* as the integer the field declares.
142 A bool is an int in Python, and `from_dict` refuses one, so a plan built from a bool could not
143 survive its own serialization.
144 """
145 if isinstance(value, bool) or not isinstance(value, int):
146 raise PlanInvalid(f"{label} must be an integer, not {type(value).__name__}.")
147 return value
150def _elements(value, kind, label: str) -> tuple:
151 """Return *value* as a tuple, rejecting any element that is not an instance of *kind*."""
152 if isinstance(value, str) or not isinstance(value, (list, tuple)):
153 raise PlanInvalid(f"{label} must be a list or tuple of {kind.__name__} objects.")
154 for item in value:
155 if not isinstance(item, kind):
156 raise PlanInvalid(f"{label} must hold {kind.__name__} objects, not {type(item).__name__}.")
157 return tuple(value)
160def _identities(value, label: str) -> tuple[str, ...]:
161 """Return *value* as a tuple of identity strings.
163 A bare string is refused because tuple("rack:1") would silently become six identities.
164 """
165 if isinstance(value, str) or not isinstance(value, (list, tuple)):
166 raise PlanInvalid(f"{label} must be a list or tuple of identity strings.")
167 for item in value:
168 if not isinstance(item, str):
169 raise PlanInvalid(f"{label} must hold identity strings, not {type(item).__name__}.")
170 return tuple(value)
173@dataclass(frozen=True)
174class Diagnostic:
175 """One structured finding attached to a plan or a unit."""
177 code: str
178 severity: str
179 identities: tuple[str, ...] = ()
180 display: Mapping[str, Any] = field(default_factory=dict)
182 def __post_init__(self):
183 """Validate the code namespace and the severity vocabulary."""
184 if not _DIAGNOSTIC_CODE.match(_plan_text(self.code, "Diagnostic code")):
185 raise PlanInvalid(f"Diagnostic code '{self.code}' must use the '<domain>.<condition>' form.")
186 if _plan_text(self.severity, "Diagnostic severity") not in Severity.ALL:
187 raise PlanInvalid(f"Unknown diagnostic severity '{self.severity}'.")
188 object.__setattr__(self, "identities", _identities(self.identities, "Diagnostic identities"))
189 object.__setattr__(self, "display", _plan_mapping(self.display, "Diagnostic display"))
191 def __hash__(self):
192 """Hash over the serialized form, which mirrors the generated equality."""
193 return hash(canonical_json(self.to_dict()))
195 @property
196 def fingerprint_data(self):
197 """Return the decision inputs: the code, the severity, and the affected identities."""
198 return {"code": self.code, "severity": self.severity, "identities": list(self.identities)}
200 def to_dict(self) -> dict:
201 """Return the serialized form."""
202 return {
203 "code": self.code,
204 "severity": self.severity,
205 "identities": list(self.identities),
206 "display": _thaw_json(self.display),
207 }
209 @classmethod
210 def from_dict(cls, data: dict) -> Diagnostic:
211 """Rebuild a diagnostic from its serialized form."""
212 return cls(
213 code=data["code"],
214 severity=data["severity"],
215 identities=tuple(data.get("identities", ())),
216 display=data.get("display", {}),
217 )
220@dataclass(frozen=True)
221class PlannedChange:
222 """One typed target write with its dependencies and target-state preconditions."""
224 identity: str
225 target_module: str
226 operation: str
227 payload: Mapping[str, Any]
228 dependencies: tuple[str, ...] = ()
229 preconditions: Mapping[str, Any] = field(default_factory=dict)
231 def __post_init__(self):
232 """Detach the mappings from planning state and reject anything a plan may not carry."""
233 if not _plan_text(self.identity, "Planned Change identity"):
234 raise PlanInvalid("A Planned Change needs a stable identity.")
235 for name in ("target_module", "operation"):
236 value = getattr(self, name)
237 if not isinstance(value, str) or not value:
238 raise PlanInvalid(f"A Planned Change needs a non-empty {name.replace('_', ' ')}.")
239 object.__setattr__(self, "dependencies", _identities(self.dependencies, "Planned Change dependencies"))
240 object.__setattr__(self, "payload", _plan_mapping(self.payload, "Planned Change payload"))
241 object.__setattr__(
242 self,
243 "preconditions",
244 _plan_mapping(self.preconditions, "Planned Change preconditions"),
245 )
247 def __hash__(self):
248 """Hash over the serialized form, which mirrors the generated equality."""
249 return hash(canonical_json(self.to_dict()))
251 @property
252 def fingerprint_data(self):
253 """Return every decision input this change contributes."""
254 return {
255 "identity": self.identity,
256 "target_module": self.target_module,
257 "operation": self.operation,
258 "payload": _thaw_json(self.payload),
259 "dependencies": list(self.dependencies),
260 "preconditions": _thaw_json(self.preconditions),
261 }
263 def to_dict(self) -> dict:
264 """Return a detached copy of the serialized form, which is exactly the fingerprint data."""
265 return json.loads(canonical_json(self.fingerprint_data))
267 @classmethod
268 def from_dict(cls, data: dict) -> PlannedChange:
269 """Rebuild a change from its serialized form."""
270 return cls(
271 identity=data["identity"],
272 target_module=data["target_module"],
273 operation=data["operation"],
274 payload=data.get("payload", {}),
275 dependencies=tuple(data.get("dependencies", ())),
276 preconditions=data.get("preconditions", {}),
277 )
280@dataclass(frozen=True)
281class SynchronizationUnit:
282 """The smallest independently reviewable and executable part of a plan."""
284 identity: str
285 disposition: str
286 changes: tuple[PlannedChange, ...] = ()
287 diagnostics: tuple[Diagnostic, ...] = ()
288 display: Mapping[str, Any] = field(default_factory=dict)
290 def __post_init__(self):
291 """Validate the disposition and detach the display data."""
292 if not _plan_text(self.identity, "Synchronization Unit identity"):
293 raise PlanInvalid("A Synchronization Unit needs a stable identity.")
294 if _plan_text(self.disposition, "Synchronization Unit disposition") not in Disposition.ALL:
295 raise PlanInvalid(f"Unknown disposition '{self.disposition}'.")
296 object.__setattr__(self, "changes", _elements(self.changes, PlannedChange, "Synchronization Unit changes"))
297 object.__setattr__(
298 self, "diagnostics", _elements(self.diagnostics, Diagnostic, "Synchronization Unit diagnostics")
299 )
300 object.__setattr__(self, "display", _plan_mapping(self.display, "Synchronization Unit display"))
302 def __hash__(self):
303 """Hash over the serialized form, which mirrors the generated equality."""
304 return hash(canonical_json(self.to_dict()))
306 @property
307 def fingerprint_data(self):
308 """Return the decision inputs, which exclude the display wording."""
309 return {
310 "identity": self.identity,
311 "disposition": self.disposition,
312 "changes": [change.fingerprint_data for change in self.changes],
313 "diagnostics": [diagnostic.fingerprint_data for diagnostic in self.diagnostics],
314 }
316 def to_dict(self) -> dict:
317 """Return the serialized form."""
318 return {
319 "identity": self.identity,
320 "disposition": self.disposition,
321 "changes": [change.to_dict() for change in self.changes],
322 "diagnostics": [diagnostic.to_dict() for diagnostic in self.diagnostics],
323 "display": _thaw_json(self.display),
324 }
326 @classmethod
327 def from_dict(cls, data: dict) -> SynchronizationUnit:
328 """Rebuild a unit from its serialized form."""
329 return cls(
330 identity=data["identity"],
331 disposition=data["disposition"],
332 changes=tuple(PlannedChange.from_dict(item) for item in data.get("changes", ())),
333 diagnostics=tuple(Diagnostic.from_dict(item) for item in data.get("diagnostics", ())),
334 display=data.get("display", {}),
335 )
338@dataclass(frozen=True)
339class ImportPlan:
340 """A serializable plan: units, diagnostics, and the inputs its fingerprint covers."""
342 units: tuple[SynchronizationUnit, ...] = ()
343 diagnostics: tuple[Diagnostic, ...] = ()
344 source_fingerprint: str = ""
345 profile_fingerprint: str = ""
346 actor: str = ""
347 planning_context: Mapping[str, Any] = field(default_factory=dict)
348 revision: int = 1
349 schema_version: int = SCHEMA_VERSION
351 def __post_init__(self):
352 """Detach the planning context from planning state."""
353 object.__setattr__(self, "units", _elements(self.units, SynchronizationUnit, "Import Plan units"))
354 object.__setattr__(self, "diagnostics", _elements(self.diagnostics, Diagnostic, "Import Plan diagnostics"))
355 object.__setattr__(self, "planning_context", _plan_mapping(self.planning_context, "Planning context"))
356 # canonical_json raises a bare TypeError past `except PlanError`, and takes a dict for a str.
357 for name in ("source_fingerprint", "profile_fingerprint", "actor"):
358 _plan_text(getattr(self, name), f"Import Plan {name}")
359 for name in ("revision", "schema_version"):
360 _plan_int(getattr(self, name), f"Import Plan {name}")
361 counts = Counter(unit.identity for unit in self.units)
362 duplicates = sorted(identity for identity, count in counts.items() if count > 1)
363 if duplicates:
364 raise PlanInvalid(f"Synchronization Unit identities must be unique: {', '.join(duplicates)}.")
366 def __hash__(self):
367 """Hash over the serialized form, which mirrors the generated equality."""
368 return hash(canonical_json(self.to_dict()))
370 @property
371 def _selection_context_data(self):
372 """Return the plan-wide inputs that invalidate every accepted unit."""
373 return {
374 "schema_version": self.schema_version,
375 "source_fingerprint": self.source_fingerprint,
376 "profile_fingerprint": self.profile_fingerprint,
377 "actor": self.actor,
378 "planning_context": _thaw_json(self.planning_context),
379 }
381 @property
382 def fingerprint_data(self):
383 """Return the decision inputs of section 4.3, which exclude display data and the revision."""
384 return {
385 **self._selection_context_data,
386 "units": [unit.fingerprint_data for unit in self.units],
387 "diagnostics": [diagnostic.fingerprint_data for diagnostic in self.diagnostics],
388 }
390 @property
391 def fingerprint(self) -> str:
392 """Return the canonical plan fingerprint."""
393 return fingerprint_of(self.fingerprint_data)
395 def unit(self, identity: str) -> SynchronizationUnit | None:
396 """Return one indexed unit without repeating a linear scan for each lookup."""
397 index = self.__dict__.get("_unit_index")
398 if index is None:
399 index = {unit.identity: unit for unit in self.units}
400 object.__setattr__(self, "_unit_index", index)
401 return index.get(identity)
403 def unit_fingerprint(self, identity: str) -> str:
404 """Return one unit's decision inputs plus the context shared by every unit."""
405 unit = self.unit(identity)
406 if unit is None:
407 raise PlanInvalid(f"The Import Plan has no Synchronization Unit '{identity}'.")
408 return fingerprint_of({**self._selection_context_data, "unit": unit.fingerprint_data})
410 def to_dict(self) -> dict:
411 """Return the serialized form the session and a job payload carry."""
412 return {
413 "schema_version": self.schema_version,
414 "units": [unit.to_dict() for unit in self.units],
415 "diagnostics": [diagnostic.to_dict() for diagnostic in self.diagnostics],
416 "source_fingerprint": self.source_fingerprint,
417 "profile_fingerprint": self.profile_fingerprint,
418 "actor": self.actor,
419 "planning_context": _thaw_json(self.planning_context),
420 "revision": self.revision,
421 }
423 @classmethod
424 def from_dict(cls, data: Any) -> ImportPlan:
425 """Rebuild a plan, rejecting a schema version this release does not execute.
427 Every other malformed payload also raises a PlanError, so one caller-side ``except PlanError``
428 covers a corrupted session entry or job payload (section 4.8).
429 """
430 try:
431 version = data.get("schema_version")
432 if not isinstance(version, int) or isinstance(version, bool) or version != SCHEMA_VERSION:
433 raise PlanSchemaMismatch(f"Import Plan schema version {version} is not version {SCHEMA_VERSION}.")
434 return cls(
435 units=tuple(SynchronizationUnit.from_dict(item) for item in data["units"]),
436 diagnostics=tuple(Diagnostic.from_dict(item) for item in data["diagnostics"]),
437 source_fingerprint=data.get("source_fingerprint", ""),
438 profile_fingerprint=data.get("profile_fingerprint", ""),
439 actor=data.get("actor", ""),
440 planning_context=data.get("planning_context", {}),
441 revision=data.get("revision", 1),
442 schema_version=version,
443 )
444 except PlanError:
445 raise
446 except (AttributeError, KeyError, TypeError, ValueError) as exc:
447 raise PlanInvalid(f"The serialized Import Plan is malformed: {exc!r}") from exc
450def executable_units(units) -> tuple[SynchronizationUnit, ...]:
451 """Return only the actionable units.
453 Section 4.6: blocked, invalid, excluded, and no-op units never enter an execution transaction.
454 """
455 return tuple(unit for unit in units if unit.disposition == Disposition.ACTIONABLE)
458def merge_changes(units, *, reconciled=()) -> tuple[PlannedChange, ...]:
459 """Merge the changes of *units* into one deterministic acyclic execution order.
461 Identical identities are shared and execute once. A conflicting payload or precondition, a
462 dangling dependency, and a cycle each make the plan invalid (section 4.4).
464 *reconciled* holds the identities a previous execution already applied. They satisfy a
465 dependency without being merged in, which is what section 4.5 means by a dependency that is
466 already reconciled. Passing an identity here never adds work to the selection.
468 This merges whatever units it is given. Filter with ``executable_units`` before executing.
469 """
470 merged: dict[str, PlannedChange] = {}
471 order: list[str] = []
472 for unit in units:
473 for change in unit.changes:
474 existing = merged.get(change.identity)
475 if existing is None:
476 merged[change.identity] = change
477 order.append(change.identity)
478 elif existing.fingerprint_data != change.fingerprint_data:
479 raise PlanInvalid(
480 f"Planned Change '{change.identity}' appears with conflicting content, so the plan cannot share it."
481 )
483 already_applied = frozenset(reconciled)
484 for identity, change in merged.items():
485 for dependency in change.dependencies:
486 if dependency not in merged and dependency not in already_applied:
487 raise PlanInvalid(
488 f"Planned Change '{identity}' depends on '{dependency}', "
489 "which the selection neither contains nor has reconciled."
490 )
492 return _topological_order(merged, order)
495def _topological_order(merged: dict, order: list[str]) -> tuple[PlannedChange, ...]:
496 """Return the changes in dependency order, breaking ties by first appearance."""
497 position = {identity: index for index, identity in enumerate(order)}
498 state: dict[str, int] = {}
499 result: list[PlannedChange] = []
501 def visit(identity: str, path: tuple[str, ...]):
502 """Depth-first visit that reports the identities forming a cycle."""
503 if state.get(identity) == 2:
504 return
505 if state.get(identity) == 1:
506 cycle = " -> ".join((*path[path.index(identity) :], identity))
507 raise PlanInvalid(f"The Planned Change dependencies form a cycle: {cycle}.")
508 state[identity] = 1
509 for dependency in sorted(
510 (dep for dep in merged[identity].dependencies if dep in merged), key=lambda dep: position[dep]
511 ):
512 visit(dependency, (*path, identity))
513 state[identity] = 2
514 result.append(merged[identity])
516 for identity in order:
517 visit(identity, ())
518 return tuple(result)
521__all__ = (
522 "SCHEMA_VERSION",
523 "Diagnostic",
524 "Disposition",
525 "FrozenDict",
526 "ImportPlan",
527 "PlanError",
528 "PlanInvalid",
529 "PlanSchemaMismatch",
530 "PlannedChange",
531 "Severity",
532 "SynchronizationUnit",
533 "canonical_json",
534 "executable_units",
535 "fingerprint_of",
536 "merge_changes",
537)