Coverage for netbox_data_import/device_field_review.py: 100.00%
197 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) 2026 Marcin Zieba <marcinpsk@gmail.com>
3"""Review exact source/NetBox value pairs for matched Device fields."""
5from __future__ import annotations
7from collections.abc import Callable, Mapping
8from dataclasses import dataclass
9from typing import Any
12def _text(value: Any) -> str:
13 """Return the display text used for an exact text comparison."""
14 if value is None:
15 return ""
16 return str(value).strip()
19def _number(value: Any) -> str:
20 """Return a stable string for values that represent a number."""
21 if value is None or value == "":
22 return ""
23 try:
24 number = float(value)
25 if number == int(number):
26 return str(int(number))
27 return str(number)
28 except (TypeError, ValueError, OverflowError):
29 return _text(value)
32def _identity(value: Any) -> str:
33 """Return the canonical identity for a model instance or scalar."""
34 if value is None:
35 return ""
36 primary_key = getattr(value, "pk", None)
37 return _text(primary_key if primary_key is not None else value)
40def _related_display(value: Any) -> str:
41 """Return a human-readable value for a related object snapshot."""
42 if value is None:
43 return ""
44 return _text(getattr(value, "name", value))
47def _device_rack_name(device) -> str:
48 """Return the raw rack name currently assigned to a device."""
49 return device.rack.name if getattr(device, "rack_id", None) else ""
52def _device_rack_display(device) -> str:
53 """Return the location-aware rack label currently assigned to a device."""
54 if not getattr(device, "rack_id", None):
55 return ""
56 rack = device.rack
57 if getattr(rack, "location_id", None):
58 return f"{rack.location} / {rack.name}"
59 return _text(rack.name)
62def _device_rack_location_id(device):
63 """Return the location that scopes a device's rack, if it has one."""
64 return getattr(device.rack, "location_id", None) if getattr(device, "rack_id", None) else None
67def _scope_rack_canonical(snapshot: dict[str, str], location_id) -> dict[str, str]:
68 """Prefix a rack canonical value with its location so both sides compare the same way."""
69 snapshot["canonical"] = f"{location_id or ''}:{snapshot['canonical']}"
70 return snapshot
73def _device_type_value(device):
74 """Return the canonical and display data for the current DeviceType."""
75 device_type = device.device_type
76 manufacturer = device_type.manufacturer
77 return (manufacturer.slug, device_type.slug, manufacturer.name, device_type.model)
80def _device_type_display(value: Any) -> str:
81 """Return a human-readable DeviceType value."""
82 if isinstance(value, (tuple, list)) and len(value) >= 4:
83 return f"{value[2]} / {value[3]}"
84 return _text(value)
87def _device_type_normalize(value: Any) -> str:
88 """Return the stable manufacturer/device-type slug pair."""
89 if isinstance(value, (tuple, list)) and len(value) >= 2:
90 return f"{_text(value[0])}/{_text(value[1])}"
91 return _text(value)
94def _device_role_value(device) -> str:
95 """Return the current DeviceRole slug."""
96 return _text(device.role.slug) if getattr(device, "role_id", None) else ""
99def _device_ip_value(target_field: str):
100 """Return a reader for one of the Device's IP fields."""
102 def read(device):
103 current = getattr(device, target_field, None)
104 return _text(current.address) if current is not None else ""
106 return read
109def _ip_normalize(value: Any) -> str:
110 """Return an address in the one spelling both sides compare on."""
111 import ipaddress
113 text = _text(value)
114 if not text:
115 return ""
116 try:
117 return str(ipaddress.ip_interface(text))
118 except ValueError:
119 return text
122@dataclass(frozen=True)
123class FieldDefinition:
124 """One source of truth for a Device field's review behavior."""
126 target_field: str
127 current_value: Callable[[Any], Any]
128 normalize: Callable[[Any], str] = _text
129 display: Callable[[Any], str] = _text
130 provided: Callable[[Any], bool] = lambda value: True
131 writable: bool = True
132 reviewable: bool = True
134 def snapshot(self, value: Any, display_override: str | None = None) -> dict[str, str]:
135 """Return the persisted canonical and display representation."""
136 return {
137 "canonical": self.normalize(value),
138 "display": self.display(value) if display_override is None else display_override,
139 }
142def _provided_nonempty(value: Any) -> bool:
143 """Return whether a writer receives a value for an optional field."""
144 return bool(_text(value))
147def _provided_optional(value: Any) -> bool:
148 """Return whether a source row explicitly supplies an optional field."""
149 return value is not None and value != ""
152def _provided_not_none(value: Any) -> bool:
153 """Return whether a source row supplied a value, including zero and empty text."""
154 return value is not None
157# Keep this registry private. Callers use DeviceFieldReviewer instead of
158# reimplementing normalization, comparison, display, or write semantics.
159_FIELD_DEFINITIONS: tuple[FieldDefinition, ...] = (
160 FieldDefinition("rack_name", _device_rack_name, display=_text),
161 FieldDefinition("u_position", lambda device: device.position, normalize=_number, display=_number),
162 FieldDefinition("face", lambda device: device.face or "", provided=_provided_optional),
163 FieldDefinition("airflow", lambda device: device.airflow or "", provided=_provided_nonempty),
164 FieldDefinition("status", lambda device: device.status or ""),
165 FieldDefinition("serial", lambda device: device.serial or "", provided=_provided_nonempty),
166 FieldDefinition("asset_tag", lambda device: device.asset_tag or "", provided=_provided_nonempty),
167 FieldDefinition(
168 "device_type",
169 _device_type_value,
170 normalize=_device_type_normalize,
171 display=_device_type_display,
172 ),
173 FieldDefinition("role", _device_role_value),
174 # The writer assigns these, so they are differences the preview has to report.
175 FieldDefinition(
176 "primary_ip4", _device_ip_value("primary_ip4"), normalize=_ip_normalize, provided=_provided_nonempty
177 ),
178 FieldDefinition(
179 "primary_ip6", _device_ip_value("primary_ip6"), normalize=_ip_normalize, provided=_provided_nonempty
180 ),
181 FieldDefinition("oob_ip", _device_ip_value("oob_ip"), normalize=_ip_normalize, provided=_provided_nonempty),
182 FieldDefinition("tenant", lambda device: device.tenant, normalize=_identity, display=_related_display),
183 FieldDefinition("location", lambda device: device.location, normalize=_identity, display=_related_display),
184 # These values are shown by the legacy field-diff helper, but the Device
185 # writer does not assign them. Keep them reviewable and non-writable so an
186 # ignored review never claims that a write was suppressed.
187 FieldDefinition("device_name", lambda device: device.name, writable=False),
188 FieldDefinition(
189 "u_height",
190 lambda device: device.device_type.u_height if getattr(device, "device_type_id", None) else None,
191 normalize=_number,
192 display=_number,
193 provided=_provided_not_none,
194 writable=False,
195 ),
196)
198_DEFINITIONS_BY_FIELD = {definition.target_field: definition for definition in _FIELD_DEFINITIONS}
201@dataclass(frozen=True)
202class DeviceFieldReview:
203 """The review state and effective proposal for one matched Device."""
205 differing: dict[str, dict[str, str]]
206 ignored: dict[str, dict[str, str]]
207 informational: dict[str, dict[str, str]]
208 matching: dict[str, dict[str, str]]
209 effective_proposal: dict[str, Any]
210 snapshots: dict[str, tuple[dict[str, str], dict[str, str]]]
213class DeviceFieldReviewer:
214 """Apply exact persisted reviews to a matched Device proposal."""
216 def __init__(self, profile, ignored_records: Mapping[tuple[str, int, str], Any] | None = None):
217 self.profile = profile
218 self._ignored_records = dict(ignored_records or {})
219 review_device_ids: dict[str, set[int]] = {}
220 for source_id, device_id, _target_field in self._ignored_records:
221 review_device_ids.setdefault(source_id, set()).add(device_id)
222 self._review_device_ids = {
223 source_id: frozenset(device_ids) for source_id, device_ids in review_device_ids.items()
224 }
226 @classmethod
227 def for_profile(cls, profile):
228 """Load the profile's current review records once for an import run."""
229 from .models import IgnoredFieldDifference
231 records = IgnoredFieldDifference.objects.filter(profile=profile)
232 return cls(
233 profile,
234 {(_text(record.source_id), record.netbox_device_id, record.target_field): record for record in records},
235 )
237 @staticmethod
238 def definition(target_field: str) -> FieldDefinition | None:
239 """Return the registered definition for one target field."""
240 return _DEFINITIONS_BY_FIELD.get(target_field)
242 @staticmethod
243 def reviewable_fields() -> frozenset[str]:
244 """Return target fields that can be reviewed and ignored."""
245 return frozenset(d.target_field for d in _FIELD_DEFINITIONS if d.reviewable)
247 @staticmethod
248 def non_writable_fields() -> frozenset[str]:
249 """Return fields shown for information but not assigned by the writer."""
250 return frozenset(d.target_field for d in _FIELD_DEFINITIONS if not d.writable)
252 @staticmethod
253 def current_snapshot(matched_device, target_field: str) -> dict[str, str] | None:
254 """Return the current canonical NetBox snapshot for one registered field."""
255 definition = _DEFINITIONS_BY_FIELD.get(target_field)
256 if definition is None:
257 return None
258 value = definition.current_value(matched_device)
259 if target_field != "rack_name":
260 return definition.snapshot(value)
261 snapshot = definition.snapshot(value, _device_rack_display(matched_device))
262 return _scope_rack_canonical(snapshot, _device_rack_location_id(matched_device))
264 def review_device_ids(self, source_id: str) -> frozenset[int]:
265 """Return unique Device IDs bound to reviews for one source row."""
266 return self._review_device_ids.get(_text(source_id), frozenset())
268 @staticmethod
269 def field_differences(
270 matched_device,
271 proposal: Mapping[str, Any],
272 *,
273 display_overrides: Mapping[str, str] | None = None,
274 ):
275 """Return the writable differences and the reported-only ones as two maps."""
276 differing, informational, _, _ = DeviceFieldReviewer._compare(
277 matched_device,
278 proposal,
279 display_overrides=display_overrides,
280 )
281 return differing, informational
283 @staticmethod
284 def field_diff(
285 matched_device,
286 proposal: Mapping[str, Any],
287 *,
288 include_informational: bool = False,
289 display_overrides: Mapping[str, str] | None = None,
290 ):
291 """Return current differences without loading persisted review records."""
292 differing, informational = DeviceFieldReviewer.field_differences(
293 matched_device,
294 proposal,
295 display_overrides=display_overrides,
296 )
297 if include_informational:
298 return {**differing, **informational}
299 return differing
301 def review(
302 self,
303 source_id: str,
304 matched_device,
305 proposal: Mapping[str, Any],
306 *,
307 display_overrides: Mapping[str, str] | None = None,
308 ) -> DeviceFieldReview:
309 """Return differing, ignored, and write-safe values for one Device."""
310 differing, informational, matching, snapshots = self._compare(
311 matched_device,
312 proposal,
313 display_overrides=display_overrides,
314 )
315 effective = dict(proposal)
316 ignored = {}
317 for target_field, (file_snapshot, netbox_snapshot) in snapshots.items():
318 record = self._ignored_records.get((_text(source_id), matched_device.pk, target_field))
319 if record is None:
320 continue
321 if (
322 record.file_snapshot.get("canonical") == file_snapshot["canonical"]
323 and record.netbox_snapshot.get("canonical") == netbox_snapshot["canonical"]
324 ):
325 ignored[target_field] = {
326 "netbox": netbox_snapshot["display"],
327 "file": file_snapshot["display"],
328 }
329 definition = _DEFINITIONS_BY_FIELD[target_field]
330 effective[target_field] = definition.current_value(matched_device)
331 differing.pop(target_field, None)
332 informational.pop(target_field, None)
333 return DeviceFieldReview(
334 differing=differing,
335 ignored=ignored,
336 informational=informational,
337 matching=matching,
338 effective_proposal=effective,
339 snapshots=snapshots,
340 )
342 @staticmethod
343 def _compare(
344 matched_device,
345 proposal: Mapping[str, Any],
346 *,
347 display_overrides: Mapping[str, str] | None = None,
348 ):
349 """Compare a proposal to a Device using the private field registry."""
350 display_overrides = display_overrides or {}
351 differing: dict[str, dict[str, str]] = {}
352 informational: dict[str, dict[str, str]] = {}
353 matching: dict[str, dict[str, str]] = {}
354 snapshots: dict[str, tuple[dict[str, str], dict[str, str]]] = {}
355 for definition in _FIELD_DEFINITIONS:
356 if definition.target_field not in proposal or not definition.provided(proposal[definition.target_field]):
357 continue
358 file_value = proposal[definition.target_field]
359 netbox_value = definition.current_value(matched_device)
360 file_snapshot = definition.snapshot(file_value, display_overrides.get(definition.target_field))
361 netbox_override = None
362 if definition.target_field == "rack_name":
363 netbox_override = _device_rack_display(matched_device)
364 _scope_rack_canonical(file_snapshot, proposal.get("_rack_location_id"))
365 netbox_snapshot = definition.snapshot(netbox_value, netbox_override)
366 if definition.target_field == "rack_name":
367 _scope_rack_canonical(netbox_snapshot, _device_rack_location_id(matched_device))
368 if file_snapshot["canonical"] == netbox_snapshot["canonical"]:
369 # Recorded, not dropped: a sync has to be able to say what it leaves alone.
370 matching[definition.target_field] = {
371 "netbox": netbox_snapshot["display"],
372 "file": file_snapshot["display"],
373 }
374 continue
375 snapshots[definition.target_field] = (file_snapshot, netbox_snapshot)
376 values = {"netbox": netbox_snapshot["display"], "file": file_snapshot["display"]}
377 if definition.writable:
378 differing[definition.target_field] = values
379 else:
380 informational[definition.target_field] = values
381 return differing, informational, matching, snapshots
384#: What a sync would do to one field. The preview and the sync modal read these same names.
385SYNC_STATE_CHANGE = "change"
386SYNC_STATE_UNCHANGED = "unchanged"
387SYNC_STATE_IGNORED = "ignored"
388SYNC_STATE_NOT_WRITTEN = "not_written"
390#: The review bucket each state is built from, in the order the operator reads them.
391_SYNC_STATE_BUCKETS: tuple[tuple[str, str], ...] = (
392 (SYNC_STATE_CHANGE, "field_diff"),
393 (SYNC_STATE_UNCHANGED, "field_matching"),
394 (SYNC_STATE_IGNORED, "field_ignored"),
395 (SYNC_STATE_NOT_WRITTEN, "field_informational"),
396)
399def sync_change_preview(extra_data: Mapping[str, Any], labels: Mapping[str, str]) -> list[dict[str, str]]:
400 """Return one matched row's reviewed fields, each labelled with what a sync would do to it."""
401 entries = []
402 for state, bucket in _SYNC_STATE_BUCKETS:
403 values = extra_data.get(bucket) or {}
404 for target_field in sorted(values):
405 pair = values[target_field]
406 entries.append(
407 {
408 "field": target_field,
409 "label": labels.get(target_field) or target_field,
410 "netbox": pair.get("netbox", ""),
411 "file": pair.get("file", ""),
412 "state": state,
413 }
414 )
415 return entries