Coverage for netbox_data_import/cable_harness.py: 98.02%
202 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"""Breakout harness facts and the one constructor of every Cable write.
5This module is the only reader of Interface `channels` and `channel_id`, and of the breakout Cable
6Profile classes. On a NetBox without channel fields it reports no channel child, so a plan there
7never meets a harness. Planning and execution build each Cable write through one constructor, so a
8write that NetBox refuses, or that puts a leg on another channel, never reaches NetBox.
9"""
11from __future__ import annotations
13from dataclasses import dataclass
14from functools import cache
15from typing import Any
17from .plan import canonical_json
18from .target_runtime import PreconditionFailed
20INTERFACE = "dcim.interface"
21NOT_CONNECTABLE = "cable.termination_not_connectable"
22CHANNELIZED = "cable.termination_channelized"
23# The mirrored Cable fields of a child whose parent holds no profiled Cable.
24_EMPTY_MIRROR = (None, "", None, ())
27class IntentRefused(PreconditionFailed):
28 """A Cable write that NetBox would refuse, or that would put a leg on another channel."""
30 def __init__(self, code: str, message: str, facts: dict | None = None, port: Any = None):
31 super().__init__(message)
32 self.code = code
33 self.facts = facts or {}
34 # The one termination the refusal is about, when it names one.
35 self.port = port
38@cache
39def supported() -> bool:
40 """Return whether the running NetBox models Interface channels."""
41 from dcim.models import Interface
43 return {"channels", "channel_id"} <= {field.name for field in Interface._meta.get_fields()}
46def _label(component) -> str:
47 return component._meta.label_lower
50def _record(component) -> list:
51 """Return the object, its id and its Device, as a Planned Change records one termination."""
52 return [_label(component), component.pk, component.device_id]
55def channel_parent(component) -> int | None:
56 """Return the parent of a channel child Interface, or None for every other termination."""
57 if not supported() or _label(component) != INTERFACE or component.channel_id is None:
58 return None
59 return component.parent_id
62def channel(component) -> int:
63 """Return the channel of one channel child Interface."""
64 return component.channel_id
67def is_channelized(component) -> bool:
68 """Return whether *component* is a channelized Interface, whose channels only a breakout Cable reaches."""
69 return supported() and _label(component) == INTERFACE and bool(component.channels)
72def is_breakout(cable, ends) -> bool:
73 """Return whether an existing Cable is a breakout: its profile fans connectors out, or it ends on a channelized Interface.
75 The import never deletes or replaces such a Cable as a Logical Cable (section 6.4).
76 """
77 profile = cable.profile_class
78 fans_out = profile is not None and len(profile.a_connectors) != len(profile.b_connectors)
79 return fans_out or any(is_channelized(end) for end in ends)
82def is_plain_interface(component) -> bool:
83 """Return whether *component* is an Interface that is neither channelized nor a channel child."""
84 return _label(component) == INTERFACE and component.channel_id is None and not component.channels
87@dataclass(frozen=True)
88class Child:
89 """One Interface whose parent is a channelized Interface, as read under the plan's locks."""
91 component: Any
92 terminated: bool
94 @property
95 def channel(self) -> int | None:
96 """Return the channel the child binds to, or None for a plain subinterface."""
97 return self.component.channel_id
99 @property
100 def mirror(self) -> tuple:
101 """Return the Cable fields NetBox mirrors onto the child, with NetBox's two empty spellings merged."""
102 row = self.component
103 return (row.cable_id, row.cable_end or "", row.cable_connector, tuple(row.cable_positions or ()))
106@dataclass(frozen=True)
107class Parent:
108 """One channelized Interface and every Interface whose parent it is."""
110 component: Any
111 children: tuple[Child, ...]
113 @property
114 def pk(self) -> int:
115 """Return the parent Interface id."""
116 return self.component.pk
118 @property
119 def channels(self) -> int:
120 """Return the channel count of the parent, 0 when it is not channelized."""
121 return self.component.channels or 0
123 def evidence(self) -> dict:
124 """Return the channel facts a unit fingerprints: the count and the full child set."""
125 return {"channels": self.channels, "children": [[child.component.pk, child.channel] for child in self.children]}
127 def faults(self, cable_profile: str) -> list[tuple[Child, str]]:
128 """Return each channel child that contradicts NetBox's channel rules, with the rule it breaks.
130 The mirror rule restates `propagate_channel_cables` and `clear_channel_cables`.
131 """
132 row = self.component
133 profiled = bool(row.cable_id and cable_profile)
134 found = []
135 for child in self.children:
136 if child.channel is None:
137 continue
138 expected = (row.cable_id, row.cable_end, row.cable_connector, (child.channel,))
139 if child.terminated:
140 found.append((child, "termination"))
141 elif child.channel > self.channels:
142 found.append((child, "channel"))
143 elif child.mirror != (expected if profiled else _EMPTY_MIRROR):
144 found.append((child, "mirror"))
145 return found
148def read_parents(parent_ids, *, lock: bool) -> dict[int, Parent]:
149 """Read each parent, then every Interface under it, locking both in that order when asked."""
150 from core.models import ObjectType
151 from dcim.models import CableTermination, Interface
153 if not parent_ids:
154 return {}
155 ids = sorted(parent_ids)
156 parents = Interface.objects.filter(pk__in=ids).order_by("pk")
157 children = Interface.objects.filter(parent_id__in=ids).order_by("pk")
158 if lock:
159 parents = parents.select_for_update(of=("self",))
160 children = children.select_for_update(of=("self",))
161 parent_rows = list(parents)
162 child_rows = list(children)
163 terminated = set(
164 CableTermination.objects.filter(
165 termination_type=ObjectType.objects.get_for_model(Interface),
166 termination_id__in=[row.pk for row in child_rows],
167 ).values_list("termination_id", flat=True)
168 )
169 by_parent: dict[int, list[Child]] = {}
170 for row in child_rows:
171 by_parent.setdefault(row.parent_id, []).append(Child(component=row, terminated=row.pk in terminated))
172 return {row.pk: Parent(component=row, children=tuple(by_parent.get(row.pk, ()))) for row in parent_rows}
175def _routes_channels(profile_class, channels: int) -> bool:
176 """Return whether a profile joins one A connector of *channels* positions, position k to B connector k."""
177 if profile_class is None:
178 return False
179 profile = profile_class()
180 numbers = range(1, channels + 1)
181 return (
182 profile.a_connectors == {1: channels}
183 and profile.b_connectors == dict.fromkeys(numbers, 1)
184 and all(
185 profile.get_mapped_position("A", 1, k) == (k, 1) and profile.get_mapped_position("B", k, 1) == (1, k)
186 for k in numbers
187 )
188 )
191@cache
192def breakout_profile(channels: int) -> str | None:
193 """Return the running instance's one Cable Profile that breaks *channels* lanes out one per B connector."""
194 from dcim.models import Cable
196 from .cable_policy import cable_profile_choices
198 matches = [
199 value
200 for value, _label in cable_profile_choices()
201 if _routes_channels(Cable(profile=value).profile_class, channels)
202 ]
203 if len(matches) > 1:
204 raise RuntimeError(f"Several Cable Profiles break out {channels} channels: {', '.join(matches)}.")
205 return matches[0] if matches else None
208@dataclass(frozen=True)
209class CableIntent:
210 """The ends and the profile of one Cable write, each end list in the order NetBox numbers connectors."""
212 a_terminations: tuple
213 b_terminations: tuple
214 profile: str
215 # A harness intent: the channel child behind each B end, and the parent's channel count.
216 children: tuple = ()
217 capacity: int = 0
219 @property
220 def trailing(self) -> tuple[int, ...]:
221 """Return the channels after the last leg that the harness leaves unused."""
222 return tuple(range(len(self.b_terminations) + 1, self.capacity + 1))
224 def snapshot(self) -> frozenset:
225 """Return the CableTermination rows NetBox writes for this intent, as `update_terminations` numbers them."""
226 from dcim.models import Cable
228 profile = Cable(profile=self.profile).profile_class()
229 rows = set()
230 for side, ends, connectors in (
231 ("A", self.a_terminations, profile.a_connectors),
232 ("B", self.b_terminations, profile.b_connectors),
233 ):
234 for connector, end in enumerate(ends, start=1):
235 rows.add((side, connector, tuple(range(1, connectors[connector] + 1)), _label(end), end.pk))
236 return frozenset(rows)
238 def proved_by(self, cable_profile: str, snapshot: frozenset) -> bool:
239 """Return whether an existing Cable with this profile and termination snapshot is this write."""
240 return cable_profile == self.profile and snapshot == self.snapshot()
243def _refuse_channel_child(end) -> None:
244 if channel_parent(end) is not None:
245 raise IntentRefused(
246 "cable.harness_unsupported",
247 "NetBox cannot cable a channel subinterface directly. The import cables its parent instead.",
248 )
251def _refuse_channelized(end) -> None:
252 """Refuse a channelized Interface as a one-position end, because NetBox mirrors no channel through it."""
253 if is_channelized(end):
254 raise IntentRefused(
255 CHANNELIZED,
256 f"{end} is channelized, so an ordinary Cable on it reaches no channel. The import cables it only as a "
257 "breakout from its channel subinterfaces.",
258 {"port": str(end)},
259 port=end,
260 )
263def _refuse_not_connectable(ends) -> None:
264 """Refuse an Interface whose type NetBox refuses to cable, as `CableTermination.clean` does."""
265 from dcim.constants import NONCONNECTABLE_IFACE_TYPES
267 for end in ends:
268 if _label(end) == INTERFACE and end.type in NONCONNECTABLE_IFACE_TYPES:
269 raise IntentRefused(
270 NOT_CONNECTABLE,
271 f"NetBox cannot cable {end}, because its Interface type is {end.type}.",
272 {"port": str(end), "interface_type": end.type},
273 port=end,
274 )
277def _refuse_marked_connected(ends, code: str) -> None:
278 """Refuse an end NetBox marks as connected, which `CableTermination.clean` refuses to cable."""
279 for end in ends:
280 if end.mark_connected:
281 raise IntentRefused(
282 code,
283 f"NetBox marks {end} as connected without a Cable, so it cannot take one.",
284 {"port": str(end), "marked_connected": True},
285 port=end,
286 )
289def ordinary_intent(first, second, *, profile: str) -> CableIntent:
290 """Return a one-to-one Cable write, refusing a channel child, a channelized parent, or an end NetBox cannot cable."""
291 for end in (first, second):
292 _refuse_channel_child(end)
293 _refuse_channelized(end)
294 _refuse_not_connectable((first, second))
295 _refuse_marked_connected((first, second), "cable.termination_occupied")
296 return CableIntent(a_terminations=(first,), b_terminations=(second,), profile=profile)
299def harness_intent(parent, legs) -> CableIntent:
300 """Return the breakout Cable write for *legs*, the (channel child, far end) pairs of *parent*.
302 The B side is ordered by channel, so B connector k carries channel k. A write that would shift a
303 leg to another connector, or that NetBox would refuse, is refused here with its diagnostic code.
304 """
305 if not legs:
306 raise ValueError("A harness intent needs at least one leg.")
307 for end in (parent, *(far for _child, far in legs)):
308 _refuse_channel_child(end)
309 if any(_label(far) == INTERFACE and far.pk == parent.pk for _child, far in legs):
310 raise IntentRefused(
311 "cable.harness_unsupported", "A leg ends on its own parent, so the trace runs inside one harness."
312 )
313 if any(is_channelized(far) for _child, far in legs):
314 raise IntentRefused("cable.harness_unsupported", "A leg ends on another channelized Interface.")
315 if any(channel_parent(child) != parent.pk for child, _far in legs):
316 raise IntentRefused("cable.harness_channel_state_invalid", "A leg is no longer a channel of this parent.")
317 ordered = sorted(legs, key=lambda leg: leg[0].channel_id)
318 numbers = [child.channel_id for child, _far in ordered]
319 repeated = sorted({number for number in numbers if numbers.count(number) > 1})
320 fars = [(_label(far), far.pk) for _child, far in ordered]
321 shared = sorted({child.channel_id for child, far in ordered if fars.count((_label(far), far.pk)) > 1})
322 if repeated or shared:
323 raise IntentRefused(
324 "cable.harness_leg_conflict",
325 "Two legs of this harness claim one channel or one far end.",
326 {"conflicts": sorted(set(repeated) | set(shared))},
327 )
328 capacity = parent.channels or 0
329 if numbers[-1] > capacity:
330 raise IntentRefused("cable.harness_channel_state_invalid", "A leg names a channel the parent does not have.")
331 models = sorted({model for model, _pk in fars})
332 if len(models) > 1:
333 raise IntentRefused(
334 "cable.harness_mixed_terminations",
335 "The far ends of this harness are of more than one kind, and NetBox cables one kind per side.",
336 {"models": models},
337 )
338 missing = sorted(set(range(1, numbers[-1] + 1)) - set(numbers))
339 if missing:
340 raise IntentRefused(
341 "cable.harness_channel_gap",
342 "A channel between the legs of this harness has no leg, so a later leg would take its connector.",
343 {"missing": missing},
344 )
345 profile = breakout_profile(capacity)
346 if profile is None:
347 raise IntentRefused(
348 "cable.harness_profile_unavailable",
349 "This NetBox instance offers no breakout Cable Profile for the parent's channel count.",
350 {"channels": capacity},
351 )
352 _refuse_not_connectable((parent, *(far for _child, far in ordered)))
353 _refuse_marked_connected((parent, *(far for _child, far in ordered)), "cable.harness_occupied")
354 return CableIntent(
355 a_terminations=(parent,),
356 b_terminations=tuple(far for _child, far in ordered),
357 profile=profile,
358 children=tuple(child for child, _far in ordered),
359 capacity=capacity,
360 )
363def harness_payload(intent: CableIntent) -> dict:
364 """Return what a harness Planned Change records so execution can rebuild the same intent."""
365 return {
366 "profile": intent.profile,
367 "parent": _record(intent.a_terminations[0]),
368 "legs": [
369 [_record(child), _record(far)] for child, far in zip(intent.children, intent.b_terminations, strict=True)
370 ],
371 }
374def locked_harness_intent(recorded, free_termination) -> CableIntent:
375 """Rebuild a recorded harness intent from locked live rows, refusing one whose facts moved.
377 *free_termination* locks one recorded termination and refuses it when it is gone, moved or cabled.
378 """
379 from dcim.models import Interface
381 parent = free_termination(*recorded["parent"])
382 fars = [free_termination(*far) for _child, far in recorded["legs"]]
383 child_ids = [child_id for (_label, child_id, _device_id), _far in recorded["legs"]]
384 children = Interface.objects.filter(pk__in=sorted(child_ids)).order_by("pk").select_for_update(of=("self",))
385 by_pk = {child.pk: child for child in children}
386 if set(child_ids) - by_pk.keys():
387 raise PreconditionFailed("A channel subinterface of this harness is gone, so the harness cannot be created.")
388 for (_label, child_id, device_id), _far in recorded["legs"]:
389 if by_pk[child_id].device_id != device_id:
390 raise PreconditionFailed(f"{by_pk[child_id]} moved to another Device after the plan was made.")
391 intent = harness_intent(parent, [(by_pk[child], far) for child, far in zip(child_ids, fars, strict=True)])
392 if canonical_json(harness_payload(intent)) != canonical_json(recorded):
393 raise PreconditionFailed("The channels of this harness changed after the plan was made.")
394 return intent
397__all__ = (
398 "CableIntent",
399 "Child",
400 "IntentRefused",
401 "Parent",
402 "breakout_profile",
403 "channel",
404 "channel_parent",
405 "harness_intent",
406 "harness_payload",
407 "is_breakout",
408 "is_channelized",
409 "is_plain_interface",
410 "locked_harness_intent",
411 "ordinary_intent",
412 "read_parents",
413 "supported",
414)