Coverage for netbox_data_import/cable_target.py: 98%
747 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-09 20:50 +0000
1# SPDX-License-Identifier: Apache-2.0
2# SPDX-FileCopyrightText: 2026 Marcin Zieba <marcinpsk@gmail.com>
3"""The Cable Target Module: Patched Path Replacement planning and writes.
5Section 6 gives one complete Source Trace one Synchronization Unit: it deletes the single direct
6Logical Cable when one exists, then creates every physical segment the trace states. It reads the
7decisions and the Cable policy the Import Profile holds, and it writes Cables and provenance rows.
9It consumes Source Traces by output kind, so it imports no Source Adapter. It reads PortMapping
10rows and never writes one, and it never creates or mutates a CablePath.
12Live target state that no Planned Change carries joins the unit through an `info` diagnostic.
13Visible reused Cables and PortMapping rows contribute identities. Hidden Cable drift contributes
14only its diagnostic code. Execution replans inside its transaction, so a material change makes the
15accepted unit stale and rolls the write back.
16"""
18from __future__ import annotations
20from dataclasses import dataclass, field
21from typing import Any
23from .catalog import OutputKind, TargetModuleKey
24from .field_keys import (
25 FRONT_PORT_KIND,
26 INTERFACE_KIND,
27 MAPPED_PEER_ROLE,
28 REAR_PORT_KIND,
29 SELECT_TERMINATION_TASK,
30 TERMINATION_ROLE,
31 claimed_termination_kind,
32 parse_termination_field_key,
33 same_device_and_cards,
34 termination_field_key,
35)
36from .object_permissions import enforce_saved_object_permission
37from .plan import Diagnostic, Disposition, PlannedChange, Severity, SynchronizationUnit
38from .target_runtime import DeletedObject, PreconditionFailed
39from .values import identity_text, source_text
41CABLE_STATUS = "connected"
42ELIGIBLE_TERMINATION_LIMIT = 20
44# Section 10.2 badge vocabulary. T8 adds the proposal states to the same set.
45UNRESOLVED = "unresolved"
46AUTOMATICALLY_RESOLVED = "automatically resolved"
47MANUALLY_RESOLVED = "manually resolved"
49CREATE_SEGMENT = "create"
50REUSE_SEGMENT = "reuse existing"
51CONFLICT_SEGMENT = "conflict"
52_CONFLICT_CODES = frozenset(
53 {
54 "cable.termination_occupied",
55 "cable.multi_termination_conflict",
56 "cable.planned_termination_conflict",
57 "cable.resolved_segment_conflict",
58 }
59)
61_KIND_BY_MODEL_NAME = {
62 "interface": INTERFACE_KIND,
63 "frontport": FRONT_PORT_KIND,
64 "rearport": REAR_PORT_KIND,
65}
66_SUPPORTED_TERMINATION_LABELS = frozenset(f"dcim.{name}" for name in _KIND_BY_MODEL_NAME)
67_READER_ACCESSOR_BY_KIND = {
68 INTERFACE_KIND: "interfaces",
69 FRONT_PORT_KIND: "front_ports",
70 REAR_PORT_KIND: "rear_ports",
71}
72_VIEW_PERMISSION_BY_KIND = {
73 INTERFACE_KIND: "dcim.view_interface",
74 FRONT_PORT_KIND: "dcim.view_frontport",
75 REAR_PORT_KIND: "dcim.view_rearport",
76}
79@dataclass(frozen=True)
80class EligibleTerminations:
81 """One capped page of eligible terminations and its uncapped matching total."""
83 candidates: tuple[Any, ...]
84 total: int
87def _object_type_label(obj) -> str:
88 """Return the ``app_label.model_name`` key one target object is recorded under."""
89 return f"{obj._meta.app_label}.{obj._meta.model_name}"
92def _model_for_label(label: str):
93 """Return the model class one recorded object-type label names."""
94 from django.apps import apps
96 app_label, _, model_name = label.partition(".")
97 return apps.get_model(app_label, model_name)
100def _object_identity(label: str, object_id: int) -> str:
101 """Return the plan identity string for one target object."""
102 return f"{label}:{object_id}"
105@dataclass(frozen=True)
106class _Termination:
107 """One source Termination Reference bound to a NetBox object."""
109 object_type: str
110 object_id: int
111 kind: str
112 device_id: int
113 display: str
115 @property
116 def key(self) -> tuple[str, int]:
117 """Return the comparison key that names this object across queries."""
118 return self.object_type, self.object_id
120 @property
121 def identity(self) -> str:
122 """Return the plan identity string for this object."""
123 return _object_identity(self.object_type, self.object_id)
125 def as_json(self) -> list:
126 """Return the serializable object and resolved Device a creation precondition carries."""
127 return [self.object_type, self.object_id, self.device_id]
130@dataclass(frozen=True)
131class _DesiredSegment:
132 """One physical Cable the trace states."""
134 index: int
135 left: _Termination
136 right: _Termination
137 cable_class: str
139 @property
140 def key(self) -> str:
141 """Return the direction-independent key two traces share for one identical segment."""
142 first, second = sorted((self.left.key, self.right.key))
143 return f"{_object_identity(*first)}|{_object_identity(*second)}"
145 @property
146 def change_identity(self) -> str:
147 """Return the Planned Change identity this segment creates under."""
148 return f"cable:create:{self.key}"
150 @property
151 def terminations(self) -> tuple[_Termination, _Termination]:
152 """Return both ends in canonical order."""
153 return self.left, self.right
155 def as_json(self) -> list:
156 """Return the sorted termination records a payload and a precondition carry."""
157 return sorted((self.left.as_json(), self.right.as_json()))
160@dataclass(frozen=True)
161class _ExistingCable:
162 """One existing Cable and the termination sets it holds."""
164 cable: Any
165 a_side: frozenset
166 b_side: frozenset
168 @property
169 def multi_termination(self) -> bool:
170 """Return whether either side holds more than one termination."""
171 return len(self.a_side) > 1 or len(self.b_side) > 1
173 @property
174 def terminations(self) -> list:
175 """Return the sorted termination pairs a precondition records."""
176 return sorted([label, object_id] for label, object_id in self.a_side | self.b_side)
178 def satisfies(self, segment: _DesiredSegment) -> bool:
179 """Return whether this Cable is exactly the unordered pair *segment* states."""
180 return self.connects(segment.left, segment.right)
182 def connects(self, first: _Termination, second: _Termination) -> bool:
183 """Return whether this Cable directly joins the two given terminations, one end each."""
184 return (self.a_side, self.b_side) in (
185 ({first.key}, {second.key}),
186 ({second.key}, {first.key}),
187 )
190def _delete_identity(cable_pk: int) -> str:
191 """Return the Planned Change identity that removes one Logical Cable.
193 The unit's dependency graph and the deletion change both name it, so one helper states it.
194 """
195 return f"cable:delete:{cable_pk}"
198@dataclass
199class _TraceAnalysis:
200 """What one Source Trace contributes, built in planning order."""
202 trace: Any
203 identity: str
204 display: dict
205 diagnostics: list = field(default_factory=list)
206 endpoints: tuple = ()
207 segments: list = field(default_factory=list)
208 proven: dict = field(default_factory=dict)
209 policies: dict = field(default_factory=dict)
210 terminations: dict = field(default_factory=dict)
211 topology_read: bool = False
212 logical_cable: Any = None
213 retains_logical_cable: bool = False
214 blocked: bool = False
215 invalid: bool = False
217 @property
218 def stopped(self) -> bool:
219 """Return whether a finding makes further planning for this trace meaningless."""
220 return self.blocked or self.invalid
222 @property
223 def pending(self) -> list:
224 """Return the desired segments no existing Cable already proves."""
225 return [segment for segment in self.segments if segment.index not in self.proven]
227 @property
228 def deleted_logical_cable(self):
229 """Return the Logical Cable this trace removes, which a Cable that proves its path is not."""
230 return None if self.retains_logical_cable else self.logical_cable
232 @property
233 def delete_identity(self) -> str | None:
234 """Return the Planned Change identity that removes this trace's Logical Cable."""
235 removed = self.deleted_logical_cable
236 return None if removed is None else _delete_identity(removed.cable.pk)
238 def error(self, code: str, display: dict, identities=()) -> None:
239 """Record one blocking or invalidating finding."""
240 self.diagnostics.append(
241 Diagnostic(code=code, severity=Severity.ERROR, identities=tuple(identities), display=display)
242 )
244 def note(self, code: str, display: dict, identities=()) -> None:
245 """Record one review note whose identities keep the unit honest about live state."""
246 self.diagnostics.append(
247 Diagnostic(code=code, severity=Severity.INFO, identities=tuple(identities), display=display)
248 )
250 def block(self, code: str, display: dict, identities=()) -> None:
251 """Record a finding an operator decision or a NetBox correction can resolve."""
252 self.error(code, display, identities)
253 self.blocked = True
255 def refuse(self, code: str, display: dict, identities=()) -> None:
256 """Record a finding no decision inside this plugin can resolve."""
257 self.error(code, display, identities)
258 self.invalid = True
261def _endpoint_label(reference) -> str:
262 """Return the operator-facing name of one Termination Reference."""
263 parts = [source_text(reference.device), source_text(reference.cards), source_text(reference.port)]
264 return " ".join(part for part in parts if part)
267def _reference_display(reference) -> dict:
268 """Return the source values one Termination Reference states."""
269 return {
270 "device": source_text(reference.device),
271 "cards": source_text(reference.cards),
272 "port": source_text(reference.port),
273 "port_class": source_text(reference.port_class),
274 }
277def _field_key(reference, role: str = TERMINATION_ROLE) -> str:
278 """Return the canonical field key one Termination Reference resolves under."""
279 return termination_field_key(
280 device=reference.device,
281 cards=reference.cards,
282 port=reference.port,
283 kind=claimed_termination_kind(reference.port_class),
284 role=role,
285 )
288def _source_record(trace, segment_index: int) -> dict:
289 """Return the provenance one Source Trace contributes to one created Cable."""
290 provenance = trace.provenance[0]
291 return {
292 "trace_identity": trace.identity,
293 "segment_index": segment_index,
294 "from_text": provenance.from_text,
295 "to_text": provenance.to_text,
296 "direction": provenance.direction,
297 "workbook_fingerprint": provenance.workbook_fingerprint,
298 "sheet": provenance.sheet,
299 "block_ordinal": provenance.block_ordinal,
300 "row_start": provenance.row_start,
301 "row_end": provenance.row_end,
302 "export_timestamp": provenance.export_timestamp,
303 }
306def _visible_devices(netbox_reader, names) -> dict[str, list]:
307 """Return the visible Devices at the import target, grouped by comparison name."""
308 from django.db.models import Q
310 wanted = {identity_text(name) for name in names if identity_text(name)}
311 if not wanted:
312 return {}
313 devices = netbox_reader.devices()
314 if netbox_reader.site is not None:
315 devices = devices.filter(site=netbox_reader.site)
316 lookup = Q()
317 for name in sorted(wanted):
318 lookup |= Q(name__iexact=name)
319 grouped: dict[str, list] = {}
320 for device in devices.filter(lookup):
321 comparison = identity_text(device.name)
322 if comparison in wanted:
323 grouped.setdefault(comparison, []).append(device)
324 return grouped
327def eligible_terminations(
328 field_key: str,
329 netbox_reader,
330 *,
331 profile,
332 search: str = "",
333 limit: int = ELIGIBLE_TERMINATION_LIMIT,
334) -> EligibleTerminations:
335 """Return one page of candidates for a canonical termination field key.
337 The termination role offers the claimed kind on the resolved Device. The mapped-peer role starts
338 from the profile's saved base resolution or the exact-name match, then offers its opposite ports.
339 The reader keeps both inside the actor's view scope. The picker and a proposal request share this
340 query.
341 """
342 parsed = parse_termination_field_key(field_key)
343 device_name, kind = parsed["device"], parsed["kind"]
344 accessor = _READER_ACCESSOR_BY_KIND[kind]
345 devices = _visible_devices(netbox_reader, [device_name]).get(identity_text(device_name), [])
346 if len(devices) != 1:
347 return EligibleTerminations(candidates=(), total=0)
348 device = devices[0]
349 candidates = getattr(netbox_reader, accessor)().filter(device_id=device.pk)
350 if parsed["role"] == MAPPED_PEER_ROLE:
351 from .models import TerminationResolution, index_digest
353 base_field_key = termination_field_key(
354 device=parsed["device"],
355 cards=parsed["cards"],
356 port=parsed["port"],
357 kind=kind,
358 role=TERMINATION_ROLE,
359 )
360 stored = (
361 TerminationResolution.objects.filter(
362 profile=profile,
363 task_type=SELECT_TERMINATION_TASK,
364 field_key=base_field_key,
365 field_key_digest=index_digest(base_field_key),
366 )
367 .select_related("selected_object_type")
368 .first()
369 )
370 if stored is None:
371 resolved = [candidate for candidate in candidates if identity_text(candidate.name) == parsed["port"]]
372 else:
373 resolved = []
374 expected_label = _object_type_label(candidates.model)
375 selected_label = f"{stored.selected_object_type.app_label}.{stored.selected_object_type.model}"
376 if selected_label == expected_label:
377 selected = candidates.filter(pk=stored.selected_object_id).first()
378 if selected is not None:
379 resolved.append(selected)
380 if len(resolved) != 1 or kind == INTERFACE_KIND:
381 return EligibleTerminations(candidates=(), total=0)
382 if kind == FRONT_PORT_KIND:
383 peer_kind = REAR_PORT_KIND
384 peer_ids = (
385 netbox_reader.port_mappings()
386 .filter(front_port_id=resolved[0].pk)
387 .values_list("rear_port_id", flat=True)
388 )
389 else:
390 peer_kind = FRONT_PORT_KIND
391 peer_ids = (
392 netbox_reader.port_mappings()
393 .filter(rear_port_id=resolved[0].pk)
394 .values_list("front_port_id", flat=True)
395 )
396 candidates = getattr(netbox_reader, _READER_ACCESSOR_BY_KIND[peer_kind])().filter(
397 device_id=device.pk,
398 pk__in=peer_ids,
399 )
400 if search:
401 candidates = candidates.filter(name__icontains=search)
402 candidates = candidates.order_by("name", "pk")
403 return EligibleTerminations(candidates=tuple(candidates[:limit]), total=candidates.count())
406class _CableBatch:
407 """Plan every Source Trace in one batch, so an identical shared segment plans once."""
409 def __init__(self, traces, profile, netbox_reader, *, lock_plan_references: bool = False):
410 self.profile = profile
411 self.reader = netbox_reader
412 self.actor = netbox_reader.actor
413 self.lock_plan_references = lock_plan_references
414 self.analyses = [self._new_analysis(trace) for trace in traces]
415 self._objects: dict[tuple[str, int], Any] = {}
416 self._components: dict[tuple[int, str], dict[str, list]] = {}
417 self._resolved: dict[str, dict[tuple, _Termination]] = {}
418 self._mappings_by_front: dict[int, list] = {}
419 self._mappings_by_rear: dict[int, list] = {}
420 self._mapping_ports: dict[tuple[str, int], Any] = {}
421 self._existing: dict[int, _ExistingCable] = {}
422 self._occupied: dict[tuple[str, int], _ExistingCable] = {}
423 self._mapping_rows: dict[str, Any] | None = None
424 self._stored = self._stored_resolutions()
425 self._devices = self._load_devices()
426 self._resolve_terminations()
427 self._load_mappings()
428 self._build_segments()
429 self._lock_segment_terminations()
430 self._load_existing_cables()
431 self._classify()
432 self._decide()
433 self._block_planned_termination_conflicts()
434 self._refuse_conflicting_creations()
435 self._deletes_by_segment = self._shared_deletes()
436 self._sources_by_segment = self._shared_sources()
438 def units(self) -> list[SynchronizationUnit]:
439 """Return one Synchronization Unit per Source Trace, in batch order."""
440 return [self._unit(analysis) for analysis in self.analyses]
442 def _new_analysis(self, trace) -> _TraceAnalysis:
443 """Return the analysis record one Source Trace starts from."""
444 summary = trace.endpoint_summary
445 provenance = trace.provenance[0]
446 display = {
447 "name": f"{_endpoint_label(summary.from_termination)} to {_endpoint_label(summary.to_termination)}",
448 "row_number": provenance.row_start,
449 "sheet": provenance.sheet,
450 "trace_identity": trace.identity,
451 "from_text": provenance.from_text,
452 "to_text": provenance.to_text,
453 "segment_count": len(trace.segments),
454 }
455 analysis = _TraceAnalysis(trace=trace, identity=f"cable:trace:{trace.identity}", display=display)
456 for error in trace.errors:
457 analysis.refuse(error.code, {"message": error.message, "row_number": error.row_number})
458 return analysis
460 @staticmethod
461 def _references(trace) -> list:
462 """Return every Termination Reference one trace resolves, endpoints included."""
463 summary = trace.endpoint_summary
464 references = [summary.from_termination, summary.to_termination]
465 for segment in trace.segments:
466 references.extend((segment.left, segment.right))
467 return references
469 def _stored_resolutions(self) -> dict[str, Any]:
470 """Return the operator's saved termination decisions, keyed by field key."""
471 from .models import TerminationResolution
473 field_keys = set()
474 for analysis in self.analyses:
475 if analysis.stopped:
476 continue
477 for reference in self._references(analysis.trace):
478 field_keys.add(_field_key(reference))
479 field_keys.add(_field_key(reference, MAPPED_PEER_ROLE))
480 if not field_keys:
481 return {}
482 from .models import index_digest
484 # The unique constraint indexes the digest, and nothing indexes the unbounded field key.
485 rows = TerminationResolution.objects.filter(
486 profile=self.profile,
487 task_type=SELECT_TERMINATION_TASK,
488 field_key_digest__in=sorted(index_digest(key) for key in field_keys),
489 ).select_related("selected_object_type")
490 return {row.field_key: row for row in rows}
492 def _load_devices(self) -> dict[str, list]:
493 """Return the visible Devices every planned trace names, grouped by comparison name."""
494 names = [
495 reference.device
496 for analysis in self.analyses
497 if not analysis.stopped
498 for reference in self._references(analysis.trace)
499 ]
500 return _visible_devices(self.reader, names)
502 def _components_for(self, device_id: int, kind: str) -> dict[str, list]:
503 """Return one Device's terminations of one kind, grouped by comparison name."""
504 cached = self._components.get((device_id, kind))
505 if cached is None:
506 cached = {}
507 for component in getattr(self.reader, _READER_ACCESSOR_BY_KIND[kind])().filter(device_id=device_id):
508 cached.setdefault(identity_text(component.name), []).append(component)
509 self._components[(device_id, kind)] = cached
510 return cached
512 def _resolve_terminations(self) -> None:
513 """Bind every Termination Reference to one NetBox object, or record why it stays open."""
514 for analysis in self.analyses:
515 if analysis.stopped:
516 continue
517 resolved: dict[tuple, _Termination] = {}
518 for reference in self._references(analysis.trace):
519 if reference.identity_key in resolved:
520 continue
521 termination = self._resolve_one(analysis, reference)
522 if termination is not None:
523 resolved[reference.identity_key] = termination
524 self._resolved[analysis.identity] = resolved
526 def _resolve_one(self, analysis: _TraceAnalysis, reference) -> _Termination | None:
527 """Return the NetBox object one Termination Reference names, or record the open decision."""
528 devices = self._devices.get(identity_text(reference.device), [])
529 if len(devices) != 1:
530 analysis.block("trace.device_unresolved", {**_reference_display(reference), "matches": len(devices)})
531 # With no resolved Device there is nothing to pick from, so the picker cannot help here.
532 self._record_resolution(
533 analysis,
534 reference,
535 UNRESOLVED,
536 None,
537 reason=f"The source names {len(devices)} matching Devices, so no port list applies.",
538 )
539 return None
540 device = devices[0]
541 stored = self._stored.get(_field_key(reference))
542 if stored is not None:
543 termination = self._stored_termination(analysis, reference, device, stored)
544 state = UNRESOLVED if termination is None else MANUALLY_RESOLVED
545 self._record_resolution(analysis, reference, state, termination)
546 return termination
547 kind = claimed_termination_kind(reference.port_class)
548 candidates = self._components_for(device.pk, kind).get(identity_text(reference.port), [])
549 if len(candidates) != 1:
550 analysis.block(
551 "cable.termination_unresolved", {**_reference_display(reference), "matches": len(candidates)}
552 )
553 self._record_resolution(analysis, reference, UNRESOLVED, None)
554 return None
555 termination = self._termination(candidates[0])
556 self._record_resolution(analysis, reference, AUTOMATICALLY_RESOLVED, termination)
557 return termination
559 @staticmethod
560 def _record_resolution(
561 analysis: _TraceAnalysis,
562 reference,
563 state: str,
564 termination,
565 reason: str = "",
566 role: str = TERMINATION_ROLE,
567 ) -> None:
568 """Record how one Termination Reference was settled, for its badge and its picker."""
569 key = _field_key(reference, role)
570 claimed = claimed_termination_kind(reference.port_class)
571 if role == MAPPED_PEER_ROLE:
572 # The picker offers the ports on the far side of the panel, which are the opposite kind.
573 kind = REAR_PORT_KIND if claimed == FRONT_PORT_KIND else FRONT_PORT_KIND
574 label = f"{_endpoint_label(reference)} (mapped peer)"
575 else:
576 kind, label = claimed, _endpoint_label(reference)
577 analysis.terminations.setdefault(
578 key,
579 {
580 "field_key": key,
581 "label": label,
582 "kind": kind,
583 "state": state,
584 "selected": "" if termination is None else termination.display,
585 "selectable": not reason,
586 "reason": reason,
587 },
588 )
590 def _stored_termination(self, analysis: _TraceAnalysis, reference, device, stored) -> _Termination | None:
591 """Return the object one saved decision selected, rechecked against current target state."""
592 label = f"{stored.selected_object_type.app_label}.{stored.selected_object_type.model}"
593 if label not in _SUPPORTED_TERMINATION_LABELS:
594 analysis.refuse(
595 "cable.unsupported_termination_kind",
596 {**_reference_display(reference), "selected_object_type": label},
597 )
598 return None
599 selected_kind = _KIND_BY_MODEL_NAME[label.partition(".")[2]]
600 claimed_kind = claimed_termination_kind(reference.port_class)
601 if selected_kind != claimed_kind:
602 analysis.block(
603 "cable.termination_kind_mismatch",
604 {
605 **_reference_display(reference),
606 "selected_display_name": stored.selected_display_name,
607 "claimed_kind": claimed_kind,
608 "selected_kind": selected_kind,
609 },
610 )
611 return None
612 accessor = _READER_ACCESSOR_BY_KIND[selected_kind]
613 # A saved selection that left the resolved Device no longer answers the question it was asked.
614 selected = getattr(self.reader, accessor)().filter(pk=stored.selected_object_id, device_id=device.pk).first()
615 if selected is None:
616 analysis.block(
617 "cable.termination_unresolved",
618 {**_reference_display(reference), "selected_display_name": stored.selected_display_name},
619 )
620 return None
621 return self._termination(selected)
623 def _termination(self, component) -> _Termination:
624 """Return the plan-side record of one resolved NetBox termination."""
625 label = _object_type_label(component)
626 self._objects[(label, component.pk)] = component
627 return _Termination(
628 object_type=label,
629 object_id=component.pk,
630 kind=_KIND_BY_MODEL_NAME[label.partition(".")[2]],
631 device_id=component.device_id,
632 display=str(component),
633 )
635 def _load_mappings(self) -> None:
636 """Read every PortMapping row on a Device this batch resolved a pass-through port on."""
637 device_ids = {
638 termination.device_id
639 for resolved in self._resolved.values()
640 for termination in resolved.values()
641 if termination.kind != INTERFACE_KIND
642 }
643 if device_ids:
644 mappings = self.reader.port_mappings().filter(device_id__in=sorted(device_ids)).order_by("pk")
645 if self.lock_plan_references:
646 mappings = mappings.select_for_update(of=("self",))
647 # Indexed on load, because every pass-through end asks for the rows of one port.
648 for row in mappings:
649 self._mappings_by_front.setdefault(row.front_port_id, []).append(row)
650 self._mappings_by_rear.setdefault(row.rear_port_id, []).append(row)
651 ids_by_kind = {
652 FRONT_PORT_KIND: set(self._mappings_by_front),
653 REAR_PORT_KIND: set(self._mappings_by_rear),
654 }
655 for kind, object_ids in ids_by_kind.items():
656 for component in getattr(self.reader, _READER_ACCESSOR_BY_KIND[kind])().filter(pk__in=object_ids):
657 self._mapping_ports[(kind, component.pk)] = component
659 def _peers_of(self, termination: _Termination) -> list:
660 """Return the PortMapping rows that link one pass-through port to its opposite side."""
661 if termination.kind == FRONT_PORT_KIND:
662 return self._mappings_by_front.get(termination.object_id, [])
663 if termination.kind == REAR_PORT_KIND:
664 return self._mappings_by_rear.get(termination.object_id, [])
665 return []
667 def _peer_termination(self, mapping, termination: _Termination) -> _Termination | None:
668 """Return the visible opposite port one PortMapping row links to *termination*."""
669 if termination.kind == FRONT_PORT_KIND:
670 peer_key = REAR_PORT_KIND, mapping.rear_port_id
671 else:
672 peer_key = FRONT_PORT_KIND, mapping.front_port_id
673 component = self._mapping_ports.get(peer_key)
674 return None if component is None else self._termination(component)
676 def _mapped_peers(self, analysis: _TraceAnalysis, reference, termination: _Termination) -> list | None:
677 """Return visible mapped peers, or block when one peer is outside the actor's view scope."""
678 peers = []
679 for mapping in self._peers_of(termination):
680 peer = self._peer_termination(mapping, termination)
681 if peer is None:
682 peer_kind = REAR_PORT_KIND if termination.kind == FRONT_PORT_KIND else FRONT_PORT_KIND
683 analysis.block(
684 "cable.permission_denied",
685 {**_reference_display(reference), "permission": _VIEW_PERMISSION_BY_KIND[peer_kind]},
686 )
687 return None
688 peers.append((peer, mapping))
689 return peers
691 def _build_segments(self) -> None:
692 """Verify every Pass-Through Claim and turn Segment Evidence into desired segments."""
693 for analysis in self.analyses:
694 if analysis.stopped:
695 continue
696 resolved = self._resolved.get(analysis.identity, {})
697 summary = analysis.trace.endpoint_summary
698 endpoints = tuple(
699 resolved.get(reference.identity_key) for reference in (summary.from_termination, summary.to_termination)
700 )
701 if any(endpoint is None for endpoint in endpoints):
702 continue
703 analysis.endpoints = endpoints
704 self._build_one(analysis, resolved)
706 def _build_one(self, analysis: _TraceAnalysis, resolved: dict) -> None:
707 """Bind one trace's Segment Evidence, substituting a mapped peer where the source repeats a port."""
708 segments = analysis.trace.segments
709 left_ends: list[_Termination] = []
710 right_ends: list[_Termination] = []
711 for segment in segments:
712 left, right = resolved.get(segment.left.identity_key), resolved.get(segment.right.identity_key)
713 if left is None or right is None:
714 return
715 left_ends.append(left)
716 right_ends.append(right)
717 for index in range(len(segments) - 1):
718 if not same_device_and_cards(segments[index].right, segments[index + 1].left):
719 continue
720 entry = self._continue_path(analysis, segments[index + 1].left, right_ends[index], left_ends[index + 1])
721 if entry is None:
722 return
723 left_ends[index + 1] = entry
724 for index, segment in enumerate(segments):
725 if left_ends[index].key != right_ends[index].key:
726 continue
727 analysis.block(
728 "cable.segment_self_connection",
729 {
730 "segment_index": index,
731 "cable_class": source_text(segment.cable_class),
732 "termination": left_ends[index].display,
733 },
734 identities=(left_ends[index].identity,),
735 )
736 return
737 analysis.segments = [
738 _DesiredSegment(
739 index=index,
740 left=left_ends[index],
741 right=right_ends[index],
742 cable_class=source_text(segment.cable_class),
743 )
744 for index, segment in enumerate(segments)
745 ]
747 def _continue_path(self, analysis: _TraceAnalysis, reference, exit_end, entry_end) -> _Termination | None:
748 """Return the termination the next cable end takes where the path passes through a panel."""
749 if exit_end.key != entry_end.key:
750 return self._verified_pass_through(analysis, reference, exit_end, entry_end)
751 mapped_peers = self._mapped_peers(analysis, reference, exit_end)
752 if mapped_peers is None:
753 return None
754 peers: dict[tuple[str, int], Any] = {}
755 for peer, row in mapped_peers:
756 peers.setdefault(peer.key, (peer, row))
757 if not peers:
758 analysis.refuse(
759 "cable.pass_through_not_mapped",
760 {**_reference_display(reference), "entry": exit_end.display, "exit": entry_end.display, "mapped": []},
761 identities=(exit_end.identity,),
762 )
763 return None
764 if len(peers) > 1:
765 return self._chosen_peer(analysis, reference, exit_end, peers)
766 peer, mapping = next(iter(peers.values()))
767 return self._substituted(analysis, reference, exit_end, peer, mapping)
769 def _substituted(self, analysis: _TraceAnalysis, reference, exit_end, peer, mapping) -> _Termination:
770 """Record one same-port continuation and return the mapped peer it substitutes."""
771 analysis.note(
772 "cable.same_port_continuation",
773 {**_reference_display(reference), "port": exit_end.display, "peer": peer.display},
774 identities=(exit_end.identity, peer.identity, _object_identity("dcim.portmapping", mapping.pk)),
775 )
776 return peer
778 def _chosen_peer(self, analysis: _TraceAnalysis, reference, exit_end, peers) -> _Termination | None:
779 """Return the mapped peer the operator selected, or block on the several NetBox offers."""
780 stored = self._stored.get(_field_key(reference, MAPPED_PEER_ROLE))
781 if stored is not None:
782 selected = (
783 f"{stored.selected_object_type.app_label}.{stored.selected_object_type.model}",
784 stored.selected_object_id,
785 )
786 if selected in peers:
787 peer, mapping = peers[selected]
788 self._record_resolution(analysis, reference, MANUALLY_RESOLVED, peer, role=MAPPED_PEER_ROLE)
789 return self._substituted(analysis, reference, exit_end, peer, mapping)
790 self._record_resolution(analysis, reference, UNRESOLVED, None, role=MAPPED_PEER_ROLE)
791 analysis.block(
792 "cable.ambiguous_mapped_peer",
793 {
794 **_reference_display(reference),
795 "port": exit_end.display,
796 "peers": sorted(peer.display for peer, _mapping in peers.values()),
797 },
798 identities=(exit_end.identity, *sorted(_object_identity(*key) for key in peers)),
799 )
800 return None
802 def _verified_pass_through(self, analysis: _TraceAnalysis, reference, exit_end, entry_end) -> _Termination | None:
803 """Return the stated entry port once a PortMapping row proves the panel joins the two ports."""
804 mapping = None
805 if {exit_end.kind, entry_end.kind} == {FRONT_PORT_KIND, REAR_PORT_KIND}:
806 front, rear = (exit_end, entry_end) if exit_end.kind == FRONT_PORT_KIND else (entry_end, exit_end)
807 mapping = next(
808 (row for row in self._peers_of(front) if row.rear_port_id == rear.object_id),
809 None,
810 )
811 if mapping is None:
812 mapped = []
813 for row in self._peers_of(exit_end):
814 peer = self._peer_termination(row, exit_end)
815 if peer is not None:
816 mapped.append(peer.display)
817 analysis.refuse(
818 "cable.pass_through_not_mapped",
819 {
820 **_reference_display(reference),
821 "entry": exit_end.display,
822 "exit": entry_end.display,
823 "mapped": sorted(mapped),
824 },
825 identities=(exit_end.identity, entry_end.identity),
826 )
827 return None
828 analysis.note(
829 "cable.pass_through_verified",
830 {**_reference_display(reference), "entry": exit_end.display, "exit": entry_end.display},
831 identities=(exit_end.identity, entry_end.identity, _object_identity("dcim.portmapping", mapping.pk)),
832 )
833 return entry_end
835 def _load_existing_cables(self) -> None:
836 """Read every Cable that already holds a termination this batch resolved."""
837 from dcim.models import Cable, CableTermination
839 wanted = {
840 termination.key for analysis in self.analyses for termination in self._analysis_terminations(analysis)
841 }
842 # Occupancy is unscoped: a Cable the actor cannot view still holds the termination.
843 cable_ids = {
844 component.cable_id for key, component in self._objects.items() if key in wanted and component.cable_id
845 }
846 if not cable_ids:
847 return
848 cables = Cable.objects.filter(pk__in=sorted(cable_ids)).order_by("pk")
849 terminations = CableTermination.objects.filter(cable_id__in=sorted(cable_ids)).order_by("pk")
850 if self.lock_plan_references:
851 cables = cables.select_for_update(of=("self",))
852 terminations = terminations.select_for_update(of=("self",))
853 # Each Cable lock precedes its CableTermination locks, as it does during deletion.
854 locked = list(cables)
855 sides: dict[int, dict[str, set]] = {}
856 for row in terminations:
857 label = _object_type_label(row.termination_type.model_class())
858 sides.setdefault(row.cable_id, {"A": set(), "B": set()})[row.cable_end].add((label, row.termination_id))
859 for cable in locked:
860 ends = sides.get(cable.pk, {"A": set(), "B": set()})
861 existing = _ExistingCable(cable=cable, a_side=frozenset(ends["A"]), b_side=frozenset(ends["B"]))
862 self._existing[cable.pk] = existing
863 for key in existing.a_side | existing.b_side:
864 self._occupied[key] = existing
866 def _lock_segment_terminations(self) -> None:
867 """Lock every desired segment termination in global object-type and primary-key order."""
868 if not self.lock_plan_references:
869 return
870 ids_by_label: dict[str, set[int]] = {}
871 for analysis in self.analyses:
872 for segment in analysis.segments:
873 for termination in segment.terminations:
874 ids_by_label.setdefault(termination.object_type, set()).add(termination.object_id)
875 for label in sorted(ids_by_label):
876 object_ids = sorted(ids_by_label[label])
877 for object_id in object_ids:
878 self._objects.pop((label, object_id), None)
879 components = (
880 _model_for_label(label).objects.filter(pk__in=object_ids).order_by("pk").select_for_update(of=("self",))
881 )
882 for component in components:
883 self._objects[(label, component.pk)] = component
885 @staticmethod
886 def _analysis_terminations(analysis: _TraceAnalysis) -> list[_Termination]:
887 """Return every resolved termination one trace's plan depends on."""
888 terminations = list(analysis.endpoints)
889 for segment in analysis.segments:
890 terminations.extend(segment.terminations)
891 return terminations
893 def _cable_diagnostic_disclosure(self, cable: Any) -> tuple[dict[str, Any], tuple[str, ...]]:
894 """Return the diagnostic fields and identity the actor may see for one Cable."""
895 cable_visible = self.actor is None or self.actor.has_perm("dcim.view_cable", cable)
896 if not cable_visible:
897 return {"cable_visible": False}, ()
898 return (
899 {"cable_visible": True, "cable": str(cable)},
900 (_object_identity("dcim.cable", cable.pk),),
901 )
903 def _classify(self) -> None:
904 """Decide which segments already exist, then which Cable is each trace's Logical Cable.
906 Every desired segment of the batch is classified before any Logical Cable is chosen, so a
907 Cable that proves one trace's segment can never be deleted as another trace's leftover.
908 """
909 pending = self._writing()
910 for analysis in pending:
911 # Only these analyses look at the live topology, so only they may report what is there.
912 analysis.topology_read = True
913 for segment in analysis.segments:
914 candidate = self._occupied.get(segment.left.key)
915 proven = candidate if candidate is not None and candidate.satisfies(segment) else None
916 if proven is not None:
917 analysis.proven[segment.index] = proven
918 self._note_reuse(analysis, segment, proven)
919 proven_ids = {item.cable.pk for analysis in pending for item in analysis.proven.values()}
920 for analysis in pending:
921 analysis.logical_cable = self._logical_cable(analysis, proven_ids)
922 if analysis.segments:
923 self._report_conflicts(analysis)
924 else:
925 self._classify_endpoint_evidence(analysis)
927 def _logical_cable(self, analysis: _TraceAnalysis, proven_ids) -> _ExistingCable | None:
928 """Return the direct endpoint-to-endpoint Cable that proves no desired segment."""
929 first, second = analysis.endpoints
930 existing = self._occupied.get(first.key)
931 if existing is not None and existing.cable.pk not in proven_ids and existing.connects(first, second):
932 return existing
933 return None
935 def _classify_endpoint_evidence(self, analysis: _TraceAnalysis) -> None:
936 """Decide an Endpoint Summary fallback, which states endpoints and no physical path."""
937 if analysis.logical_cable is None:
938 analysis.block(
939 "trace.endpoint_evidence_only",
940 {"name": analysis.display["name"]},
941 identities=tuple(endpoint.identity for endpoint in analysis.endpoints),
942 )
943 return
944 # The stated endpoints are already joined, so the evidence is satisfied and nothing is removed.
945 cable_display, cable_identities = self._cable_diagnostic_disclosure(analysis.logical_cable.cable)
946 analysis.note(
947 "cable.segment_reused",
948 {"segment_index": 0, **cable_display},
949 identities=cable_identities,
950 )
951 analysis.retains_logical_cable = True
953 def _note_reuse(self, analysis: _TraceAnalysis, segment: _DesiredSegment, proven: _ExistingCable) -> None:
954 """Record the proven physical segment, so its live state joins the unit fingerprint."""
955 cable_display, cable_identities = self._cable_diagnostic_disclosure(proven.cable)
956 analysis.note(
957 "cable.segment_reused",
958 {"segment_index": segment.index, **cable_display},
959 identities=(*cable_identities, segment.left.identity, segment.right.identity),
960 )
961 drift = self._attribute_drift(segment, proven.cable)
962 if drift:
963 drift_display = drift if cable_display["cable_visible"] else {}
964 analysis.note(
965 "cable.attribute_drift",
966 {"segment_index": segment.index, **cable_display, **drift_display},
967 identities=cable_identities,
968 )
970 def _attribute_drift(self, segment: _DesiredSegment, cable) -> dict:
971 """Return the reused Cable attributes that differ from what this import would have written."""
972 mapping = self._cable_class_mapping(segment.cable_class)
973 drift = {}
974 if cable.status != CABLE_STATUS:
975 drift["status"] = cable.status
976 if mapping is not None and mapping.cable_type_resolved and (cable.type or None) != mapping.cable_type:
977 drift["type"] = cable.type or ""
978 if mapping is not None and mapping.cable_profile_resolved and (cable.profile or None) != mapping.cable_profile:
979 drift["profile"] = cable.profile or ""
980 if cable.label:
981 drift["label"] = cable.label
982 return drift
984 def _report_conflicts(self, analysis: _TraceAnalysis) -> None:
985 """Block the trace when a Cable this import may not touch holds a termination it needs."""
986 removed = analysis.deleted_logical_cable
987 logical_id = None if removed is None else removed.cable.pk
988 for segment in analysis.pending:
989 for termination in segment.terminations:
990 occupying = self._occupied.get(termination.key)
991 if occupying is None or occupying.cable.pk == logical_id:
992 continue
993 code = (
994 "cable.multi_termination_conflict" if occupying.multi_termination else "cable.termination_occupied"
995 )
996 cable_display, cable_identities = self._cable_diagnostic_disclosure(occupying.cable)
997 display = {"segment_index": segment.index, "port": termination.display, **cable_display}
998 identities = [termination.identity, *cable_identities]
999 analysis.block(code, display, identities=identities)
1001 def _cable_class_mapping(self, cable_class: str):
1002 """Return the Cable policy row one CableClass value carries, or None."""
1003 if self._mapping_rows is None:
1004 from .models import CableClassMapping
1006 rows = CableClassMapping.objects.filter(profile=self.profile)
1007 self._mapping_rows = {row.cable_class: row for row in rows}
1008 return self._mapping_rows.get(cable_class)
1010 def _decide(self) -> None:
1011 """Settle the Cable policy and the write permissions every actionable trace needs."""
1012 for analysis in self.analyses:
1013 if analysis.stopped or not analysis.endpoints:
1014 continue
1015 for segment in analysis.pending:
1016 policy = self._cable_policy(analysis, segment)
1017 if policy is not None:
1018 analysis.policies[segment.index] = policy
1019 self._check_permissions(analysis)
1021 def _cable_policy(self, analysis: _TraceAnalysis, segment: _DesiredSegment) -> dict | None:
1022 """Return the Cable Type and Cable Profile one new segment is written with."""
1023 from .models import cable_class_mapping_choice_errors
1025 display = {"segment_index": segment.index, "cable_class": segment.cable_class}
1026 mapping = self._cable_class_mapping(segment.cable_class)
1027 if mapping is None or not (mapping.cable_type_resolved and mapping.cable_profile_resolved):
1028 analysis.block("cable.cableclass_unmapped", display)
1029 return None
1030 errors = cable_class_mapping_choice_errors(mapping.cable_type, mapping.cable_profile)
1031 for error in errors.values():
1032 analysis.block(error.code, {**display, "message": error.messages[0]})
1033 if errors:
1034 return None
1035 return {"cable_type": mapping.cable_type, "cable_profile": mapping.cable_profile}
1037 def _check_permissions(self, analysis: _TraceAnalysis) -> None:
1038 """Block the trace when the actor may not make every Cable write it asks for."""
1039 if self.actor is None:
1040 return
1041 if analysis.pending and not self.actor.has_perm("dcim.add_cable"):
1042 analysis.block("cable.permission_denied", {"permission": "dcim.add_cable"})
1043 return
1044 logical = analysis.deleted_logical_cable
1045 if logical is not None and not self.actor.has_perm("dcim.delete_cable", logical.cable):
1046 cable_display, cable_identities = self._cable_diagnostic_disclosure(logical.cable)
1047 analysis.block(
1048 "cable.permission_denied",
1049 {"permission": "dcim.delete_cable", **cable_display},
1050 identities=cable_identities,
1051 )
1053 def _block_planned_termination_conflicts(self) -> None:
1054 """Block every actionable trace that competes for one free termination."""
1055 claims: dict[tuple[str, int], list[tuple[_TraceAnalysis, _DesiredSegment, _Termination]]] = {}
1056 for analysis in self._writing():
1057 for segment in analysis.pending:
1058 for termination in segment.terminations:
1059 claims.setdefault(termination.key, []).append((analysis, segment, termination))
1060 for records in claims.values():
1061 for analysis, segment, termination in records:
1062 competitors = [
1063 other
1064 for other, other_segment, _other_termination in records
1065 if other is not analysis and other_segment.key != segment.key
1066 ]
1067 for competitor in competitors:
1068 analysis.block(
1069 "cable.planned_termination_conflict",
1070 {
1071 "segment_index": segment.index,
1072 "termination": termination.display,
1073 "competing_trace": competitor.display["name"],
1074 },
1075 identities=(termination.identity,),
1076 )
1078 def _refuse_conflicting_creations(self) -> None:
1079 """Invalidate traces that resolve one shared segment to different Cable policies."""
1080 contributors: dict[str, list] = {}
1081 for analysis in self.analyses:
1082 if analysis.invalid or not analysis.endpoints:
1083 continue
1084 for segment in analysis.pending:
1085 policy = analysis.policies.get(segment.index)
1086 if policy is not None:
1087 contributors.setdefault(segment.key, []).append((analysis, segment, policy))
1088 for records in contributors.values():
1089 policies = {
1090 (segment.cable_class, policy["cable_type"], policy["cable_profile"])
1091 for _analysis, segment, policy in records
1092 }
1093 if len(policies) < 2:
1094 continue
1095 for analysis, segment, policy in records:
1096 analysis.refuse(
1097 "cable.resolved_segment_conflict",
1098 {
1099 "segment_index": segment.index,
1100 "cable_class": segment.cable_class,
1101 "cable_type": policy["cable_type"],
1102 "cable_profile": policy["cable_profile"],
1103 "terminations": segment.as_json(),
1104 },
1105 identities=(segment.left.identity, segment.right.identity),
1106 )
1108 def _shared_deletes(self) -> dict[str, tuple[str, ...]]:
1109 """Return, per desired segment, every Logical Cable deletion its creation waits for.
1111 Two traces that state one identical segment share one Planned Change, so the change names
1112 the deletions of both and reads the same in either unit.
1113 """
1114 deletes: dict[str, set] = {}
1115 for analysis in self._writing():
1116 if analysis.delete_identity is not None:
1117 for segment in analysis.pending:
1118 deletes.setdefault(segment.key, set()).add(analysis.delete_identity)
1119 return {key: tuple(sorted(values)) for key, values in deletes.items()}
1121 def _shared_sources(self) -> dict[str, list]:
1122 """Return, per desired segment, the provenance of every Source Trace that states it."""
1123 sources: dict[str, list] = {}
1124 for analysis in self.analyses:
1125 # Unlike _shared_deletes: a repaired trace reuses the Cable, so creation records it or nothing does.
1126 if analysis.invalid:
1127 continue
1128 for segment in analysis.pending:
1129 sources.setdefault(segment.key, []).append(_source_record(analysis.trace, segment.index))
1130 return {
1131 key: sorted(records, key=lambda record: (record["trace_identity"], record["segment_index"]))
1132 for key, records in sources.items()
1133 }
1135 def _writing(self) -> list[_TraceAnalysis]:
1136 """Return the analyses that still contribute writes to this plan."""
1137 return [analysis for analysis in self.analyses if not analysis.stopped and analysis.endpoints]
1139 def _unit(self, analysis: _TraceAnalysis) -> SynchronizationUnit:
1140 """Return the one Synchronization Unit one Source Trace produces."""
1141 changes = self._changes(analysis)
1142 if analysis.invalid:
1143 disposition = Disposition.INVALID
1144 elif analysis.blocked:
1145 disposition = Disposition.BLOCKED
1146 elif changes:
1147 disposition = Disposition.ACTIONABLE
1148 else:
1149 disposition = Disposition.NO_OP
1150 return SynchronizationUnit(
1151 identity=analysis.identity,
1152 disposition=disposition,
1153 changes=changes,
1154 diagnostics=tuple(analysis.diagnostics),
1155 display={
1156 **analysis.display,
1157 "detail": self._detail(analysis, disposition),
1158 "trace": self._workspace(analysis, bool(changes)),
1159 },
1160 )
1162 def _workspace(self, analysis: _TraceAnalysis, writes: bool) -> dict:
1163 """Return what the three review workspace panels show for one Source Trace."""
1164 summary = analysis.trace.endpoint_summary
1165 resolved = {segment.index: segment for segment in analysis.segments}
1166 by_reference = self._resolved.get(analysis.identity, {})
1167 conflicted = {
1168 diagnostic.display.get("segment_index")
1169 for diagnostic in analysis.diagnostics
1170 if diagnostic.code in _CONFLICT_CODES
1171 }
1172 segments = []
1173 stated_segments = list(analysis.trace.segments)
1174 for index, stated in enumerate(stated_segments):
1175 planned = resolved.get(index)
1176 source_left, source_right = _endpoint_label(stated.left), _endpoint_label(stated.right)
1177 left = source_left if planned is None else planned.left.display
1178 right = source_right if planned is None else planned.right.display
1179 segments.append(
1180 {
1181 "index": index,
1182 "cable_class": source_text(stated.cable_class),
1183 "source_left": source_left,
1184 "source_right": source_right,
1185 "left": left,
1186 "right": right,
1187 # A path that re-enters one panel implies a claim the source never states.
1188 "pass_through": index > 0 and same_device_and_cards(stated_segments[index - 1].right, stated.left),
1189 # Planning substituted the entry port only where the source re-used one port.
1190 "substituted": self._entered_through_claim(planned, by_reference.get(stated.left.identity_key)),
1191 "status": self._segment_status(analysis, index, conflicted, planned, writes),
1192 }
1193 )
1194 return {
1195 "identity": analysis.trace.identity,
1196 "endpoints": {
1197 "from": _endpoint_label(summary.from_termination),
1198 "to": _endpoint_label(summary.to_termination),
1199 },
1200 "segments": segments,
1201 "logical_cable": self._logical_cable_display(analysis),
1202 "topology_known": analysis.topology_read,
1203 # A unit with no changes proposes nothing, so the panel must not offer to delete one.
1204 "deletes_logical_cable": writes and analysis.deleted_logical_cable is not None,
1205 "terminations": list(analysis.terminations.values()),
1206 }
1208 @staticmethod
1209 def _entered_through_claim(planned: _DesiredSegment | None, stated: _Termination | None) -> bool:
1210 """Return whether planning entered this segment through a port the source never stated."""
1211 return planned is not None and stated is not None and planned.left.key != stated.key
1213 @staticmethod
1214 def _segment_status(analysis: _TraceAnalysis, index: int, conflicted: set, planned, writes: bool) -> str:
1215 """Return the verdict the proposed panel shows for one segment, or none for an unplanned one."""
1216 if index in conflicted:
1217 return CONFLICT_SEGMENT
1218 if planned is None:
1219 return ""
1220 if index in analysis.proven:
1221 return REUSE_SEGMENT
1222 # A blocked unit contributes no change, so nothing here is going to be created.
1223 return CREATE_SEGMENT if writes else ""
1225 def _logical_cable_display(self, analysis: _TraceAnalysis) -> dict | None:
1226 """Return the Logical Cable NetBox holds for this trace, as far as the actor may see it."""
1227 if analysis.logical_cable is None:
1228 return None
1229 cable = analysis.logical_cable.cable
1230 disclosure, _identities = self._cable_diagnostic_disclosure(cable)
1231 if not disclosure["cable_visible"]:
1232 return {"visible": False, "display": "", "description": "", "tags": []}
1233 # Section 6.3 reviews what the deletion removes, which the deletion payload also carries.
1234 return {
1235 "visible": True,
1236 "display": disclosure["cable"],
1237 "description": cable.description,
1238 "tags": sorted(cable.tags.values_list("name", flat=True)),
1239 }
1241 def _changes(self, analysis: _TraceAnalysis) -> tuple[PlannedChange, ...]:
1242 """Return the deletion and the creations one actionable trace performs, in that order."""
1243 if analysis.stopped:
1244 return ()
1245 changes = []
1246 if analysis.deleted_logical_cable is not None:
1247 changes.append(self._delete_change(analysis.deleted_logical_cable))
1248 changes.extend(self._create_change(segment, analysis.policies[segment.index]) for segment in analysis.pending)
1249 return tuple(changes)
1251 @staticmethod
1252 def _delete_change(logical: _ExistingCable) -> PlannedChange:
1253 """Return the one deletion a Patched Path Replacement ever performs."""
1254 return PlannedChange(
1255 identity=_delete_identity(logical.cable.pk),
1256 target_module=CableModule.key,
1257 operation="delete",
1258 payload={
1259 "cable_id": logical.cable.pk,
1260 "display": str(logical.cable),
1261 "description": logical.cable.description,
1262 "tags": sorted(logical.cable.tags.values_list("name", flat=True)),
1263 },
1264 preconditions={"cable_id": logical.cable.pk, "terminations": logical.terminations},
1265 )
1267 def _create_change(self, segment: _DesiredSegment, policy: dict) -> PlannedChange:
1268 """Return the creation of one physical segment, shared by every trace that states it."""
1269 return PlannedChange(
1270 identity=segment.change_identity,
1271 target_module=CableModule.key,
1272 operation="create",
1273 payload={
1274 "terminations": segment.as_json(),
1275 "status": CABLE_STATUS,
1276 "cable_class": segment.cable_class,
1277 "sources": self._sources_by_segment.get(segment.key, []),
1278 **policy,
1279 },
1280 dependencies=self._deletes_by_segment.get(segment.key, ()),
1281 preconditions={"terminations": segment.as_json()},
1282 )
1284 @staticmethod
1285 def _detail(analysis: _TraceAnalysis, disposition: str) -> str:
1286 """Return the operator wording one unit shows before the trace workspace exists."""
1287 if disposition == Disposition.NO_OP:
1288 return "The stated path already exists."
1289 if disposition != Disposition.ACTIONABLE:
1290 return ""
1291 count = len(analysis.pending)
1292 if analysis.deleted_logical_cable is not None:
1293 return f"Would replace the logical cable with {count} physical segment(s)."
1294 return f"Would create {count} physical segment(s)."
1297class CableModule:
1298 """Plan and write the Cables one Source Trace states, as a Patched Path Replacement."""
1300 key = TargetModuleKey.CABLE
1301 consumes = frozenset({OutputKind.SOURCE_TRACE})
1303 def plan(
1304 self,
1305 source_batch,
1306 profile,
1307 catalog,
1308 netbox_reader,
1309 *,
1310 lock_plan_references: bool = False,
1311 ) -> list[SynchronizationUnit]:
1312 """Return one unit per trace and optionally lock rows that prove the plan."""
1313 del catalog
1314 if not (self.consumes & source_batch.output_kinds):
1315 return []
1316 return _CableBatch(
1317 source_batch.rows,
1318 profile,
1319 netbox_reader,
1320 lock_plan_references=lock_plan_references,
1321 ).units()
1323 def apply(self, planned_change: PlannedChange, execution_context) -> Any:
1324 """Apply one Cable change, having locked its rows and rechecked its preconditions."""
1325 if planned_change.operation == "delete":
1326 return self._delete(planned_change, execution_context)
1327 if planned_change.operation == "create":
1328 return self._create(planned_change, execution_context)
1329 raise PreconditionFailed(f"The Cable Target Module cannot apply operation '{planned_change.operation}'.")
1331 @staticmethod
1332 def _delete(planned_change: PlannedChange, execution_context) -> DeletedObject:
1333 """Remove the one direct Logical Cable this unit replaces."""
1334 from dcim.models import Cable
1336 cable_id = planned_change.preconditions["cable_id"]
1337 cable = Cable.objects.filter(pk=cable_id).select_for_update(of=("self",)).first()
1338 if cable is None:
1339 raise PreconditionFailed(f"Cable {cable_id} is gone, so the logical cable cannot be replaced.")
1340 current = _cable_terminations(cable_id)
1341 if current != [list(item) for item in planned_change.preconditions["terminations"]]:
1342 raise PreconditionFailed(f"Cable {cable_id} was re-terminated after the plan was made.")
1343 enforce_saved_object_permission(cable, execution_context.actor, "delete")
1344 # The audit row is the only record left of this Cable, so it keeps what the row carried.
1345 snapshot = DeletedObject(
1346 object_type="dcim.cable",
1347 object_id=cable_id,
1348 display=str(cable),
1349 detail={
1350 "terminations": current,
1351 "description": cable.description,
1352 "tags": sorted(cable.tags.values_list("name", flat=True)),
1353 },
1354 )
1355 cable.delete()
1356 return snapshot
1358 @staticmethod
1359 def _create(planned_change: PlannedChange, execution_context) -> Any:
1360 """Create one physical Cable segment and the provenance rows its Source Traces earn."""
1361 from dcim.models import Cable
1363 payload = planned_change.payload
1364 ends = [
1365 CableModule._free_termination(label, object_id, device_id)
1366 for label, object_id, device_id in payload["terminations"]
1367 ]
1368 cable = Cable(
1369 type=payload["cable_type"],
1370 profile=payload["cable_profile"] or "",
1371 status=payload["status"],
1372 a_terminations=[ends[0]],
1373 b_terminations=[ends[1]],
1374 )
1375 cable.full_clean()
1376 cable.save()
1377 enforce_saved_object_permission(cable, execution_context.actor, "add")
1378 CableModule._store_provenance(cable, payload, execution_context.profile)
1379 return cable
1381 @staticmethod
1382 def _free_termination(label: str, object_id: int, device_id: int):
1383 """Return one locked termination that is still free and on its resolved Device."""
1384 component = _model_for_label(label).objects.filter(pk=object_id).select_for_update(of=("self",)).first()
1385 if component is None:
1386 raise PreconditionFailed(f"{label} {object_id} is gone, so the segment cannot be created.")
1387 if component.device_id != device_id:
1388 raise PreconditionFailed(f"{label} {object_id} moved to another Device after the plan was made.")
1389 if component.cable_id is not None:
1390 raise PreconditionFailed(f"{component} received a cable after the plan was made.")
1391 return component
1393 @staticmethod
1394 def _store_provenance(cable, payload, profile) -> None:
1395 """Write one provenance row per Source Trace that states this segment."""
1396 from .models import CableImportSource, index_digest
1398 for record in payload["sources"]:
1399 # The unique constraint carries the digest, so the lookup matches it.
1400 CableImportSource.objects.update_or_create(
1401 cable=cable,
1402 profile=profile,
1403 trace_key=index_digest(record["trace_identity"]),
1404 defaults={
1405 "trace_identity": record["trace_identity"],
1406 "segment_index": record["segment_index"],
1407 "from_text": record["from_text"],
1408 "to_text": record["to_text"],
1409 "direction": record["direction"],
1410 "workbook_fingerprint": record["workbook_fingerprint"],
1411 "sheet": record["sheet"],
1412 "block_ordinal": record["block_ordinal"],
1413 "row_start": record["row_start"],
1414 "row_end": record["row_end"],
1415 "export_timestamp": record["export_timestamp"],
1416 },
1417 )
1420def _cable_terminations(cable_id: int) -> list:
1421 """Return one Cable's termination pairs in the order a precondition records them."""
1422 from dcim.models import CableTermination
1424 return sorted(
1425 [_object_type_label(row.termination_type.model_class()), row.termination_id]
1426 for row in CableTermination.objects.filter(cable_id=cable_id).order_by("pk").select_for_update(of=("self",))
1427 )
1430__all__ = (
1431 "AUTOMATICALLY_RESOLVED",
1432 "CABLE_STATUS",
1433 "CONFLICT_SEGMENT",
1434 "CREATE_SEGMENT",
1435 "ELIGIBLE_TERMINATION_LIMIT",
1436 "MANUALLY_RESOLVED",
1437 "REUSE_SEGMENT",
1438 "UNRESOLVED",
1439 "CableModule",
1440 "EligibleTerminations",
1441 "eligible_terminations",
1442)