Coverage for netbox_data_import/device_field_review.py: 100%
181 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# 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 effective_proposal: dict[str, Any]
209 snapshots: dict[str, tuple[dict[str, str], dict[str, str]]]
212class DeviceFieldReviewer:
213 """Apply exact persisted reviews to a matched Device proposal."""
215 def __init__(self, profile, ignored_records: Mapping[tuple[str, int, str], Any] | None = None):
216 self.profile = profile
217 self._ignored_records = dict(ignored_records or {})
218 review_device_ids: dict[str, set[int]] = {}
219 for source_id, device_id, _target_field in self._ignored_records:
220 review_device_ids.setdefault(source_id, set()).add(device_id)
221 self._review_device_ids = {
222 source_id: frozenset(device_ids) for source_id, device_ids in review_device_ids.items()
223 }
225 @classmethod
226 def for_profile(cls, profile):
227 """Load the profile's current review records once for an import run."""
228 from .models import IgnoredFieldDifference
230 records = IgnoredFieldDifference.objects.filter(profile=profile)
231 return cls(
232 profile,
233 {(_text(record.source_id), record.netbox_device_id, record.target_field): record for record in records},
234 )
236 @staticmethod
237 def definition(target_field: str) -> FieldDefinition | None:
238 """Return the registered definition for one target field."""
239 return _DEFINITIONS_BY_FIELD.get(target_field)
241 @staticmethod
242 def reviewable_fields() -> frozenset[str]:
243 """Return target fields that can be reviewed and ignored."""
244 return frozenset(d.target_field for d in _FIELD_DEFINITIONS if d.reviewable)
246 @staticmethod
247 def non_writable_fields() -> frozenset[str]:
248 """Return fields shown for information but not assigned by the writer."""
249 return frozenset(d.target_field for d in _FIELD_DEFINITIONS if not d.writable)
251 @staticmethod
252 def current_snapshot(matched_device, target_field: str) -> dict[str, str] | None:
253 """Return the current canonical NetBox snapshot for one registered field."""
254 definition = _DEFINITIONS_BY_FIELD.get(target_field)
255 if definition is None:
256 return None
257 value = definition.current_value(matched_device)
258 if target_field != "rack_name":
259 return definition.snapshot(value)
260 snapshot = definition.snapshot(value, _device_rack_display(matched_device))
261 return _scope_rack_canonical(snapshot, _device_rack_location_id(matched_device))
263 def review_device_ids(self, source_id: str) -> frozenset[int]:
264 """Return unique Device IDs bound to reviews for one source row."""
265 return self._review_device_ids.get(_text(source_id), frozenset())
267 @staticmethod
268 def field_differences(
269 matched_device,
270 proposal: Mapping[str, Any],
271 *,
272 display_overrides: Mapping[str, str] | None = None,
273 ):
274 """Return the writable differences and the reported-only ones as two maps."""
275 differing, informational, _ = DeviceFieldReviewer._compare(
276 matched_device,
277 proposal,
278 display_overrides=display_overrides,
279 )
280 return differing, informational
282 @staticmethod
283 def field_diff(
284 matched_device,
285 proposal: Mapping[str, Any],
286 *,
287 include_informational: bool = False,
288 display_overrides: Mapping[str, str] | None = None,
289 ):
290 """Return current differences without loading persisted review records."""
291 differing, informational = DeviceFieldReviewer.field_differences(
292 matched_device,
293 proposal,
294 display_overrides=display_overrides,
295 )
296 if include_informational:
297 return {**differing, **informational}
298 return differing
300 def review(
301 self,
302 source_id: str,
303 matched_device,
304 proposal: Mapping[str, Any],
305 *,
306 display_overrides: Mapping[str, str] | None = None,
307 ) -> DeviceFieldReview:
308 """Return differing, ignored, and write-safe values for one Device."""
309 differing, informational, snapshots = self._compare(
310 matched_device,
311 proposal,
312 display_overrides=display_overrides,
313 )
314 effective = dict(proposal)
315 ignored = {}
316 for target_field, (file_snapshot, netbox_snapshot) in snapshots.items():
317 record = self._ignored_records.get((_text(source_id), matched_device.pk, target_field))
318 if record is None:
319 continue
320 if (
321 record.file_snapshot.get("canonical") == file_snapshot["canonical"]
322 and record.netbox_snapshot.get("canonical") == netbox_snapshot["canonical"]
323 ):
324 ignored[target_field] = {
325 "netbox": netbox_snapshot["display"],
326 "file": file_snapshot["display"],
327 }
328 definition = _DEFINITIONS_BY_FIELD[target_field]
329 effective[target_field] = definition.current_value(matched_device)
330 differing.pop(target_field, None)
331 informational.pop(target_field, None)
332 return DeviceFieldReview(
333 differing=differing,
334 ignored=ignored,
335 informational=informational,
336 effective_proposal=effective,
337 snapshots=snapshots,
338 )
340 @staticmethod
341 def _compare(
342 matched_device,
343 proposal: Mapping[str, Any],
344 *,
345 display_overrides: Mapping[str, str] | None = None,
346 ):
347 """Compare a proposal to a Device using the private field registry."""
348 display_overrides = display_overrides or {}
349 differing: dict[str, dict[str, str]] = {}
350 informational: dict[str, dict[str, str]] = {}
351 snapshots: dict[str, tuple[dict[str, str], dict[str, str]]] = {}
352 for definition in _FIELD_DEFINITIONS:
353 if definition.target_field not in proposal or not definition.provided(proposal[definition.target_field]):
354 continue
355 file_value = proposal[definition.target_field]
356 netbox_value = definition.current_value(matched_device)
357 file_snapshot = definition.snapshot(file_value, display_overrides.get(definition.target_field))
358 netbox_override = None
359 if definition.target_field == "rack_name":
360 netbox_override = _device_rack_display(matched_device)
361 _scope_rack_canonical(file_snapshot, proposal.get("_rack_location_id"))
362 netbox_snapshot = definition.snapshot(netbox_value, netbox_override)
363 if definition.target_field == "rack_name":
364 _scope_rack_canonical(netbox_snapshot, _device_rack_location_id(matched_device))
365 if file_snapshot["canonical"] == netbox_snapshot["canonical"]:
366 continue
367 snapshots[definition.target_field] = (file_snapshot, netbox_snapshot)
368 values = {"netbox": netbox_snapshot["display"], "file": file_snapshot["display"]}
369 if definition.writable:
370 differing[definition.target_field] = values
371 else:
372 informational[definition.target_field] = values
373 return differing, informational, snapshots