Coverage for netbox_interface_name_rules / family / domain.py: 99.17%
112 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-10-16 10:13 +0000
1# SPDX-License-Identifier: Apache-2.0
2# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com>
3"""Immutable domain values for interface-family operations."""
5from dataclasses import dataclass
6from enum import StrEnum
9class FamilyTopology(StrEnum):
10 """Installed interface-family topology."""
12 FLAT = "flat"
13 CHANNELIZED = "channelized"
16class FamilyStatus(StrEnum):
17 """Result of a family or member operation."""
19 CHANGED = "changed"
20 UNCHANGED = "unchanged"
21 BLOCKED = "blocked"
22 STALE = "stale"
23 UNSUPPORTED = "unsupported"
24 FAILED = "failed"
27class RunScope(StrEnum):
28 """The interfaces an automatic run may touch: an install those with a raw name, a forced reapply all."""
30 INSTALL = "install"
31 FORCED = "forced"
34class MemberRole(StrEnum):
35 """A member's role in its installed family."""
37 PARENT = "parent"
38 CHANNEL = "channel"
39 FLAT_MEMBER = "flat_member"
42@dataclass(frozen=True, slots=True)
43class InterfaceSnapshot:
44 """Live interface facts that execution must revalidate."""
46 pk: int
47 device_id: int
48 module_id: int | None
49 name: str
50 parent_id: int | None
51 channel_id: int | None
52 channels: int | None
54 @classmethod
55 def from_interface(cls, interface):
56 """Capture the protected facts from a NetBox interface row."""
57 return cls(
58 pk=interface.pk,
59 device_id=interface.device_id,
60 module_id=interface.module_id,
61 name=interface.name,
62 parent_id=getattr(interface, "parent_id", None),
63 channel_id=getattr(interface, "channel_id", None),
64 channels=getattr(interface, "channels", None),
65 )
68@dataclass(frozen=True, slots=True)
69class PlannedMember:
70 """One installed member and its intended name."""
72 snapshot: InterfaceSnapshot
73 target_name: str | None
74 role: MemberRole
75 reason: str = ""
78@dataclass(frozen=True, slots=True)
79class InstalledFamilyPlan:
80 """An executable rename plan for one installed interface family."""
82 family_id: str
83 topology: FamilyTopology
84 device_id: int
85 module_id: int | None
86 members: tuple[PlannedMember, ...]
87 parent_pk: int | None = None
88 precondition_status: FamilyStatus | None = None
89 precondition_reason: str = ""
91 @property
92 def member_pks(self) -> tuple[int, ...]:
93 """Return member primary keys in plan order."""
94 return tuple(member.snapshot.pk for member in self.members)
97@dataclass(frozen=True, slots=True)
98class InstalledFamilyPlanSet:
99 """Exactly one immutable plan for each discovered installed family."""
101 module_id: int
102 plans: tuple[InstalledFamilyPlan, ...]
104 @property
105 def member_pks(self) -> frozenset[int]:
106 """Return every interface claimed by the plan set."""
107 return frozenset(pk for plan in self.plans for pk in plan.member_pks)
110@dataclass(frozen=True, slots=True)
111class PlannedChannel:
112 """One channel row a structural plan creates under its parent, or takes over from *holder* (ADR 0017)."""
114 channel_id: int
115 name: str
116 holder: InterfaceSnapshot | None = None
119@dataclass(frozen=True, slots=True)
120class StructuralFamilyPlan:
121 """An executable plan that turns one plain interface into a channelized family.
123 *taken_over_names* are the names of the rows that the families planned on the module take over,
124 which the module's flat surplus does not count (ADR 0017). *family_bases* are the primary keys of
125 the rows another family is built on, which execution refuses to take over under lock as planning did.
126 """
128 family_id: str
129 device_id: int
130 module_id: int
131 module_type_id: int
132 base: InterfaceSnapshot
133 parent_target_name: str
134 channel_count: int
135 channels: tuple[PlannedChannel, ...]
136 family_bases: frozenset[int]
137 precondition_status: FamilyStatus | None = None
138 precondition_reason: str = ""
139 taken_over_names: tuple[str, ...] = ()
141 @property
142 def topology(self) -> FamilyTopology:
143 """Return the topology this plan installs."""
144 return FamilyTopology.CHANNELIZED
146 @property
147 def channel_names(self) -> tuple[str, ...]:
148 """Return every channel name in creation order."""
149 return tuple(channel.name for channel in self.channels)
151 @property
152 def target_names(self) -> tuple[str, ...]:
153 """Return the parent name and every channel name in creation order."""
154 return (self.parent_target_name, *self.channel_names)
156 @property
157 def holders(self) -> tuple[InterfaceSnapshot, ...]:
158 """Return the snapshot of every existing row that has one of the channel names, once each, in creation order."""
159 return tuple(dict.fromkeys(channel.holder for channel in self.channels if channel.holder is not None))
162@dataclass(frozen=True, slots=True)
163class FlatCreationPlan:
164 """An executable plan that expands one plain interface into a flat breakout family.
166 *members* are the rows of a half-built family that the claim gave the base's template, which
167 the family keeps under the names they carry.
168 """
170 family_id: str
171 device_id: int
172 module_id: int
173 base: InterfaceSnapshot
174 target_names: tuple[str, ...]
175 precondition_status: FamilyStatus | None = None
176 precondition_reason: str = ""
177 members: tuple[InterfaceSnapshot, ...] = ()
179 @property
180 def topology(self) -> FamilyTopology:
181 """Return the topology this plan installs."""
182 return FamilyTopology.FLAT
184 @property
185 def member_pks(self) -> tuple[int, ...]:
186 """Return the base and every planned member, by primary key."""
187 return (self.base.pk, *(member.pk for member in self.members))
190@dataclass(frozen=True, slots=True)
191class ProspectiveMember:
192 """One member of a family planned from names alone."""
194 source_name: str | None
195 target_name: str
196 role: MemberRole
197 channel_id: int | None = None
198 reason: str = ""
201@dataclass(frozen=True, slots=True)
202class ProspectiveFamilyPlan:
203 """What a rule intends for one family of named interfaces.
205 A plan whose *base_name* is set describes a family the rule builds out of that one name, so the
206 name expands to every target. A plan without one renames members that already exist.
207 """
209 family_id: str
210 topology: FamilyTopology
211 base_name: str | None
212 members: tuple[ProspectiveMember, ...]
213 precondition_status: FamilyStatus | None = None
214 precondition_reason: str = ""
216 @property
217 def source_names(self) -> tuple[str, ...]:
218 """Return the name of every member that already exists, in plan order."""
219 return tuple(member.source_name for member in self.members if member.source_name is not None)
221 @property
222 def target_names(self) -> tuple[str, ...]:
223 """Return every intended name, in plan order."""
224 return tuple(member.target_name for member in self.members)
227@dataclass(frozen=True, slots=True)
228class ProspectiveFamilyPlanSet:
229 """Exactly one prospective plan for each family the rule intends on a module."""
231 module_id: int
232 plans: tuple[ProspectiveFamilyPlan, ...]
234 def predicted_names(self, source_name: str) -> tuple[str, ...]:
235 """Return the names *source_name* becomes, or the name itself when no plan claims it."""
236 for plan in self.plans:
237 if plan.base_name == source_name:
238 return plan.target_names
239 for member in plan.members:
240 if member.source_name == source_name:
241 return (member.target_name,)
242 return (source_name,)
245@dataclass(frozen=True, slots=True)
246class MemberOutcome:
247 """Result facts for one planned family member."""
249 interface_pk: int
250 current_name: str
251 target_name: str | None
252 status: FamilyStatus
253 reason: str = ""
256@dataclass(frozen=True, slots=True)
257class FamilyOutcome:
258 """Execution result for one installed family."""
260 family_id: str
261 topology: FamilyTopology
262 status: FamilyStatus
263 members: tuple[MemberOutcome, ...]
264 reason: str = ""
266 @property
267 def changed_count(self) -> int:
268 """Return the number of members renamed by this operation."""
269 return sum(member.status == FamilyStatus.CHANGED for member in self.members)
272@dataclass(frozen=True, slots=True)
273class InstalledPlanSetOutcome:
274 """Execution results for an installed family plan set."""
276 families: tuple[FamilyOutcome, ...]
278 @property
279 def changed_count(self) -> int:
280 """Return the number of members renamed across all families."""
281 return sum(family.changed_count for family in self.families)
284@dataclass(frozen=True, slots=True)
285class ConversionMember: # pragma: no cover - requires channelization support
286 """One installed flat sibling and the channel identifier it takes."""
288 snapshot: InterfaceSnapshot
289 channel_id: int
292@dataclass(frozen=True, slots=True)
293class ConversionPlan: # pragma: no cover - requires channelization support
294 """An executable plan that turns one installed flat family into a channelized family.
296 The ch-0 row keeps its primary key and becomes the parent, giving up its name to a new
297 channel-1 row; each sibling is retyped in place as a channel under the name it already has.
299 *channel_names* is every name the rule spells for the family, whether or not the module still
300 carries a row for it, so a family with a gap can still say what it would have become.
301 """
303 family_id: str
304 device_id: int
305 module_id: int
306 base: InterfaceSnapshot
307 parent_target_name: str
308 channel_names: tuple[str, ...]
309 siblings: tuple[ConversionMember, ...]
310 precondition_status: FamilyStatus | None = None
311 precondition_reason: str = ""
313 @property
314 def topology(self) -> FamilyTopology:
315 """Return the topology this plan installs."""
316 return FamilyTopology.CHANNELIZED
318 @property
319 def channel_count(self) -> int:
320 """Return how many channels the converted parent declares."""
321 return len(self.channel_names)
323 @property
324 def member_pks(self) -> tuple[int, ...]:
325 """Return the primary key of every row the plan rewrites, in family order."""
326 return (self.base.pk, *(sibling.snapshot.pk for sibling in self.siblings))
328 @property
329 def current_names(self) -> tuple[str, ...]:
330 """Return the names the family carries now, in family order."""
331 return (self.base.name, *(sibling.snapshot.name for sibling in self.siblings))
333 @property
334 def sibling_target_names(self) -> tuple[str, ...]:
335 """Return the name each installed sibling keeps as a channel, in family order."""
336 return tuple(sibling.snapshot.name for sibling in self.siblings)
338 @property
339 def missing_names(self) -> tuple[str, ...]:
340 """Return every name the family needs that this module carries no row for."""
341 present = {self.base.name, *(sibling.snapshot.name for sibling in self.siblings)}
342 return tuple(name for name in self.channel_names if name not in present)
344 @property
345 def target_names(self) -> tuple[str, ...]:
346 """Return the parent name and every channel name, in family order."""
347 return (self.parent_target_name, *self.channel_names)
350@dataclass(frozen=True, slots=True)
351class PlannedName: # pragma: no cover - requires channelization support
352 """One name a family plan intends, and the role it fills in that family."""
354 name: str
355 role: str
356 channel_id: int | None = None
359@dataclass(frozen=True, slots=True)
360class ConversionCandidate: # pragma: no cover - requires channelization support
361 """One flat family an operator may convert, and what converting it would do.
363 *module* and *interface* are the live rows the Apply page links to and submits; every name,
364 role and reason on the candidate comes from the immutable plan instead.
365 """
367 plan: ConversionPlan
368 module: object
369 interface: object
370 reason: str = ""
372 @property
373 def convertible(self) -> bool:
374 """Return whether a dry run of this family succeeded."""
375 return not self.reason
377 @property
378 def status_label(self) -> str:
379 """Return the operator-facing result of the conversion preview."""
380 if self.convertible:
381 return "Convertible"
382 if self.plan.precondition_status == FamilyStatus.UNSUPPORTED:
383 return "Unsupported"
384 return "Blocked"
386 @property
387 def current_name(self) -> str:
388 """Return the name of the ch-0 row the confirm form submits."""
389 return self.plan.base.name
391 @property
392 def current_names(self) -> tuple[str, ...]:
393 """Return the names the family carries now."""
394 return self.plan.current_names
396 @property
397 def new_names(self) -> tuple[str, ...]:
398 """Return the names the family would carry once converted."""
399 return self.plan.target_names
401 @property
402 def name_details(self) -> tuple[PlannedName, ...]:
403 """Describe every intended name so the page can render a family as a family."""
404 return (
405 PlannedName(self.plan.parent_target_name, "parent"),
406 *(
407 PlannedName(name, "channel", channel_id)
408 for channel_id, name in enumerate(self.plan.channel_names, start=1)
409 ),
410 )
412 @property
413 def metadata_note(self) -> str:
414 """Return the sentence the Apply page shows about where the ch-0 configuration ends up."""
415 return (
416 f"The interface VRF, addresses, VLANs, MTU, description and tags on {self.plan.base.name} move to the new "
417 f"channel 1 interface that takes over that name; custom field values are copied. The physical "
418 f"row keeps its ID and becomes the parent {self.plan.parent_target_name}, so automation keyed "
419 f"on that interface ID will address the parent afterwards."
420 )
423@dataclass(frozen=True, slots=True)
424class ConversionPreview:
425 """Every flat family a conversion scan examined, and whether it left any unexamined."""
427 candidates: tuple[ConversionCandidate, ...]
428 has_more: bool = False