Coverage for netbox_data_import/review_workspace.py: 98.77%
813 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"""Present an Import Plan, resolve explicit review commands, and persist review decisions."""
5from __future__ import annotations
7from collections import Counter
8from collections.abc import Callable, Mapping
9from dataclasses import dataclass, replace
10from functools import cached_property
11from types import MappingProxyType
12from typing import Any
14from .cable_disclosure import (
15 CABLE_SEGMENT_OVERRIDE_ROW,
16 DISCLOSURE_SOURCE,
17 FINDING_HIDDEN,
18 POLICY_HIDDEN,
19 POLICY_VISIBLE,
20 TERMINATION_SOURCES,
21)
22from .cable_policy import cable_media_family_label, cable_profile_label, cable_type_label
23from .cable_target import REUSE_SEGMENT, UNRESOLVED
24from .field_keys import TERMINATION_ROLE, is_outside_claim, parse_termination_field_key
25from .identity import identity_in, identity_text
26from .import_engine import ImportEngine
27from .models import (
28 CableClassMapping,
29 CableSegmentOverride,
30 ImportProfile,
31 PortClassMapping,
32 TerminationResolution,
33 TraceDeviceResolution,
34 TraceLocationResolution,
35 index_digest,
36 locked_profile_policy,
37)
38from .object_permissions import (
39 POLICY_WRITE_REFUSED,
40 assess_permission_scoped_save,
41 delete_permission_scoped_objects,
42 save_permission_scoped_object,
43)
44from .plan import (
45 Diagnostic,
46 Disposition,
47 ImportPlan,
48 PlannedChange,
49 Severity,
50 SynchronizationUnit,
51 missing_companions,
52)
53from .trace_device_resolution import source_device_key
54from .values import (
55 effective_device_name,
56 has_below_rack_position,
57 normalize_for_compare,
58 source_position,
59 source_text,
60 translation_maps,
61)
62from .public_refusal import PublicRefusal
65TERMINATION_UNRESOLVABLE = "That termination cannot be resolved here."
66# The mark a selection carries when its NetBox model is outside the source PortClass claim (section 10.2).
67OUTSIDE_CLAIM_MARK = "Differs from the source PortClass"
70class IneligibleDeviceSelection(Exception):
71 """The selected Device is absent from the candidates offered for this request."""
74def save_termination_resolution_and_replan(
75 *,
76 profile,
77 source_document,
78 actor,
79 planning_context,
80 task_type,
81 field_key,
82 source,
83 selected_object_type,
84 selected_object_id,
85 selected_display_name,
86 device,
87 reviewed_fingerprint,
88):
89 """Persist one manual termination selection with its source spelling and its Device, then replan."""
90 values = {
91 "source_device": source["device"],
92 "source_cards": source["cards"],
93 "source_port": source["port"],
94 "selected_object_type": selected_object_type,
95 "selected_object_id": selected_object_id,
96 "selected_display_name": selected_display_name,
97 }
98 with locked_profile_policy(profile.pk):
99 locked_profile = ImportProfile.objects.get(pk=profile.pk)
100 refuse_moved_policy(locked_profile, reviewed_fingerprint)
101 lookup = {
102 "profile": locked_profile,
103 "task_type": task_type,
104 "field_key": field_key,
105 }
106 candidate = TerminationResolution(**lookup, **values)
107 candidate.full_clean(validate_unique=False, validate_constraints=False)
108 save_permission_scoped_object(
109 actor,
110 TerminationResolution,
111 lookup,
112 values,
113 )
114 pin_trace_device(profile=locked_profile, actor=actor, source_device=source["device"], device=device)
115 # atomic-exit-safe: decision-saved-and-replanned
116 return ImportEngine.plan(locked_profile, source_document, actor, planning_context)
119def _save_trace_device_resolution(actor, lookup, values) -> None:
120 """Validate and save one Trace Device Resolution inside the actor's permission scope."""
121 candidate = TraceDeviceResolution(**lookup, **values)
122 candidate.full_clean(validate_unique=False, validate_constraints=False)
123 save_permission_scoped_object(actor, TraceDeviceResolution, lookup, values)
126def pin_trace_device(*, profile, actor, source_device, device) -> None:
127 """Save the Device a termination choice was made on, unless a decision already names its source Device.
129 A Device matched by name has no saved decision, so a rename would leave the saved port without its Device.
130 An operator who may not save the decision still saves the termination, which then follows the name.
131 """
132 key = source_device_key(source_device)
133 digest = index_digest(key)
134 if TraceDeviceResolution.objects.filter(profile=profile, source_device_key_digest=digest).exists():
135 return
136 lookup = {"profile": profile, "source_device_key": key, "source_device_key_digest": digest}
137 values = {
138 "source_device_label": source_device,
139 "selected_device_id": device.pk,
140 "selected_display_name": str(device),
141 }
142 if not assess_permission_scoped_save(actor, TraceDeviceResolution, lookup, values).allowed:
143 return
144 _save_trace_device_resolution(actor, lookup, values)
147class ProfilePolicyMoved(PublicRefusal):
148 """Another operator changed this profile's policy after the reviewed preview was planned."""
151PROFILE_POLICY_MOVED = (
152 "This profile's policy changed since this preview was planned. Re-read from NetBox, then make the decision again."
153)
156class UnacceptablePolicyDecision(Exception):
157 """A profile policy decision does not validate, or it would overwrite a row the actor cannot view."""
159 def __init__(self, errors):
160 self.errors = errors
161 super().__init__("; ".join(errors))
164def _refuse_blind_overwrite(actor, row) -> None:
165 """Refuse a blind policy write against a row this actor cannot read."""
166 if row is None or row.__class__.objects.restrict(actor, "view").filter(pk=row.pk).exists():
167 return
168 raise UnacceptablePolicyDecision([POLICY_WRITE_REFUSED])
171def refuse_moved_policy(locked_profile, reviewed_fingerprint) -> None:
172 """Refuse a decision made against a policy that has already moved under this preview.
174 A preview revision is per browser session, so it cannot see another operator's profile edit.
175 """
176 if locked_profile.planning_fingerprint != reviewed_fingerprint:
177 raise ProfilePolicyMoved(PROFILE_POLICY_MOVED)
180def _form_messages(form) -> list:
181 """Return every field and non-field message one refused policy form reports."""
182 return [message for messages in form.errors.values() for message in messages]
185def save_cable_class_mapping_and_replan(
186 *,
187 profile,
188 source_document,
189 actor,
190 planning_context,
191 cable_class,
192 data,
193 reviewed_fingerprint,
194):
195 """Persist one CableClass policy decision, then request a fresh Import Plan."""
196 from .forms import CableClassMappingForm
198 with locked_profile_policy(profile.pk):
199 locked_profile = ImportProfile.objects.get(pk=profile.pk)
200 lookup = {"profile": locked_profile, "cable_class": cable_class}
201 # The row is read under the lock, so the form validates what the write will replace.
202 stored = CableClassMapping.objects.filter(**lookup).first()
203 _refuse_blind_overwrite(actor, stored)
204 refuse_moved_policy(locked_profile, reviewed_fingerprint)
205 instance = stored or CableClassMapping(**lookup)
206 form = CableClassMappingForm({**data, "cable_class": cable_class}, instance=instance)
207 if not form.is_valid():
208 raise UnacceptablePolicyDecision(_form_messages(form))
209 save_permission_scoped_object(
210 actor,
211 CableClassMapping,
212 lookup,
213 {
214 "cable_type_resolved": form.instance.cable_type_resolved,
215 "cable_type": form.instance.cable_type,
216 "cable_profile_resolved": form.instance.cable_profile_resolved,
217 "cable_profile": form.instance.cable_profile,
218 },
219 )
220 # atomic-exit-safe: policy-saved-and-replanned
221 return ImportEngine.plan(locked_profile, source_document, actor, planning_context)
224def save_port_class_mapping_and_replan(
225 *,
226 profile,
227 source_document,
228 actor,
229 planning_context,
230 source_label,
231 port_class,
232 reviewed_fingerprint,
233):
234 """Persist the fixed PortClass one source label maps to on the profile, then request a fresh Import Plan."""
235 from .forms import PortClassMappingForm
237 with locked_profile_policy(profile.pk):
238 locked_profile = ImportProfile.objects.get(pk=profile.pk)
239 lookup = {"profile": locked_profile, "source_label": source_label}
240 stored = PortClassMapping.objects.filter(**lookup).first()
241 _refuse_blind_overwrite(actor, stored)
242 refuse_moved_policy(locked_profile, reviewed_fingerprint)
243 form = PortClassMappingForm(
244 {"source_label": source_label, "port_class": port_class},
245 instance=stored or PortClassMapping(**lookup),
246 )
247 if not form.is_valid():
248 raise UnacceptablePolicyDecision(_form_messages(form))
249 save_permission_scoped_object(actor, PortClassMapping, lookup, {"port_class": form.instance.port_class})
250 # atomic-exit-safe: port-class-mapping-saved-and-replanned
251 return ImportEngine.plan(locked_profile, source_document, actor, planning_context)
254def clear_port_class_mapping_and_replan(
255 *,
256 profile,
257 source_document,
258 actor,
259 planning_context,
260 source_label,
261 reviewed_fingerprint,
262):
263 """Drop the profile mapping of one source PortClass label, then request a fresh Import Plan."""
264 with locked_profile_policy(profile.pk):
265 locked_profile = ImportProfile.objects.get(pk=profile.pk)
266 stored = PortClassMapping.objects.filter(profile=locked_profile, source_label=source_label).first()
267 _refuse_blind_overwrite(actor, stored)
268 refuse_moved_policy(locked_profile, reviewed_fingerprint)
269 if stored is None:
270 raise UnacceptablePolicyDecision([f"The profile maps no PortClass label '{source_label}'."])
271 delete_permission_scoped_objects(actor, PortClassMapping.objects.filter(pk=stored.pk))
272 # atomic-exit-safe: port-class-mapping-cleared-and-replanned
273 return ImportEngine.plan(locked_profile, source_document, actor, planning_context)
276def save_cable_segment_override_and_replan(
277 *,
278 profile,
279 source_document,
280 actor,
281 planning_context,
282 segment_key,
283 trace_identity,
284 segment_index,
285 cable_class,
286 data,
287 reviewed_fingerprint,
288 breakout_leg,
289):
290 """Force one planned segment's Cable policy, then request a fresh Import Plan."""
291 from .forms import CableSegmentOverrideForm
293 with locked_profile_policy(profile.pk):
294 locked_profile = ImportProfile.objects.get(pk=profile.pk)
295 lookup = {"profile": locked_profile, "segment_key": segment_key}
296 # The row is read under the lock, so the form validates what the write will replace.
297 stored = CableSegmentOverride.objects.filter(**lookup).first()
298 deciding = stored or CableClassMapping.objects.filter(profile=locked_profile, cable_class=cable_class).first()
299 _refuse_blind_overwrite(actor, deciding)
300 refuse_moved_policy(locked_profile, reviewed_fingerprint)
301 instance = stored or CableSegmentOverride(**lookup)
302 instance.source_trace_identity = trace_identity
303 instance.segment_index = segment_index
304 form = CableSegmentOverrideForm(data, instance=instance, breakout_leg=breakout_leg)
305 if not form.is_valid():
306 raise UnacceptablePolicyDecision(_form_messages(form))
307 save_permission_scoped_object(
308 actor,
309 CableSegmentOverride,
310 lookup,
311 {
312 "cable_type": form.instance.cable_type,
313 "cable_profile": form.instance.cable_profile,
314 "source_trace_identity": trace_identity,
315 "segment_index": segment_index,
316 },
317 )
318 # atomic-exit-safe: segment-override-saved-and-replanned
319 return ImportEngine.plan(locked_profile, source_document, actor, planning_context)
322def clear_cable_segment_override_and_replan(
323 *,
324 profile,
325 source_document,
326 actor,
327 planning_context,
328 segment_key,
329 reviewed_fingerprint,
330):
331 """Drop one segment override, so the CableClass policy decides that segment again."""
332 with locked_profile_policy(profile.pk):
333 locked_profile = ImportProfile.objects.get(pk=profile.pk)
334 stored = CableSegmentOverride.objects.filter(profile=locked_profile, segment_key=segment_key).first()
335 _refuse_blind_overwrite(actor, stored)
336 refuse_moved_policy(locked_profile, reviewed_fingerprint)
337 if stored is not None:
338 delete_permission_scoped_objects(actor, CableSegmentOverride.objects.filter(pk=stored.pk))
339 # atomic-exit-safe: segment-override-cleared-and-replanned
340 return ImportEngine.plan(locked_profile, source_document, actor, planning_context)
343def save_trace_device_resolution_and_replan(
344 *,
345 profile,
346 source_document,
347 actor,
348 planning_context,
349 evidence,
350 selected_device_id,
351 search,
352 limit,
353 offset,
354 reviewed_fingerprint,
355):
356 """Persist one offered Device selection, then request a fresh Import Plan."""
357 from .netbox_reader import NetBoxReader
358 from .trace_device_resolution import eligible_trace_devices
360 with locked_profile_policy(profile.pk):
361 locked_profile = ImportProfile.objects.get(pk=profile.pk)
362 refuse_moved_policy(locked_profile, reviewed_fingerprint)
363 reader = NetBoxReader.for_actor(actor).for_planning_context(
364 planning_context, output_kinds=locked_profile.output_kinds
365 )
366 offered = eligible_trace_devices(
367 profile=locked_profile,
368 reader=reader,
369 evidence=evidence,
370 search=search,
371 limit=limit,
372 offset=offset,
373 lock_rows=True,
374 )
375 chosen = next(
376 (candidate.device for candidate in offered.candidates if candidate.device.pk == selected_device_id),
377 None,
378 )
379 if chosen is None:
380 raise IneligibleDeviceSelection("That Device is not one of the eligible candidates.")
381 lookup = {
382 "profile": locked_profile,
383 "source_device_key": evidence.key,
384 "source_device_key_digest": index_digest(evidence.key),
385 }
386 values = {
387 "source_device_label": evidence.labels[0] if evidence.labels else "",
388 "selected_device_id": chosen.pk,
389 "selected_display_name": str(chosen),
390 }
391 _save_trace_device_resolution(actor, lookup, values)
392 plan = ImportEngine.plan(locked_profile, source_document, actor, planning_context)
393 # atomic-exit-safe: device-decision-saved-and-replanned
394 return plan, chosen
397class IneligibleLocationSelection(Exception):
398 """The selected Location is not a visible Location of the selected Site."""
401def save_trace_location_resolution_and_replan(
402 *,
403 profile,
404 source_document,
405 actor,
406 planning_context,
407 source_location_path,
408 selected_location_id,
409 reviewed_fingerprint,
410):
411 """Map one source Location path to a visible Location of the selected Site, then replan."""
412 from .netbox_reader import NetBoxReader
413 from .trace_location_resolution import site_locations, source_location_key, stored_location_row
415 key = source_location_key(source_location_path)
416 with locked_profile_policy(profile.pk):
417 locked_profile = ImportProfile.objects.get(pk=profile.pk)
418 lookup = {
419 "profile": locked_profile,
420 "source_location_key": key,
421 "source_location_key_digest": index_digest(key),
422 }
423 stored = stored_location_row(locked_profile, key)
424 _refuse_blind_overwrite(actor, stored)
425 refuse_moved_policy(locked_profile, reviewed_fingerprint)
426 reader = NetBoxReader.for_actor(actor).for_planning_context(
427 planning_context, output_kinds=locked_profile.output_kinds
428 )
429 # The recheck locks the target, so it cannot leave the Site before this decision commits.
430 location = site_locations(reader).filter(pk=selected_location_id).select_for_update(of=("self",)).first()
431 if location is None:
432 raise IneligibleLocationSelection(f"Location {selected_location_id} is not a visible Location of the Site.")
433 values = {
434 "source_location_path": source_location_path,
435 "selected_location_id": location.pk,
436 "selected_display_name": str(location),
437 }
438 TraceLocationResolution(**lookup, **values).full_clean(validate_unique=False, validate_constraints=False)
439 save_permission_scoped_object(actor, TraceLocationResolution, lookup, values)
440 # atomic-exit-safe: location-mapping-saved-and-replanned
441 return ImportEngine.plan(locked_profile, source_document, actor, planning_context)
444def clear_trace_location_resolution_and_replan(
445 *,
446 profile,
447 source_document,
448 actor,
449 planning_context,
450 source_location_key,
451 reviewed_fingerprint,
452):
453 """Drop one source Location mapping, so the path gives no Location evidence, then replan."""
454 from .trace_location_resolution import stored_location_row
456 with locked_profile_policy(profile.pk):
457 locked_profile = ImportProfile.objects.get(pk=profile.pk)
458 stored = stored_location_row(locked_profile, source_location_key)
459 _refuse_blind_overwrite(actor, stored)
460 refuse_moved_policy(locked_profile, reviewed_fingerprint)
461 if stored is not None:
462 delete_permission_scoped_objects(actor, TraceLocationResolution.objects.filter(pk=stored.pk))
463 # atomic-exit-safe: location-mapping-cleared-and-replanned
464 return ImportEngine.plan(locked_profile, source_document, actor, planning_context)
467def _media_family_message(display) -> str:
468 """Return the wording for one media mismatch from its redacted segments."""
469 segments = display["segments"]
470 statements = []
471 for segment in segments:
472 position = segment["segment_index"] + 1
473 if not segment["visible"]:
474 hidden = POLICY_HIDDEN if segment["origin"] == "policy" else "a Cable you cannot view"
475 statements.append(f"segment {position} uses {hidden}")
476 continue
477 family = cable_media_family_label(segment["family"])
478 statement = f"segment {position} is {cable_type_label(segment['cable_type'])} ({family})"
479 if segment["retained"]:
480 statement += ", on the Cable this import keeps"
481 statements.append(statement)
482 remedy = (
483 "Correct those Cables in NetBox, then re-read."
484 if all(segment["retained"] for segment in segments)
485 else "Force the segment that states the wrong medium, or correct the source."
486 )
487 return f"Verified pass-throughs join these segments, and {'; '.join(statements)}. {remedy}"
490def _with_article(noun: str) -> str:
491 return f"{'an' if noun[:1] in 'aeiou' else 'a'} {noun}"
494def _kind_mismatch_message(display) -> str:
495 """Return the statement that the selected NetBox model differs from the kind the source PortClass claims."""
496 selected = _termination_model_name(display["selected_object_type"])
497 claimed = display["claimed_kind"].replace("_", " ")
498 return (
499 f"The selected NetBox port is {_with_article(selected)}, "
500 f"and the source PortClass '{display['port_class']}' claims {_with_article(claimed)}."
501 )
504def _channels(numbers) -> str:
505 return ", ".join(str(number) for number in numbers)
508def _harness_state_message(display) -> str:
509 """Return the wording that says which channels of one breakout this plan states."""
510 return (
511 f"{display['parent']} breaks out as one Cable. This source states channel(s) "
512 f"{_channels(display['legs'])} of {display['channels']}."
513 )
516def _missing_channels(display) -> str:
517 """Return the subject that names the channels a harness finding misses."""
518 return f"Channel(s) {_channels(display['missing'])}"
521def _harness_unsupported_message(display) -> str:
522 """Return the wording for a trace the import cannot write as one breakout Cable."""
523 # The constructor refuses a whole harness, so its finding names no single trace.
524 subject = f"The trace '{display['trace']}'" if "trace" in display else "A leg of this breakout"
525 return (
526 f"{subject} joins two channel subinterfaces, two breakouts, or a channel subinterface and its own parent, "
527 "or names the parent as an ordinary port. The import cannot write that as one breakout Cable."
528 )
531_CHANNEL_STATE_REPAIRS = {
532 "mirror": "does not mirror the Cable of its parent. Save the parent again in NetBox, or run manage.py trace_paths",
533 "termination": "holds its own Cable termination, which NetBox does not permit. Remove that Cable in NetBox",
534 "channel": "names a channel its parent does not have. Correct the channel in NetBox",
535}
538def _channel_state_message(display) -> str:
539 """Return the wording for one channel subinterface whose NetBox state contradicts its parent."""
540 repair = _CHANNEL_STATE_REPAIRS.get(display.get("fault", ""))
541 if repair is None:
542 return "A leg of this breakout is no longer a channel of its parent. Re-read the preview."
543 return f"Channel subinterface {display['port']} {repair}, then re-read."
546# Operator wording per code: a sentence, or a function of the presented display.
547_DIAGNOSTIC_MESSAGES: dict[str, str | Callable[[Mapping[str, Any]], str]] = {
548 "cable.ambiguous_mapped_peer": (
549 "NetBox maps this port to several peer ports. Choose the peer port this trace continues through."
550 ),
551 "cable.attribute_drift": "The existing Cable carries attributes this import would not have written.",
552 "cable.cableclass_unmapped": "No Cable policy maps this CableClass. Set the Cable policy for it.",
553 FINDING_HIDDEN: "A port this trace reaches is outside your view, so this preview does not describe it.",
554 "cable.harness_channel_gap": lambda display: (
555 f"{_missing_channels(display)} between the legs of this breakout have no leg. "
556 "NetBox numbers connectors by position, so a later leg would take the wrong connector. "
557 "Resolve the missing lane to its channel subinterface, or correct the source."
558 ),
559 "cable.harness_channel_state_invalid": _channel_state_message,
560 "cable.harness_leg_conflict": (
561 "Two legs of this breakout claim one channel or one far end. Resolve each lane to its own channel subinterface."
562 ),
563 "cable.harness_member_unchannelized": lambda display: (
564 f"The trace '{display['trace']}' names this breakout's card and reaches {display['port']}, which is not a "
565 "channel subinterface. If it is a lane of this breakout, channelize it in NetBox, then re-read."
566 ),
567 "cable.harness_member_unresolved": lambda display: (
568 f"The trace '{display['trace']}' names this breakout's card but is not resolved. "
569 "Resolve it, so the breakout Cable is written with every lane."
570 ),
571 "cable.harness_mixed_terminations": (
572 "The legs of this breakout reach ports of more than one kind, and NetBox cables one kind on each side "
573 "of a Cable. Make the source state the same first hop for every lane, for example its panel front port."
574 ),
575 "cable.harness_occupied": lambda display: (
576 "NetBox marks a port this breakout needs as connected without a Cable, so it cannot take one. "
577 "Clear Mark connected on that port in NetBox, then re-read."
578 if display.get("marked_connected")
579 else "Another Cable holds a port this breakout needs, and the import never edits a breakout Cable. "
580 "Correct that Cable in NetBox, then re-read."
581 ),
582 "cable.harness_partial": lambda display: (
583 f"{_missing_channels(display)} of this breakout have no leg in this source, so the breakout Cable leaves "
584 "them unused. A per-module export confirms only the legs it states. A leg added later needs the Cable "
585 "edited in NetBox, because the import never edits a breakout Cable."
586 ),
587 "cable.harness_profile_unavailable": lambda display: (
588 f"This NetBox instance offers no breakout Cable Profile for a parent with {display['channels']} channels."
589 ),
590 "cable.harness_state": _harness_state_message,
591 "cable.harness_type_conflict": (
592 "The legs of this breakout resolve to different Cable Types. "
593 "Make their CableClass policies or segment overrides agree."
594 ),
595 "cable.harness_unsupported": _harness_unsupported_message,
596 "cable.incompatible_terminations": (
597 "NetBox cannot cable these two terminations together. Choose another termination for the end that is wrong."
598 ),
599 "cable.media_family_mismatch": _media_family_message,
600 "cable.multi_termination_conflict": (
601 "A Cable with several terminations on one side holds a port this trace needs. Correct that Cable in NetBox."
602 ),
603 "cable.pass_through_not_mapped": (
604 "No PortMapping joins these two ports, so the stated pass-through cannot be true. "
605 "Correct the source, or add the PortMapping in NetBox."
606 ),
607 "cable.pass_through_verified": "A PortMapping proves the stated pass-through.",
608 "cable.permission_denied": "Permission denied: you cannot make one of the Cable changes this trace needs.",
609 "cable.planned_termination_conflict": lambda display: (
610 "Another segment of this trace plans a different Cable on this termination, and a termination holds "
611 "one Cable. Correct the source path."
612 if "competing_segment" in display
613 else "Another Source Trace plans a Cable on this termination. Resolve this trace to a different termination."
614 ),
615 "cable.policy_stale": "The selected Cable policy value is no longer offered by this NetBox instance.",
616 "cable.profile_incompatible": "The selected Cable Profile does not permit one termination on each side.",
617 "cable.resolved_segment_conflict": (
618 "Two Source Traces give one shared segment different Cable policies. "
619 "Force one policy on the segment, or make the CableClass policies agree."
620 ),
621 "cable.segment_override_lost": (
622 "This segment now resolves to different ports, so the Cable policy forced on it no longer applies."
623 ),
624 "cable.same_port_continuation": "A mapped peer port continues the path where the source repeats one port.",
625 "cable.segment_reused": "An existing Cable already proves this segment, so the import keeps it.",
626 "cable.segment_self_connection": "Both ends of this segment name one termination. Correct the source path.",
627 "cable.termination_channelized": (
628 "This Interface is channelized, so an ordinary Cable on it reaches no channel. "
629 "Choose the channel subinterface of this lane, so the import writes the breakout Cable on this Interface."
630 ),
631 "cable.termination_kind_mismatch": _kind_mismatch_message,
632 "cable.termination_not_connectable": (
633 "NetBox cannot cable this Interface because of its type. "
634 "Choose another termination for this end, or correct the Interface type in NetBox."
635 ),
636 "cable.termination_occupied": lambda display: (
637 "NetBox marks this termination as connected without a Cable, so it cannot take one. "
638 "Clear Mark connected on it in NetBox, or resolve this trace to a free termination."
639 if display.get("marked_connected")
640 else "Another Cable already occupies this termination. "
641 "Remove that Cable, or resolve this trace to a free termination."
642 ),
643 "cable.termination_unresolved": (
644 "No single port on the resolved Device matches this name. Choose the termination for it."
645 ),
646 "cable.unsupported_termination_kind": (
647 "The saved selection is not a kind of port this import can cable. Choose the termination again."
648 ),
649 "device.add_permission": "Permission denied: dcim.add_device",
650 "device.already_bound": "Another source row is already linked to this device.",
651 "device.ambiguous_asset_tag": "Multiple devices have this asset tag.",
652 "device.ambiguous_field_review": "Saved field reviews identify more than one device.",
653 "device.ambiguous_name": "Multiple devices have this name.",
654 "device.ambiguous_serial": "Multiple devices have this serial number.",
655 "device.ambiguous_stored_source_id": "More than one device stores this source ID.",
656 "device.contact_invalid": "The primary contact selection is invalid.",
657 "device.contact_permission": "Permission denied: cannot read or write this row's primary contact.",
658 "device.contact_resolution_required": "Choose the primary contact values before importing.",
659 "device.class_ignored": "Ignored source class",
660 "device.class_unmapped": "No class-to-role mapping exists for this source class.",
661 "device.cross_site_match": "A strong identity matches a device at another site.",
662 "device.device_type_missing": (
663 "Add the Device Type in NetBox, or map the source make and model to an existing Device Type."
664 ),
665 "device.device_type_slug_collision": "A stored device type already uses the slug this model derives.",
666 "device.device_type_mapping_ambiguous": (
667 "Device Type mappings with this make and model name different Device Types. "
668 "Delete one of them, or make them name one Device Type."
669 ),
670 "device.manufacturer_mapping_ambiguous": (
671 "Manufacturer mappings with this make name different manufacturers. "
672 "Delete one of them, or make them name one manufacturer."
673 ),
674 "device.duplicate_asset_tag": "The asset tag appears more than once in this import.",
675 "device.duplicate_name": "The device name appears more than once in this import.",
676 "device.duplicate_serial": "The serial number appears more than once in this import.",
677 "device.duplicate_source_id": "The source ID appears more than once in this import.",
678 "device.face_required": "A rack position needs a device face.",
679 "device.ignored": "Ignored device",
680 "device.inaccessible_match": "Permission denied: dcim.view_device",
681 "device.below_rack": "Skipped because the source position is below rack unit one.",
682 "device.change_permission": "Permission denied: dcim.change_device",
683 "device.missing_name": "Missing device name",
684 "device.name_placement_conflict": "The name matches a device at another placement.",
685 "device.name_unplaced_match": "The name matches a device with no recorded placement.",
686 "device.rack_ambiguous": "Multiple racks have this name at the import target.",
687 "device.rack_missing": "The target rack does not exist.",
688 "device.rack_position_claimed": "Another row claims this rack position.",
689 "device.rack_position_occupied": "The rack position is occupied.",
690 "device.rack_required": "A rack position needs a target rack.",
691 "device.role_permission": "Permission denied: dcim.add_devicerole",
692 "device.role_unconfigured": "No device role is configured for this source class.",
693 "device.zero_u_review_conflict": "A saved review keeps a rack position on a 0U device type.",
694 "device.unparseable_ip": "The source value is not a valid IP address.",
695 "device.validation_failed": "The planned device does not pass NetBox validation.",
696 "profile.dangling_reference": "The import profile names something NetBox no longer offers.",
697 "rack.add_permission": "Permission denied: dcim.add_rack",
698 "rack.change_permission": "Permission denied: dcim.change_rack",
699 "rack.inaccessible_match": "Permission denied: dcim.view_rack",
700 "rack.duplicate_name": "The rack name appears more than once in this import.",
701 "rack.duplicate_source_id": "The source ID appears more than once in this import.",
702 "rack.ignored": "Ignored rack",
703 "rack.missing_name": "Missing rack name",
704 "rack.ambiguous_name": "Multiple racks have this name at the import target.",
705 "rack.validation_failed": "The planned rack does not pass NetBox validation.",
706 "trace.device_unresolved": "No single Device matches this name. Choose the Device.",
707 "trace.device_resolution_stale": (
708 "The saved Device is no longer available at this import target. Choose the Device again."
709 ),
710 "trace.endpoint_evidence_only": (
711 "This trace states its two endpoints and no physical path, and no Cable joins them. "
712 "Add the Segment Evidence rows the path needs."
713 ),
714}
716_IDENTITY_CONFLICTS = {
717 "device.already_bound": "device_already_bound",
718 "device.ambiguous_asset_tag": "ambiguous_asset_tag",
719 "device.ambiguous_field_review": "ambiguous_field_review",
720 "device.ambiguous_name": "ambiguous_name",
721 "device.ambiguous_serial": "ambiguous_serial",
722 "device.ambiguous_stored_source_id": "ambiguous_source_id",
723 "device.cross_site_match": "cross_site_match",
724 "device.duplicate_asset_tag": "duplicate_asset_tag",
725 "device.duplicate_name": "duplicate_name",
726 "device.duplicate_serial": "duplicate_serial",
727 "device.duplicate_source_id": "duplicate_source_id",
728 "device.inaccessible_match": "permission_denied",
729 "device.name_placement_conflict": "name_placement_conflict",
730 "device.name_unplaced_match": "name_placement_conflict",
731 "device.rack_position_claimed": "rack_position_occupied",
732 "device.rack_position_occupied": "rack_position_occupied",
733 "rack.ambiguous_name": "ambiguous_rack",
734 "rack.duplicate_name": "duplicate_rack",
735 "rack.duplicate_source_id": "duplicate_source_id",
736 "rack.validation_failed": "rack_validation_failed",
737}
740# One table: the actions each identity conflict offers, and the extra_data keys each action runs on.
741_CONFLICT_ACTIONS = MappingProxyType(
742 {
743 "duplicate_name": (("use_name", ("suggested_name",)),),
744 "duplicate_serial": (("ignore_serial", ("duplicate_serial",)),),
745 "name_placement_conflict": (("use_name", ("suggested_name",)),),
746 "rack_position_occupied": (("ignore_position", ("u_position",)), ("ignore_row", ())),
747 }
748)
751def _offered_actions(extra_data, source_id) -> list[str]:
752 """Return the row actions the preview can run for one unit, in the order the table states them."""
753 if not source_id:
754 return []
755 offered: list[str] = []
756 for conflict in extra_data.get("identity_conflicts") or ():
757 for action, required in _CONFLICT_ACTIONS.get(conflict, ()):
758 if action not in offered and all(extra_data.get(key) for key in required):
759 offered.append(action)
760 return offered
763def _operation(unit: SynchronizationUnit) -> str:
764 """Return the preview action one unit displays."""
765 if unit.disposition == Disposition.ACTIONABLE:
766 return unit.changes[-1].operation if unit.changes else "update"
767 if unit.disposition == Disposition.NO_OP:
768 return "skip"
769 if unit.disposition == Disposition.EXCLUDED:
770 return "ignore"
771 return "error"
774def _object_type(unit: SynchronizationUnit) -> str:
775 """Return the target type one unit represents."""
776 if unit.changes:
777 return unit.changes[-1].target_module
778 return unit.identity.partition(":")[0]
781def _blocking(unit: SynchronizationUnit) -> list:
782 """Return the unit's error diagnostics: the first states the row, the rest are what it needs."""
783 return [item for item in unit.diagnostics if item.severity == Severity.ERROR]
786def _states_a_trace(unit: SynchronizationUnit) -> bool:
787 """Return whether one unit carries a Source Trace, which is what the workspace lists."""
788 return unit.display.get("trace") is not None
791def _diagnostic_message(diagnostic) -> str:
792 """Return the operator wording for one diagnostic, which must be a presentation copy."""
793 wording = _DIAGNOSTIC_MESSAGES.get(diagnostic.code, diagnostic.code)
794 message = str(diagnostic.display.get("message") or "") or (
795 wording(diagnostic.display) if callable(wording) else wording
796 )
797 if diagnostic.code == "cable.segment_override_lost":
798 source = diagnostic.display.get(DISCLOSURE_SOURCE)
799 if (
800 diagnostic.display.get(POLICY_VISIBLE) is True
801 and isinstance(source, dict)
802 and source.get("kind") == CABLE_SEGMENT_OVERRIDE_ROW
803 and type(source.get("pk")) is int
804 ):
805 return f"Stored Cable policy override {source['pk']}: {message}"
806 return message
809def _detail(unit: SynchronizationUnit, action: str, object_type: str, name: str) -> str:
810 """Return stable operator wording for one unit."""
811 if unit.diagnostics:
812 blocking = _blocking(unit)
813 return _diagnostic_message(blocking[0] if blocking else unit.diagnostics[0])
814 if detail := unit.display.get("detail"):
815 return str(detail)
816 verb = {
817 "create": "Would create",
818 "update": "Would update",
819 "skip": "No changes for",
820 "ignore": "Ignored",
821 }.get(action, action.title())
822 return f"{verb} {object_type} '{name}'"
825@dataclass(frozen=True)
826class WorkspaceUnit:
827 """View-supplied presentation data for one Synchronization Unit."""
829 identity: str
830 disposition: str
831 row_number: int | None
832 source_id: str
833 name: str
834 action: str
835 object_type: str
836 detail: str
837 rack_name: str
838 netbox_url: str
839 extra_data: dict[str, Any]
840 source_row: dict[str, Any]
842 @classmethod
843 def from_unit(cls, unit: SynchronizationUnit) -> WorkspaceUnit:
844 """Build presentation data without changing the accepted plan."""
845 unit_data = unit.to_dict()
846 display = unit_data["display"]
847 source_row = dict(display.get("source_row") or {})
848 object_type = _object_type(unit)
849 action = _operation(unit)
850 name = str(display.get("name") or display.get("device_name") or display.get("rack_name") or "")
851 rack_name = str(display.get("rack_name") or source_row.get("rack_name") or "")
852 source_id = str(display.get("source_id") or source_row.get("source_id") or "")
853 row_number = display.get("row_number", source_row.get("_row_number"))
854 extra_data = dict(display.get("extra_data") or {})
855 for key, value in source_row.items():
856 if key.startswith("_"):
857 continue
858 extra_data.setdefault(f"source_{key}", value)
859 if unit.changes:
860 change = unit.changes[-1]
861 change_data = change.to_dict()
862 extra_data.update({key: value for key, value in change_data["payload"].items() if key not in extra_data})
863 object_id = change.preconditions.get(f"{object_type}_id")
864 if object_id is not None:
865 extra_data.setdefault(f"netbox_{object_type}_id", object_id)
866 if unit.diagnostics:
867 diagnostic = unit.diagnostics[0]
868 diagnostic_display = diagnostic.to_dict()["display"]
869 structural = {"device_name", "extra_data", "message", "name", "object_type", "source_row"}
870 extra_data.update({key: value for key, value in diagnostic_display.items() if key not in structural})
871 if conflict := _IDENTITY_CONFLICTS.get(diagnostic.code):
872 extra_data.setdefault("identity_conflict", conflict)
873 rows = diagnostic_display.get("rows")
874 if rows:
875 row_key = {
876 "device.duplicate_asset_tag": "duplicate_asset_tag_rows",
877 "device.duplicate_serial": "duplicate_serial_rows",
878 "device.duplicate_source_id": "duplicate_source_id_rows",
879 "rack.duplicate_source_id": "duplicate_source_id_rows",
880 }.get(diagnostic.code)
881 if row_key:
882 extra_data[row_key] = [number for number in rows if number != row_number]
883 # The first error states the row; the rest are what it still needs. A warning is neither.
884 blocking = _blocking(unit)
885 extra_data["other_issues"] = [
886 {"code": item.code, "message": _diagnostic_message(item)} for item in blocking[1:]
887 ]
888 # A row action reads these facts, so a second problem is settled like a first one.
889 extra_data["identity_conflicts"] = [
890 conflict for item in blocking if (conflict := _IDENTITY_CONFLICTS.get(item.code))
891 ]
892 for item in blocking[1:]:
893 secondary = item.to_dict()["display"]
894 for key, value in secondary.items():
895 if key not in structural:
896 extra_data.setdefault(key, value)
897 for key, value in (secondary.get("extra_data") or {}).items():
898 extra_data.setdefault(key, value)
899 extra_data["offered_actions"] = _offered_actions(extra_data, source_id)
900 return cls(
901 identity=unit.identity,
902 disposition=unit.disposition,
903 row_number=row_number,
904 source_id=source_id,
905 name=name,
906 action=action,
907 object_type=object_type,
908 detail=_detail(unit, action, object_type, name),
909 rack_name=rack_name,
910 netbox_url=str(display.get("netbox_url") or ""),
911 extra_data=extra_data,
912 source_row=source_row,
913 )
916@dataclass(frozen=True)
917class AutoMatchSummary:
918 """Counts returned by the device auto-match review command."""
920 matched: int = 0
921 probable: int = 0
922 ambiguous: int = 0
923 placement_conflicts: int = 0
924 already: int = 0
925 skipped: int = 0
927 def message(self) -> str:
928 """Return the existing operator-facing summary."""
929 parts = []
930 if self.matched:
931 parts.append(f"{self.matched} auto-matched (serial/asset_tag/name)")
932 if self.probable:
933 parts.append(f"{self.probable} probable name match(es): use Link button to confirm")
934 if self.ambiguous:
935 parts.append(f"{self.ambiguous} ambiguous (multiple devices)")
936 if self.placement_conflicts:
937 parts.append(f"{self.placement_conflicts} placement conflict(s)")
938 if self.already:
939 parts.append(f"{self.already} already matched")
940 if self.skipped:
941 parts.append(f"{self.skipped} skipped (permission denied or concurrent change)")
942 return f"Auto-match: {', '.join(parts) or 'nothing found'}."
945def _resolve_strong_identity(devices, serial: str, asset_tag: str):
946 """Resolve serial and asset tag to one device, or report ambiguity."""
947 serial_matches = list(devices.filter(serial=serial)[:2]) if serial else []
948 asset_matches = list(devices.filter(identity_in("asset_tag", [identity_text(asset_tag)]))[:2]) if asset_tag else []
949 if len(serial_matches) > 1 or len(asset_matches) > 1:
950 return None, None, True
951 serial_device = serial_matches[0] if serial_matches else None
952 asset_device = asset_matches[0] if asset_matches else None
953 if serial_device is not None and asset_device is not None and serial_device.pk != asset_device.pk:
954 return None, None, True
955 if serial_device is not None:
956 return serial_device, "serial", False
957 if asset_device is not None:
958 return asset_device, "asset tag", False
959 return None, None, False
962def _match_existing_device(device_model, visible_devices, name, serial, asset_tag, site, tenant_id):
963 """Return one exact device match and its method, or report ambiguity."""
964 device, method, ambiguous = _resolve_strong_identity(device_model.objects, serial, asset_tag)
965 if ambiguous:
966 return None, None, True
967 if device is None and name:
968 tenant_filter = {"tenant_id": tenant_id} if tenant_id is not None else {"tenant__isnull": True}
969 matches = list(
970 device_model.objects.filter(site=site, **tenant_filter).filter(identity_in("name", [identity_text(name)]))[
971 :2
972 ]
973 )
974 if len(matches) > 1:
975 return None, None, True
976 if matches:
977 device, method = matches[0], "name"
978 if device is not None and (device.site_id != site.pk or not visible_devices.filter(pk=device.pk).exists()):
979 return None, None, True
980 return device, method, False
983def _device_placement_differs(device, source_location_id, rack_name, position, face) -> bool:
984 """Return whether source placement differs from one matched NetBox device."""
985 device_rack_name = device.rack.name if device.rack_id else ""
986 device_rack_location_id = device.rack.location_id if device.rack_id else None
987 return (
988 device.location_id != source_location_id
989 or (device.rack_id is not None and device_rack_location_id != source_location_id)
990 or identity_text(device_rack_name) != identity_text(rack_name)
991 or normalize_for_compare(device.position) != normalize_for_compare(position)
992 or (face is not None and (device.face or None) != face)
993 )
996# EXCLUDED is absent because only `cable_target` sets `display["trace"]`, and it assigns none.
997_SUMMARY_KEYS = {
998 Disposition.ACTIONABLE: "actionable",
999 Disposition.BLOCKED: "blocked",
1000 Disposition.INVALID: "invalid",
1001 Disposition.NO_OP: "no_change",
1002}
1005_SYNC_URL_NAME = "plugins:netbox_data_import:trace_sync"
1006_SYNC_ALL_URL_NAME = "plugins:netbox_data_import:trace_sync_all"
1007SYNC_ALL_NOTHING = "This preview has no actionable trace to synchronize."
1008SYNC_DEPENDENCY_HELD = "This trace depends on a change that cannot synchronize."
1009SYNC_SELECTION_CHANGED = (
1010 "The traces this sync includes changed since the page loaded. Reload the page and review them again."
1011)
1012SYNC_SELECTION_INVALID = "The sync request names no valid trace selection."
1013SYNC_NOTHING = "That trace has no changes to synchronize."
1014SYNC_COMPANIONS = "On the same breakout Cable as this trace"
1015SYNC_TOGETHER = "Synchronized together with this trace"
1016SYNC_DEPENDENCY_OWNERS = "Needed for a change this trace depends on"
1017BREAKOUT_PROFILE = "Set by the breakout"
1020@dataclass(frozen=True)
1021class TraceAction:
1022 """One review command, always visible, carrying its reason when it cannot run.
1024 `url_name` has no default: the page posts every action to it, so a command that named none
1025 would inherit whichever endpoint the template happened to hardcode.
1026 """
1028 key: str
1029 label: str
1030 enabled: bool
1031 url_name: str
1032 reason: str = ""
1035def _count(number: int, noun: str) -> str:
1036 return f"{number} {noun}{'' if number == 1 else 's'}"
1039@dataclass(frozen=True)
1040class SyncAllSelection:
1041 """What `Sync all` queues: each actionable trace that can sync, with every unit its changes wait on.
1043 `held_back` counts actionable traces left out because a unit they depend on cannot sync.
1044 """
1046 traces: tuple[str, ...]
1047 units: tuple[str, ...]
1048 blocked: int
1049 invalid: int
1050 held_back: int
1052 @property
1053 def label(self) -> str:
1054 """Return the button text, which states how many traces the command synchronizes."""
1055 return f"Sync {_count(len(self.traces), 'actionable trace')}"
1057 @property
1058 def action(self) -> TraceAction:
1059 """Return the command, disabled with its reason when no trace can sync."""
1060 return TraceAction(
1061 key="sync_all",
1062 label=self.label,
1063 enabled=bool(self.units),
1064 url_name=_SYNC_ALL_URL_NAME,
1065 reason="" if self.units else SYNC_ALL_NOTHING,
1066 )
1068 @property
1069 def unsynced_note(self) -> str:
1070 """Return how many traces the command leaves unsynced and why, or nothing when it leaves none."""
1071 parts = [
1072 f"{number} {why}"
1073 for number, why in (
1074 (self.blocked, "blocked"),
1075 (self.invalid, "invalid"),
1076 (self.held_back, "with a dependency that cannot sync"),
1077 )
1078 if number
1079 ]
1080 total = self.blocked + self.invalid + self.held_back
1081 if not total:
1082 return ""
1083 return f"{_count(total, 'trace')} {'stays' if total == 1 else 'stay'} unsynced: {', '.join(parts)}."
1086def held_sync(action: TraceAction, reason: str) -> TraceAction:
1087 """Refuse a sync action the view would reject, so the page cannot offer what the POST refuses."""
1088 return replace(action, enabled=False, reason=reason) if reason and action.enabled else action
1091def with_blocked_sync(trace: TraceWorkspaceUnit, reason: str) -> TraceWorkspaceUnit:
1092 """Hold the trace's own sync action for *reason*."""
1093 actions = tuple(held_sync(action, reason) if action.key == "sync" else action for action in trace.actions)
1094 return replace(trace, actions=actions)
1097@dataclass(frozen=True)
1098class SyncMember:
1099 """One other unit a trace's sync executes, and why: a companion, or one its changes wait on."""
1101 identity: str
1102 name: str
1103 # "companion" on a breakout the viewer sees, "together" on one they do not, or "dependency".
1104 kind: str
1105 is_trace: bool
1108CONFLICT_SEGMENT = "conflict"
1109# Findings that make a segment a conflict; only a finding the viewer may read can say so.
1110_CONFLICT_CODES = frozenset(
1111 {
1112 "cable.termination_occupied",
1113 "cable.multi_termination_conflict",
1114 "cable.planned_termination_conflict",
1115 "cable.resolved_segment_conflict",
1116 "cable.harness_occupied",
1117 }
1118)
1119# Findings that keep a resolved termination in attention, so its picker can choose another.
1120_REOPEN_CODES = frozenset(
1121 {"cable.termination_not_connectable", "cable.termination_channelized", "cable.incompatible_terminations"}
1122)
1125def _source_key(source) -> tuple:
1126 return (source.get("kind"), source.get("pk"))
1129def _presented_terminations(terminations, diagnostics) -> list[dict[str, Any]]:
1130 """Return each termination, in attention only while a finding the viewer may read still puts it there."""
1131 sources: set[tuple] = set()
1132 field_keys: set[Any] = set()
1133 for item in diagnostics:
1134 if item.code in _REOPEN_CODES:
1135 sources.update(_source_key(source) for source in item.display.get(TERMINATION_SOURCES) or ())
1136 field_keys.update(item.display.get(key) for key in ("left_field_key", "right_field_key"))
1137 return [
1138 {
1139 **item,
1140 "incompatible": bool(item["incompatible"])
1141 and (item["field_key"] in field_keys or _source_key(item.get(DISCLOSURE_SOURCE) or {}) in sources),
1142 "selected_model": _termination_model_name(item["selected_type"]),
1143 "outside_claim": OUTSIDE_CLAIM_MARK if _selection_outside_claim(item) else "",
1144 }
1145 for item in terminations
1146 ]
1149def _selection_outside_claim(item) -> bool:
1150 """Return whether one presented termination selected a model its source PortClass claim does not admit."""
1151 if not item["selected_type"]:
1152 return False
1153 parsed = parse_termination_field_key(item["field_key"])
1154 return parsed["role"] == TERMINATION_ROLE and is_outside_claim(parsed["kind"], item["selected_type"])
1157def _presented_segments(segments, diagnostics) -> list[dict[str, Any]]:
1158 """Return each segment as a conflict when a readable finding says so, and with the breakout profile of a leg."""
1159 state = next((item.display for item in diagnostics if item.code == "cable.harness_state"), None)
1160 conflicted = {item.display.get("segment_index") for item in diagnostics if item.code in _CONFLICT_CODES}
1161 # Each kept Cable carries its own reuse note, so a segment reads as kept only while that note is readable.
1162 reused = {item.display.get("segment_index") for item in diagnostics if item.code == "cable.segment_reused"}
1163 text = BREAKOUT_PROFILE
1164 # A hidden harness hides the profile its channel count selects, and a blocked harness selects none.
1165 if state is not None and state["profile"]:
1166 text = f"{BREAKOUT_PROFILE}: {cable_profile_label(state['profile'])}"
1167 return [
1168 {
1169 **segment,
1170 "breakout_profile": text if segment["breakout_leg"] else "",
1171 "status": CONFLICT_SEGMENT
1172 if segment["index"] in conflicted
1173 else ("" if segment["status"] == REUSE_SEGMENT and segment["index"] not in reused else segment["status"]),
1174 "retained": segment["retained"] and segment["index"] in reused,
1175 }
1176 for segment in segments
1177 ]
1180def _breakout_policies(policies, segments) -> list[dict[str, Any]]:
1181 """Return each CableClass decision, naming the breakout profile on a class that only the harness leg states."""
1182 leg = next((segment for segment in segments if segment["breakout_profile"]), None)
1183 ordinary = {segment["cable_class"] for segment in segments if segment is not leg}
1184 return [
1185 {
1186 **policy,
1187 "breakout_profile": "" if leg is None or policy["cable_class"] in ordinary else leg["breakout_profile"],
1188 }
1189 for policy in policies
1190 ]
1193def _termination_model_name(label: str) -> str:
1194 """Return the NetBox name of the model one resolved termination selected, or nothing while open."""
1195 if not label:
1196 return ""
1197 from django.apps import apps
1199 return str(apps.get_model(label)._meta.verbose_name)
1202@dataclass(frozen=True)
1203class TraceWorkspaceUnit:
1204 """One Source Trace as the review workspace shows it."""
1206 identity: str
1207 trace_identity: str
1208 disposition: str
1209 name: str
1210 row_number: int | None
1211 sheet: str
1212 endpoints: dict[str, str]
1213 segments: list[dict[str, Any]]
1214 cable_policies: list[dict[str, Any]]
1215 port_classes: list[dict[str, str]]
1216 logical_cable: dict[str, Any] | None
1217 deletes_logical_cable: bool
1218 resolution_started: bool
1219 topology_known: bool
1220 devices: list[dict[str, Any]]
1221 terminations: list[dict[str, Any]]
1222 findings: list[dict[str, str]]
1223 actions: tuple[TraceAction, ...]
1224 sync_members: tuple[SyncMember, ...] = ()
1226 @property
1227 def sync_selection(self) -> tuple[str, ...]:
1228 """Return what the sync form posts: this trace, then each member in the order the sync selects it."""
1229 return (self.identity, *(member.identity for member in self.sync_members))
1231 @property
1232 def sync_groups(self) -> tuple[tuple[str, tuple[SyncMember, ...]], ...]:
1233 """Return the members under why the sync needs them, leaving out an empty group."""
1234 groups = (
1235 (label, tuple(member for member in self.sync_members if member.kind == kind))
1236 for kind, label in (
1237 ("companion", SYNC_COMPANIONS),
1238 ("together", SYNC_TOGETHER),
1239 ("dependency", SYNC_DEPENDENCY_OWNERS),
1240 )
1241 )
1242 return tuple((label, members) for label, members in groups if members)
1244 @classmethod
1245 def from_unit(cls, unit: SynchronizationUnit) -> TraceWorkspaceUnit:
1246 """Build the workspace entry one trace unit states, without recomputing its plan."""
1247 display = unit.to_dict()["display"]
1248 workspace = dict(display.get("trace") or {})
1249 findings = [
1250 {"code": item.code, "message": _diagnostic_message(item), "severity": item.severity}
1251 for item in unit.diagnostics
1252 ]
1253 segments = _presented_segments(workspace.get("segments") or (), unit.diagnostics)
1254 detail = str(display.get("detail") or "")
1255 # "The path exists" repeats the reuse notes, so it stands only while each of them is readable.
1256 if any(
1257 stored["status"] == REUSE_SEGMENT and presented["status"] != REUSE_SEGMENT
1258 for stored, presented in zip(workspace.get("segments") or (), segments, strict=True)
1259 ):
1260 detail = SYNC_NOTHING
1261 return cls(
1262 identity=unit.identity,
1263 trace_identity=str(workspace.get("identity") or ""),
1264 disposition=unit.disposition,
1265 name=str(display.get("name") or ""),
1266 row_number=display.get("row_number"),
1267 sheet=str(display.get("sheet") or ""),
1268 endpoints=dict(workspace.get("endpoints") or {}),
1269 segments=segments,
1270 cable_policies=_breakout_policies(workspace["cable_policies"], segments),
1271 port_classes=[dict(item) for item in workspace.get("port_classes") or ()],
1272 logical_cable=workspace.get("logical_cable"),
1273 deletes_logical_cable=bool(workspace.get("deletes_logical_cable")),
1274 resolution_started=bool(workspace.get("resolution_started")),
1275 topology_known=bool(workspace.get("topology_known")),
1276 devices=[dict(item) for item in workspace.get("devices") or ()],
1277 terminations=_presented_terminations(workspace.get("terminations") or (), unit.diagnostics),
1278 findings=findings,
1279 actions=cls._actions(unit, findings, detail),
1280 )
1282 @staticmethod
1283 def _actions(unit: SynchronizationUnit, findings: list[dict[str, str]], detail: str) -> tuple[TraceAction, ...]:
1284 """Return every review command, each stating why it cannot run when it cannot."""
1285 blocking = next((item["message"] for item in findings if item["severity"] == Severity.ERROR), "")
1286 if unit.disposition == Disposition.ACTIONABLE:
1287 sync = TraceAction(key="sync", label="Sync with dependencies", enabled=True, url_name=_SYNC_URL_NAME)
1288 else:
1289 reason = blocking or detail or f"This trace is {unit.disposition}."
1290 sync = TraceAction(
1291 key="sync", label="Sync with dependencies", enabled=False, url_name=_SYNC_URL_NAME, reason=reason
1292 )
1293 return (sync,)
1296def _present_object_reviews(units, viewer) -> tuple:
1297 """Hide cached object facts when the viewer can no longer read their Device or Rack."""
1298 from dcim.models import Device, Rack
1300 # Use the same projection as rendering, including changes and secondary diagnostics.
1301 projections = [WorkspaceUnit.from_unit(unit) for unit in units]
1302 references = [projection.extra_data for projection in projections]
1303 visible_ids = {}
1304 for object_type, model in (("device", Device), ("rack", Rack)):
1305 key = f"netbox_{object_type}_id"
1306 reviewed_ids = {data[key] for data in references if data.get(key) is not None}
1307 visible_ids[object_type] = set(
1308 model.objects.restrict(viewer, "view").filter(pk__in=reviewed_ids).values_list("pk", flat=True)
1309 )
1310 presented = []
1311 for unit, projection, references_for_unit in zip(units, projections, references, strict=True):
1312 hidden_type = next(
1313 (
1314 object_type
1315 for object_type, ids in visible_ids.items()
1316 if (object_id := references_for_unit.get(f"netbox_{object_type}_id")) is not None
1317 and object_id not in ids
1318 ),
1319 None,
1320 )
1321 if hidden_type is None:
1322 presented.append((unit, projection))
1323 continue
1324 # Only source evidence survives: details and field snapshots can name the hidden object.
1325 display = {
1326 key: value
1327 for key, value in unit.display.items()
1328 if key in {"object_type", "row_number", "source_id", "name", "rack_name", "source_row"}
1329 }
1330 redacted = replace(
1331 unit,
1332 disposition=Disposition.INVALID,
1333 changes=(),
1334 display=display,
1335 diagnostics=(
1336 Diagnostic(
1337 code=f"{hidden_type}.inaccessible_match",
1338 severity=Severity.ERROR,
1339 identities=(unit.identity,),
1340 display=display,
1341 ),
1342 ),
1343 )
1345 presented.append((redacted, WorkspaceUnit.from_unit(redacted)))
1346 return tuple(presented)
1349class ReviewWorkspace:
1350 """Read-only presentation of the accepted Import Plan."""
1352 def __init__(self, plan: ImportPlan, viewer):
1353 if viewer is None:
1354 raise TypeError("ReviewWorkspace requires a live viewer.")
1355 self.plan = plan
1356 self._viewer = viewer
1358 @cached_property
1359 def _presentation(self) -> tuple:
1360 """Build redacted units and their projections only when a command needs presentation."""
1361 from .cable_disclosure import present_units
1363 return _present_object_reviews(present_units(self.plan.units, self._viewer), self._viewer)
1365 @cached_property
1366 def _presentation_units(self) -> tuple:
1367 """Return the viewer's redacted plan units."""
1368 return tuple(unit for unit, _projection in self._presentation)
1370 @cached_property
1371 def units(self) -> tuple[WorkspaceUnit, ...]:
1372 """Return the presentation of every unit, in plan order."""
1373 return tuple(projection for _unit, projection in self._presentation)
1375 @classmethod
1376 def from_dict(cls, data: dict, viewer) -> ReviewWorkspace:
1377 """Restore a workspace from a serialized Import Plan."""
1378 return cls(ImportPlan.from_dict(data), viewer)
1380 @property
1381 def counts(self) -> MappingProxyType:
1382 """Return preview counts in the existing template vocabulary."""
1383 counts: dict[str, int] = {}
1384 for unit in self.units:
1385 if unit.action == "error":
1386 key = "errors"
1387 elif unit.action == "skip":
1388 key = "skipped"
1389 elif unit.action == "ignore":
1390 key = "ignored"
1391 elif unit.action in {"create", "update"}:
1392 key = f"{unit.object_type}s_{unit.action}d"
1393 else:
1394 continue
1395 counts[key] = counts.get(key, 0) + 1
1396 return MappingProxyType(counts)
1398 @property
1399 def has_errors(self) -> bool:
1400 """Return whether any unit cannot execute."""
1401 return any(unit.action == "error" for unit in self.units)
1403 @property
1404 def has_traces(self) -> bool:
1405 """Return whether the plan holds a Source Trace, without building one workspace entry."""
1406 return any(_states_a_trace(unit) for unit in self.plan.units)
1408 @cached_property
1409 def traces(self) -> tuple[TraceWorkspaceUnit, ...]:
1410 """Return one workspace entry per Source Trace, in plan order.
1412 Cached because one page reads it twice, and each build reserializes every change.
1413 """
1414 traces = []
1415 for unit in self._presentation_units:
1416 if not _states_a_trace(unit):
1417 continue
1418 trace = TraceWorkspaceUnit.from_unit(unit)
1419 if unit.disposition == Disposition.ACTIONABLE:
1420 closure = self._sync_closure(unit.identity)
1421 if self.cannot_sync(tuple(closure)):
1422 trace = with_blocked_sync(trace, SYNC_DEPENDENCY_HELD)
1423 else:
1424 trace = replace(trace, sync_members=self._sync_members(unit.identity, closure))
1425 traces.append(trace)
1426 return tuple(traces)
1428 @cached_property
1429 def port_class_labels(self) -> MappingProxyType:
1430 """Return each PortClass label outside the fixed vocabulary the preview states, with its mapped PortClass."""
1431 labels: dict[str, str] = {}
1432 for trace in self.traces:
1433 for item in trace.port_classes:
1434 labels.setdefault(item["label"], item["port_class"])
1435 return MappingProxyType(dict(sorted(labels.items())))
1437 @cached_property
1438 def termination_sources(self) -> MappingProxyType:
1439 """Return each field source, or None when its occurrences state different spellings."""
1440 sources: dict[str, dict | None] = {}
1441 for trace in self.traces:
1442 for item in trace.terminations:
1443 key, source = item["field_key"], item["source"]
1444 if key not in sources:
1445 sources[key] = source
1446 elif sources[key] != source:
1447 sources[key] = None
1448 return MappingProxyType(sources)
1450 def sync_selection(self, identity: str) -> tuple[str, ...]:
1451 """Return the unit, its companions, and transitively one owner of each change they wait on and do not carry.
1453 The engine refuses a selection that lacks a dependency or a companion, so a review command that
1454 synchronizes one trace has to carry both, visibly: the page lists them and the POST must match.
1455 """
1456 return tuple(self._sync_closure(identity))
1458 def _sync_members(self, identity: str, closure: Mapping[str, bool]) -> tuple[SyncMember, ...]:
1459 """Return every unit of *closure* other than *identity*, by the name the trace list shows."""
1460 return tuple(
1461 SyncMember(
1462 identity=member,
1463 name=str(self._presented[member].display.get("name") or ""),
1464 kind=self._member_kind(identity, member, companion),
1465 is_trace=_states_a_trace(self._presented[member]),
1466 )
1467 for member, companion in closure.items()
1468 if member != identity
1469 )
1471 def _member_kind(self, identity: str, member: str, companion: bool) -> str:
1472 """Name a companion's breakout only when the viewer sees the harness of both legs."""
1473 if not companion:
1474 return "dependency"
1475 return "companion" if self._shows_harness(identity) and self._shows_harness(member) else "together"
1477 def _shows_harness(self, identity: str) -> bool:
1478 """Return whether presentation kept the unit's harness state, which hides with any port of the harness."""
1479 return any(item.code == "cable.harness_state" for item in self._presented[identity].diagnostics)
1481 def _sync_closure(self, identity: str) -> dict[str, bool]:
1482 """Return `sync_selection` in order, each unit mapped to whether it joins as a companion of *identity*."""
1483 selected = self.plan.unit(identity)
1484 if selected is None or selected.disposition != Disposition.ACTIONABLE:
1485 return {}
1486 chosen: dict[str, bool] = {}
1487 carried: set[str] = set()
1488 queue: list[str] = []
1490 def choose(first: str, companion: bool) -> None:
1491 pending = [first]
1492 while pending:
1493 unit = pending.pop()
1494 if unit in chosen:
1495 continue
1496 chosen[unit] = companion
1497 carried.update(change.identity for change in self._planned_changes(unit))
1498 queue.append(unit)
1499 pending.extend(reversed(self._presented[unit].selected_with))
1501 choose(identity, True)
1502 while queue:
1503 for change in self._planned_changes(queue.pop()):
1504 for dependency in change.dependencies:
1505 if dependency in carried:
1506 continue
1507 # Identical changes are shared (section 4.4): take an owner whose own closure can sync.
1508 owners = self._change_owners.get(dependency, ())
1509 owner = next((unit for unit in owners if unit in self._executable_units), None)
1510 if owner is not None:
1511 choose(owner, False)
1512 return chosen
1514 @cached_property
1515 def _change_owners(self) -> dict[str, tuple[str, ...]]:
1516 """Return every unit that carries each planned change, in plan order, built once for every sync check."""
1517 owners: dict[str, list[str]] = {}
1518 for unit in self.plan.units:
1519 for change in unit.changes:
1520 owners.setdefault(change.identity, []).append(unit.identity)
1521 return {identity: tuple(units) for identity, units in owners.items()}
1523 @cached_property
1524 def _executable_units(self) -> frozenset[str]:
1525 """Return the largest set of actionable units that carries every dependency and holds every companion.
1527 A unit outside it has no selection that `cannot_sync` accepts, so no closure may pick it as an owner.
1528 """
1529 units = {identity for identity, unit in self._presented.items() if unit.disposition == Disposition.ACTIONABLE}
1530 owners_left: dict[str, int] = {}
1531 dependents: dict[str, list[str]] = {}
1532 named_by: dict[str, list[str]] = {}
1533 for identity in units:
1534 for change in self._planned_changes(identity):
1535 owners_left[change.identity] = owners_left.get(change.identity, 0) + 1
1536 for dependency in change.dependencies:
1537 dependents.setdefault(dependency, []).append(identity)
1538 for member in self._presented[identity].selected_with:
1539 named_by.setdefault(member, []).append(identity)
1540 stuck = [
1541 identity
1542 for dependency, waiting in dependents.items()
1543 if not owners_left.get(dependency)
1544 for identity in waiting
1545 ]
1546 stuck.extend(identity for member, naming in named_by.items() if member not in units for identity in naming)
1547 while stuck:
1548 identity = stuck.pop()
1549 if identity not in units:
1550 continue
1551 units.remove(identity)
1552 stuck.extend(named_by.get(identity, ()))
1553 for change in self._planned_changes(identity):
1554 owners_left[change.identity] -= 1
1555 if not owners_left[change.identity]:
1556 stuck.extend(dependents.get(change.identity, ()))
1557 return frozenset(units)
1559 @cached_property
1560 def _presented(self) -> dict[str, SynchronizationUnit]:
1561 """Return each unit as the viewer sees it, by identity."""
1562 return {unit.identity: unit for unit in self._presentation_units}
1564 def _planned_changes(self, identity: str) -> tuple[PlannedChange, ...]:
1565 """Return the changes of one plan unit, read from the plan's own index."""
1566 unit = self.plan.unit(identity)
1567 if unit is None:
1568 raise KeyError(identity)
1569 return unit.changes
1571 def cannot_sync(self, selection: tuple[str, ...]) -> bool:
1572 """Return whether the engine refuses *selection*.
1574 It refuses a unit the viewer does not see as actionable, a dependency that no unit in the
1575 selection carries, and a companion the selection lacks. Sync all, the per-trace action and its
1576 POST all ask this one question.
1577 """
1578 carried = {change.identity for identity in selection for change in self._planned_changes(identity)}
1579 return bool(missing_companions(self._presented, selection)) or not all(
1580 self._presented[identity].disposition == Disposition.ACTIONABLE
1581 and all(set(change.dependencies) <= carried for change in self._planned_changes(identity))
1582 for identity in selection
1583 )
1585 @cached_property
1586 def sync_all(self) -> SyncAllSelection:
1587 """Return every actionable trace whose dependency closure can execute, in plan order.
1589 The engine refuses a unit that is not actionable and a dependency the selection lacks, so a
1590 trace that would need either is left out and counted, and never stops the others.
1591 """
1592 traces: list[str] = []
1593 chosen: set[str] = set()
1594 held_back = 0
1595 for trace in self.traces:
1596 if trace.disposition != Disposition.ACTIONABLE:
1597 continue
1598 closure = self.sync_selection(trace.identity)
1599 if closure and not self.cannot_sync(closure):
1600 traces.append(trace.identity)
1601 chosen.update(closure)
1602 else:
1603 held_back += 1
1604 return SyncAllSelection(
1605 traces=tuple(traces),
1606 units=tuple(unit.identity for unit in self.plan.units if unit.identity in chosen),
1607 blocked=sum(1 for trace in self.traces if trace.disposition == Disposition.BLOCKED),
1608 invalid=sum(1 for trace in self.traces if trace.disposition == Disposition.INVALID),
1609 held_back=held_back,
1610 )
1612 @property
1613 def trace_dispositions(self) -> tuple[tuple[str, int], ...]:
1614 """Return each disposition the trace list holds with its trace count, in summary order."""
1615 counts = Counter(trace.disposition for trace in self.traces)
1616 return tuple((disposition, counts[disposition]) for disposition in _SUMMARY_KEYS if counts[disposition])
1618 @property
1619 def trace_summary(self) -> dict[str, int]:
1620 """Return the summary strip: what the reviewer still has to work through."""
1621 traces = self.traces
1622 summary = {
1623 "traces": len(traces),
1624 "unresolved_terminations": 0,
1625 "resolved_terminations": 0,
1626 }
1627 # A template cannot resolve a key with a hyphen, so the strip names each disposition itself.
1628 for disposition, key in _SUMMARY_KEYS.items():
1629 summary[key] = sum(1 for trace in traces if trace.disposition == disposition)
1630 for trace in traces:
1631 for termination in trace.terminations:
1632 key = (
1633 "unresolved_terminations"
1634 if termination.get("state", UNRESOLVED) == UNRESOLVED
1635 else "resolved_terminations"
1636 )
1637 summary[key] += 1
1638 return summary
1640 @property
1641 def rack_groups(self) -> dict:
1642 """Group Rack and Device units for the existing rack-card view."""
1643 groups: dict[str, dict[str, Any]] = {}
1644 for unit in self.units:
1645 if unit.object_type == "rack":
1646 if not unit.name:
1647 continue
1648 groups.setdefault(unit.name, {"rack_row": None, "devices": []})["rack_row"] = unit
1649 elif unit.object_type == "device":
1650 groups.setdefault(unit.rack_name or "(No rack)", {"rack_row": None, "devices": []})["devices"].append(
1651 unit
1652 )
1654 def placement_sort_key(unit):
1655 position = source_position(unit.extra_data.get("u_position"))
1656 return position is None, position or 0
1658 for group in groups.values():
1659 group["devices"].sort(key=placement_sort_key)
1660 return groups
1662 @property
1663 def source_rows(self) -> list[dict[str, Any]]:
1664 """Return the source rows carried as display data, once per row number."""
1665 rows: dict[int, dict[str, Any]] = {}
1666 for unit in self.units:
1667 if unit.row_number is not None and unit.source_row:
1668 rows.setdefault(unit.row_number, dict(unit.source_row))
1669 return [rows[number] for number in sorted(rows)]
1671 @property
1672 def unused_columns(self) -> list[dict[str, Any]]:
1673 """Return unmapped source columns carried by plan diagnostics."""
1674 columns = [
1675 dict(diagnostic.display)
1676 for diagnostic in self.plan.diagnostics
1677 if diagnostic.code.endswith(".unused_column")
1678 ]
1679 columns.sort(key=lambda column: -int(column.get("count") or 0))
1680 return columns
1682 def with_units(self, units) -> ReviewWorkspace:
1683 """Return a presentation-only copy with replaced units."""
1684 workspace = object.__new__(type(self))
1685 workspace.plan = self.plan
1686 workspace._presentation_units = self._presentation_units
1687 workspace.units = tuple(units)
1688 return workspace
1690 @staticmethod
1691 def replace_unit(unit: WorkspaceUnit, **values) -> WorkspaceUnit:
1692 """Return a presentation-only copy of one unit."""
1693 return replace(unit, **values)
1696def auto_match_devices(workspace, profile, actor, target) -> AutoMatchSummary: # noqa: C901
1697 """Save safe exact device matches for every eligible source row of one reviewed plan.
1699 A preview command runs this under the Preview Coordinator, which replans afterwards.
1700 """
1701 from dcim.models import Device
1702 from django.core.exceptions import ValidationError
1703 from django.db import IntegrityError
1705 from .models import DeviceExistingMatch
1706 from .object_permissions import ObjectPermissionDenied, save_permission_scoped_object
1708 site = target["site"]
1709 tenant_id = target["tenant"].pk if target["tenant"] else None
1710 visible_devices = Device.objects.restrict(actor, "view")
1711 ignored_source_ids = set(profile.ignored_devices.values_list("source_id", flat=True))
1712 class_mappings = {mapping.source_class: mapping for mapping in profile.class_role_mappings.all()}
1713 eligible_rows = []
1714 for row in workspace.source_rows:
1715 source_id = source_text(row.get("source_id"))
1716 mapping = class_mappings.get(source_text(row.get("device_class")))
1717 if (
1718 mapping is None
1719 or mapping.creates_rack
1720 or mapping.ignore
1721 or source_id in ignored_source_ids
1722 or has_below_rack_position(row)
1723 ):
1724 continue
1725 eligible_rows.append(row)
1727 bound_device_by_source = dict(profile.device_matches.values_list("source_id", "netbox_device_id"))
1728 bound_source_by_device = {device_id: source for source, device_id in bound_device_by_source.items()}
1729 source_counts: dict[str, int] = {}
1730 name_counts: dict[str, int] = {}
1731 serial_counts: dict[str, int] = {}
1732 asset_tag_counts: dict[str, int] = {}
1733 for row in eligible_rows:
1734 values = (
1735 (source_text(row.get("source_id")), source_counts),
1736 (identity_text(effective_device_name(row)), name_counts),
1737 (source_text(row.get("serial")), serial_counts),
1738 (identity_text(source_text(row.get("asset_tag"))[:50]), asset_tag_counts),
1739 )
1740 for value, counts in values:
1741 if value:
1742 counts[value] = counts.get(value, 0) + 1
1744 counts = {
1745 "matched": 0,
1746 "probable": 0,
1747 "ambiguous": 0,
1748 "placement_conflicts": 0,
1749 "already": 0,
1750 "skipped": 0,
1751 }
1752 side_map, _, _ = translation_maps()
1753 for row in eligible_rows:
1754 source_id = source_text(row.get("source_id"))
1755 name = effective_device_name(row)
1756 serial = source_text(row.get("serial"))
1757 asset_tag = source_text(row.get("asset_tag"))[:50]
1758 if not source_id:
1759 continue
1760 if source_counts.get(source_id, 0) > 1:
1761 counts["ambiguous"] += 1
1762 continue
1763 if source_id in bound_device_by_source:
1764 counts["already"] += 1
1765 continue
1767 device, method, ambiguous = _match_existing_device(
1768 Device,
1769 visible_devices,
1770 name if name_counts.get(identity_text(name), 0) == 1 else "",
1771 serial if serial_counts.get(serial, 0) == 1 else "",
1772 asset_tag if asset_tag_counts.get(identity_text(asset_tag), 0) == 1 else "",
1773 site,
1774 tenant_id,
1775 )
1776 if ambiguous:
1777 counts["ambiguous"] += 1
1778 continue
1779 if device is not None and method == "name":
1780 face = side_map.get(source_text(row.get("face")).lower())
1781 if _device_placement_differs(
1782 device,
1783 target["location"].pk if target["location"] else None,
1784 source_text(row.get("rack_name")),
1785 source_position(row.get("u_position")),
1786 face,
1787 ):
1788 counts["placement_conflicts"] += 1
1789 continue
1790 if device is not None:
1791 bound_source = bound_source_by_device.get(device.pk)
1792 if bound_source is not None and bound_source != source_id:
1793 counts["ambiguous"] += 1
1794 continue
1795 try:
1796 save_permission_scoped_object(
1797 actor,
1798 DeviceExistingMatch,
1799 {"profile": profile, "source_id": source_id},
1800 {
1801 "netbox_device_id": device.pk,
1802 "device_name": device.name,
1803 "source_asset_tag": asset_tag,
1804 },
1805 on_existing="reject",
1806 )
1807 except (ValidationError, IntegrityError, ObjectPermissionDenied):
1808 counts["skipped"] += 1
1809 continue
1810 bound_device_by_source[source_id] = device.pk
1811 bound_source_by_device[device.pk] = source_id
1812 counts["matched"] += 1
1813 continue
1814 if name:
1815 short_name = name.split(" - ")[-1].strip() if " - " in name else name
1816 tenant_filter = {"tenant_id": tenant_id} if tenant_id is not None else {"tenant__isnull": True}
1817 if visible_devices.filter(site=site, name__icontains=short_name, **tenant_filter).exists():
1818 counts["probable"] += 1
1819 return AutoMatchSummary(**counts)
1822__all__ = (
1823 "SYNC_ALL_NOTHING",
1824 "AutoMatchSummary",
1825 "ReviewWorkspace",
1826 "SyncAllSelection",
1827 "WorkspaceUnit",
1828 "auto_match_devices",
1829 "refuse_moved_policy",
1830)