Coverage for netbox_data_import/plan.py: 100.00%

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 Preview Coordinator, 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 logging

18import re

19from collections import Counter

20from collections.abc import Callable, Mapping, Sequence

21from dataclasses import dataclass, field

22from typing import Any

24SCHEMA_VERSION = 8

26logger = logging.getLogger(__name__)

28_DIAGNOSTIC_CODE = re.compile(r"^[a-z0-9]+(?:_[a-z0-9]+)*\.[a-z0-9]+(?:_[a-z0-9]+)*$")

30# Diagnostic code namespace -> the display validator its Target Module registers at application start.

31_DISPLAY_VALIDATORS: dict[str, Callable[[str, dict], None]] = {}

34class PlanError(Exception):

35 """Base class for the Import Plan failures the coordinator reports."""

38class PlanInvalid(PlanError):

39 """A plan violates a structural invariant and cannot execute."""

42class PlanSchemaMismatch(PlanError):

43 """A serialized plan states a schema version this release does not execute."""

46class Disposition:

47 """The exactly-one state a Synchronization Unit carries (section 4.2)."""

49 ACTIONABLE = "actionable"

50 NO_OP = "no-op"

51 BLOCKED = "blocked"

52 INVALID = "invalid"

53 EXCLUDED = "excluded"

55 ALL = frozenset({ACTIONABLE, NO_OP, BLOCKED, INVALID, EXCLUDED})

58class Severity:

59 """Diagnostic severity (section 4.2)."""

61 INFO = "info"

62 WARNING = "warning"

63 ERROR = "error"

65 ALL = frozenset({INFO, WARNING, ERROR})

68def canonical_json(value: Any) -> str:

69 """Return the canonical serialization used for every fingerprint.

71 ``allow_nan`` stays off: NaN and Infinity are not JSON, PostgreSQL rejects them in a JSONField,

72 and NaN never equals itself, which would make two identical shared changes look conflicting.

73 """

74 return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False)

77def fingerprint_of(value: Any) -> str:

78 """Return the SHA-256 digest of the canonical serialization of *value*."""

79 return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()

82class FrozenDict(dict):

83 """The JSON object a plan hands out: JSON data, closed to mutation.

85 A frozen plan value stays JSON data, because a Target Module writes one into a JSONField.

86 """

88 __slots__ = ()

90 def _immutable(self, *args, **kwargs):

91 """Refuse one mutation, so a shared plan value cannot change under its other readers."""

92 raise TypeError("An Import Plan value cannot be changed.")

94 __setitem__ = _immutable

95 __delitem__ = _immutable

96 __ior__ = _immutable

97 clear = _immutable

98 pop = _immutable

99 popitem = _immutable

100 setdefault = _immutable

101 update = _immutable

102

103

104def _freeze_json(value: Any) -> Any:

105 """Return recursively immutable JSON data."""

106 if isinstance(value, dict):

107 return FrozenDict({key: _freeze_json(item) for key, item in value.items()})

108 if isinstance(value, list):

109 return tuple(_freeze_json(item) for item in value)

110 return value

111

112

113def _thaw_json(value: Any) -> Any:

114 """Return detached plain JSON data from an immutable plan value."""

115 if isinstance(value, Mapping):

116 return {key: _thaw_json(item) for key, item in value.items()}

117 if isinstance(value, tuple):

118 return [_thaw_json(item) for item in value]

119 return value

120

121

122def _frozen_json(value: Any, label: str) -> Any:

123 """Validate, detach, and recursively freeze JSON data entering a plan."""

124 try:

125 return _freeze_json(json.loads(canonical_json(value)))

126 except (TypeError, ValueError) as exc:

127 logger.warning("%s is not JSON-serializable plan data.", label, exc_info=True)

128 raise PlanInvalid(f"{label} must be JSON-serializable plan data.") from exc

129

130

131def register_diagnostic_display(namespace: str, validator: Callable[[str, dict], None]) -> None:

132 """Register the validator that refuses an unknown display field for every code in *namespace*.

133

134 The validator raises ValueError. A namespace no Target Module registers is not checked: only a

135 registered display carries row text that a render must recheck.

136 """

137 registered = _DISPLAY_VALIDATORS.setdefault(namespace, validator)

138 if registered is not validator:

139 raise ValueError(f"Diagnostic namespace '{namespace}' already has a display validator.")

140

141

142def validate_diagnostic_display(code: str, display: Mapping[str, Any]) -> None:

143 """Raise ValueError when the validator registered for the code's namespace refuses *display*."""

144 validator = _DISPLAY_VALIDATORS.get(code.partition(".")[0])

145 if validator is not None:

146 validator(code, _thaw_json(display))

147

148

149def _plan_mapping(value: Any, label: str) -> Mapping[str, Any]:

150 """Return *value* as the immutable JSON object the field declares."""

151 frozen = _frozen_json(value, label)

152 if not isinstance(frozen, Mapping):

153 raise PlanInvalid(f"{label} must be a JSON object, not {type(value).__name__}.")

154 return frozen

155

156

157def _plan_text(value, label: str) -> str:

158 """Return *value* as the string the field declares."""

159 if not isinstance(value, str):

160 raise PlanInvalid(f"{label} must be a string, not {type(value).__name__}.")

161 return value

162

163

164def _plan_int(value, label: str) -> int:

165 """Return *value* as the integer the field declares.

166

167 A bool is an int in Python, and `from_dict` refuses one, so a plan built from a bool could not

168 survive its own serialization.

169 """

170 if isinstance(value, bool) or not isinstance(value, int):

171 raise PlanInvalid(f"{label} must be an integer, not {type(value).__name__}.")

172 return value

173

174

175def _elements(value, kind, label: str) -> tuple:

176 """Return *value* as a tuple, rejecting any element that is not an instance of *kind*."""

177 if isinstance(value, str) or not isinstance(value, (list, tuple)):

178 raise PlanInvalid(f"{label} must be a list or tuple of {kind.__name__} objects.")

179 for item in value:

180 if not isinstance(item, kind):

181 raise PlanInvalid(f"{label} must hold {kind.__name__} objects, not {type(item).__name__}.")

182 return tuple(value)

183

184

185def _identities(value, label: str) -> tuple[str, ...]:

186 """Return *value* as a tuple of identity strings.

187

188 A bare string is refused because tuple("rack:1") would silently become six identities.

189 """

190 if isinstance(value, str) or not isinstance(value, (list, tuple)):

191 raise PlanInvalid(f"{label} must be a list or tuple of identity strings.")

192 for item in value:

193 if not isinstance(item, str):

194 raise PlanInvalid(f"{label} must hold identity strings, not {type(item).__name__}.")

195 return tuple(value)

196

197

198@dataclass(frozen=True)

199class Diagnostic:

200 """One structured finding attached to a plan or a unit."""

201

202 code: str

203 severity: str

204 identities: tuple[str, ...] = ()

205 display: Mapping[str, Any] = field(default_factory=dict)

206 # The facts the finding was decided from, which the fingerprint carries and `display` does not.

207 evidence: Mapping[str, Any] = field(default_factory=dict)

208

209 def __post_init__(self):

210 """Validate the code namespace and the severity vocabulary."""

211 if not _DIAGNOSTIC_CODE.match(_plan_text(self.code, "Diagnostic code")):

212 raise PlanInvalid(f"Diagnostic code '{self.code}' must use the '<domain>.<condition>' form.")

213 if _plan_text(self.severity, "Diagnostic severity") not in Severity.ALL:

214 raise PlanInvalid(f"Unknown diagnostic severity '{self.severity}'.")

215 object.__setattr__(self, "identities", _identities(self.identities, "Diagnostic identities"))

216 object.__setattr__(self, "display", _plan_mapping(self.display, "Diagnostic display"))

217 object.__setattr__(self, "evidence", _plan_mapping(self.evidence, "Diagnostic evidence"))

218

219 def __hash__(self):

220 """Hash over the serialized form, which mirrors the generated equality."""

221 return hash(canonical_json(self.to_dict()))

222

223 @property

224 def fingerprint_data(self):

225 """Return the decision inputs: the code, the severity, the identities, and the evidence.

226

227 A finding about live state the identities do not name, such as the media two Cables carry,

228 would otherwise leave the fingerprint unchanged while the world moved under an accepted plan.

229 """

230 return {

231 "code": self.code,

232 "severity": self.severity,

233 "identities": list(self.identities),

234 "evidence": _thaw_json(self.evidence),

235 }

236

237 def to_dict(self) -> dict:

238 """Return the serialized form."""

239 return {

240 "code": self.code,

241 "severity": self.severity,

242 "identities": list(self.identities),

243 "display": _thaw_json(self.display),

244 "evidence": _thaw_json(self.evidence),

245 }

246

247 @classmethod

248 def from_dict(cls, data: dict) -> Diagnostic:

249 """Rebuild a diagnostic, refusing a display field its namespace's registered validator does not permit.

250

251 A cached plan can predate a display change, and an unknown field can hold a value no render

252 rechecks, so the plan has to be planned again instead of reused.

253 """

254 diagnostic = cls(

255 code=data["code"],

256 severity=data["severity"],

257 identities=tuple(data.get("identities", ())),

258 display=data.get("display", {}),

259 evidence=data.get("evidence", {}),

260 )

261 validate_diagnostic_display(diagnostic.code, diagnostic.display)

262 return diagnostic

263

264

265@dataclass(frozen=True)

266class PlannedChange:

267 """One typed target write with its dependencies and target-state preconditions."""

268

269 identity: str

270 target_module: str

271 operation: str

272 payload: Mapping[str, Any]

273 dependencies: tuple[str, ...] = ()

274 preconditions: Mapping[str, Any] = field(default_factory=dict)

275

276 def __post_init__(self):

277 """Detach the mappings from planning state and reject anything a plan may not carry."""

278 if not _plan_text(self.identity, "Planned Change identity"):

279 raise PlanInvalid("A Planned Change needs a stable identity.")

280 for name in ("target_module", "operation"):

281 value = getattr(self, name)

282 if not isinstance(value, str) or not value:

283 raise PlanInvalid(f"A Planned Change needs a non-empty {name.replace('_', ' ')}.")

284 object.__setattr__(self, "dependencies", _identities(self.dependencies, "Planned Change dependencies"))

285 object.__setattr__(self, "payload", _plan_mapping(self.payload, "Planned Change payload"))

286 object.__setattr__(

287 self,

288 "preconditions",

289 _plan_mapping(self.preconditions, "Planned Change preconditions"),

290 )

291

292 def __hash__(self):

293 """Hash over the serialized form, which mirrors the generated equality."""

294 return hash(canonical_json(self.to_dict()))

295

296 @property

297 def fingerprint_data(self):

298 """Return every decision input this change contributes."""

299 return {

300 "identity": self.identity,

301 "target_module": self.target_module,

302 "operation": self.operation,

303 "payload": _thaw_json(self.payload),

304 "dependencies": list(self.dependencies),

305 "preconditions": _thaw_json(self.preconditions),

306 }

307

308 def to_dict(self) -> dict:

309 """Return a detached copy of the serialized form, which is exactly the fingerprint data."""

310 return json.loads(canonical_json(self.fingerprint_data))

311

312 @classmethod

313 def from_dict(cls, data: dict) -> PlannedChange:

314 """Rebuild a change from its serialized form."""

315 return cls(

316 identity=data["identity"],

317 target_module=data["target_module"],

318 operation=data["operation"],

319 payload=data.get("payload", {}),

320 dependencies=tuple(data.get("dependencies", ())),

321 preconditions=data.get("preconditions", {}),

322 )

323

324

325@dataclass(frozen=True)

326class SynchronizationUnit:

327 """The smallest independently reviewable and executable part of a plan."""

328

329 identity: str

330 disposition: str

331 changes: tuple[PlannedChange, ...] = ()

332 diagnostics: tuple[Diagnostic, ...] = ()

333 display: Mapping[str, Any] = field(default_factory=dict)

334 # The units every selection must hold together with this one (section 4.5).

335 selected_with: tuple[str, ...] = ()

336

337 def __post_init__(self):

338 """Validate the disposition and detach the display data."""

339 if not _plan_text(self.identity, "Synchronization Unit identity"):

340 raise PlanInvalid("A Synchronization Unit needs a stable identity.")

341 if _plan_text(self.disposition, "Synchronization Unit disposition") not in Disposition.ALL:

342 raise PlanInvalid(f"Unknown disposition '{self.disposition}'.")

343 object.__setattr__(self, "changes", _elements(self.changes, PlannedChange, "Synchronization Unit changes"))

344 object.__setattr__(

345 self, "diagnostics", _elements(self.diagnostics, Diagnostic, "Synchronization Unit diagnostics")

346 )

347 object.__setattr__(self, "display", _plan_mapping(self.display, "Synchronization Unit display"))

348 object.__setattr__(self, "selected_with", _identities(self.selected_with, "Synchronization Unit selected_with"))

349

350 def __hash__(self):

351 """Hash over the serialized form, which mirrors the generated equality."""

352 return hash(canonical_json(self.to_dict()))

353

354 def _companions(self) -> dict:

355 """Return `selected_with` as a serialized field, or nothing, so a unit without one keeps its bytes."""

356 return {"selected_with": list(self.selected_with)} if self.selected_with else {}

357

358 @property

359 def fingerprint_data(self):

360 """Return the decision inputs, which exclude the display wording."""

361 return {

362 "identity": self.identity,

363 "disposition": self.disposition,

364 "changes": [change.fingerprint_data for change in self.changes],

365 "diagnostics": [diagnostic.fingerprint_data for diagnostic in self.diagnostics],

366 **self._companions(),

367 }

368

369 def to_dict(self) -> dict:

370 """Return the serialized form."""

371 return {

372 "identity": self.identity,

373 "disposition": self.disposition,

374 "changes": [change.to_dict() for change in self.changes],

375 "diagnostics": [diagnostic.to_dict() for diagnostic in self.diagnostics],

376 "display": _thaw_json(self.display),

377 **self._companions(),

378 }

379

380 @classmethod

381 def from_dict(cls, data: dict) -> SynchronizationUnit:

382 """Rebuild a unit from its serialized form."""

383 return cls(

384 identity=data["identity"],

385 disposition=data["disposition"],

386 changes=tuple(PlannedChange.from_dict(item) for item in data.get("changes", ())),

387 diagnostics=tuple(Diagnostic.from_dict(item) for item in data.get("diagnostics", ())),

388 display=data.get("display", {}),

389 selected_with=tuple(data.get("selected_with", ())),

390 )

391

392

393def is_current_schema_version(version: Any) -> bool:

394 """Return whether a serialized plan's ``schema_version`` is the one this release executes."""

395 return isinstance(version, int) and not isinstance(version, bool) and version == SCHEMA_VERSION

396

397

398@dataclass(frozen=True)

399class ImportPlan:

400 """A serializable plan: units, diagnostics, and the inputs its fingerprint covers."""

401

402 units: tuple[SynchronizationUnit, ...] = ()

403 diagnostics: tuple[Diagnostic, ...] = ()

404 source_fingerprint: str = ""

405 profile_fingerprint: str = ""

406 actor: str = ""

407 planning_context: Mapping[str, Any] = field(default_factory=dict)

408 revision: int = 1

409 schema_version: int = SCHEMA_VERSION

410

411 def __post_init__(self):

412 """Detach the planning context from planning state."""

413 object.__setattr__(self, "units", _elements(self.units, SynchronizationUnit, "Import Plan units"))

414 object.__setattr__(self, "diagnostics", _elements(self.diagnostics, Diagnostic, "Import Plan diagnostics"))

415 object.__setattr__(self, "planning_context", _plan_mapping(self.planning_context, "Planning context"))

416 # canonical_json raises a bare TypeError past `except PlanError`, and takes a dict for a str.

417 for name in ("source_fingerprint", "profile_fingerprint", "actor"):

418 _plan_text(getattr(self, name), f"Import Plan {name}")

419 for name in ("revision", "schema_version"):

420 _plan_int(getattr(self, name), f"Import Plan {name}")

421 counts = Counter(unit.identity for unit in self.units)

422 duplicates = sorted(identity for identity, count in counts.items() if count > 1)

423 if duplicates:

424 raise PlanInvalid(f"Synchronization Unit identities must be unique: {', '.join(duplicates)}.")

425 unknown = sorted({member for unit in self.units for member in unit.selected_with} - counts.keys())

426 if unknown:

427 raise PlanInvalid(

428 f"A selected_with member names no Synchronization Unit of the plan: {', '.join(unknown)}."

429 )

430

431 def __hash__(self):

432 """Hash over the serialized form, which mirrors the generated equality."""

433 return hash(canonical_json(self.to_dict()))

434

435 @property

436 def _selection_context_data(self):

437 """Return the plan-wide inputs that invalidate every accepted unit."""

438 return {

439 "schema_version": self.schema_version,

440 "source_fingerprint": self.source_fingerprint,

441 "profile_fingerprint": self.profile_fingerprint,

442 "actor": self.actor,

443 "planning_context": _thaw_json(self.planning_context),

444 }

445

446 @property

447 def fingerprint_data(self):

448 """Return the decision inputs of section 4.3, which exclude display data and the revision."""

449 return {

450 **self._selection_context_data,

451 "units": [unit.fingerprint_data for unit in self.units],

452 "diagnostics": [diagnostic.fingerprint_data for diagnostic in self.diagnostics],

453 }

454

455 @property

456 def fingerprint(self) -> str:

457 """Return the canonical plan fingerprint."""

458 return fingerprint_of(self.fingerprint_data)

459

460 def unit(self, identity: str) -> SynchronizationUnit | None:

461 """Return one indexed unit without repeating a linear scan for each lookup."""

462 index = self.__dict__.get("_unit_index")

463 if index is None:

464 index = {unit.identity: unit for unit in self.units}

465 object.__setattr__(self, "_unit_index", index)

466 return index.get(identity)

467

468 def unit_fingerprint(self, identity: str) -> str:

469 """Return one unit's decision inputs plus the context shared by every unit."""

470 unit = self.unit(identity)

471 if unit is None:

472 raise PlanInvalid(f"The Import Plan has no Synchronization Unit '{identity}'.")

473 return fingerprint_of({**self._selection_context_data, "unit": unit.fingerprint_data})

474

475 def to_dict(self) -> dict:

476 """Return the serialized form the Preview Coordinator and a job payload carry."""

477 return {

478 "schema_version": self.schema_version,

479 "units": [unit.to_dict() for unit in self.units],

480 "diagnostics": [diagnostic.to_dict() for diagnostic in self.diagnostics],

481 "source_fingerprint": self.source_fingerprint,

482 "profile_fingerprint": self.profile_fingerprint,

483 "actor": self.actor,

484 "planning_context": _thaw_json(self.planning_context),

485 "revision": self.revision,

486 }

487

488 @classmethod

489 def from_dict(cls, data: Any) -> ImportPlan:

490 """Rebuild a plan, rejecting a schema version this release does not execute.

491

492 Every other malformed payload also raises a PlanError, so one caller-side ``except PlanError``

493 covers a corrupted stored preview or job payload (section 4.8).

494 """

495 try:

496 version = data.get("schema_version")

497 if not is_current_schema_version(version):

498 raise PlanSchemaMismatch(f"Import Plan schema version {version} is not version {SCHEMA_VERSION}.")

499 return cls(

500 units=tuple(SynchronizationUnit.from_dict(item) for item in data["units"]),

501 diagnostics=tuple(Diagnostic.from_dict(item) for item in data["diagnostics"]),

502 source_fingerprint=data.get("source_fingerprint", ""),

503 profile_fingerprint=data.get("profile_fingerprint", ""),

504 actor=data.get("actor", ""),

505 planning_context=data.get("planning_context", {}),

506 revision=data.get("revision", 1),

507 schema_version=version,

508 )

509 except PlanError:

510 raise

511 except (AttributeError, KeyError, TypeError, ValueError) as exc:

512 logger.warning("Refused a malformed serialized Import Plan.", exc_info=True)

513 raise PlanInvalid("The serialized Import Plan is malformed.") from exc

514

515

516def executable_units(units) -> tuple[SynchronizationUnit, ...]:

517 """Return only the actionable units.

518

519 Section 4.6: blocked, invalid, excluded, and no-op units never enter an execution transaction.

520 """

521 return tuple(unit for unit in units if unit.disposition == Disposition.ACTIONABLE)

522

523

524def missing_companions(

525 units: Mapping[str, SynchronizationUnit], selection: Sequence[str]

526) -> dict[str, tuple[str, ...]]:

527 """Return, for each selected unit, the `selected_with` members the selection lacks or that are not actionable.

528

529 *units* maps each identity to its unit. An empty result means the selection honours co-selection.

530 It never adds a unit: a review command shows the operator every unit it selects (section 4.5).

531 """

532 chosen = frozenset(selection)

533 missing = {}

534 for identity in selection:

535 absent = tuple(

536 member

537 for member in units[identity].selected_with

538 if member not in chosen or units[member].disposition != Disposition.ACTIONABLE

539 )

540 if absent:

541 missing[identity] = absent

542 return missing

543

544

545def merge_changes(units, *, reconciled=()) -> tuple[PlannedChange, ...]:

546 """Merge the changes of *units* into one deterministic acyclic execution order.

547

548 Identical identities are shared and execute once. A conflicting payload or precondition, a

549 dangling dependency, and a cycle each make the plan invalid (section 4.4).

550

551 *reconciled* holds the identities a previous execution already applied. They satisfy a

552 dependency without being merged in, which is what section 4.5 means by a dependency that is

553 already reconciled. Passing an identity here never adds work to the selection.

554

555 This merges whatever units it is given. Filter with ``executable_units`` before executing.

556 """

557 merged: dict[str, PlannedChange] = {}

558 order: list[str] = []

559 for unit in units:

560 for change in unit.changes:

561 existing = merged.get(change.identity)

562 if existing is None:

563 merged[change.identity] = change

564 order.append(change.identity)

565 elif existing.fingerprint_data != change.fingerprint_data:

566 raise PlanInvalid(

567 f"Planned Change '{change.identity}' appears with conflicting content, so the plan cannot share it."

568 )

569

570 already_applied = frozenset(reconciled)

571 for identity, change in merged.items():

572 for dependency in change.dependencies:

573 if dependency not in merged and dependency not in already_applied:

574 raise PlanInvalid(

575 f"Planned Change '{identity}' depends on '{dependency}', "

576 "which the selection neither contains nor has reconciled."

577 )

578

579 return _topological_order(merged, order)

580

581

582def _topological_order(merged: dict, order: list[str]) -> tuple[PlannedChange, ...]:

583 """Return the changes in dependency order, breaking ties by first appearance."""

584 position = {identity: index for index, identity in enumerate(order)}

585 state: dict[str, int] = {}

586 result: list[PlannedChange] = []

587

588 def visit(identity: str, path: tuple[str, ...]):

589 """Depth-first visit that reports the identities forming a cycle."""

590 if state.get(identity) == 2:

591 return

592 if state.get(identity) == 1:

593 cycle = " -> ".join((*path[path.index(identity) :], identity))

594 raise PlanInvalid(f"The Planned Change dependencies form a cycle: {cycle}.")

595 state[identity] = 1

596 for dependency in sorted(

597 (dep for dep in merged[identity].dependencies if dep in merged), key=lambda dep: position[dep]

598 ):

599 visit(dependency, (*path, identity))

600 state[identity] = 2

601 result.append(merged[identity])

602

603 for identity in order:

604 visit(identity, ())

605 return tuple(result)

606

607

608__all__ = (

609 "SCHEMA_VERSION",

610 "Diagnostic",

611 "Disposition",

612 "FrozenDict",

613 "ImportPlan",

614 "PlanError",

615 "PlanInvalid",

616 "PlanSchemaMismatch",

617 "PlannedChange",

618 "Severity",

619 "SynchronizationUnit",

620 "canonical_json",

621 "executable_units",

622 "fingerprint_of",

623 "is_current_schema_version",

624 "merge_changes",

625 "missing_companions",

626 "register_diagnostic_display",

627 "validate_diagnostic_display",

628)