Coverage for netbox_data_import/cable_target.py: 99.25%
1197 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"""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 collections.abc import Mapping
21from dataclasses import dataclass, field
22from functools import partial
23from typing import Any
25from . import cable_harness
26from .cable_disclosure import (
27 DEVICE_ROW,
28 WORDING_CODES,
29 DISCLOSURE_SOURCE,
30 SEGMENT_END_SOURCES,
31 wording_sources,
32 device_source,
33 disclosed_cable,
34 disclosed_policy,
35 disclosure_source,
36 segment_end_sources,
37 termination_sources,
38)
39from .catalog import OutputKind, TargetModuleKey
40from .field_keys import (
41 ADMITTED_TERMINATION_MODELS,
42 CABLE_END_KINDS,
43 CABLE_END_MODELS,
44 FRONT_PORT_MODEL,
45 MAPPED_PEER_ROLE,
46 REAR_PORT_MODEL,
47 SELECT_TERMINATION_TASK,
48 TERMINATION_ROLE,
49 claimed_termination_kind,
50 is_outside_claim,
51 parse_termination_field_key,
52 same_device_and_cards,
53 termination_field_key,
54)
55from .identity import CANONICAL_NAME, identity_text, matching_search, with_name_identity
56from .object_permissions import enforce_saved_object_permission
57from .plan import (
58 Diagnostic,
59 Disposition,
60 PlannedChange,
61 Severity,
62 SynchronizationUnit,
63 fingerprint_of,
64 validate_diagnostic_display,
65)
66from .target_runtime import DeletedObject, PreconditionFailed
67from .trace_device_resolution import (
68 STALE as DEVICE_STALE,
69 collect_trace_device_evidence,
70 resolve_trace_devices,
71 resolved_trace_device,
72 source_device_key,
73)
74from .values import source_text
76CABLE_STATUS = "connected"
77ELIGIBLE_TERMINATION_LIMIT = 20
79# Section 10.2 badge vocabulary. T8 adds the proposal states to the same set.
80UNRESOLVED = "unresolved"
81AUTOMATICALLY_RESOLVED = "automatically resolved"
82MANUALLY_RESOLVED = "manually resolved"
84CREATE_SEGMENT = "create"
85REUSE_SEGMENT = "reuse existing"
87# A PortMapping joins a front port to a rear port; no other Cable End Kind passes a path through.
88_PASS_THROUGH_PEERS = {FRONT_PORT_MODEL: REAR_PORT_MODEL, REAR_PORT_MODEL: FRONT_PORT_MODEL}
91@dataclass(frozen=True)
92class EligibleTerminations:
93 """One capped page of eligible terminations and its uncapped matching total."""
95 candidates: tuple[Any, ...]
96 total: int
97 # A mapped-peer read offers the opposite ports of a resolved base, so no claim applies to them.
98 claimed_kind: str | None = None
100 def outside_claim(self, candidate) -> bool:
101 """Return whether one offered candidate is a model the source PortClass claim does not admit."""
102 return self.claimed_kind is not None and is_outside_claim(self.claimed_kind, candidate._meta.label_lower)
105def _model_for_label(label: str):
106 """Return the model class one recorded object-type label names."""
107 from django.apps import apps
109 app_label, _, model_name = label.partition(".")
110 return apps.get_model(app_label, model_name)
113def _view_permission(label: str) -> str:
114 """Return the NetBox view permission for the model one object-type label names."""
115 app_label, _, model_name = label.partition(".")
116 return f"{app_label}.view_{model_name}"
119def _stored_label(stored) -> str:
120 """Return the object-type label one saved termination decision selected."""
121 return f"{stored.selected_object_type.app_label}.{stored.selected_object_type.model}"
124def _object_identity(label: str, object_id: int) -> str:
125 """Return the plan identity string for one target object."""
126 return f"{label}:{object_id}"
129@dataclass(frozen=True)
130class _Termination:
131 """One source Termination Reference bound to a NetBox object."""
133 object_type: str
134 object_id: int
135 device_id: int
136 display: str
137 topology_display: str
139 @property
140 def model_name(self) -> str:
141 """Return the NetBox model name, the form NetBox's compatible-termination table uses."""
142 return self.object_type.partition(".")[2]
144 @property
145 def key(self) -> tuple[str, int]:
146 """Return the comparison key that names this object across queries."""
147 return self.object_type, self.object_id
149 @property
150 def identity(self) -> str:
151 """Return the plan identity string for this object."""
152 return _object_identity(self.object_type, self.object_id)
154 def as_json(self) -> list:
155 """Return the serializable object and resolved Device a creation precondition carries."""
156 return [self.object_type, self.object_id, self.device_id]
159@dataclass(frozen=True)
160class _DesiredSegment:
161 """One physical Cable the trace states."""
163 index: int
164 left: _Termination
165 right: _Termination
166 cable_class: str
168 @property
169 def key(self) -> str:
170 """Return the direction-independent key two traces share for one identical segment."""
171 first, second = sorted((self.left.key, self.right.key))
172 return f"{_object_identity(*first)}|{_object_identity(*second)}"
174 @property
175 def change_identity(self) -> str:
176 """Return the Planned Change identity this segment creates under."""
177 return f"cable:create:{self.key}"
179 @property
180 def terminations(self) -> tuple[_Termination, _Termination]:
181 """Return both ends in canonical order."""
182 return self.left, self.right
184 def as_json(self) -> list:
185 """Return the sorted termination records a payload and a precondition carry."""
186 return sorted((self.left.as_json(), self.right.as_json()))
189@dataclass(frozen=True)
190class _ExistingCable:
191 """One existing Cable and the termination sets it holds."""
193 cable: Any
194 a_side: frozenset
195 b_side: frozenset
196 # Side, connector, positions and object of every CableTermination row: what a harness reuse compares.
197 snapshot: frozenset = frozenset()
199 @property
200 def multi_termination(self) -> bool:
201 """Return whether either side holds more than one termination."""
202 return len(self.a_side) > 1 or len(self.b_side) > 1
204 @property
205 def terminations(self) -> list:
206 """Return the sorted termination pairs a precondition records."""
207 return sorted([label, object_id] for label, object_id in self.a_side | self.b_side)
209 def satisfies(self, segment: _DesiredSegment) -> bool:
210 """Return whether this Cable is exactly the unordered pair *segment* states."""
211 return self.connects(segment.left, segment.right)
213 def connects(self, first: _Termination, second: _Termination) -> bool:
214 """Return whether this Cable directly joins the two given terminations, one end each."""
215 return (self.a_side, self.b_side) in (
216 ({first.key}, {second.key}),
217 ({second.key}, {first.key}),
218 )
221def _delete_identity(cable_pk: int) -> str:
222 """Return the Planned Change identity that removes one Logical Cable.
224 The unit's dependency graph and the deletion change both name it, so one helper states it.
225 """
226 return f"cable:delete:{cable_pk}"
229def _cable_review_snapshot(cable) -> dict:
230 """Read the Logical Cable metadata that review and deletion must share."""
231 return {
232 "display": str(cable),
233 "description": cable.description,
234 "tags": tuple(sorted(cable.tags.values_list("name", flat=True))),
235 }
238def _deleted_cable_review_fingerprint(cable) -> str:
239 """Hash the review metadata whose plaintext must not survive Cable deletion."""
240 return fingerprint_of(_cable_review_snapshot(cable))
243@dataclass(frozen=True, eq=False)
244class _Leg:
245 """One trace's segment from a channel child to the far end its parent's harness cables."""
247 analysis: Any
248 segment: _DesiredSegment
249 child: _Termination
250 far: _Termination
253@dataclass(eq=False)
254class _Harness:
255 """One channelized parent, the legs the batch states on it, and the traces that block it."""
257 parent: cable_harness.Parent
258 termination: _Termination
259 legs: list = field(default_factory=list)
260 unsupported: list = field(default_factory=list)
261 unresolved: list = field(default_factory=list)
262 unchannelized: list = field(default_factory=list)
263 intent: cable_harness.CableIntent | None = None
264 cable_type: str | None = None
265 reused: _ExistingCable | None = None
266 # Every port of the harness with its Device, and the sources they give every finding about it.
267 wording_ends: list = field(default_factory=list)
268 wording: dict = field(default_factory=dict)
270 @property
271 def change_identity(self) -> str:
272 """Return the Planned Change identity every leg unit carries for this harness."""
273 return f"cable:create:harness:{self.termination.identity}"
275 @property
276 def pending(self) -> bool:
277 """Return whether the harness can be written and no existing Cable already is it."""
278 return self.intent is not None and self.reused is None
280 @property
281 def ends(self) -> list[_Termination]:
282 """Return the parent and every far end the harness claims."""
283 return [self.termination, *(leg.far for leg in self.legs)]
285 def settle_wording(self, resolved: Mapping[str, Mapping[tuple, _Termination]]) -> None:
286 """Bind every finding about this harness to every port it involves, each with its Device.
288 A finding on one leg can describe the parent, a sibling child, another leg's far end or a trace
289 that blocks the harness, so the whole harness is one disclosure unit (section 6.6).
290 """
291 ends = [self.termination, *(end for leg in self.legs for end in (leg.child, leg.far))]
292 blockers = (*self.unsupported, *self.unresolved, *(analysis for analysis, _port in self.unchannelized))
293 ends.extend(end for analysis in blockers for end in resolved.get(analysis.identity, {}).values())
294 children = [
295 ((child.component._meta.label_lower, child.component.pk), child.component.device_id)
296 for child in self.parent.children
297 if child.channel is not None
298 ]
299 self.wording_ends = [*((end.key, end.device_id) for end in ends), *children]
300 self.wording = wording_sources(self.wording_ends)
302 def companions(self, leg: _Leg) -> tuple[str, ...]:
303 """Return the unit identity of every other leg, which must be selected together with *leg*."""
304 return tuple(sorted(other.analysis.identity for other in self.legs if other is not leg))
307@dataclass
308class _TraceAnalysis:
309 """What one Source Trace contributes, built in planning order."""
311 trace: Any
312 identity: str
313 display: dict
314 diagnostics: list = field(default_factory=list)
315 endpoints: tuple = ()
316 segments: list = field(default_factory=list)
317 segment_pairs_resolved: bool = False
318 proven: dict = field(default_factory=dict)
319 policies: dict = field(default_factory=dict)
320 # Index i means a verified PortMapping joins segment i to segment i + 1.
321 joined: set = field(default_factory=set)
322 devices: dict = field(default_factory=dict)
323 terminations: dict = field(default_factory=dict)
324 resolution_started: bool = False
325 topology_read: bool = False
326 logical_cable: Any = None
327 retains_logical_cable: bool = False
328 blocked: bool = False
329 invalid: bool = False
330 # The one segment that leaves a channel child, which the harness of its parent writes.
331 leg: Any = None
332 harness: Any = None
333 selected_with: tuple = ()
335 @property
336 def stopped(self) -> bool:
337 """Return whether a finding makes further planning for this trace meaningless."""
338 return self.blocked or self.invalid
340 def is_leg(self, segment: _DesiredSegment) -> bool:
341 """Return whether *segment* is this trace's harness leg rather than an ordinary segment."""
342 return self.leg is not None and segment.index == self.leg.segment.index
344 @property
345 def pending(self) -> list:
346 """Return the ordinary desired segments no existing Cable already proves."""
347 return [segment for segment in self.segments if segment.index not in self.proven and not self.is_leg(segment)]
349 @property
350 def harness_pending(self) -> bool:
351 """Return whether this trace's harness still has to be created."""
352 return self.harness is not None and self.harness.pending
354 @property
355 def deleted_logical_cable(self):
356 """Return the Logical Cable this trace removes, which a Cable that proves its path is not."""
357 return None if self.retains_logical_cable else self.logical_cable
359 @property
360 def delete_identity(self) -> str | None:
361 """Return the Planned Change identity that removes this trace's Logical Cable."""
362 removed = self.deleted_logical_cable
363 return None if removed is None else _delete_identity(removed.cable.pk)
365 def error(self, code: str, display: dict, identities=(), evidence=None) -> None:
366 """Record one blocking or invalidating finding."""
367 validate_diagnostic_display(code, display)
368 self.diagnostics.append(
369 Diagnostic(
370 code=code,
371 severity=Severity.ERROR,
372 identities=tuple(identities),
373 display=display,
374 evidence=evidence or {},
375 )
376 )
378 def note(self, code: str, display: dict, identities=(), evidence=None) -> None:
379 """Record one review note whose identities keep the unit honest about live state."""
380 validate_diagnostic_display(code, display)
381 self.diagnostics.append(
382 Diagnostic(
383 code=code,
384 severity=Severity.INFO,
385 identities=tuple(identities),
386 display=display,
387 evidence=evidence or {},
388 )
389 )
391 def warn(self, code: str, display: dict, identities=(), evidence=None) -> None:
392 """Record a finding the operator has to see, which changes no disposition."""
393 validate_diagnostic_display(code, display)
394 self.diagnostics.append(
395 Diagnostic(
396 code=code,
397 severity=Severity.WARNING,
398 identities=tuple(identities),
399 display=display,
400 evidence=evidence or {},
401 )
402 )
404 def block(self, code: str, display: dict, identities=(), evidence=None) -> None:
405 """Record a finding an operator decision or a NetBox correction can resolve."""
406 self.error(code, display, identities, evidence)
407 self.blocked = True
409 def refuse(self, code: str, display: dict, identities=()) -> None:
410 """Record a finding no decision inside this plugin can resolve."""
411 self.error(code, display, identities)
412 self.invalid = True
414 def reopen(self, termination_key: tuple[str, int]) -> None:
415 """Keep each end resolved to one termination in attention, so its picker can choose another."""
416 source = disclosure_source(*termination_key)
417 for record in self.terminations.values():
418 if record.get(DISCLOSURE_SOURCE) == source:
419 record["incompatible"] = True
422def _endpoint_label(reference) -> str:
423 """Return the operator-facing name of one Termination Reference."""
424 parts = [source_text(reference.device), source_text(reference.cards), source_text(reference.port)]
425 return " ".join(part for part in parts if part)
428def _reference_display(reference) -> dict:
429 """Return the source values one Termination Reference states."""
430 return {
431 "device": source_text(reference.device),
432 "cards": source_text(reference.cards),
433 "port": source_text(reference.port),
434 "port_class": source_text(reference.port_class),
435 }
438def _field_key(reference, role: str = TERMINATION_ROLE) -> str:
439 """Return the canonical field key one Termination Reference resolves under."""
440 return termination_field_key(
441 device=reference.device,
442 cards=reference.cards,
443 port=reference.port,
444 kind=claimed_termination_kind(reference.port_class),
445 role=role,
446 )
449def _source_record(trace, segment_index: int) -> dict:
450 """Return the provenance one Source Trace contributes to one created Cable."""
451 provenance = trace.provenance[0]
452 return {
453 "trace_identity": trace.identity,
454 "segment_index": segment_index,
455 "from_text": provenance.from_text,
456 "to_text": provenance.to_text,
457 "direction": provenance.direction,
458 "workbook_fingerprint": provenance.workbook_fingerprint,
459 "sheet": provenance.sheet,
460 "block_ordinal": provenance.block_ordinal,
461 "row_start": provenance.row_start,
462 "row_end": provenance.row_end,
463 "export_timestamp": provenance.export_timestamp,
464 }
467def resolved_device_for(field_key: str, netbox_reader, *, profile, _lock_rows=False):
468 """Return the one visible Device a termination field key names, or None when it names no single one.
470 A proposal freezes this reference and revalidates it, so the retrieval and the freshness check
471 have to agree on what "the resolved Device" means.
472 """
473 key = parse_termination_field_key(field_key)["device"]
474 return resolved_trace_device(profile=profile, reader=netbox_reader, key=key, lock_rows=_lock_rows)
477def _named_terminations(netbox_reader, label: str, device_id: int, identities) -> dict[str, list]:
478 """Return one Device's rows of one model whose name identity is wanted, each marked as visible or not.
480 The read is unscoped, so the actor's view scope cannot change which model the exact-name rule
481 stops at. Only the visibility mark comes from the actor's scope.
482 """
483 from django.db.models import Exists, OuterRef
485 rows = (
486 with_name_identity(_model_for_label(label).objects.filter(device_id=device_id))
487 .filter(**{f"{CANONICAL_NAME}__in": sorted(identities)})
488 .annotate(actor_may_view=Exists(netbox_reader.terminations(label).filter(pk=OuterRef("pk"))))
489 .order_by("pk")
490 )
491 named: dict[str, list] = {}
492 for row in rows:
493 named.setdefault(getattr(row, CANONICAL_NAME), []).append(row)
494 return named
497def _exact_name_match(kind: str, identity: str, named_for) -> Any | None:
498 """Return the one row the section 6.1 exact-name rule resolves a port *identity* to, or None.
500 The rule stops at the first admitted model with any match. Several matches or a hidden one leave
501 the field open there, and never fall through to a later model.
502 """
503 for label in ADMITTED_TERMINATION_MODELS[kind]:
504 matches = named_for(label).get(identity, [])
505 if matches:
506 return matches[0] if len(matches) == 1 and matches[0].actor_may_view else None
507 return None
510def _base_termination(stored, kind: str, identity: str, device_id: int, netbox_reader, named_for) -> Any | None:
511 """Return the row one termination role resolves to, or None while it stays open.
513 A saved selection decides first, and an unavailable one never falls back to the exact-name rule.
514 """
515 if stored is None:
516 return _exact_name_match(kind, identity, named_for)
517 label = _stored_label(stored)
518 if label not in CABLE_END_KINDS:
519 return None
520 return netbox_reader.terminations(label).filter(pk=stored.selected_object_id, device_id=device_id).first()
523def _mapped_peer_sources(parsed: dict, device, netbox_reader, profile) -> tuple:
524 """Return the opposite ports one resolved pass-through port maps to, as (label, rows) sources."""
525 from .models import TerminationResolution, index_digest
527 base_field_key = termination_field_key(
528 device=parsed["device"],
529 cards=parsed["cards"],
530 port=parsed["port"],
531 kind=parsed["kind"],
532 role=TERMINATION_ROLE,
533 )
534 stored = (
535 TerminationResolution.objects.filter(
536 profile=profile,
537 task_type=SELECT_TERMINATION_TASK,
538 field_key=base_field_key,
539 field_key_digest=index_digest(base_field_key),
540 )
541 .select_related("selected_object_type")
542 .first()
543 )
544 identity = parsed["port"]
545 base = _base_termination(
546 stored,
547 parsed["kind"],
548 identity,
549 device.pk,
550 netbox_reader,
551 partial(_named_terminations, netbox_reader, device_id=device.pk, identities=(identity,)),
552 )
553 if base is None or base._meta.label_lower not in _PASS_THROUGH_PEERS:
554 return ()
555 base_label = base._meta.label_lower
556 mappings = netbox_reader.port_mappings()
557 if base_label == FRONT_PORT_MODEL:
558 peer_ids = mappings.filter(front_port_id=base.pk).values_list("rear_port_id", flat=True)
559 else:
560 peer_ids = mappings.filter(rear_port_id=base.pk).values_list("front_port_id", flat=True)
561 peer_label = _PASS_THROUGH_PEERS[base_label]
562 return ((peer_label, netbox_reader.terminations(peer_label).filter(device_id=device.pk, pk__in=peer_ids)),)
565_RESOLVED_DEVICE_UNSET = object()
568def eligible_terminations(
569 field_key: str,
570 netbox_reader,
571 *,
572 profile,
573 search: str = "",
574 limit: int = ELIGIBLE_TERMINATION_LIMIT,
575 offset: int = 0,
576 _resolved_device=_RESOLVED_DEVICE_UNSET,
577 _lock_rows: bool = False,
578) -> EligibleTerminations:
579 """Return one page of candidates for a canonical termination field key.
581 The termination role offers every Cable End Kind on the resolved Device, as one set with one
582 count, one bound, and one order, the claimed models first (section 6.1). The mapped-peer role
583 starts from the base rule, the profile's saved selection or else the exact-name match, then
584 offers its opposite ports. The reader keeps both inside the actor's view scope. The picker and a
585 proposal request share this query. The models count and page in the database as one set, so no
586 read holds more than *limit* rows.
587 """
588 parsed = parse_termination_field_key(field_key)
589 device = (
590 resolved_device_for(field_key, netbox_reader, profile=profile, _lock_rows=_lock_rows)
591 if _resolved_device is _RESOLVED_DEVICE_UNSET
592 else _resolved_device
593 )
594 claimed_kind = None if parsed["role"] == MAPPED_PEER_ROLE else parsed["kind"]
595 if device is None:
596 return EligibleTerminations(candidates=(), total=0, claimed_kind=claimed_kind)
597 if claimed_kind is None:
598 sources = _mapped_peer_sources(parsed, device, netbox_reader, profile)
599 else:
600 sources = tuple(
601 (label, netbox_reader.terminations(label).filter(device_id=device.pk)) for label in CABLE_END_MODELS
602 )
603 wanted = identity_text(search)
604 sources = tuple((label, matching_search(rows, search)) for label, rows in sources)
605 if _lock_rows:
606 # One global order, object type then id, as the segment locks take it.
607 for _label, rows in sorted(sources, key=lambda source: source[0]):
608 tuple(rows.order_by("pk").select_for_update(of=("self",)).values_list("pk", flat=True))
609 total = sum(rows.count() for _label, rows in sources)
610 if offset >= total:
611 return EligibleTerminations(candidates=(), total=total, claimed_kind=claimed_kind)
612 ranked = [
613 _ranked_keys(rows, wanted, order, unclaimed=claimed_kind is not None and is_outside_claim(claimed_kind, label))
614 for order, (label, rows) in enumerate(sources)
615 ]
616 combined = ranked[0].union(*ranked[1:], all=True) if len(ranked) > 1 else ranked[0]
617 page = combined.order_by("-_ndi_exact", "_ndi_unclaimed", "_ndi_name", "_ndi_order", "_ndi_pk")[
618 offset : offset + limit
619 ]
620 wanted_ids: dict[int, list[int]] = {}
621 for _exact, _unclaimed, _name, order, pk in page:
622 wanted_ids.setdefault(order, []).append(pk)
623 loaded = {order: sources[order][1].in_bulk(ids) for order, ids in wanted_ids.items()}
624 # A row deleted, moved, or renamed after the ranking query is no longer offered.
625 return EligibleTerminations(
626 candidates=tuple(loaded[order][pk] for _exact, _unclaimed, _name, order, pk in page if pk in loaded[order]),
627 total=total,
628 claimed_kind=claimed_kind,
629 )
632def _ranked_keys(rows, wanted: str, order: int, *, unclaimed: bool):
633 """Return one model's order keys for the combined order, which the database pages as one set.
635 The bytewise collation orders names the same way across every Cable End Kind.
636 """
637 from django.db.models import Case, F, IntegerField, Value, When
638 from django.db.models.functions import Collate
640 if wanted:
641 exact = Case(When(**{CANONICAL_NAME: wanted}, then=Value(1)), default=Value(0), output_field=IntegerField())
642 rows = with_name_identity(rows)
643 else:
644 exact = Value(0, output_field=IntegerField())
645 return (
646 rows.order_by()
647 .annotate(
648 _ndi_exact=exact,
649 _ndi_unclaimed=Value(int(unclaimed), output_field=IntegerField()),
650 _ndi_name=Collate("name", "C"),
651 _ndi_order=Value(order, output_field=IntegerField()),
652 _ndi_pk=F("pk"),
653 )
654 .values_list("_ndi_exact", "_ndi_unclaimed", "_ndi_name", "_ndi_order", "_ndi_pk")
655 )
658class _CableBatch:
659 """Plan every Source Trace in one batch, so an identical shared segment plans once."""
661 def __init__(self, traces, profile, netbox_reader, *, lock_plan_references: bool = False):
662 traces = tuple(traces)
663 self.profile = profile
664 self.reader = netbox_reader
665 self.actor = netbox_reader.actor
666 self.lock_plan_references = lock_plan_references
667 self.analyses = [self._new_analysis(trace) for trace in traces]
668 self._objects: dict[tuple[str, int], Any] = {}
669 self._visible: dict[tuple[str, int], bool] = {}
670 self._named: dict[tuple[int, str], dict[str, list]] = {}
671 self._asked_identities: dict[int, set[str]] = {}
672 self._resolved: dict[str, dict[tuple, _Termination]] = {}
673 self._mappings_by_front: dict[int, list] = {}
674 self._mappings_by_rear: dict[int, list] = {}
675 self._mapping_ports: dict[tuple[str, int], Any] = {}
676 self._existing: dict[int, _ExistingCable] = {}
677 self._cable_review_snapshots: dict[int, dict] = {}
678 self._occupied: dict[tuple[str, int], _ExistingCable] = {}
679 self._mapping_rows: dict[str, Any] | None = None
680 self._overrides: dict[str, Any] = {}
681 self._overrides_by_trace: dict[str, list] = {}
682 self._parents: dict[int, cable_harness.Parent] = {}
683 self._harnesses: dict[int, _Harness] = {}
684 self._unsupported_traces: list[tuple[_TraceAnalysis, _Harness, int]] = []
685 self._stored = self._stored_resolutions()
686 active_traces = (analysis.trace for analysis in self.analyses if not analysis.stopped)
687 self._device_evidence = collect_trace_device_evidence(active_traces)
688 self._device_resolutions = resolve_trace_devices(
689 profile=self.profile,
690 reader=self.reader,
691 evidence=self._device_evidence,
692 lock_rows=self.lock_plan_references,
693 )
694 self._device_displays = {
695 resolution.device.pk: str(resolution.device)
696 for resolution in self._device_resolutions.values()
697 if resolution.device is not None
698 }
699 self._collect_asked_ports()
700 self._resolve_terminations()
701 self._load_mappings()
702 self._build_segments()
703 self._load_overrides()
704 self._lock_harness_parents()
705 self._lock_segment_terminations()
706 self._find_legs()
707 self._load_existing_cables()
708 self._classify()
709 self._plan_harnesses()
710 self._decide()
711 self._assess_media()
712 self._block_planned_termination_conflicts()
713 self._block_conflicting_creations()
714 self._deletes_by_segment = self._shared_deletes()
715 self._sources_by_segment = self._shared_sources()
717 def units(self) -> list[SynchronizationUnit]:
718 """Return one Synchronization Unit per Source Trace, in batch order."""
719 return [self._unit(analysis) for analysis in self.analyses]
721 def _new_analysis(self, trace) -> _TraceAnalysis:
722 """Return the analysis record one Source Trace starts from."""
723 summary = trace.endpoint_summary
724 provenance = trace.provenance[0]
725 display = {
726 "name": f"{_endpoint_label(summary.from_termination)} to {_endpoint_label(summary.to_termination)}",
727 "row_number": provenance.row_start,
728 "sheet": provenance.sheet,
729 "trace_identity": trace.identity,
730 "from_text": provenance.from_text,
731 "to_text": provenance.to_text,
732 "occurrences": [{"sheet": item.sheet, "row_number": item.row_start} for item in trace.provenance],
733 "segment_count": len(trace.segments),
734 }
735 analysis = _TraceAnalysis(trace=trace, identity=f"cable:trace:{trace.identity}", display=display)
736 for error in trace.errors:
737 analysis.refuse(error.code, {"message": error.message, "row_number": error.row_number})
738 return analysis
740 @staticmethod
741 def _references(trace) -> list:
742 """Return every Termination Reference one trace resolves, endpoints included."""
743 summary = trace.endpoint_summary
744 references = [summary.from_termination, summary.to_termination]
745 for segment in trace.segments:
746 references.extend((segment.left, segment.right))
747 return references
749 def _stored_resolutions(self) -> dict[str, Any]:
750 """Return the operator's saved termination decisions, keyed by field key."""
751 from .models import TerminationResolution
753 field_keys = set()
754 for analysis in self.analyses:
755 if analysis.stopped:
756 continue
757 for reference in self._references(analysis.trace):
758 field_keys.add(_field_key(reference))
759 field_keys.add(_field_key(reference, MAPPED_PEER_ROLE))
760 if not field_keys:
761 return {}
762 from .models import index_digest
764 # The unique constraint indexes the digest, and nothing indexes the unbounded field key.
765 rows = TerminationResolution.objects.filter(
766 profile=self.profile,
767 task_type=SELECT_TERMINATION_TASK,
768 field_key_digest__in=sorted(index_digest(key) for key in field_keys),
769 ).select_related("selected_object_type")
770 return {row.field_key: row for row in rows}
772 def _collect_asked_ports(self) -> None:
773 """Collect, per resolved Device, the name identity of every port the exact-name rule is asked about."""
774 for analysis in self.analyses:
775 if analysis.stopped:
776 continue
777 for reference in self._references(analysis.trace):
778 device = self._device_resolutions[source_device_key(reference.device)].device
779 if device is not None and _field_key(reference) not in self._stored:
780 self._asked_identities.setdefault(device.pk, set()).add(identity_text(reference.port))
782 def _named_for(self, device_id: int, label: str) -> dict[str, list]:
783 """Return one Device's asked rows of one model by name identity, read once per batch."""
784 if (device_id, label) not in self._named:
785 self._named[(device_id, label)] = _named_terminations(
786 self.reader, label, device_id, self._asked_identities[device_id]
787 )
788 return self._named[(device_id, label)]
790 def _resolve_terminations(self) -> None:
791 """Bind every Termination Reference to one NetBox object, or record why it stays open."""
792 for analysis in self.analyses:
793 if analysis.stopped:
794 continue
795 analysis.resolution_started = True
796 resolved: dict[tuple, _Termination] = {}
797 for reference in self._references(analysis.trace):
798 if reference.identity_key in resolved:
799 continue
800 termination = self._resolve_one(analysis, reference)
801 if termination is not None:
802 resolved[reference.identity_key] = termination
803 self._resolved[analysis.identity] = resolved
805 def _resolve_one(self, analysis: _TraceAnalysis, reference) -> _Termination | None:
806 """Return the NetBox object one Termination Reference names, or record the open decision."""
807 resolution = self._device_resolutions[source_device_key(reference.device)]
808 question = resolution.to_question()
809 if resolution.device is not None:
810 question[DISCLOSURE_SOURCE] = device_source(resolution.device.pk)
811 analysis.devices.setdefault(resolution.evidence.key, question)
812 if resolution.device is None:
813 code = "trace.device_resolution_stale" if resolution.state == DEVICE_STALE else "trace.device_unresolved"
814 analysis.block(
815 code,
816 {**_reference_display(reference), "matches": resolution.exact_match_count},
817 )
818 # With no resolved Device there is nothing to pick from, so the picker cannot help here.
819 self._record_resolution(
820 analysis,
821 reference,
822 UNRESOLVED,
823 None,
824 reason="Resolve the source Device before choosing one of its ports.",
825 )
826 return None
827 device = resolution.device
828 stored = self._stored.get(_field_key(reference))
829 if stored is not None and _stored_label(stored) not in CABLE_END_KINDS:
830 analysis.refuse(
831 "cable.unsupported_termination_kind",
832 {**_reference_display(reference), "selected_object_type": _stored_label(stored)},
833 )
834 self._record_resolution(analysis, reference, UNRESOLVED, None)
835 return None
836 claimed_kind = claimed_termination_kind(reference.port_class)
837 base = _base_termination(
838 stored,
839 claimed_kind,
840 identity_text(reference.port),
841 device.pk,
842 self.reader,
843 partial(self._named_for, device.pk),
844 )
845 if base is None:
846 # A count would disclose hidden matches, and a saved name may belong to a hidden port.
847 analysis.block("cable.termination_unresolved", _reference_display(reference))
848 self._record_resolution(analysis, reference, UNRESOLVED, None)
849 return None
850 termination = self._termination(base)
851 if stored is not None and is_outside_claim(claimed_kind, termination.object_type):
852 analysis.warn(
853 "cable.termination_kind_mismatch",
854 {
855 **_reference_display(reference),
856 "claimed_kind": claimed_kind,
857 "selected_object_type": termination.object_type,
858 **wording_sources([(termination.key, termination.device_id)]),
859 },
860 identities=(termination.identity,),
861 )
862 state = AUTOMATICALLY_RESOLVED if stored is None else MANUALLY_RESOLVED
863 self._record_resolution(analysis, reference, state, termination)
864 return termination
866 @staticmethod
867 def _record_resolution(
868 analysis: _TraceAnalysis,
869 reference,
870 state: str,
871 termination,
872 reason: str = "",
873 role: str = TERMINATION_ROLE,
874 ) -> None:
875 """Record how one Termination Reference was settled, for its badge and its picker."""
876 key = _field_key(reference, role)
877 label = _endpoint_label(reference)
878 record = {
879 "field_key": key,
880 "label": f"{label} (mapped peer)" if role == MAPPED_PEER_ROLE else label,
881 "state": state,
882 "selected": "" if termination is None else termination.display,
883 "selected_type": "" if termination is None else termination.object_type,
884 # A saved decision keeps this spelling beside its key, so a later identity change can rekey it.
885 "source": {"device": reference.device, "cards": reference.cards, "port": reference.port},
886 "selectable": not reason,
887 "reason": reason,
888 # An end NetBox cannot cable, alone or with its peer, reopens its picker however it was resolved.
889 "incompatible": False,
890 }
891 if termination is not None:
892 record[DISCLOSURE_SOURCE] = disclosure_source(*termination.key)
893 analysis.terminations.setdefault(key, record)
895 def _termination(self, component) -> _Termination:
896 """Return the plan-side record of one resolved NetBox termination."""
897 label = component._meta.label_lower
898 self._objects[(label, component.pk)] = component
899 return _Termination(
900 object_type=label,
901 object_id=component.pk,
902 device_id=component.device_id,
903 display=str(component),
904 topology_display=f"{self._device_displays[component.device_id]} {component.name}",
905 )
907 def _load_mappings(self) -> None:
908 """Read every PortMapping row on a Device this batch resolved a pass-through port on."""
909 device_ids = {
910 termination.device_id
911 for resolved in self._resolved.values()
912 for termination in resolved.values()
913 if termination.object_type in _PASS_THROUGH_PEERS
914 }
915 if device_ids:
916 mappings = self.reader.port_mappings().filter(device_id__in=sorted(device_ids)).order_by("pk")
917 if self.lock_plan_references:
918 mappings = mappings.select_for_update(of=("self",))
919 # Indexed on load, because every pass-through end asks for the rows of one port.
920 for row in mappings:
921 self._mappings_by_front.setdefault(row.front_port_id, []).append(row)
922 self._mappings_by_rear.setdefault(row.rear_port_id, []).append(row)
923 ids_by_label = {
924 FRONT_PORT_MODEL: set(self._mappings_by_front),
925 REAR_PORT_MODEL: set(self._mappings_by_rear),
926 }
927 for label, object_ids in ids_by_label.items():
928 for component in self.reader.terminations(label).filter(pk__in=object_ids):
929 self._mapping_ports[(label, component.pk)] = component
931 def _peers_of(self, termination: _Termination) -> list:
932 """Return the PortMapping rows that link one pass-through port to its opposite side."""
933 if termination.object_type == FRONT_PORT_MODEL:
934 return self._mappings_by_front.get(termination.object_id, [])
935 if termination.object_type == REAR_PORT_MODEL:
936 return self._mappings_by_rear.get(termination.object_id, [])
937 return []
939 def _peer_termination(self, mapping, termination: _Termination) -> _Termination | None:
940 """Return the visible opposite port one PortMapping row links to *termination*."""
941 if termination.object_type == FRONT_PORT_MODEL:
942 peer_key = REAR_PORT_MODEL, mapping.rear_port_id
943 else:
944 peer_key = FRONT_PORT_MODEL, mapping.front_port_id
945 component = self._mapping_ports.get(peer_key)
946 return None if component is None else self._termination(component)
948 def _mapped_peers(self, analysis: _TraceAnalysis, reference, termination: _Termination) -> list | None:
949 """Return visible mapped peers, or block when one peer is outside the actor's view scope."""
950 peers = []
951 for mapping in self._peers_of(termination):
952 peer = self._peer_termination(mapping, termination)
953 if peer is None:
954 analysis.block(
955 "cable.permission_denied",
956 {
957 **_reference_display(reference),
958 "permission": _view_permission(_PASS_THROUGH_PEERS[termination.object_type]),
959 },
960 )
961 return None
962 peers.append((peer, mapping))
963 return peers
965 def _build_segments(self) -> None:
966 """Verify every Pass-Through Claim and turn Segment Evidence into desired segments."""
967 for analysis in self.analyses:
968 if analysis.stopped:
969 continue
970 resolved = self._resolved.get(analysis.identity, {})
971 summary = analysis.trace.endpoint_summary
972 endpoints = tuple(
973 resolved.get(reference.identity_key) for reference in (summary.from_termination, summary.to_termination)
974 )
975 if any(endpoint is None for endpoint in endpoints):
976 continue
977 analysis.endpoints = endpoints
978 self._build_one(analysis, resolved)
980 def _build_one(self, analysis: _TraceAnalysis, resolved: dict) -> None:
981 """Bind one trace's Segment Evidence, substituting a mapped peer where the source repeats a port."""
982 segments = analysis.trace.segments
983 left_ends: list[_Termination] = []
984 right_ends: list[_Termination] = []
985 for segment in segments:
986 left, right = resolved.get(segment.left.identity_key), resolved.get(segment.right.identity_key)
987 if left is None or right is None:
988 return
989 left_ends.append(left)
990 right_ends.append(right)
991 for index in range(len(segments) - 1):
992 if not same_device_and_cards(segments[index].right, segments[index + 1].left):
993 continue
994 entry = self._continue_path(analysis, segments[index + 1].left, right_ends[index], left_ends[index + 1])
995 if entry is None:
996 return
997 left_ends[index + 1] = entry
998 analysis.joined.add(index)
999 analysis.segment_pairs_resolved = True
1000 analysis.segments = [
1001 _DesiredSegment(
1002 index=index,
1003 left=left_ends[index],
1004 right=right_ends[index],
1005 cable_class=source_text(segment.cable_class),
1006 )
1007 for index, segment in enumerate(segments)
1008 ]
1009 for index, segment in enumerate(segments):
1010 if left_ends[index].key != right_ends[index].key:
1011 continue
1012 analysis.block(
1013 "cable.segment_self_connection",
1014 {
1015 "segment_index": index,
1016 "cable_class": source_text(segment.cable_class),
1017 "termination": left_ends[index].display,
1018 **termination_sources(left_ends[index].key),
1019 },
1020 identities=(left_ends[index].identity,),
1021 )
1022 return
1023 self._block_incompatible_segments(analysis)
1025 @staticmethod
1026 def _block_incompatible_segments(analysis: _TraceAnalysis) -> None:
1027 """Block each segment whose two ends NetBox's own compatible-termination table does not pair."""
1028 from dcim.constants import COMPATIBLE_TERMINATION_TYPES
1030 stated = analysis.trace.segments
1031 for segment in analysis.segments:
1032 if segment.right.model_name in COMPATIBLE_TERMINATION_TYPES[segment.left.model_name]:
1033 continue
1034 ends = (_field_key(stated[segment.index].left), _field_key(stated[segment.index].right))
1035 for key in ends:
1036 analysis.terminations[key]["incompatible"] = True
1037 analysis.block(
1038 "cable.incompatible_terminations",
1039 {
1040 "segment_index": segment.index,
1041 "left_model": segment.left.object_type,
1042 "right_model": segment.right.object_type,
1043 "left_field_key": ends[0],
1044 "right_field_key": ends[1],
1045 **termination_sources(segment.left.key, segment.right.key),
1046 },
1047 identities=(segment.left.identity, segment.right.identity),
1048 )
1050 def _continue_path(self, analysis: _TraceAnalysis, reference, exit_end, entry_end) -> _Termination | None:
1051 """Return the termination the next cable end takes where the path passes through a panel."""
1052 if exit_end.key != entry_end.key:
1053 return self._verified_pass_through(analysis, reference, exit_end, entry_end)
1054 mapped_peers = self._mapped_peers(analysis, reference, exit_end)
1055 if mapped_peers is None:
1056 return None
1057 peers: dict[tuple[str, int], Any] = {}
1058 for peer, row in mapped_peers:
1059 peers.setdefault(peer.key, (peer, row))
1060 if not peers:
1061 analysis.refuse(
1062 "cable.pass_through_not_mapped",
1063 {
1064 **_reference_display(reference),
1065 "entry": exit_end.display,
1066 "exit": entry_end.display,
1067 "mapped": [],
1068 **termination_sources(exit_end.key, entry_end.key),
1069 },
1070 identities=(exit_end.identity,),
1071 )
1072 return None
1073 if len(peers) > 1:
1074 return self._chosen_peer(analysis, reference, exit_end, peers)
1075 peer, mapping = next(iter(peers.values()))
1076 return self._substituted(analysis, reference, exit_end, peer, mapping)
1078 def _substituted(self, analysis: _TraceAnalysis, reference, exit_end, peer, mapping) -> _Termination:
1079 """Record one same-port continuation and return the mapped peer it substitutes."""
1080 analysis.note(
1081 "cable.same_port_continuation",
1082 {
1083 **_reference_display(reference),
1084 "port": exit_end.display,
1085 "peer": peer.display,
1086 **termination_sources(exit_end.key, peer.key),
1087 },
1088 identities=(exit_end.identity, peer.identity, _object_identity("dcim.portmapping", mapping.pk)),
1089 )
1090 return peer
1092 def _chosen_peer(self, analysis: _TraceAnalysis, reference, exit_end, peers) -> _Termination | None:
1093 """Return the mapped peer the operator selected, or block on the several NetBox offers."""
1094 stored = self._stored.get(_field_key(reference, MAPPED_PEER_ROLE))
1095 if stored is not None:
1096 selected = (_stored_label(stored), stored.selected_object_id)
1097 if selected in peers:
1098 peer, mapping = peers[selected]
1099 self._record_resolution(analysis, reference, MANUALLY_RESOLVED, peer, role=MAPPED_PEER_ROLE)
1100 return self._substituted(analysis, reference, exit_end, peer, mapping)
1101 self._record_resolution(analysis, reference, UNRESOLVED, None, role=MAPPED_PEER_ROLE)
1102 analysis.block(
1103 "cable.ambiguous_mapped_peer",
1104 {
1105 **_reference_display(reference),
1106 "port": exit_end.display,
1107 "peers": sorted(peer.display for peer, _mapping in peers.values()),
1108 **termination_sources(exit_end.key, *sorted(peers)),
1109 },
1110 identities=(exit_end.identity, *sorted(_object_identity(*key) for key in peers)),
1111 )
1112 return None
1114 def _verified_pass_through(self, analysis: _TraceAnalysis, reference, exit_end, entry_end) -> _Termination | None:
1115 """Return the stated entry port once a PortMapping row proves the panel joins the two ports."""
1116 mapping = None
1117 if {exit_end.object_type, entry_end.object_type} == {FRONT_PORT_MODEL, REAR_PORT_MODEL}:
1118 front, rear = (exit_end, entry_end) if exit_end.object_type == FRONT_PORT_MODEL else (entry_end, exit_end)
1119 mapping = next(
1120 (row for row in self._peers_of(front) if row.rear_port_id == rear.object_id),
1121 None,
1122 )
1123 if mapping is None:
1124 mapped = []
1125 for row in self._peers_of(exit_end):
1126 peer = self._peer_termination(row, exit_end)
1127 if peer is not None:
1128 mapped.append(peer)
1129 analysis.refuse(
1130 "cable.pass_through_not_mapped",
1131 {
1132 **_reference_display(reference),
1133 "entry": exit_end.display,
1134 "exit": entry_end.display,
1135 "mapped": sorted(peer.display for peer in mapped),
1136 **termination_sources(exit_end.key, entry_end.key, *sorted({peer.key for peer in mapped})),
1137 },
1138 identities=(exit_end.identity, entry_end.identity),
1139 )
1140 return None
1141 analysis.note(
1142 "cable.pass_through_verified",
1143 {
1144 **_reference_display(reference),
1145 "entry": exit_end.display,
1146 "exit": entry_end.display,
1147 **termination_sources(exit_end.key, entry_end.key),
1148 },
1149 identities=(exit_end.identity, entry_end.identity, _object_identity("dcim.portmapping", mapping.pk)),
1150 )
1151 return entry_end
1153 def _load_existing_cables(self) -> None:
1154 """Read every Cable that already holds a termination this batch resolved."""
1155 from dcim.models import Cable, CableTermination
1157 wanted = {
1158 termination.key for analysis in self.analyses for termination in self._analysis_terminations(analysis)
1159 }
1160 wanted.update(harness.termination.key for harness in self._harnesses.values())
1161 # Occupancy is unscoped: a Cable the actor cannot view still holds the termination.
1162 cable_ids = {
1163 component.cable_id for key, component in self._objects.items() if key in wanted and component.cable_id
1164 }
1165 if not cable_ids:
1166 return
1167 cables = Cable.objects.filter(pk__in=sorted(cable_ids)).order_by("pk")
1168 terminations = CableTermination.objects.filter(cable_id__in=sorted(cable_ids)).order_by("pk")
1169 if self.lock_plan_references:
1170 cables = cables.select_for_update(of=("self",))
1171 terminations = terminations.select_for_update(of=("self",))
1172 # Each Cable lock precedes its CableTermination locks, as it does during deletion.
1173 locked = list(cables)
1174 sides: dict[int, dict[str, set]] = {}
1175 snapshots: dict[int, set] = {}
1176 for row in terminations:
1177 label = row.termination_type.model_class()._meta.label_lower
1178 sides.setdefault(row.cable_id, {"A": set(), "B": set()})[row.cable_end].add((label, row.termination_id))
1179 snapshots.setdefault(row.cable_id, set()).add(
1180 (row.cable_end, row.connector, tuple(row.positions or ()), label, row.termination_id)
1181 )
1182 if self.lock_plan_references:
1183 from core.models import ObjectType
1184 from django.db import connection
1185 from extras.models import TaggedItem
1187 associations = list(
1188 TaggedItem.objects.filter(
1189 content_type_id=ObjectType.objects.get_for_model(Cable).pk,
1190 object_id__in=sorted(cable_ids),
1191 )
1192 .order_by("pk")
1193 .select_for_update(of=("self",))
1194 )
1195 tag_ids = sorted({association.tag_id for association in associations})
1196 if tag_ids:
1197 with connection.cursor() as cursor:
1198 cursor.execute(
1199 "SELECT id FROM extras_tag WHERE id = ANY(%s) ORDER BY id FOR SHARE",
1200 [tag_ids],
1201 )
1202 cursor.fetchall()
1203 for cable in locked:
1204 ends = sides.get(cable.pk, {"A": set(), "B": set()})
1205 existing = _ExistingCable(
1206 cable=cable,
1207 a_side=frozenset(ends["A"]),
1208 b_side=frozenset(ends["B"]),
1209 snapshot=frozenset(snapshots.get(cable.pk, ())),
1210 )
1211 self._existing[cable.pk] = existing
1212 for key in existing.a_side | existing.b_side:
1213 self._occupied[key] = existing
1215 def _lock_segment_terminations(self) -> None:
1216 """Lock every desired segment termination in global object-type and primary-key order."""
1217 if not self.lock_plan_references:
1218 return
1219 ids_by_label: dict[str, set[int]] = {}
1220 for analysis in self.analyses:
1221 for segment in analysis.segments:
1222 for termination in segment.terminations:
1223 ids_by_label.setdefault(termination.object_type, set()).add(termination.object_id)
1224 for label in sorted(ids_by_label):
1225 object_ids = sorted(ids_by_label[label])
1226 for object_id in object_ids:
1227 self._objects.pop((label, object_id), None)
1228 components = (
1229 _model_for_label(label).objects.filter(pk__in=object_ids).order_by("pk").select_for_update(of=("self",))
1230 )
1231 for component in components:
1232 self._objects[(label, component.pk)] = component
1234 def _channel_parent(self, termination: _Termination) -> int | None:
1235 """Return the parent of one resolved termination that is a channel child, from its latest row."""
1236 component = self._objects.get(termination.key)
1237 return None if component is None else cable_harness.channel_parent(component)
1239 def _child_parents(self) -> set[int]:
1240 """Return the parent of every resolved termination that is a channel child."""
1241 return {
1242 parent
1243 for resolved in self._resolved.values()
1244 for termination in resolved.values()
1245 if (parent := self._channel_parent(termination)) is not None
1246 }
1248 def _lock_harness_parents(self) -> None:
1249 """Lock each parent of a resolved channel child, then every Interface under it, before the segment locks."""
1250 if self.lock_plan_references:
1251 self._parents = cable_harness.read_parents(self._child_parents(), lock=True)
1253 def _find_legs(self) -> None:
1254 """Give each trace that reaches a channel child to its parent's harness, as a leg or as a blocker.
1256 A child can move to another parent between the first read and its lock, so that parent is read,
1257 and locked, here.
1258 """
1259 parents = self._child_parents()
1260 self._parents.update(cable_harness.read_parents(parents - self._parents.keys(), lock=self.lock_plan_references))
1261 self._forget_orphaned_children()
1262 for analysis in self.analyses:
1263 self._place_trace(analysis)
1264 for analysis in self.analyses:
1265 self._block_harness_of_ordinary_parent(analysis)
1266 for parent_pk in sorted(self._harnesses):
1267 self._gather_cohort(self._harnesses[parent_pk])
1268 for parent_pk in sorted(self._harnesses):
1269 self._harnesses[parent_pk].settle_wording(self._resolved)
1270 # Its wording names the whole harness, which is known only once every trace is placed.
1271 for analysis, harness, segment_index in self._unsupported_traces:
1272 analysis.block(
1273 "cable.harness_unsupported",
1274 self._harness_display(harness, segment_index, trace=analysis.display["name"]),
1275 identities=(harness.termination.identity,),
1276 )
1278 def _forget_orphaned_children(self) -> None:
1279 """Forget each resolved channel child whose parent row is gone, as any row gone since resolution.
1281 An unlocked preview reads a parent after its children, so another transaction can delete both
1282 in between. The child is then no leg, its write skips the row, and execution refuses the plan.
1283 """
1284 for key, component in list(self._objects.items()):
1285 parent = cable_harness.channel_parent(component)
1286 if parent is not None and parent not in self._parents:
1287 del self._objects[key]
1289 def _harness(self, parent_pk: int) -> _Harness:
1290 """Return the harness of one parent, created on first use."""
1291 if parent_pk not in self._harnesses:
1292 parent = self._parents[parent_pk]
1293 self._harnesses[parent_pk] = _Harness(
1294 parent=parent, termination=self._channel_termination(parent.component)
1295 )
1296 return self._harnesses[parent_pk]
1298 def _channel_termination(self, component) -> _Termination:
1299 """Return the plan record of a parent or a child, which no trace may have resolved a Device for."""
1300 # On a virtual chassis a parent and its children can sit on different member Devices.
1301 self._device_displays.setdefault(component.device_id, str(component.device))
1302 return self._termination(component)
1304 def _place_trace(self, analysis: _TraceAnalysis) -> None:
1305 """Record one trace as the leg of its parent's harness, or as a trace that blocks that harness."""
1306 touched = sorted(
1307 {
1308 parent
1309 for termination in self._resolved.get(analysis.identity, {}).values()
1310 if (parent := self._channel_parent(termination)) is not None
1311 }
1312 )
1313 if not touched:
1314 return
1315 legs = [
1316 (segment, end, parent)
1317 for segment in analysis.segments
1318 for end in segment.terminations
1319 if (parent := self._channel_parent(end)) is not None
1320 ]
1321 if len(legs) > 1:
1322 # A child at both ends of one segment, or a trace between two harnesses.
1323 self._unsupported_traces.append((analysis, self._harness(touched[0]), legs[0][0].index))
1324 for parent in touched:
1325 self._harness(parent).unsupported.append(analysis)
1326 return
1327 if not legs:
1328 # The trace names a channel child but no physical first hop from it.
1329 for parent in touched:
1330 self._harness(parent).unresolved.append(analysis)
1331 return
1332 ((segment, child, parent),) = legs
1333 harness = self._harness(parent)
1334 far = segment.right if child.key == segment.left.key else segment.left
1335 analysis.leg = _Leg(analysis=analysis, segment=segment, child=child, far=far)
1336 analysis.harness = harness
1337 harness.legs.append(analysis.leg)
1339 def _block_harness_of_ordinary_parent(self, analysis: _TraceAnalysis) -> None:
1340 """Block the harness of each parent this trace names as an ordinary segment end: the source states two paths on it."""
1341 for segment in analysis.segments:
1342 if analysis.is_leg(segment):
1343 continue
1344 for end in segment.terminations:
1345 harness = self._harnesses.get(end.object_id) if end.object_type == cable_harness.INTERFACE else None
1346 if harness is not None and analysis not in harness.unsupported:
1347 harness.unsupported.append(analysis)
1349 def _gather_cohort(self, harness: _Harness) -> None:
1350 """Read the source cohorts of one harness as negative evidence (section 6.4).
1352 A cohort is the device and cards label of a reference that resolves to a child of the parent.
1353 Any trace with an unresolved reference in a cohort blocks the harness, so a lane the source
1354 states but the plan cannot resolve never leaves a shorter harness behind.
1355 """
1356 members = [leg.analysis for leg in harness.legs] + harness.unsupported + harness.unresolved
1357 cohorts = {
1358 reference.identity_key[:2]
1359 for analysis in members
1360 for reference in self._references(analysis.trace)
1361 if (termination := self._resolved.get(analysis.identity, {}).get(reference.identity_key)) is not None
1362 and self._channel_parent(termination) == harness.parent.pk
1363 }
1364 for analysis in self.analyses:
1365 resolved = self._resolved.get(analysis.identity, {})
1366 keys = {reference.identity_key for reference in self._references(analysis.trace)}
1367 for key in sorted(key for key in keys if key[:2] in cohorts):
1368 termination = resolved.get(key)
1369 component = None if termination is None else self._objects.get(termination.key)
1370 if termination is None and analysis.identity not in {other.identity for other in harness.unresolved}:
1371 harness.unresolved.append(analysis)
1372 elif component is not None and cable_harness.is_plain_interface(component):
1373 harness.unchannelized.append((analysis, termination))
1375 @staticmethod
1376 def _harness_display(harness: _Harness, segment_index: int, *ports: _Termination, **fields) -> dict:
1377 """Return the display every finding about a harness carries: the leg, the parent, and the ports it names."""
1378 return {
1379 "segment_index": segment_index,
1380 "parent": harness.termination.display,
1381 **fields,
1382 **harness.wording,
1383 # The parent's Device authorizes its channel facts too: on a virtual chassis it can be hidden alone.
1384 **termination_sources(
1385 harness.termination.key, (DEVICE_ROW, harness.parent.component.device_id), *(port.key for port in ports)
1386 ),
1387 }
1389 def _plan_harnesses(self) -> None:
1390 """Decide each harness once, then give each of its leg units the same findings."""
1391 for parent_pk in sorted(self._harnesses):
1392 harness = self._harnesses[parent_pk]
1393 if not harness.legs:
1394 # Only blocked traces touch this parent, and each already carries its own finding.
1395 continue
1396 self._refuse_deleted_far_ends(harness)
1397 harness.legs.sort(
1398 key=lambda leg: (cable_harness.channel(self._objects[leg.child.key]), leg.analysis.identity)
1399 )
1400 findings = self._harness_findings(harness)
1401 missing = self._parent_permission_missing(harness)
1402 for leg in harness.legs:
1403 if missing is not None:
1404 self._hide_leg(leg, harness, missing)
1405 continue
1406 for kind, code, ports, fields in findings:
1407 getattr(leg.analysis, kind)(
1408 code,
1409 self._harness_display(harness, leg.segment.index, *ports, **fields),
1410 identities=(harness.termination.identity, *(port.identity for port in ports)),
1411 )
1412 if code == cable_harness.NOT_CONNECTABLE:
1413 leg.analysis.reopen(ports[0].key)
1414 self._settle_leg(leg, harness)
1416 def _refuse_deleted_far_ends(self, harness: _Harness) -> None:
1417 """Refuse a harness whose far end the locked reread no longer found, as the segment write refuses one.
1419 A child row that is gone makes its trace no leg, so only a far end can be missing here.
1420 """
1421 for leg in harness.legs:
1422 if leg.far.key not in self._objects:
1423 label, object_id = leg.far.key
1424 raise PreconditionFailed(f"{label} {object_id} is gone, so the breakout Cable cannot be created.")
1426 def _harness_findings(self, harness: _Harness) -> list[tuple[str, str, tuple, dict]]:
1427 """Return each finding of one harness as (severity method, code, ports it names, display fields)."""
1428 parent = harness.parent
1429 on_parent = self._existing.get(parent.component.cable_id)
1430 faults = parent.faults("" if on_parent is None else on_parent.cable.profile)
1431 if faults:
1432 return [
1433 (
1434 "refuse",
1435 "cable.harness_channel_state_invalid",
1436 (self._channel_termination(child.component),),
1437 {"port": str(child.component), "channel": child.channel, "fault": fault},
1438 )
1439 for child, fault in faults
1440 ]
1441 findings: list[tuple[str, str, tuple, dict]] = [
1442 ("block", "cable.harness_unsupported", (), {"trace": analysis.display["name"]})
1443 for analysis in harness.unsupported
1444 ]
1445 findings.extend(
1446 ("block", "cable.harness_member_unresolved", (), {"trace": analysis.display["name"]})
1447 for analysis in harness.unresolved
1448 )
1449 try:
1450 harness.intent = cable_harness.harness_intent(
1451 parent.component, [(self._objects[leg.child.key], self._objects[leg.far.key]) for leg in harness.legs]
1452 )
1453 except cable_harness.IntentRefused as refusal:
1454 kind = "refuse" if refusal.code == "cable.harness_channel_state_invalid" else "block"
1455 ports = () if refusal.port is None else (self._channel_termination(refusal.port),)
1456 findings.append((kind, refusal.code, ports, dict(refusal.facts)))
1457 if self._harness_cable_types(harness):
1458 findings.append(("block", "cable.harness_type_conflict", (), {}))
1459 findings.extend(self._harness_occupancy(harness))
1460 findings.extend(
1461 (
1462 "warn",
1463 "cable.harness_member_unchannelized",
1464 (port,),
1465 {"trace": analysis.display["name"], "port": port.display},
1466 )
1467 for analysis, port in harness.unchannelized
1468 )
1469 if harness.intent is not None and harness.intent.trailing:
1470 findings.append(("warn", "cable.harness_partial", (), {"missing": list(harness.intent.trailing)}))
1471 return findings
1473 def _harness_cable_types(self, harness: _Harness) -> bool:
1474 """Settle the one Cable Type every leg's policy agrees on, and return whether the legs disagree.
1476 Only the Cable Type dimension applies to a leg: the channel topology decides the profile.
1477 """
1478 from .cable_policy import policy_choice_errors
1480 types = set()
1481 for leg in harness.legs:
1482 row = self._policy_row(leg.segment)
1483 display = {"segment_index": leg.segment.index, "cable_class": leg.segment.cable_class}
1484 if row is None or not row.cable_type_resolved:
1485 leg.analysis.block("cable.cableclass_unmapped", display)
1486 elif policy_choice_errors(row.cable_type, None):
1487 leg.analysis.block("cable.policy_stale", display)
1488 else:
1489 types.add(row.cable_type)
1490 if len(types) == 1:
1491 harness.cable_type = next(iter(types))
1492 return len(types) > 1
1494 def _harness_occupancy(self, harness: _Harness) -> list[tuple[str, str, tuple, dict]]:
1495 """Reuse the Cable on the parent when its termination snapshot is the harness, else block each held end."""
1496 on_parent = self._occupied.get(harness.termination.key)
1497 if (
1498 harness.intent is not None
1499 and on_parent is not None
1500 and harness.intent.proved_by(on_parent.cable.profile, on_parent.snapshot)
1501 ):
1502 harness.reused = on_parent
1503 return []
1504 findings = []
1505 for port in harness.ends:
1506 occupying = self._occupied.get(port.key)
1507 if occupying is not None:
1508 cable_display, _identities = self._cable_diagnostic_disclosure(occupying.cable)
1509 findings.append(("block", "cable.harness_occupied", (port,), {"port": port.display, **cable_display}))
1510 return findings
1512 def _parent_permission_missing(self, harness: _Harness) -> str | None:
1513 """Return the view permission the actor lacks on the parent Interface or on its Device (section 6.6)."""
1514 parent = harness.parent.component
1515 if not self.reader.terminations(cable_harness.INTERFACE).filter(pk=parent.pk).exists():
1516 return _view_permission(cable_harness.INTERFACE)
1517 # On a virtual chassis the parent can sit on another member Device than its children.
1518 if not self.reader.devices().filter(pk=parent.device_id).exists():
1519 return _view_permission("dcim.device")
1520 return None
1522 def _hide_leg(self, leg: _Leg, harness: _Harness, permission: str) -> None:
1523 """Block one leg whose parent, or its Device, the actor cannot view, and state no fact of that parent.
1525 As for a hidden Cable or mapped peer, only the evidence holds the parent's facts, so a change
1526 to them still moves the unit.
1527 """
1528 # Reuse is a fact of the parent, so a hidden leg names its companions whatever the parent holds.
1529 leg.analysis.selected_with = harness.companions(leg)
1530 leg.analysis.block(
1531 "cable.permission_denied",
1532 {"permission": permission},
1533 evidence=self._harness_evidence(harness),
1534 )
1536 def _leg_channels(self, harness: _Harness) -> list[int]:
1537 """Return the channel of each leg, in leg order."""
1538 return [cable_harness.channel(self._objects[leg.child.key]) for leg in harness.legs]
1540 def _harness_evidence(self, harness: _Harness) -> dict:
1541 """Return the harness facts every leg unit fingerprints (section 6.4)."""
1542 return {
1543 "parent": harness.termination.identity,
1544 **harness.parent.evidence(),
1545 # The child's Device is read from its latest row: in a virtual chassis a child can change member.
1546 "legs": [
1547 [number, _object_identity("dcim.device", self._objects[leg.child.key].device_id), leg.far.identity]
1548 for number, leg in zip(self._leg_channels(harness), harness.legs, strict=True)
1549 ],
1550 "profile": "" if harness.intent is None else harness.intent.profile,
1551 "reused": harness.reused is not None,
1552 }
1554 def _settle_leg(self, leg: _Leg, harness: _Harness) -> None:
1555 """Record what the harness means for one leg unit: reuse or companions, then its fingerprinted state."""
1556 analysis = leg.analysis
1557 if harness.reused is not None:
1558 analysis.proven[leg.segment.index] = harness.reused
1559 cable_display, cable_identities = self._cable_diagnostic_disclosure(harness.reused.cable)
1560 analysis.note(
1561 "cable.segment_reused",
1562 # The Cable proves this segment only as the harness, so the reuse is a fact of the harness.
1563 {"segment_index": leg.segment.index, **cable_display, **harness.wording},
1564 identities=(*cable_identities, harness.termination.identity, leg.far.identity),
1565 )
1566 else:
1567 analysis.selected_with = harness.companions(leg)
1568 evidence = self._harness_evidence(harness)
1569 analysis.note(
1570 "cable.harness_state",
1571 self._harness_display(
1572 harness,
1573 leg.segment.index,
1574 channels=harness.parent.channels,
1575 legs=self._leg_channels(harness),
1576 profile=evidence["profile"],
1577 ),
1578 identities=(harness.termination.identity, leg.child.identity, leg.far.identity),
1579 evidence=evidence,
1580 )
1582 @staticmethod
1583 def _analysis_terminations(analysis: _TraceAnalysis) -> list[_Termination]:
1584 """Return every resolved termination one trace's plan depends on."""
1585 terminations = list(analysis.endpoints)
1586 for segment in analysis.segments:
1587 terminations.extend(segment.terminations)
1588 return terminations
1590 def _may_view(self, row: Any) -> bool:
1591 """Return whether the actor may read one row, asking NetBox once per row in this batch."""
1592 if self.actor is None:
1593 return True
1594 key = (row._meta.label_lower, row.pk)
1595 if key not in self._visible:
1596 # NetBox answers each object-level check with its own query.
1597 self._visible[key] = self.actor.has_perm(f"{row._meta.app_label}.view_{row._meta.model_name}", row)
1598 return self._visible[key]
1600 def _cable_diagnostic_disclosure(self, cable: Any) -> tuple[dict[str, Any], tuple[str, ...]]:
1601 """Return the diagnostic fields and identity the actor may see for one Cable."""
1602 cable_visible = self._may_view(cable)
1603 if not cable_visible:
1604 return {"cable_visible": False}, ()
1605 return (
1606 disclosed_cable(cable),
1607 (_object_identity("dcim.cable", cable.pk),),
1608 )
1610 def _classify(self) -> None:
1611 """Decide which segments already exist, then which Cable is each trace's Logical Cable.
1613 Every desired segment of the batch is classified before any Logical Cable is chosen, so a
1614 Cable that proves one trace's segment can never be deleted as another trace's leftover.
1615 """
1616 pending = self._writing()
1617 for analysis in pending:
1618 # Only these analyses look at the live topology, so only they may report what is there.
1619 analysis.topology_read = True
1620 for segment in analysis.pending:
1621 candidate = self._occupied.get(segment.left.key)
1622 proven = candidate if candidate is not None and candidate.satisfies(segment) else None
1623 if proven is not None:
1624 analysis.proven[segment.index] = proven
1625 self._note_reuse(analysis, segment, proven)
1626 proven_ids = {item.cable.pk for analysis in pending for item in analysis.proven.values()}
1627 for analysis in pending:
1628 analysis.logical_cable = self._logical_cable(analysis, proven_ids)
1629 if analysis.segments:
1630 self._report_conflicts(analysis)
1631 else:
1632 self._classify_endpoint_evidence(analysis)
1634 def _logical_cable(self, analysis: _TraceAnalysis, proven_ids) -> _ExistingCable | None:
1635 """Return the direct endpoint-to-endpoint Cable that proves no desired segment and is no breakout."""
1636 first, second = analysis.endpoints
1637 existing = self._occupied.get(first.key)
1638 if (
1639 existing is not None
1640 and existing.cable.pk not in proven_ids
1641 and existing.connects(first, second)
1642 and not cable_harness.is_breakout(existing.cable, (self._objects[first.key], self._objects[second.key]))
1643 ):
1644 return existing
1645 return None
1647 def _classify_endpoint_evidence(self, analysis: _TraceAnalysis) -> None:
1648 """Decide an Endpoint Summary fallback, which states endpoints and no physical path."""
1649 if analysis.logical_cable is None:
1650 analysis.block(
1651 "trace.endpoint_evidence_only",
1652 {"name": analysis.display["name"]},
1653 identities=tuple(endpoint.identity for endpoint in analysis.endpoints),
1654 )
1655 return
1656 # The stated endpoints are already joined, so the evidence is satisfied and nothing is removed.
1657 cable_display, cable_identities = self._cable_diagnostic_disclosure(analysis.logical_cable.cable)
1658 analysis.note(
1659 "cable.segment_reused",
1660 {"segment_index": 0, **cable_display},
1661 identities=cable_identities,
1662 )
1663 analysis.retains_logical_cable = True
1665 def _note_reuse(self, analysis: _TraceAnalysis, segment: _DesiredSegment, proven: _ExistingCable) -> None:
1666 """Record the proven physical segment, so its live state joins the unit fingerprint."""
1667 cable_display, cable_identities = self._cable_diagnostic_disclosure(proven.cable)
1668 analysis.note(
1669 "cable.segment_reused",
1670 {"segment_index": segment.index, **cable_display},
1671 identities=(*cable_identities, segment.left.identity, segment.right.identity),
1672 )
1673 drift = self._attribute_drift(segment, proven.cable)
1674 if drift:
1675 drift_display = drift if cable_display["cable_visible"] else {}
1676 analysis.note(
1677 "cable.attribute_drift",
1678 {"segment_index": segment.index, **cable_display, **drift_display},
1679 identities=cable_identities,
1680 )
1682 def _attribute_drift(self, segment: _DesiredSegment, cable) -> dict:
1683 """Return the reused Cable attributes that differ from what this import would have written."""
1684 row = self._policy_row(segment)
1685 drift = {}
1686 if cable.status != CABLE_STATUS:
1687 drift["status"] = cable.status
1688 if row is not None and row.cable_type_resolved and (cable.type or None) != row.cable_type:
1689 drift["type"] = cable.type or ""
1690 if row is not None and row.cable_profile_resolved and (cable.profile or None) != row.cable_profile:
1691 drift["profile"] = cable.profile or ""
1692 if cable.label:
1693 drift["label"] = cable.label
1694 return drift
1696 def _report_conflicts(self, analysis: _TraceAnalysis) -> None:
1697 """Block the trace when a Cable this import may not touch holds a termination it needs."""
1698 removed = analysis.deleted_logical_cable
1699 logical_id = None if removed is None else removed.cable.pk
1700 for segment in analysis.pending:
1701 for termination in segment.terminations:
1702 occupying = self._occupied.get(termination.key)
1703 if occupying is None or occupying.cable.pk == logical_id:
1704 continue
1705 code = (
1706 "cable.multi_termination_conflict" if occupying.multi_termination else "cable.termination_occupied"
1707 )
1708 cable_display, cable_identities = self._cable_diagnostic_disclosure(occupying.cable)
1709 display = {
1710 "segment_index": segment.index,
1711 "port": termination.display,
1712 **termination_sources(termination.key),
1713 **cable_display,
1714 }
1715 identities = [termination.identity, *cable_identities]
1716 analysis.block(code, display, identities=identities)
1718 def _cable_class_mapping(self, cable_class: str):
1719 """Return the Cable policy row one CableClass value carries, or None."""
1720 if self._mapping_rows is None:
1721 from .models import CableClassMapping
1723 rows = CableClassMapping.objects.filter(profile=self.profile)
1724 self._mapping_rows = {row.cable_class: row for row in rows}
1725 return self._mapping_rows.get(cable_class)
1727 def _policy_row(self, segment: _DesiredSegment):
1728 """Return the stored decision that governs one segment, or None while nothing decides it."""
1729 from .cable_policy import policy_in_force
1731 return policy_in_force(self._overrides.get(segment.key), self._cable_class_mapping(segment.cable_class))
1733 def _load_overrides(self) -> None:
1734 """Read every override this batch can apply, and every one its own traces may have lost."""
1735 from django.db.models import Q
1737 from .models import CableSegmentOverride
1739 keys = {segment.key for analysis in self.analyses for segment in analysis.segments}
1740 identities = {analysis.trace.identity for analysis in self.analyses}
1741 rows = CableSegmentOverride.objects.filter(profile=self.profile).filter(
1742 Q(segment_key__in=keys) | Q(source_trace_identity__in=identities)
1743 )
1744 self._overrides = {row.segment_key: row for row in rows}
1745 # Uniqueness is by pair, so one trace and one position can own several overrides at once.
1746 self._overrides_by_trace = {}
1747 for row in rows:
1748 self._overrides_by_trace.setdefault(row.source_trace_identity, []).append(row)
1750 def _report_lost_overrides(self, analysis: _TraceAnalysis) -> None:
1751 """Report each override of this trace that no longer governs any segment it states.
1753 The comparison is by pair, not by position: a trace identity names its endpoints, so a
1754 revised path renumbers every segment after the one it inserts without losing anything.
1755 """
1756 stated = {segment.key for segment in analysis.segments}
1757 for stored in self._overrides_by_trace.get(analysis.trace.identity, ()):
1758 if stored.segment_key in stated:
1759 continue
1760 analysis.note(
1761 "cable.segment_override_lost",
1762 disclosed_policy(
1763 stored,
1764 self._may_view(stored),
1765 {
1766 "segment_index": stored.segment_index,
1767 "cable_type": stored.cable_type_display(),
1768 "cable_profile": stored.cable_profile_display(),
1769 },
1770 ),
1771 )
1773 def _decide(self) -> None:
1774 """Settle the Cable policy and the write permissions every actionable trace needs."""
1775 for analysis in self.analyses:
1776 # A refused Segment Evidence block also leaves no segments, and states no path at all.
1777 if (analysis.trace.valid and not analysis.trace.segments) or analysis.segment_pairs_resolved:
1778 self._report_lost_overrides(analysis)
1779 if analysis.stopped or not analysis.endpoints:
1780 continue
1781 for segment in analysis.pending:
1782 policy = self._cable_policy(analysis, segment)
1783 if policy is not None:
1784 analysis.policies[segment.index] = policy
1785 self._check_write(analysis, segment, policy)
1786 self._check_permissions(analysis)
1788 def _check_write(self, analysis: _TraceAnalysis, segment: _DesiredSegment, policy: dict) -> None:
1789 """Build one segment's Cable write through the one constructor, and block the trace when it refuses."""
1790 ends = [self._objects.get(end.key) for end in segment.terminations]
1791 # A row gone since resolution is refused by the write itself.
1792 if None in ends:
1793 return
1794 try:
1795 cable_harness.ordinary_intent(*ends, profile=policy["cable_profile"] or "")
1796 except cable_harness.IntentRefused as refusal:
1797 (port,) = [end for end in segment.terminations if self._objects[end.key] is refusal.port]
1798 analysis.block(
1799 refusal.code,
1800 {
1801 "segment_index": segment.index,
1802 **refusal.facts,
1803 **termination_sources(port.key),
1804 # An ordinary segment's refusal describes the refused port alone, so only that port authorizes it.
1805 **(wording_sources([(port.key, port.device_id)]) if refusal.code in WORDING_CODES else {}),
1806 },
1807 identities=(port.identity,),
1808 )
1809 if refusal.code in {cable_harness.NOT_CONNECTABLE, cable_harness.CHANNELIZED}:
1810 analysis.reopen(port.key)
1812 def _cable_policy(self, analysis: _TraceAnalysis, segment: _DesiredSegment) -> dict | None:
1813 """Return the Cable Type and Cable Profile one new segment is written with."""
1814 from .cable_policy import policy_choice_errors
1816 display = {"segment_index": segment.index, "cable_class": segment.cable_class}
1817 row = self._policy_row(segment)
1818 policy = None if row is None else row.decided_policy()
1819 if policy is None:
1820 analysis.block("cable.cableclass_unmapped", display)
1821 return None
1822 errors = policy_choice_errors(policy["cable_type"], policy["cable_profile"])
1823 for error in errors.values():
1824 analysis.block(error.code, display)
1825 if errors:
1826 return None
1827 return policy
1829 def _assess_media(self) -> None:
1830 """Report each run of verified pass-throughs whose segments state two known media."""
1831 for analysis in self.analyses:
1832 if not analysis.topology_read or not analysis.segment_pairs_resolved or not analysis.endpoints:
1833 continue
1834 observations = [self._media_observation(analysis, segment) for segment in analysis.segments]
1835 for span in self._media_spans(analysis, observations):
1836 self._report_media_mismatch(analysis, span)
1838 def _media_observation(self, analysis: _TraceAnalysis, segment: _DesiredSegment) -> dict:
1839 """Return what one segment says about the medium of the run it belongs to."""
1840 from .cable_policy import cable_profile_splits_a_span, decisive_media_family, policy_choice_errors
1842 proven = analysis.proven.get(segment.index)
1843 visible = True
1844 source = None
1845 if proven is not None:
1846 cable_type, cable_profile = proven.cable.type or "", proven.cable.profile or ""
1847 disclosure = self._cable_diagnostic_disclosure(proven.cable)[0]
1848 visible = disclosure["cable_visible"]
1849 source = disclosure.get("disclosure_source")
1850 origin = "cable"
1851 else:
1852 row = self._policy_row(segment)
1853 policy = analysis.policies.get(segment.index)
1854 if policy is None and row is not None:
1855 decided = row.decided_policy()
1856 if decided is not None and not policy_choice_errors(decided["cable_type"], decided["cable_profile"]):
1857 policy = decided
1858 policy = policy or {}
1859 cable_type, cable_profile = policy.get("cable_type") or "", policy.get("cable_profile") or ""
1860 disclosure = disclosed_policy(row, self._may_view(row), {}) if row is not None else {}
1861 visible = DISCLOSURE_SOURCE in disclosure
1862 source = disclosure.get(DISCLOSURE_SOURCE)
1863 origin = "policy"
1864 return {
1865 "segment_index": segment.index,
1866 "cable_type": cable_type,
1867 "family": decisive_media_family(cable_type),
1868 "retained": proven is not None,
1869 "visible": visible,
1870 # A breakout Cable fans out, so a leg ends the run whatever its policy states.
1871 "splits": analysis.is_leg(segment) or cable_profile_splits_a_span(cable_profile),
1872 "disclosure_source": source,
1873 "origin": origin,
1874 }
1876 @staticmethod
1877 def _media_spans(analysis: _TraceAnalysis, observations: list) -> list:
1878 """Return each run of segments a verified mapping joins, cut where a Cable fans out.
1880 A shared Device name proves nothing: only a PortMapping the plan verified joins two segments.
1881 """
1882 spans: list[list[dict]] = [[]]
1883 for observation in observations:
1884 if observation["splits"]:
1885 spans.append([])
1886 continue
1887 if not (spans[-1] and (observation["segment_index"] - 1) in analysis.joined):
1888 spans.append([])
1889 spans[-1].append(observation)
1890 return [span for span in spans if len(span) > 1]
1892 def _report_media_mismatch(self, analysis: _TraceAnalysis, span: list) -> None:
1893 """Warn where one verified passive run carries two media families, and say who can fix it."""
1894 decided = [item for item in span if item["family"]]
1895 if len({item["family"] for item in decided}) < 2:
1896 return
1897 resolved = {segment.index: segment for segment in analysis.segments}
1898 # A hidden Cable still decides the run, and still states nothing about itself.
1899 analysis.warn(
1900 "cable.media_family_mismatch",
1901 {"segments": [self._media_display(item, include_source=True) for item in decided]},
1902 identities=tuple(
1903 identity
1904 for item in decided
1905 for identity in (
1906 resolved[item["segment_index"]].left.identity,
1907 resolved[item["segment_index"]].right.identity,
1908 )
1909 ),
1910 evidence={"segments": [self._media_display(item, for_evidence=True) for item in decided]},
1911 )
1913 @staticmethod
1914 def _media_display(observation: dict, *, include_source: bool = False, for_evidence: bool = False) -> dict:
1915 """Return stable media facts, with row identity only in presentation data."""
1916 visible = observation["visible"] or (for_evidence and observation["origin"] == "policy")
1917 record = {
1918 "segment_index": observation["segment_index"],
1919 "retained": observation["retained"],
1920 "visible": visible,
1921 "origin": observation["origin"],
1922 }
1923 if not visible:
1924 return record
1925 record.update(cable_type=observation["cable_type"], family=observation["family"])
1926 if include_source and observation["disclosure_source"] is not None:
1927 record["disclosure_source"] = observation["disclosure_source"]
1928 return record
1930 def _check_permissions(self, analysis: _TraceAnalysis) -> None:
1931 """Block the trace when the actor may not make every Cable write it asks for."""
1932 if self.actor is None:
1933 return
1934 if (analysis.pending or analysis.harness_pending) and not self.actor.has_perm("dcim.add_cable"):
1935 analysis.block("cable.permission_denied", {"permission": "dcim.add_cable"})
1936 return
1937 logical = analysis.deleted_logical_cable
1938 if logical is not None and not self.actor.has_perm("dcim.delete_cable", logical.cable):
1939 cable_display, cable_identities = self._cable_diagnostic_disclosure(logical.cable)
1940 analysis.block(
1941 "cable.permission_denied",
1942 {"permission": "dcim.delete_cable", **cable_display},
1943 identities=cable_identities,
1944 )
1946 def _block_planned_termination_conflicts(self) -> None:
1947 """Block every actionable trace whose planned Cables compete for one free termination.
1949 A claim is keyed by the Cable it creates, so traces that share one segment or one harness agree.
1950 Two different Cables also conflict inside one trace, such as a harness and a segment back to its far end.
1951 """
1952 claims: dict[tuple[str, int], list[tuple[_TraceAnalysis, str, int, _Termination, _Harness | None]]] = {}
1953 for analysis in self._writing():
1954 for segment in analysis.pending:
1955 for termination in segment.terminations:
1956 claims.setdefault(termination.key, []).append(
1957 (analysis, segment.key, segment.index, termination, None)
1958 )
1959 if analysis.harness_pending:
1960 harness, index = analysis.harness, analysis.leg.segment.index
1961 for termination in harness.ends:
1962 claims.setdefault(termination.key, []).append(
1963 (analysis, harness.change_identity, index, termination, harness)
1964 )
1965 for records in claims.values():
1966 for analysis, claim, index, termination, harness in records:
1967 for other, other_claim, other_index, _other_termination, other_harness in records:
1968 if other_claim == claim:
1969 continue
1970 competitor = (
1971 {"competing_segment": other_index}
1972 if other is analysis
1973 else {"competing_trace": other.display["name"]}
1974 )
1975 # A claim that only a harness makes is a fact of that harness, so the conflict hides with it.
1976 harnesses = [item for item in (harness, other_harness) if item is not None]
1977 analysis.block(
1978 "cable.planned_termination_conflict",
1979 {
1980 "segment_index": index,
1981 "termination": termination.display,
1982 **competitor,
1983 **termination_sources(termination.key),
1984 **(
1985 wording_sources([end for item in harnesses for end in item.wording_ends])
1986 if harnesses
1987 else {}
1988 ),
1989 },
1990 identities=(termination.identity,),
1991 )
1993 def _block_conflicting_creations(self) -> None:
1994 """Block traces that resolve one shared segment to different Cable policies."""
1995 contributors: dict[str, list] = {}
1996 for analysis in self.analyses:
1997 if analysis.invalid or not analysis.endpoints:
1998 continue
1999 for segment in analysis.pending:
2000 policy = analysis.policies.get(segment.index)
2001 if policy is not None:
2002 contributors.setdefault(segment.key, []).append((analysis, segment, policy))
2003 for records in contributors.values():
2004 # One shared Planned Change carries one payload, so only the effective policy may differ.
2005 policies = {(policy["cable_type"], policy["cable_profile"]) for _analysis, _segment, policy in records}
2006 if len(policies) < 2:
2007 continue
2008 for analysis, segment, policy in records:
2009 row = self._policy_row(segment)
2010 analysis.block(
2011 "cable.resolved_segment_conflict",
2012 disclosed_policy(
2013 row,
2014 self._may_view(row),
2015 {
2016 "segment_index": segment.index,
2017 "cable_type": policy["cable_type"],
2018 "cable_profile": policy["cable_profile"],
2019 "terminations": segment.as_json(),
2020 **termination_sources(segment.left.key, segment.right.key),
2021 },
2022 ),
2023 identities=(segment.left.identity, segment.right.identity),
2024 )
2026 def _shared_deletes(self) -> dict[str, tuple[str, ...]]:
2027 """Return, per desired segment, every Logical Cable deletion its creation waits for.
2029 Two traces that state one identical segment share one Planned Change, so the change names
2030 the deletions of both and reads the same in either unit.
2031 """
2032 deletes: dict[str, set] = {}
2033 for analysis in self._writing():
2034 if analysis.delete_identity is not None:
2035 for segment in analysis.pending:
2036 deletes.setdefault(segment.key, set()).add(analysis.delete_identity)
2037 return {key: tuple(sorted(values)) for key, values in deletes.items()}
2039 def _shared_sources(self) -> dict[str, list]:
2040 """Return, per desired segment, the provenance of every Source Trace that states it."""
2041 sources: dict[str, list] = {}
2042 for analysis in self.analyses:
2043 # Unlike _shared_deletes: a repaired trace reuses the Cable, so creation records it or nothing does.
2044 if analysis.invalid:
2045 continue
2046 for segment in analysis.pending:
2047 sources.setdefault(segment.key, []).append(_source_record(analysis.trace, segment.index))
2048 return {
2049 key: sorted(records, key=lambda record: (record["trace_identity"], record["segment_index"]))
2050 for key, records in sources.items()
2051 }
2053 def _writing(self) -> list[_TraceAnalysis]:
2054 """Return the analyses that still contribute writes to this plan."""
2055 return [analysis for analysis in self.analyses if not analysis.stopped and analysis.endpoints]
2057 def _unit(self, analysis: _TraceAnalysis) -> SynchronizationUnit:
2058 """Return the one Synchronization Unit one Source Trace produces."""
2059 changes = self._changes(analysis)
2060 if analysis.invalid:
2061 disposition = Disposition.INVALID
2062 elif analysis.blocked:
2063 disposition = Disposition.BLOCKED
2064 elif changes:
2065 disposition = Disposition.ACTIONABLE
2066 else:
2067 disposition = Disposition.NO_OP
2068 return SynchronizationUnit(
2069 identity=analysis.identity,
2070 disposition=disposition,
2071 changes=changes,
2072 diagnostics=tuple(analysis.diagnostics),
2073 display={
2074 **analysis.display,
2075 "detail": self._detail(analysis, disposition),
2076 "trace": self._workspace(analysis, bool(changes)),
2077 },
2078 selected_with=analysis.selected_with,
2079 )
2081 def _workspace(self, analysis: _TraceAnalysis, writes: bool) -> dict:
2082 """Return what the three review workspace panels show for one Source Trace."""
2083 summary = analysis.trace.endpoint_summary
2084 resolved = {segment.index: segment for segment in analysis.segments}
2085 by_reference = self._resolved.get(analysis.identity, {})
2086 segments = []
2087 stated_segments = list(analysis.trace.segments)
2088 for index, stated in enumerate(stated_segments):
2089 planned = resolved.get(index)
2090 source_left, source_right = _endpoint_label(stated.left), _endpoint_label(stated.right)
2091 left = source_left if planned is None else planned.left.topology_display
2092 right = source_right if planned is None else planned.right.topology_display
2093 segments.append(
2094 {
2095 "index": index,
2096 "cable_class": source_text(stated.cable_class),
2097 "source_left": source_left,
2098 "source_right": source_right,
2099 "left": left,
2100 "right": right,
2101 # A path that re-enters one panel implies a claim the source never states.
2102 "pass_through": index > 0 and same_device_and_cards(stated_segments[index - 1].right, stated.left),
2103 # Planning substituted the entry port only where the source re-used one port.
2104 "substituted": self._entered_through_claim(planned, by_reference.get(stated.left.identity_key)),
2105 # Presentation adds "conflict" from the findings the viewer may read.
2106 "status": self._segment_status(analysis, index, planned, writes),
2107 # An override cannot change a Cable this plan keeps, so the panel says which is which.
2108 "retained": index in analysis.proven,
2109 # The end that holds the channel child, so presentation hides the mark with that end.
2110 "breakout_leg": self._leg_end(analysis, planned),
2111 **self._policy_display(planned),
2112 **self._end_sources(planned),
2113 }
2114 )
2115 cable_classes = list(dict.fromkeys(source_text(segment.cable_class) for segment in stated_segments))
2116 return {
2117 "identity": analysis.trace.identity,
2118 "endpoints": {
2119 "from": _endpoint_label(summary.from_termination),
2120 "to": _endpoint_label(summary.to_termination),
2121 },
2122 "segments": segments,
2123 "cable_policies": [self._cable_class_display(cable_class) for cable_class in cable_classes],
2124 "port_classes": [
2125 {"label": item.label, "port_class": item.port_class} for item in analysis.trace.port_class_labels
2126 ],
2127 "logical_cable": self._logical_cable_display(analysis),
2128 "resolution_started": analysis.resolution_started,
2129 "topology_known": analysis.topology_read,
2130 # A unit with no changes proposes nothing, so the panel must not offer to delete one.
2131 "deletes_logical_cable": writes and analysis.deleted_logical_cable is not None,
2132 "devices": list(analysis.devices.values()),
2133 "terminations": list(analysis.terminations.values()),
2134 }
2136 def _cable_class_display(self, cable_class: str) -> dict:
2137 """Return one CableClass decision for the workspace policy table."""
2138 row = self._cable_class_mapping(cable_class)
2139 display = {
2140 "cable_class": cable_class,
2141 "cable_type": "Unresolved" if row is None else row.cable_type_display(),
2142 "cable_profile": "Unresolved" if row is None else row.cable_profile_display(),
2143 "policy": (None if row is None else row.decided_policy()) or {},
2144 }
2145 return display if row is None else disclosed_policy(row, self._may_view(row), display)
2147 @staticmethod
2148 def _leg_end(analysis: _TraceAnalysis, planned: _DesiredSegment | None) -> str:
2149 """Return the segment end that holds the harness leg's channel child, or nothing for an ordinary segment."""
2150 if planned is None or not analysis.is_leg(planned):
2151 return ""
2152 return "left" if planned.left.key == analysis.leg.child.key else "right"
2154 def _policy_display(self, planned: _DesiredSegment | None) -> dict:
2155 """Return the Cable policy in force for one resolved segment, and whether it is an override."""
2156 if planned is None:
2157 return {"segment_key": "", "overridden": False, "cable_type": "", "cable_profile": "", "policy": {}}
2158 row = self._policy_row(planned)
2159 display = {
2160 "segment_key": planned.key,
2161 "overridden": planned.key in self._overrides,
2162 "cable_type": "Unresolved" if row is None else row.cable_type_display(),
2163 "cable_profile": "Unresolved" if row is None else row.cable_profile_display(),
2164 # The stored values, not their labels, so the workspace can offer what is in force.
2165 "policy": (None if row is None else row.decided_policy()) or {},
2166 }
2167 return display if row is None else disclosed_policy(row, self._may_view(row), display)
2169 @staticmethod
2170 def _end_sources(planned: _DesiredSegment | None) -> dict:
2171 """Return the rows that authorize a planned segment's two NetBox Device and port names."""
2172 if planned is None:
2173 return {}
2174 return {
2175 source_key: segment_end_sources(end.key, end.device_id)
2176 for source_key, end in zip(SEGMENT_END_SOURCES.values(), planned.terminations, strict=True)
2177 }
2179 @staticmethod
2180 def _entered_through_claim(planned: _DesiredSegment | None, stated: _Termination | None) -> bool:
2181 """Return whether planning entered this segment through a port the source never stated."""
2182 return planned is not None and stated is not None and planned.left.key != stated.key
2184 @staticmethod
2185 def _segment_status(analysis: _TraceAnalysis, index: int, planned, writes: bool) -> str:
2186 """Return the verdict the proposed panel shows for one segment, or none for an unplanned one."""
2187 if planned is None:
2188 return ""
2189 if index in analysis.proven:
2190 return REUSE_SEGMENT
2191 # A blocked unit contributes no change, so nothing here is going to be created.
2192 return CREATE_SEGMENT if writes else ""
2194 def _logical_cable_display(self, analysis: _TraceAnalysis) -> dict | None:
2195 """Return the Logical Cable NetBox holds for this trace, as far as the actor may see it."""
2196 if analysis.logical_cable is None:
2197 return None
2198 cable = analysis.logical_cable.cable
2199 disclosure, _identities = self._cable_diagnostic_disclosure(cable)
2200 if not disclosure["cable_visible"]:
2201 return {"visible": False, "display": "", "description": "", "tags": []}
2202 review = self._cable_review_snapshot(cable)
2203 return {
2204 "visible": True,
2205 "display": review["display"],
2206 "description": review["description"],
2207 "tags": review["tags"],
2208 "disclosure_source": disclosure["disclosure_source"],
2209 }
2211 def _cable_review_snapshot(self, cable) -> dict:
2212 """Reuse one Cable review value for all changes and displays in this plan."""
2213 if cable.pk not in self._cable_review_snapshots:
2214 self._cable_review_snapshots[cable.pk] = _cable_review_snapshot(cable)
2215 return self._cable_review_snapshots[cable.pk]
2217 def _changes(self, analysis: _TraceAnalysis) -> tuple[PlannedChange, ...]:
2218 """Return the deletion and the creations one actionable trace performs, in that order."""
2219 if analysis.stopped:
2220 return ()
2221 changes = []
2222 if analysis.deleted_logical_cable is not None:
2223 changes.append(self._delete_change(analysis.deleted_logical_cable))
2224 for segment in analysis.segments:
2225 if analysis.is_leg(segment):
2226 if analysis.harness_pending:
2227 changes.append(self._harness_change(analysis.harness))
2228 elif segment.index not in analysis.proven:
2229 changes.append(self._create_change(segment, analysis.policies[segment.index]))
2230 return tuple(changes)
2232 def _delete_change(self, logical: _ExistingCable) -> PlannedChange:
2233 """Return the one deletion a Patched Path Replacement ever performs."""
2234 return PlannedChange(
2235 identity=_delete_identity(logical.cable.pk),
2236 target_module=CableModule.key,
2237 operation="delete",
2238 payload={"cable_id": logical.cable.pk},
2239 preconditions={
2240 "cable_id": logical.cable.pk,
2241 "terminations": logical.terminations,
2242 "review_fingerprint": fingerprint_of(self._cable_review_snapshot(logical.cable)),
2243 },
2244 )
2246 def _create_change(self, segment: _DesiredSegment, policy: dict) -> PlannedChange:
2247 """Return the creation of one physical segment, shared by every trace that states it."""
2248 return PlannedChange(
2249 identity=segment.change_identity,
2250 target_module=CableModule.key,
2251 operation="create",
2252 payload={
2253 "terminations": segment.as_json(),
2254 "status": CABLE_STATUS,
2255 "sources": self._sources_by_segment.get(segment.key, []),
2256 **policy,
2257 },
2258 dependencies=self._deletes_by_segment.get(segment.key, ()),
2259 preconditions={"terminations": segment.as_json()},
2260 )
2262 def _harness_change(self, harness: _Harness) -> PlannedChange:
2263 """Return the one creation every leg unit of a harness carries, identical in each unit."""
2264 if harness.intent is None:
2265 raise RuntimeError("A harness that cannot be written has no Planned Change.")
2266 sources = sorted(
2267 (_source_record(leg.analysis.trace, leg.segment.index) for leg in harness.legs),
2268 key=lambda record: (record["trace_identity"], record["segment_index"]),
2269 )
2270 return PlannedChange(
2271 identity=harness.change_identity,
2272 target_module=CableModule.key,
2273 operation="create",
2274 payload={
2275 "harness": cable_harness.harness_payload(harness.intent),
2276 "cable_type": harness.cable_type,
2277 "status": CABLE_STATUS,
2278 "sources": sources,
2279 },
2280 preconditions={"terminations": sorted(end.as_json() for end in harness.ends)},
2281 )
2283 @staticmethod
2284 def _detail(analysis: _TraceAnalysis, disposition: str) -> str:
2285 """Return the operator wording one unit shows before the trace workspace exists."""
2286 if disposition == Disposition.NO_OP:
2287 return "The stated path already exists."
2288 if disposition != Disposition.ACTIONABLE:
2289 return ""
2290 count = len(analysis.pending) + analysis.harness_pending
2291 if analysis.deleted_logical_cable is not None:
2292 return f"Would replace the logical cable with {count} physical segment(s)."
2293 return f"Would create {count} physical segment(s)."
2296class CableModule:
2297 """Plan and write the Cables one Source Trace states, as a Patched Path Replacement."""
2299 key = TargetModuleKey.CABLE
2300 consumes = frozenset({OutputKind.SOURCE_TRACE})
2302 def plan(
2303 self,
2304 source_batch,
2305 profile,
2306 catalog,
2307 netbox_reader,
2308 *,
2309 lock_plan_references: bool = False,
2310 ) -> list[SynchronizationUnit]:
2311 """Return one unit per trace and optionally lock rows that prove the plan."""
2312 del catalog
2313 if not (self.consumes & source_batch.output_kinds):
2314 return []
2315 return _CableBatch(
2316 source_batch.rows,
2317 profile,
2318 netbox_reader,
2319 lock_plan_references=lock_plan_references,
2320 ).units()
2322 def apply(self, planned_change: PlannedChange, execution_context) -> Any:
2323 """Apply one Cable change, having locked its rows and rechecked its preconditions."""
2324 if planned_change.operation == "delete":
2325 return self._delete(planned_change, execution_context)
2326 if planned_change.operation == "create":
2327 return self._create(planned_change, execution_context)
2328 raise PreconditionFailed(f"The Cable Target Module cannot apply operation '{planned_change.operation}'.")
2330 @staticmethod
2331 def _delete(planned_change: PlannedChange, execution_context) -> DeletedObject:
2332 """Remove the one direct Logical Cable this unit replaces."""
2333 from dcim.models import Cable
2335 cable_id = planned_change.preconditions["cable_id"]
2336 cable = Cable.objects.filter(pk=cable_id).select_for_update(of=("self",)).first()
2337 if cable is None:
2338 raise PreconditionFailed(f"Cable {cable_id} is gone, so the logical cable cannot be replaced.")
2339 current = _cable_terminations(cable_id)
2340 if current != [list(item) for item in planned_change.preconditions["terminations"]]:
2341 raise PreconditionFailed(f"Cable {cable_id} was re-terminated after the plan was made.")
2342 ends = [_model_for_label(label).objects.get(pk=object_id) for label, object_id in current]
2343 if cable_harness.is_breakout(cable, ends):
2344 raise PreconditionFailed(f"Cable {cable_id} is a breakout Cable, and the import never deletes it.")
2345 if _deleted_cable_review_fingerprint(cable) != planned_change.preconditions["review_fingerprint"]:
2346 raise PreconditionFailed(f"Cable {cable_id} changed after the plan was made.")
2347 enforce_saved_object_permission(cable, execution_context.actor, "delete")
2348 snapshot = DeletedObject(
2349 object_type="dcim.cable",
2350 object_id=cable_id,
2351 )
2352 cable.delete()
2353 return snapshot
2355 @staticmethod
2356 def _create(planned_change: PlannedChange, execution_context) -> Any:
2357 """Create one physical Cable segment or one harness, and the provenance rows its Source Traces earn.
2359 The write is built from locked live rows through the same constructor planning used.
2360 """
2361 from dcim.models import Cable
2363 payload = planned_change.payload
2364 if "harness" in payload:
2365 intent = cable_harness.locked_harness_intent(payload["harness"], CableModule._free_termination)
2366 else:
2367 first, second = (CableModule._free_termination(*end) for end in payload["terminations"])
2368 intent = cable_harness.ordinary_intent(first, second, profile=payload["cable_profile"] or "")
2369 cable = Cable(
2370 type=payload["cable_type"],
2371 profile=intent.profile,
2372 status=payload["status"],
2373 a_terminations=list(intent.a_terminations),
2374 b_terminations=list(intent.b_terminations),
2375 )
2376 cable.full_clean()
2377 cable.save()
2378 enforce_saved_object_permission(cable, execution_context.actor, "add")
2379 CableModule._store_provenance(cable, payload, execution_context.profile)
2380 return cable
2382 @staticmethod
2383 def _free_termination(label: str, object_id: int, device_id: int):
2384 """Return one locked termination that is still free and on its resolved Device."""
2385 component = _model_for_label(label).objects.filter(pk=object_id).select_for_update(of=("self",)).first()
2386 if component is None:
2387 raise PreconditionFailed(f"{label} {object_id} is gone, so the segment cannot be created.")
2388 if component.device_id != device_id:
2389 raise PreconditionFailed(f"{label} {object_id} moved to another Device after the plan was made.")
2390 if component.cable_id is not None:
2391 raise PreconditionFailed(f"{component} received a cable after the plan was made.")
2392 return component
2394 @staticmethod
2395 def _store_provenance(cable, payload, profile) -> None:
2396 """Write one provenance row per Source Trace that states this segment."""
2397 from .models import CableImportSource, index_digest
2399 for record in payload["sources"]:
2400 # The unique constraint carries the digest, so the lookup matches it.
2401 CableImportSource.objects.update_or_create(
2402 cable=cable,
2403 profile=profile,
2404 trace_key=index_digest(record["trace_identity"]),
2405 defaults={
2406 "trace_identity": record["trace_identity"],
2407 "segment_index": record["segment_index"],
2408 "from_text": record["from_text"],
2409 "to_text": record["to_text"],
2410 "direction": record["direction"],
2411 "workbook_fingerprint": record["workbook_fingerprint"],
2412 "sheet": record["sheet"],
2413 "block_ordinal": record["block_ordinal"],
2414 "row_start": record["row_start"],
2415 "row_end": record["row_end"],
2416 "export_timestamp": record["export_timestamp"],
2417 },
2418 )
2421def _cable_terminations(cable_id: int) -> list:
2422 """Return one Cable's termination pairs in the order a precondition records them."""
2423 from dcim.models import CableTermination
2425 return sorted(
2426 [row.termination_type.model_class()._meta.label_lower, row.termination_id]
2427 for row in CableTermination.objects.filter(cable_id=cable_id).order_by("pk").select_for_update(of=("self",))
2428 )
2431__all__ = (
2432 "AUTOMATICALLY_RESOLVED",
2433 "CABLE_STATUS",
2434 "CREATE_SEGMENT",
2435 "ELIGIBLE_TERMINATION_LIMIT",
2436 "MANUALLY_RESOLVED",
2437 "REUSE_SEGMENT",
2438 "UNRESOLVED",
2439 "CableModule",
2440 "EligibleTerminations",
2441 "eligible_terminations",
2442)