Coverage for netbox_data_import/trace_device_resolution.py: 98.19%
221 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"""Resolve Source Trace Device labels inside one permission-scoped import target."""
5from __future__ import annotations
7from collections.abc import Iterable, Mapping
8from dataclasses import dataclass
9from decimal import Decimal
10from functools import reduce
11from operator import or_
12from typing import Any
14from django.db.models import BigIntegerField, Case, F, IntegerField, Q, Value, When
16from .identity import CANONICAL_NAME, identity_in, identity_text, matching_search, with_name_identity
17from .trace_location_resolution import MAPPED, decided_location_mappings, site_locations, source_location_key
18from .values import normalize_for_compare, source_position
20AUTOMATICALLY_RESOLVED = "automatically resolved"
21MANUALLY_RESOLVED = "manually resolved"
22UNRESOLVED = "unresolved"
23STALE = "stale"
25_SERIALIZED_EVIDENCE_FIELDS = frozenset({"key", "labels", "locations", "racks", "u_positions"})
26_DEVICE_QUESTION_PRESENTATION_FIELDS = frozenset(
27 {"label", "state", "state_style", "selected", "selectable", "reason", "exact_match_count", "disclosure_source"}
28)
31@dataclass(frozen=True)
32class DeviceEvidence:
33 """Aggregate every source fact stated for one canonical Device label."""
35 key: str
36 labels: tuple[str, ...]
37 locations: tuple[str, ...]
38 racks: tuple[str, ...]
39 u_positions: tuple[str, ...]
41 @classmethod
42 def from_dict(cls, value: Mapping[str, Any]) -> DeviceEvidence:
43 """Restore evidence carried by an accepted Import Plan."""
44 if not isinstance(value, Mapping):
45 raise TypeError("Device evidence must be an object.")
46 missing = _SERIALIZED_EVIDENCE_FIELDS - value.keys()
47 if missing:
48 raise ValueError(f"Device evidence is missing fields: {', '.join(sorted(missing))}.")
49 unknown = value.keys() - _SERIALIZED_EVIDENCE_FIELDS - _DEVICE_QUESTION_PRESENTATION_FIELDS
50 if unknown:
51 raise ValueError(f"Device evidence has unknown fields: {', '.join(sorted(unknown))}.")
52 raw_key = value["key"]
53 if not isinstance(raw_key, str):
54 raise TypeError("Device evidence key must be a string.")
55 key = source_device_key(raw_key)
56 if not key:
57 raise ValueError("A Device question needs a source Device key.")
58 return cls(
59 key=key,
60 labels=_serialized_source_values(value, "labels"),
61 locations=_serialized_source_values(value, "locations"),
62 racks=_serialized_source_values(value, "racks"),
63 u_positions=_serialized_source_values(value, "u_positions"),
64 )
66 def to_dict(self) -> dict[str, Any]:
67 """Return the JSON-safe evidence stored in an Import Plan."""
68 return {
69 "key": self.key,
70 "labels": list(self.labels),
71 "locations": list(self.locations),
72 "racks": list(self.racks),
73 "u_positions": list(self.u_positions),
74 }
77@dataclass(frozen=True)
78class DeviceResolution:
79 """State how one source Device label resolves in NetBox."""
81 evidence: DeviceEvidence
82 state: str
83 device: Any | None
84 exact_match_count: int
85 reason: str = ""
87 def to_question(self) -> dict[str, Any]:
88 """Return the server-authored Device question carried by a trace plan."""
89 state_style = {
90 AUTOMATICALLY_RESOLVED: "auto",
91 MANUALLY_RESOLVED: "manual",
92 UNRESOLVED: "unresolved",
93 STALE: "stale",
94 }[self.state]
95 return {
96 **self.evidence.to_dict(),
97 "label": self.evidence.labels[0] if self.evidence.labels else self.evidence.key,
98 "state": self.state,
99 "state_style": state_style,
100 "selected": str(self.device) if self.device is not None else "",
101 "selectable": self.device is None,
102 "reason": self.reason,
103 "exact_match_count": self.exact_match_count,
104 }
107@dataclass(frozen=True)
108class CandidateFact:
109 """One source fact and the visible NetBox value it was compared with."""
111 fact: str
112 source: str
113 netbox: str
114 # Only a Location fact compares through a mapping, so only it names the mapped Location.
115 mapped: str = ""
117 def to_dict(self) -> dict[str, str]:
118 """Return the JSON-safe explanation the Device picker renders."""
119 return {"fact": self.fact, "source": self.source, "mapped": self.mapped, "netbox": self.netbox}
122@dataclass(frozen=True)
123class ImportLocationHint:
124 """The import-page Location that holds a candidate's placement Location."""
126 location: str
127 netbox: str
129 def to_dict(self) -> dict[str, str]:
130 """Return the JSON-safe hint the Device picker renders apart from source evidence."""
131 return {"location": self.location, "netbox": self.netbox}
134@dataclass(frozen=True)
135class DeviceCandidate:
136 """One visible Device and the source evidence that supports or contradicts it."""
138 device: Any
139 matched: tuple[CandidateFact, ...]
140 conflicting: tuple[CandidateFact, ...]
141 import_location: ImportLocationHint | None
144@dataclass(frozen=True)
145class DeviceCandidatePage:
146 """One bounded candidate page and its uncapped total."""
148 candidates: tuple[DeviceCandidate, ...]
149 total: int
152def source_device_key(label: str) -> str:
153 """Return the profile-wide identity of one source Device label."""
154 return identity_text(label)
157def _source_values(values: Iterable[str]) -> tuple[str, ...]:
158 """Return one source spelling per nonempty identity, in identity order."""
159 found: dict[str, str] = {}
160 for value in values:
161 if key := identity_text(value):
162 found.setdefault(key, value)
163 return tuple(found[key] for key in sorted(found))
166def _serialized_source_values(value: Mapping[str, Any], field: str) -> tuple[str, ...]:
167 """Validate and restore one list of source facts from an Import Plan."""
168 values = value[field]
169 if not isinstance(values, (list, tuple)) or any(not isinstance(item, str) for item in values):
170 raise TypeError(f"Device evidence {field} must be a list or tuple of strings.")
171 return _source_values(values)
174def _trace_references(trace) -> tuple:
175 """Return every Termination Reference carried by one Source Trace."""
176 summary = trace.endpoint_summary
177 references = [summary.from_termination, summary.to_termination]
178 for segment in trace.segments:
179 references.extend((segment.left, segment.right))
180 references.extend(trace.corroboration)
181 return tuple(references)
184def collect_trace_device_evidence(traces: Iterable[Any]) -> dict[str, DeviceEvidence]:
185 """Aggregate Device labels and placement hints across one Source Trace batch."""
186 collected: dict[str, dict[str, list[str]]] = {}
187 for trace in traces:
188 for reference in _trace_references(trace):
189 key = source_device_key(reference.device)
190 if not key:
191 continue
192 values = collected.setdefault(
193 key,
194 {"labels": [], "locations": [], "racks": [], "u_positions": []},
195 )
196 values["labels"].append(reference.device)
197 values["locations"].append(reference.location)
198 values["racks"].append(reference.rack)
199 values["u_positions"].append(reference.u_position)
200 return {
201 key: DeviceEvidence(
202 key=key,
203 labels=_source_values(values["labels"]),
204 locations=_source_values(values["locations"]),
205 racks=_source_values(values["racks"]),
206 u_positions=_source_values(values["u_positions"]),
207 )
208 for key, values in collected.items()
209 }
212def _target_devices(reader):
213 """Return visible Devices inside the Cable Target Module's selected Site."""
214 devices = reader.devices()
215 if reader.site is not None:
216 devices = devices.filter(site=reader.site)
217 return devices
220def resolve_trace_devices(
221 *,
222 profile,
223 reader,
224 evidence: Mapping[str, DeviceEvidence],
225 lock_rows: bool = False,
226) -> dict[str, DeviceResolution]:
227 """Resolve every source Device key with saved-choice precedence and bulk reads."""
228 from .models import TraceDeviceResolution, index_digest
230 keys = tuple(evidence)
231 if not keys:
232 return {}
233 stored_rows = TraceDeviceResolution.objects.filter(
234 profile=profile,
235 source_device_key_digest__in=[index_digest(key) for key in keys],
236 )
237 stored = {}
238 for row in stored_rows:
239 if row.source_device_key not in evidence:
240 raise ValueError("A Trace Device Resolution digest does not match its source Device key.")
241 stored[row.source_device_key] = row
243 # A stored key needs its name lookup too, so a stale choice can report the matches it now has.
244 devices = with_name_identity(_target_devices(reader)).filter(
245 Q(pk__in={row.selected_device_id for row in stored.values()}) | Q(identity_in("name", keys))
246 )
247 if lock_rows:
248 devices = devices.order_by("pk").select_for_update(of=("self",))
249 by_id = {}
250 by_name: dict[str, list[Any]] = {}
251 for device in devices:
252 by_id[device.pk] = device
253 by_name.setdefault(getattr(device, CANONICAL_NAME), []).append(device)
255 outcomes = {}
256 for key, facts in evidence.items():
257 saved = stored.get(key)
258 if saved is not None:
259 device = by_id.get(saved.selected_device_id)
260 if device is None:
261 outcomes[key] = DeviceResolution(
262 evidence=facts,
263 state=STALE,
264 device=None,
265 exact_match_count=len(by_name.get(key, ())),
266 reason="The saved Device is no longer available at this import target. Choose it again.",
267 )
268 else:
269 outcomes[key] = DeviceResolution(
270 evidence=facts,
271 state=MANUALLY_RESOLVED,
272 device=device,
273 exact_match_count=len(by_name.get(key, ())),
274 )
275 continue
276 matches = by_name.get(key, [])
277 device = matches[0] if len(matches) == 1 else None
278 outcomes[key] = DeviceResolution(
279 evidence=facts,
280 state=AUTOMATICALLY_RESOLVED if device is not None else UNRESOLVED,
281 device=device,
282 exact_match_count=len(matches),
283 reason=(
284 ""
285 if device is not None
286 else f"The source name matches {len(matches)} Devices at this import target. Choose the Device."
287 ),
288 )
289 return outcomes
292def resolved_trace_device(*, profile, reader, key: str, lock_rows: bool = False):
293 """Return the Device one source Device key resolves to, or None when its decision is open."""
294 if not key:
295 return None
296 evidence = {key: DeviceEvidence(key=key, labels=(), locations=(), racks=(), u_positions=())}
297 return resolve_trace_devices(profile=profile, reader=reader, evidence=evidence, lock_rows=lock_rows)[key].device
300def _site_racks(reader):
301 """Return the Racks the actor may view inside the selected Site."""
302 racks = reader.racks()
303 return racks.filter(site=reader.site) if reader.site is not None else racks
306def _flag(condition) -> Case:
307 """Return 1 for a Device row that meets *condition*, else 0."""
308 return Case(When(condition, then=Value(1)), default=Value(0), output_field=IntegerField())
311def _placement_location(reader) -> Case:
312 """Return a Device's visible placement Location: its own, else its visible Rack's when it has none."""
313 visible_locations = site_locations(reader).values("pk")
314 return Case(
315 When(location_id__in=visible_locations, then=F("location_id")),
316 When(
317 Q(location_id__isnull=True, rack_id__in=_site_racks(reader).values("pk"))
318 & Q(rack__location_id__in=visible_locations),
319 then=F("rack__location_id"),
320 ),
321 default=Value(None),
322 output_field=BigIntegerField(),
323 )
326def _in_subtree(location) -> Q:
327 """Match a visible placement inside *location*'s subtree, whatever lies between them."""
328 return Q(_ndi_placement_location__in=location.get_descendants(include_self=True).values("pk"))
331def _mapped_paths(profile, reader, evidence: DeviceEvidence) -> tuple[tuple[str, Any], ...]:
332 """Return each source Location path of *evidence* whose deciding row maps to a visible Location, with it."""
333 paths = {source_location_key(path): path for path in evidence.locations}
334 mappings = decided_location_mappings(profile=profile, reader=reader, paths=paths.values())
335 return tuple((paths[key], mapping.location) for key, mapping in sorted(mappings.items()) if mapping.state == MAPPED)
338def _with_evidence(devices, reader, evidence, *, exact_names, matching_racks, mapped_paths):
339 """Annotate the per-fact match and conflict flags that rank and explain each candidate."""
340 positions = [source_position(value) for value in evidence.u_positions]
341 positions = [Decimal(str(value)) for value in positions if value is not None]
342 visible_racks = _site_racks(reader).values("pk")
343 path_names = tuple(f"_ndi_location_path_{index}" for index in range(len(mapped_paths)))
344 devices = with_name_identity(devices).annotate(_ndi_placement_location=_placement_location(reader))
345 devices = devices.annotate(
346 **{name: _flag(_in_subtree(location)) for name, (_path, location) in zip(path_names, mapped_paths, strict=True)}
347 )
348 zero = Value(0, output_field=IntegerField())
349 location_match = location_conflict = zero
350 if path_names:
351 # Location scores once when any mapped path matches (section 6.1), and conflicts once when any does not.
352 location_match = _flag(reduce(or_, (Q(**{name: 1}) for name in path_names)))
353 location_conflict = _flag(
354 Q(_ndi_placement_location__isnull=False) & reduce(or_, (Q(**{name: 0}) for name in path_names))
355 )
356 devices = devices.annotate(
357 _ndi_exact_name=_flag(Q(**{f"{CANONICAL_NAME}__in": exact_names})),
358 _ndi_rack_match=_flag(Q(rack_id__in=matching_racks)) if matching_racks else zero,
359 _ndi_rack_conflict=(
360 _flag(Q(rack_id__in=visible_racks) & ~Q(rack_id__in=matching_racks))
361 if matching_racks
362 else _flag(Q(rack_id__in=visible_racks))
363 if evidence.racks
364 else zero
365 ),
366 _ndi_position_match=_flag(Q(position__in=positions)) if positions else zero,
367 _ndi_position_conflict=(
368 _flag(Q(position__isnull=False) & ~Q(position__in=positions))
369 if positions
370 else _flag(Q(position__isnull=False))
371 if evidence.u_positions
372 else zero
373 ),
374 _ndi_location_match=location_match,
375 _ndi_location_conflict=location_conflict,
376 _ndi_import_location=_flag(_in_subtree(reader.location)) if reader.location is not None else zero,
377 )
378 return devices.annotate(
379 _ndi_hint_score=F("_ndi_rack_match") + F("_ndi_location_match") + F("_ndi_position_match"),
380 _ndi_conflict_score=F("_ndi_rack_conflict") + F("_ndi_location_conflict") + F("_ndi_position_conflict"),
381 )
384def _explain(device, evidence, *, mapped_paths, rack_names, location_names, import_location) -> DeviceCandidate:
385 """Explain one ranked candidate from the same flags that ranked it, naming both values of each fact."""
386 matched: list[CandidateFact] = []
387 conflicting: list[CandidateFact] = []
388 if device._ndi_exact_name:
389 matched.append(CandidateFact("name", source=", ".join(evidence.labels), netbox=device.name))
390 placement = location_names.get(device._ndi_placement_location, "")
391 for index, (path, location) in enumerate(mapped_paths):
392 fact = CandidateFact("location", source=path, mapped=location.name, netbox=placement)
393 if getattr(device, f"_ndi_location_path_{index}"):
394 matched.append(fact)
395 elif device._ndi_placement_location is not None:
396 conflicting.append(fact)
397 rack = CandidateFact("rack", source=", ".join(evidence.racks), netbox=rack_names.get(device.rack_id, ""))
398 if device._ndi_rack_match:
399 matched.append(rack)
400 if device._ndi_rack_conflict:
401 conflicting.append(rack)
402 position = CandidateFact(
403 "U position", source=", ".join(evidence.u_positions), netbox=normalize_for_compare(device.position)
404 )
405 if device._ndi_position_match:
406 matched.append(position)
407 if device._ndi_position_conflict:
408 conflicting.append(position)
409 return DeviceCandidate(
410 device=device,
411 matched=tuple(matched),
412 conflicting=tuple(conflicting),
413 import_location=(
414 ImportLocationHint(location=import_location.name, netbox=placement) if device._ndi_import_location else None
415 ),
416 )
419def eligible_trace_devices(
420 *,
421 profile,
422 reader,
423 evidence: DeviceEvidence,
424 search: str = "",
425 limit: int,
426 offset: int = 0,
427 lock_rows: bool = False,
428) -> DeviceCandidatePage:
429 """Return a deterministic bounded page of visible Device candidates."""
430 devices = matching_search(_target_devices(reader), search)
431 wanted_racks = {identity_text(value) for value in evidence.racks} - {""}
432 matching_racks = set(_site_racks(reader).filter(identity_in("name", wanted_racks)).values_list("pk", flat=True))
433 mapped_paths = _mapped_paths(profile, reader, evidence)
434 devices = _with_evidence(
435 devices,
436 reader,
437 evidence,
438 exact_names={evidence.key},
439 matching_racks=matching_racks,
440 mapped_paths=mapped_paths,
441 ).order_by("-_ndi_exact_name", "-_ndi_hint_score", "_ndi_conflict_score", "-_ndi_import_location", "name", "pk")
442 total = devices.count()
443 selected: tuple[Any, ...] = ()
444 if offset < total and lock_rows:
445 ranked_ids = tuple(devices.values_list("pk", flat=True)[offset : offset + limit])
446 locked = {
447 device.pk: device
448 for device in devices.filter(pk__in=ranked_ids).order_by("pk").select_for_update(of=("self",))
449 }
450 selected = tuple(locked[pk] for pk in ranked_ids if pk in locked)
451 elif offset < total:
452 selected = tuple(devices[offset : offset + limit])
453 rack_names = dict(
454 _site_racks(reader).filter(pk__in={device.rack_id for device in selected}).values_list("pk", "name")
455 )
456 location_names = dict(
457 site_locations(reader)
458 .filter(pk__in={device._ndi_placement_location for device in selected})
459 .values_list("pk", "name")
460 )
461 return DeviceCandidatePage(
462 candidates=tuple(
463 _explain(
464 device,
465 evidence,
466 mapped_paths=mapped_paths,
467 rack_names=rack_names,
468 location_names=location_names,
469 import_location=reader.location,
470 )
471 for device in selected
472 ),
473 total=total,
474 )
477__all__ = (
478 "AUTOMATICALLY_RESOLVED",
479 "MANUALLY_RESOLVED",
480 "STALE",
481 "UNRESOLVED",
482 "CandidateFact",
483 "DeviceCandidate",
484 "DeviceCandidatePage",
485 "DeviceEvidence",
486 "DeviceResolution",
487 "ImportLocationHint",
488 "collect_trace_device_evidence",
489 "eligible_trace_devices",
490 "resolve_trace_devices",
491 "resolved_trace_device",
492 "source_device_key",
493)