Coverage for netbox_data_import/cable_disclosure.py: 100.00%
281 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"""Apply live row permissions to cached Cable presentation data."""
5from dataclasses import replace
6from types import MappingProxyType
7from typing import NamedTuple
9from .catalog import TargetModuleKey
10from .field_keys import CABLE_END_KINDS
11from .plan import ImportPlan, register_diagnostic_display
13CABLE_ROW = "dcim.cable"
14CABLE_CLASS_MAPPING_ROW = "netbox_data_import.cableclassmapping"
15CABLE_SEGMENT_OVERRIDE_ROW = "netbox_data_import.cablesegmentoverride"
16DEVICE_HIDDEN = "a Device you cannot view"
17DEVICE_ROW = "dcim.device"
18DISCLOSURE_SOURCE = "disclosure_source"
19POLICY_HIDDEN = "a policy you cannot view"
20POLICY_VISIBLE = "policy_visible"
21TERMINATION_HIDDEN = "a termination you cannot view"
22TERMINATION_SOURCES = "termination_sources"
23# A termination display may also name a Device, when that Device alone can hide the display's facts.
24TERMINATION_SOURCE_KINDS = (*sorted(CABLE_END_KINDS), DEVICE_ROW)
26# A segment names its two planned ends apart, so one hidden end leaves the other readable.
27SEGMENT_END_SOURCES = MappingProxyType({"left": "left_sources", "right": "right_sources"})
28# The rows a finding's wording describes; `wording_sources` builds them.
29WORDING_SOURCES = "wording_sources"
30# What a finding becomes when its viewer cannot see the rows its WORDING_SOURCES name.
31FINDING_HIDDEN = "cable.finding_hidden"
33_REFERENCE_FIELDS = frozenset({"device", "cards", "port", "port_class"})
34# These codes replace the source `port` with the NetBox port they resolved.
35_RESOLVED_PORT_FIELDS = _REFERENCE_FIELDS - {"port"}
38class _DisplaySchema(NamedTuple):
39 """The display fields one diagnostic may carry, grouped by the row that authorizes them."""
41 public: frozenset
42 cable: frozenset = frozenset()
43 policy: frozenset = frozenset()
44 termination: frozenset = frozenset()
45 # The wording states facts of ports, so the finding renders only when its WORDING_SOURCES are visible.
46 wording: bool = False
47 # The finding carries WORDING_SOURCES only when a harness gives rise to it, and then follows the same rule.
48 wording_optional: bool = False
51def _harness_schema(public: frozenset, *, cable: frozenset = frozenset(), termination: frozenset = frozenset()):
52 """Return the schema of one harness finding, which always names its leg and its parent Interface.
54 Its wording describes the harness as a whole, so every port of the harness authorizes it (section 6.6).
55 """
56 return _DisplaySchema(public | {"segment_index"}, cable=cable, termination=termination | {"parent"}, wording=True)
59_DIAGNOSTIC_DISCLOSURE_FIELDS = MappingProxyType(
60 {
61 "cable.ambiguous_mapped_peer": _DisplaySchema(_RESOLVED_PORT_FIELDS, termination=frozenset({"port", "peers"})),
62 "cable.attribute_drift": _DisplaySchema(
63 frozenset({"segment_index"}), cable=frozenset({"cable", "status", "type", "profile", "label"})
64 ),
65 "cable.cableclass_unmapped": _DisplaySchema(frozenset({"segment_index", "cable_class"})),
66 FINDING_HIDDEN: _DisplaySchema(frozenset()),
67 "cable.harness_channel_gap": _harness_schema(frozenset(), termination=frozenset({"missing"})),
68 "cable.harness_channel_state_invalid": _harness_schema(
69 frozenset({"channel", "fault"}), termination=frozenset({"port"})
70 ),
71 "cable.harness_leg_conflict": _harness_schema(frozenset({"conflicts"})),
72 "cable.harness_member_unchannelized": _harness_schema(frozenset({"trace"}), termination=frozenset({"port"})),
73 "cable.harness_member_unresolved": _harness_schema(frozenset({"trace"})),
74 "cable.harness_mixed_terminations": _harness_schema(frozenset({"models"})),
75 "cable.harness_occupied": _harness_schema(
76 frozenset({"marked_connected"}), cable=frozenset({"cable"}), termination=frozenset({"port"})
77 ),
78 "cable.harness_partial": _harness_schema(frozenset(), termination=frozenset({"missing"})),
79 "cable.harness_profile_unavailable": _harness_schema(frozenset(), termination=frozenset({"channels"})),
80 "cable.harness_state": _harness_schema(frozenset(), termination=frozenset({"channels", "legs", "profile"})),
81 "cable.harness_type_conflict": _harness_schema(frozenset()),
82 "cable.harness_unsupported": _harness_schema(frozenset({"trace"})),
83 "cable.incompatible_terminations": _DisplaySchema(
84 frozenset({"segment_index", "left_field_key", "right_field_key"}),
85 termination=frozenset({"left_model", "right_model"}),
86 ),
87 "cable.media_family_mismatch": _DisplaySchema(
88 frozenset(), cable=frozenset({"segments"}), policy=frozenset({"segments"})
89 ),
90 "cable.multi_termination_conflict": _DisplaySchema(
91 frozenset({"segment_index"}), cable=frozenset({"cable"}), termination=frozenset({"port"})
92 ),
93 "cable.pass_through_not_mapped": _DisplaySchema(
94 _REFERENCE_FIELDS, termination=frozenset({"entry", "exit", "mapped"})
95 ),
96 "cable.pass_through_verified": _DisplaySchema(_REFERENCE_FIELDS, termination=frozenset({"entry", "exit"})),
97 "cable.permission_denied": _DisplaySchema(_REFERENCE_FIELDS | {"permission"}, cable=frozenset({"cable"})),
98 "cable.planned_termination_conflict": _DisplaySchema(
99 frozenset({"segment_index", "competing_trace", "competing_segment"}),
100 termination=frozenset({"termination"}),
101 wording_optional=True,
102 ),
103 "cable.policy_stale": _DisplaySchema(frozenset({"segment_index", "cable_class"})),
104 "cable.profile_incompatible": _DisplaySchema(frozenset({"segment_index", "cable_class"})),
105 "cable.resolved_segment_conflict": _DisplaySchema(
106 frozenset({"segment_index"}),
107 policy=frozenset({"cable_type", "cable_profile"}),
108 termination=frozenset({"terminations"}),
109 ),
110 "cable.same_port_continuation": _DisplaySchema(_RESOLVED_PORT_FIELDS, termination=frozenset({"port", "peer"})),
111 "cable.segment_override_lost": _DisplaySchema(
112 frozenset({"segment_index"}), policy=frozenset({"cable_type", "cable_profile"})
113 ),
114 "cable.segment_reused": _DisplaySchema(
115 frozenset({"segment_index"}), cable=frozenset({"cable"}), wording_optional=True
116 ),
117 "cable.segment_self_connection": _DisplaySchema(
118 frozenset({"segment_index", "cable_class"}), termination=frozenset({"termination"})
119 ),
120 "cable.termination_channelized": _DisplaySchema(
121 frozenset({"segment_index"}), termination=frozenset({"port"}), wording=True
122 ),
123 "cable.termination_kind_mismatch": _DisplaySchema(
124 _REFERENCE_FIELDS | {"claimed_kind", "selected_object_type"}, wording=True
125 ),
126 "cable.termination_not_connectable": _DisplaySchema(
127 frozenset({"segment_index"}), termination=frozenset({"port", "interface_type", "parent"}), wording=True
128 ),
129 "cable.termination_occupied": _DisplaySchema(
130 frozenset({"segment_index", "marked_connected"}),
131 cable=frozenset({"cable"}),
132 termination=frozenset({"port"}),
133 ),
134 "cable.termination_unresolved": _DisplaySchema(_REFERENCE_FIELDS),
135 "cable.unsupported_termination_kind": _DisplaySchema(_REFERENCE_FIELDS | {"selected_object_type"}),
136 }
137)
139CABLE_DIAGNOSTIC_FIELDS = MappingProxyType(
140 {
141 code: fields.public | fields.cable | fields.policy | fields.termination
142 for code, fields in _DIAGNOSTIC_DISCLOSURE_FIELDS.items()
143 }
144)
146CABLE_DIAGNOSTIC_DISCLOSURES = MappingProxyType(
147 {code: fields.cable for code, fields in _DIAGNOSTIC_DISCLOSURE_FIELDS.items() if fields.cable}
148)
150POLICY_DIAGNOSTIC_DISCLOSURES = MappingProxyType(
151 {code: fields.policy for code, fields in _DIAGNOSTIC_DISCLOSURE_FIELDS.items() if fields.policy}
152)
154# Every code whose wording states facts of ports, so its display must carry WORDING_SOURCES.
155WORDING_CODES = frozenset(code for code, fields in _DIAGNOSTIC_DISCLOSURE_FIELDS.items() if fields.wording)
157TERMINATION_DIAGNOSTIC_DISCLOSURES = MappingProxyType(
158 {code: fields.termination for code, fields in _DIAGNOSTIC_DISCLOSURE_FIELDS.items() if fields.termination}
159)
162def disclosure_source(row_kind: str, row_pk: int) -> dict:
163 """Return the plan-side identity for one row that supplied display text."""
164 return {"kind": row_kind, "pk": row_pk}
167def device_source(device_id: int) -> dict:
168 """Return the source a display needs when it names one resolved Device."""
169 return disclosure_source(DEVICE_ROW, device_id)
172def segment_end_sources(port: tuple[str, int], device_id: int) -> list:
173 """Return the rows that authorize one planned segment end, which names its Device and its port."""
174 return [disclosure_source(*port), device_source(device_id)]
177def wording_sources(ends) -> dict:
178 """Return the sources a finding's wording needs: each ``(port, device_id)`` end, the port followed by its Device."""
179 return {
180 WORDING_SOURCES: [
181 source for port, device_id in dict.fromkeys(ends) for source in segment_end_sources(port, device_id)
182 ]
183 }
186def termination_sources(*terminations) -> dict:
187 """Return the sources a display needs when it names the given resolved terminations."""
188 return {TERMINATION_SOURCES: [disclosure_source(label, object_id) for label, object_id in terminations]}
191def disclosed_cable(cable) -> dict:
192 """Return the fields a planner may record for one visible Cable."""
193 return {
194 "cable_visible": True,
195 "cable": str(cable),
196 DISCLOSURE_SOURCE: disclosure_source(CABLE_ROW, cable.pk),
197 }
200def policy_row_kind(row) -> str:
201 """Return the disclosure kind for one supported Cable policy row."""
202 from .models import CableClassMapping, CableSegmentOverride
204 if isinstance(row, CableClassMapping):
205 return CABLE_CLASS_MAPPING_ROW
206 if isinstance(row, CableSegmentOverride):
207 return CABLE_SEGMENT_OVERRIDE_ROW
208 raise TypeError(f"Unsupported Cable policy row: {type(row).__name__}")
211def disclosed_policy(row, visible: bool, display: dict) -> dict:
212 """Return policy display values only when the planning viewer may read their row."""
213 if not visible:
214 return _redact_policy(display)
215 return {
216 **display,
217 POLICY_VISIBLE: True,
218 DISCLOSURE_SOURCE: disclosure_source(policy_row_kind(row), row.pk),
219 }
222def policy_row_is_disclosed(display: dict, row) -> bool:
223 """Return whether a presented value names this policy row as its visible source."""
224 return (
225 display.get(POLICY_VISIBLE) is True
226 and _source_pk(display.get(DISCLOSURE_SOURCE), policy_row_kind(row)) == row.pk
227 )
230def validate_diagnostic_disclosures(code: str, display: dict) -> None:
231 """Reject diagnostic display fields that bypass the shared disclosure vocabulary."""
232 fields = _DIAGNOSTIC_DISCLOSURE_FIELDS.get(code)
233 if fields is None:
234 raise ValueError(f"Diagnostic '{code}' has no registered display schema.")
235 public, cable, policy, termination, wording, wording_optional = fields
236 if code == "cable.media_family_mismatch":
237 _validate_media_segments(display)
238 return
239 metadata = set()
240 if cable:
241 metadata.update({DISCLOSURE_SOURCE, "cable_visible"})
242 if policy:
243 metadata.update({DISCLOSURE_SOURCE, POLICY_VISIBLE})
244 if termination:
245 metadata.add(TERMINATION_SOURCES)
246 if not _well_formed_termination_sources(display.get(TERMINATION_SOURCES)):
247 raise ValueError(f"Diagnostic '{code}' names terminations without their sources.")
248 if wording or (wording_optional and WORDING_SOURCES in display):
249 metadata.add(WORDING_SOURCES)
250 _validate_wording_sources(code, display)
251 unknown = set(display) - set(public) - set(cable) - set(policy) - set(termination) - metadata
252 if unknown:
253 names = ", ".join(sorted(unknown))
254 raise ValueError(f"Diagnostic '{code}' has unregistered display fields: {names}.")
255 _validate_row_flags(code, display, cable, policy)
258def _validate_wording_sources(code: str, display: dict) -> None:
259 """Reject wording sources that are not port and Device pairs, or that leave out a port the display names."""
260 listed = display.get(WORDING_SOURCES)
261 sources = list(listed) if isinstance(listed, (list, tuple)) else []
262 pairs = list(zip(sources[::2], sources[1::2], strict=False))
263 if (
264 not pairs
265 or len(sources) != 2 * len(pairs)
266 or not all(
267 _well_formed_termination_sources([port])
268 and _source_pk(port, DEVICE_ROW) is None
269 and _source_pk(device, DEVICE_ROW) is not None
270 for port, device in pairs
271 )
272 ):
273 raise ValueError(f"Diagnostic '{code}' states a fact of a port without its sources.")
274 named = list(display.get(TERMINATION_SOURCES) or ())
275 ports = {(port["kind"], port["pk"]) for port, _device in pairs}
276 if any((source["kind"], source["pk"]) not in ports for source in named if _source_pk(source, DEVICE_ROW) is None):
277 raise ValueError(f"Diagnostic '{code}' names a port its wording sources leave out.")
280def _validate_row_flags(code: str, display: dict, cable: frozenset, policy: frozenset) -> None:
281 """Reject Cable and policy fields whose visibility flag and source disagree."""
282 source = display.get(DISCLOSURE_SOURCE)
283 if set(display) & set(cable) or "cable_visible" in display:
284 visible = display.get("cable_visible")
285 if not isinstance(visible, bool):
286 raise ValueError(f"Diagnostic '{code}' has Cable fields without a visibility flag.")
287 if visible and _source_pk(source, CABLE_ROW) is None:
288 raise ValueError(f"Diagnostic '{code}' has visible Cable fields without a Cable source.")
289 if not visible and source is not None:
290 raise ValueError(f"Diagnostic '{code}' has a source for a hidden Cable.")
291 if set(display) & set(policy) or POLICY_VISIBLE in display:
292 visible = display.get(POLICY_VISIBLE)
293 if not isinstance(visible, bool):
294 raise ValueError(f"Diagnostic '{code}' has policy fields without a visibility flag.")
295 row_kinds = (CABLE_CLASS_MAPPING_ROW, CABLE_SEGMENT_OVERRIDE_ROW)
296 if visible and not any(_source_pk(source, kind) is not None for kind in row_kinds):
297 raise ValueError(f"Diagnostic '{code}' has visible policy fields without a policy source.")
298 if not visible and source is not None:
299 raise ValueError(f"Diagnostic '{code}' has a source for a hidden policy.")
302def _validate_media_segments(display: dict) -> None:
303 """Reject media observations that omit provenance or add unregistered nested fields."""
304 if set(display) != {"segments"} or not isinstance(display["segments"], list):
305 raise ValueError("Diagnostic 'cable.media_family_mismatch' has an invalid display schema.")
306 base = {"segment_index", "retained", "visible", "origin"}
307 disclosed = {"cable_type", "family", DISCLOSURE_SOURCE}
308 for segment in display["segments"]:
309 if not isinstance(segment, dict) or frozenset(segment) not in {frozenset(base), frozenset(base | disclosed)}:
310 raise ValueError("Diagnostic 'cable.media_family_mismatch' has invalid segment fields.")
311 origin = segment.get("origin")
312 if origin not in {"cable", "policy"} or not isinstance(segment.get("visible"), bool):
313 raise ValueError("Diagnostic 'cable.media_family_mismatch' has an invalid segment origin.")
314 if segment["visible"] is False:
315 if set(segment) != base:
316 raise ValueError("Diagnostic 'cable.media_family_mismatch' discloses a hidden segment.")
317 continue
318 row_kinds = (CABLE_ROW,) if origin == "cable" else (CABLE_CLASS_MAPPING_ROW, CABLE_SEGMENT_OVERRIDE_ROW)
319 if not any(_source_pk(segment.get(DISCLOSURE_SOURCE), kind) is not None for kind in row_kinds):
320 raise ValueError("Diagnostic 'cable.media_family_mismatch' has a visible segment without its source.")
323def _source_pk(value, row_kind: str) -> int | None:
324 if not isinstance(value, dict) or value.get("kind") != row_kind:
325 return None
326 row_pk = value.get("pk")
327 return row_pk if isinstance(row_pk, int) and not isinstance(row_pk, bool) else None
330def _source_is_visible(value, row_kinds, visible_row_ids: dict[str, set[int]]) -> bool:
331 """Return whether one well-formed source names a row visible in its expected kind."""
332 return any(
333 (row_pk := _source_pk(value, row_kind)) is not None and row_pk in visible_row_ids.get(row_kind, ())
334 for row_kind in row_kinds
335 )
338def _well_formed_termination_sources(sources) -> bool:
339 """Return whether *sources* is a nonempty list of sources that each name a Cable End Kind row or its Device."""
340 return (
341 isinstance(sources, list)
342 and bool(sources)
343 and all(any(_source_pk(source, kind) is not None for kind in TERMINATION_SOURCE_KINDS) for source in sources)
344 )
347def _terminations_are_visible(sources, visible_row_ids: dict[str, set[int]]) -> bool:
348 """Return whether every termination, and every Device, one display names is live and visible."""
349 return _well_formed_termination_sources(sources) and all(
350 _source_is_visible(source, TERMINATION_SOURCE_KINDS, visible_row_ids) for source in sources
351 )
354def _identity_key(identity) -> tuple[str, int] | None:
355 """Return the Cable End Kind row one plan identity names, or None for any other identity."""
356 label, _, object_id = str(identity).rpartition(":")
357 if label not in CABLE_END_KINDS or not object_id.isdigit():
358 return None
359 return label, int(object_id)
362def _row_sources(value: dict) -> list:
363 """Return every row source one display mapping carries."""
364 sources = [value.get(DISCLOSURE_SOURCE)]
365 for key in (TERMINATION_SOURCES, WORDING_SOURCES, *SEGMENT_END_SOURCES.values()):
366 listed = value.get(key)
367 # A frozen plan holds a list as a tuple.
368 if isinstance(listed, (list, tuple)):
369 sources.extend(listed)
370 return sources
373def _end_is_visible(sources, visible_row_ids: dict[str, set[int]]) -> bool:
374 """Return whether a planned end's port and Device are both live and visible."""
375 return (
376 isinstance(sources, (list, tuple))
377 and len(sources) == 2
378 and _source_is_visible(sources[0], CABLE_END_KINDS, visible_row_ids)
379 and _source_is_visible(sources[1], (DEVICE_ROW,), visible_row_ids)
380 )
383def _row_ids(units) -> dict[str, set[int]]:
384 row_ids: dict[str, set[int]] = {
385 CABLE_ROW: set(),
386 CABLE_CLASS_MAPPING_ROW: set(),
387 CABLE_SEGMENT_OVERRIDE_ROW: set(),
388 DEVICE_ROW: set(),
389 **{kind: set() for kind in sorted(CABLE_END_KINDS)},
390 }
392 def collect(value) -> None:
393 if isinstance(value, dict):
394 for source in _row_sources(value):
395 for row_kind, ids in row_ids.items():
396 if row_pk := _source_pk(source, row_kind):
397 ids.add(row_pk)
398 for child in value.values():
399 collect(child)
400 elif isinstance(value, (list, tuple)):
401 for child in value:
402 collect(child)
404 for unit in units:
405 collect(unit.display)
406 for diagnostic in unit.diagnostics:
407 collect(diagnostic.display)
408 for identity in diagnostic.identities:
409 if key := _identity_key(identity):
410 row_ids[key[0]].add(key[1])
411 return row_ids
414def _redact_cable(display: dict, keys: frozenset[str]) -> dict:
415 for key in keys:
416 display.pop(key, None)
417 display.pop(DISCLOSURE_SOURCE, None)
418 display["cable_visible"] = False
419 return display
422def _redact_policy(display: dict) -> dict:
423 display = dict(display)
424 for key in ("cable_type", "cable_profile"):
425 if key in display:
426 display[key] = POLICY_HIDDEN
427 if "policy" in display:
428 display["policy"] = {}
429 display.pop(DISCLOSURE_SOURCE, None)
430 display[POLICY_VISIBLE] = False
431 return display
434def _media_segment(segment: dict, visible_row_ids: dict[str, set[int]]) -> dict:
435 source = segment.get(DISCLOSURE_SOURCE)
436 origin = segment.get("origin")
437 row_kinds = (
438 (CABLE_ROW,)
439 if origin == "cable"
440 else (CABLE_CLASS_MAPPING_ROW, CABLE_SEGMENT_OVERRIDE_ROW)
441 if origin == "policy"
442 else ()
443 )
444 if segment.get("visible") is True and _source_is_visible(source, row_kinds, visible_row_ids):
445 return segment
446 return {
447 "segment_index": segment["segment_index"],
448 "retained": segment["retained"],
449 "visible": False,
450 "origin": origin if origin in {"cable", "policy"} else "cable",
451 }
454def _redact_terminations(display: dict, keys: frozenset[str]) -> dict:
455 for key in keys & set(display):
456 display[key] = [] if isinstance(display[key], list) else TERMINATION_HIDDEN
457 display.pop(TERMINATION_SOURCES, None)
458 return display
461def _visible_identities(identities, visible_row_ids: dict[str, set[int]]) -> tuple:
462 """Return the identities a viewer may read: a hidden or deleted termination drops out."""
463 return tuple(
464 identity
465 for identity in identities
466 if (key := _identity_key(identity)) is None or key[1] in visible_row_ids.get(key[0], ())
467 )
470def _diagnostic(diagnostic, visible_row_ids: dict[str, set[int]]):
471 display = diagnostic.to_dict()["display"]
472 identities = _visible_identities(diagnostic.identities, visible_row_ids)
473 schema = _DIAGNOSTIC_DISCLOSURE_FIELDS.get(diagnostic.code)
474 if (
475 schema is not None
476 and (schema.wording or (schema.wording_optional and WORDING_SOURCES in display))
477 and not _terminations_are_visible(display[WORDING_SOURCES], visible_row_ids)
478 ):
479 return replace(diagnostic, code=FINDING_HIDDEN, display={}, identities=identities)
480 if diagnostic.code == "cable.media_family_mismatch":
481 display["segments"] = [_media_segment(segment, visible_row_ids) for segment in display["segments"]]
482 elif (
483 (keys := CABLE_DIAGNOSTIC_DISCLOSURES.get(diagnostic.code))
484 and display.get("cable_visible") is True
485 and not _source_is_visible(display.get(DISCLOSURE_SOURCE), (CABLE_ROW,), visible_row_ids)
486 ):
487 display = _redact_cable(display, keys)
488 if (
489 diagnostic.code in POLICY_DIAGNOSTIC_DISCLOSURES
490 and display.get(POLICY_VISIBLE) is True
491 and not _source_is_visible(
492 display.get(DISCLOSURE_SOURCE),
493 (CABLE_CLASS_MAPPING_ROW, CABLE_SEGMENT_OVERRIDE_ROW),
494 visible_row_ids,
495 )
496 ):
497 display = _redact_policy(display)
498 keys = TERMINATION_DIAGNOSTIC_DISCLOSURES.get(diagnostic.code)
499 if keys and not _terminations_are_visible(display.get(TERMINATION_SOURCES), visible_row_ids):
500 display = _redact_terminations(display, keys)
501 return replace(diagnostic, display=display, identities=identities)
504def _redact_segment_ends(segment: dict, visible_row_ids: dict[str, set[int]]) -> None:
505 """Hide each end of one planned segment whose port or Device the viewer cannot see."""
506 for end, source_key in SEGMENT_END_SOURCES.items():
507 if not _end_is_visible(segment.get(source_key), visible_row_ids):
508 segment.pop(source_key, None)
509 segment[end] = TERMINATION_HIDDEN
510 # Being a leg is a fact of the channel child, so it hides with that end.
511 if segment["breakout_leg"] == end:
512 segment["breakout_leg"] = ""
515def _unit(unit, visible_row_ids: dict[str, set[int]]):
516 display = unit.to_dict()["display"]
517 trace = display.get("trace")
518 if trace is not None:
519 logical = trace.get("logical_cable")
520 if (
521 logical is not None
522 and logical.get("visible") is True
523 and not _source_is_visible(logical.get(DISCLOSURE_SOURCE), (CABLE_ROW,), visible_row_ids)
524 ):
525 trace["logical_cable"] = {"visible": False, "display": "", "description": "", "tags": []}
526 for segment in trace.get("segments") or ():
527 if segment.get(POLICY_VISIBLE) is True and not _source_is_visible(
528 segment.get(DISCLOSURE_SOURCE),
529 (CABLE_CLASS_MAPPING_ROW, CABLE_SEGMENT_OVERRIDE_ROW),
530 visible_row_ids,
531 ):
532 segment.pop(DISCLOSURE_SOURCE, None)
533 segment.update(_redact_policy(segment))
534 for policy in trace.get("cable_policies") or ():
535 if policy.get(POLICY_VISIBLE) is True and not _source_is_visible(
536 policy.get(DISCLOSURE_SOURCE), (CABLE_CLASS_MAPPING_ROW,), visible_row_ids
537 ):
538 policy.pop(DISCLOSURE_SOURCE, None)
539 policy.update(_redact_policy(policy))
540 for question in trace.get("devices") or ():
541 if question.get("selected") and not _source_is_visible(
542 question.get(DISCLOSURE_SOURCE), (DEVICE_ROW,), visible_row_ids
543 ):
544 question.pop(DISCLOSURE_SOURCE, None)
545 question["selected"] = DEVICE_HIDDEN
546 for field in trace.get("terminations") or ():
547 if field.get("selected") and not _source_is_visible(
548 field.get(DISCLOSURE_SOURCE), CABLE_END_KINDS, visible_row_ids
549 ):
550 field.pop(DISCLOSURE_SOURCE, None)
551 field.update(selected=TERMINATION_HIDDEN, selected_type="")
552 for segment in trace.get("segments") or ():
553 # Only a planned segment names NetBox ports; an unplanned one repeats the source text.
554 if segment.get("segment_key"):
555 _redact_segment_ends(segment, visible_row_ids)
556 diagnostics = tuple(_diagnostic(item, visible_row_ids) for item in unit.diagnostics)
557 return replace(unit, diagnostics=diagnostics, display=display)
560def present_units(units, viewer) -> tuple:
561 """Return presentation copies after one live visibility query per referenced row kind."""
562 if viewer is None:
563 raise TypeError("ReviewWorkspace requires a live viewer.")
564 units = tuple(units)
565 from django.apps import apps
567 row_ids = _row_ids(units)
568 visible = {}
569 for row_kind, ids in row_ids.items():
570 model = apps.get_model(row_kind)
571 visible[row_kind] = (
572 set(model.objects.restrict(viewer, "view").filter(pk__in=sorted(ids)).values_list("pk", flat=True))
573 if ids
574 else set()
575 )
576 return tuple(_unit(unit, visible) for unit in units)
579def redact_deleted_cables(plan_data: dict) -> dict:
580 """Return an execution copy whose deleted Cable rows authorize no stored display text."""
581 plan = ImportPlan.from_dict(plan_data)
582 visible = _row_ids(plan.units)
583 deleted_ids = {
584 change.payload["cable_id"]
585 for unit in plan.units
586 for change in unit.changes
587 if change.target_module == "cable" and change.operation == "delete"
588 }
589 visible[CABLE_ROW].difference_update(deleted_ids)
590 units = tuple(_unit(unit, visible) for unit in plan.units)
591 return replace(plan, units=units).to_dict()
594def register() -> None:
595 """Register the Cable display vocabulary, so a cached plan with an unknown Cable display field is refused."""
596 register_diagnostic_display(TargetModuleKey.CABLE, validate_diagnostic_disclosures)
599__all__ = (
600 "CABLE_CLASS_MAPPING_ROW",
601 "CABLE_DIAGNOSTIC_DISCLOSURES",
602 "CABLE_DIAGNOSTIC_FIELDS",
603 "CABLE_ROW",
604 "CABLE_SEGMENT_OVERRIDE_ROW",
605 "DEVICE_HIDDEN",
606 "DEVICE_ROW",
607 "DISCLOSURE_SOURCE",
608 "FINDING_HIDDEN",
609 "POLICY_DIAGNOSTIC_DISCLOSURES",
610 "POLICY_HIDDEN",
611 "POLICY_VISIBLE",
612 "SEGMENT_END_SOURCES",
613 "TERMINATION_DIAGNOSTIC_DISCLOSURES",
614 "TERMINATION_HIDDEN",
615 "TERMINATION_SOURCES",
616 "WORDING_CODES",
617 "WORDING_SOURCES",
618 "device_source",
619 "disclosed_cable",
620 "disclosed_policy",
621 "disclosure_source",
622 "policy_row_is_disclosed",
623 "present_units",
624 "redact_deleted_cables",
625 "register",
626 "segment_end_sources",
627 "termination_sources",
628 "validate_diagnostic_disclosures",
629 "wording_sources",
630)