Coverage for netbox_interface_name_rules / family / prospective.py: 100.00%
105 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"""Plan interface families for interfaces described by name alone.
5A prospective plan says what a rule intends, using the same naming rules installed execution
6follows. It carries no row snapshot, so it can describe a family NetBox has not created yet and
7the executors refuse to take one.
8"""
10import logging
11from dataclasses import dataclass
13from ..choices import BreakoutModeChoices
14from ..name_template import evaluate_name_template
15from .capabilities import supports_channelization
16from .domain import (
17 FamilyStatus,
18 FamilyTopology,
19 MemberRole,
20 ProspectiveFamilyPlan,
21 ProspectiveFamilyPlanSet,
22 ProspectiveMember,
23)
24from .names import COLLISION_REASON, repeated_name_refusal, repeated_names
25from .structural import UNSUPPORTED_REASON, has_flat_expansion, module_taken_over_names
26from .targets import (
27 UNCLAIMED_BASE_REASON,
28 breaks_out,
29 channelized_family_names,
30 channelized_family_targets,
31 flat_family_names,
32 template_channel_suffixes,
33)
34from .template_names import resolved_template_names
36logger = logging.getLogger(__name__)
38FLAT_EXPANSION_REASON = "the module already carries a flat breakout family"
41@dataclass(frozen=True, slots=True)
42class ProspectiveInterface:
43 """One interface a prospective plan reasons about, described without a row."""
45 name: str
46 parent_name: str | None = None
47 channel_id: int | None = None
48 channels: int | None = None
51def describe_interfaces(interfaces) -> tuple[ProspectiveInterface, ...]:
52 """Describe live interface rows for prospective planning."""
53 interfaces = tuple(interfaces)
54 names_by_pk = {interface.pk: interface.name for interface in interfaces}
55 return tuple(
56 ProspectiveInterface(
57 name=interface.name,
58 parent_name=names_by_pk.get(getattr(interface, "parent_id", None)),
59 channel_id=getattr(interface, "channel_id", None),
60 channels=getattr(interface, "channels", None),
61 )
62 for interface in interfaces
63 )
66def describe_template_interfaces(templates, names=()) -> tuple[ProspectiveInterface, ...]:
67 """Describe the interfaces a module type's templates produce, plus any extra *names*.
69 A channel is paired with its parent only where that parent declares a channel count, because a
70 template family is what the parent's capacity defines. A name no template resolves to is
71 described as a plain interface.
72 """
73 parents = {template.pk: template for template in templates if template.channels is not None}
74 described = {}
75 for template in templates:
76 parent = parents.get(template.parent_id) if template.channel_id is not None else None
77 described[template.resolved] = ProspectiveInterface(
78 name=template.resolved,
79 parent_name=None if parent is None else parent.resolved,
80 channel_id=None if parent is None else template.channel_id,
81 channels=template.channels,
82 )
83 described.update(
84 {name: ProspectiveInterface(name=name) for name in names if name not in described},
85 )
86 return tuple(described.values())
89def describe_module_interfaces(module, names=()) -> tuple[ProspectiveInterface, ...]:
90 """Describe the interfaces *module*'s templates produce, plus any extra *names*.
92 A release that cannot model channelized families describes the names alone: its templates carry
93 no family structure, so reading them would buy nothing.
94 """
95 if not supports_channelization():
96 return tuple(ProspectiveInterface(name=name) for name in names)
97 return describe_template_interfaces( # pragma: no cover - requires channelization support
98 resolved_template_names(module), names
99 )
102def _partition(interfaces):
103 """Split described interfaces into ``(roots, children_by_parent_name)``.
105 A channel is never an independent candidate: it belongs to the interface it names as its
106 parent, and one whose parent is not described here is planned by nothing at all.
107 """
108 described = {interface.name for interface in interfaces}
109 roots = []
110 children: dict[str, list] = {}
111 for interface in interfaces:
112 if interface.channel_id is None:
113 roots.append(interface)
114 elif interface.parent_name in described: # pragma: no cover - requires channelization support
115 children.setdefault(interface.parent_name, []).append(interface)
116 return roots, children
119def _is_family_root(rule, interface, children):
120 """Return whether *interface* owns a family the rule renames as a unit.
122 A breakout rule renames only a family NetBox models with a channel count; on anything else it
123 builds a new family from the interface itself. Any other rule carries whatever channels the
124 interface has along with it.
125 """
126 if rule.channel_count > 0:
127 return interface.channels is not None
128 return bool(children)
131def _rename_plan(rule, variables, parent, children, suffixes, bases): # pragma: no cover - channelization only
132 """Plan the rename of a family the templates or rows already describe."""
133 children = sorted(children, key=lambda child: child.channel_id)
134 targets = channelized_family_targets(
135 rule,
136 variables,
137 parent.name,
138 bases.base_for(parent.name),
139 parent.channels,
140 tuple((child.name, child.channel_id) for child in children),
141 suffixes,
142 )
143 members = [ProspectiveMember(source_name=parent.name, target_name=targets.parent_name, role=MemberRole.PARENT)]
144 members.extend(
145 ProspectiveMember(
146 # A channel whose target cannot be derived keeps the name it has.
147 source_name=child.name,
148 target_name=child.name if target is None else target,
149 role=MemberRole.CHANNEL,
150 channel_id=child.channel_id,
151 reason=reason,
152 )
153 for child, (target, reason) in zip(children, targets.channels, strict=True)
154 )
155 return ProspectiveFamilyPlan(
156 family_id=f"channelized:{parent.name}",
157 topology=FamilyTopology.CHANNELIZED,
158 base_name=None,
159 members=tuple(members),
160 precondition_status=targets.status,
161 precondition_reason=targets.reason,
162 )
165def _refused_creation(base_name, topology, role, status, reason):
166 """Plan a family that will not be built, so the base keeps the name it has."""
167 return ProspectiveFamilyPlan(
168 family_id=f"{topology}:{base_name}",
169 topology=topology,
170 base_name=base_name,
171 members=(ProspectiveMember(source_name=base_name, target_name=base_name, role=role),),
172 precondition_status=status,
173 precondition_reason=reason,
174 )
177def _unclaimed(base_name, topology, role):
178 """Plan a family that will not be built because no template gives *base_name* a raw name."""
179 return _refused_creation(base_name, topology, role, FamilyStatus.BLOCKED, UNCLAIMED_BASE_REASON)
182def _simple_plan(rule, variables, base_name, base):
183 """Plan the rename of one interface that owns no family."""
184 if base is None:
185 return _unclaimed(base_name, FamilyTopology.FLAT, MemberRole.FLAT_MEMBER)
186 try:
187 target_name = evaluate_name_template(rule.name_template, {**variables, "base": base})
188 except (TypeError, ValueError) as error:
189 return _refused_creation(
190 base_name, FamilyTopology.FLAT, MemberRole.FLAT_MEMBER, FamilyStatus.FAILED, str(error)
191 )
192 return ProspectiveFamilyPlan(
193 family_id=f"flat:{base_name}",
194 topology=FamilyTopology.FLAT,
195 base_name=base_name,
196 members=(ProspectiveMember(source_name=base_name, target_name=target_name, role=MemberRole.FLAT_MEMBER),),
197 )
200def _flat_creation_plan(rule, variables, base_name, base):
201 """Plan the flat sibling family a breakout rule creates on one plain interface."""
202 if base is None:
203 return _unclaimed(base_name, FamilyTopology.FLAT, MemberRole.FLAT_MEMBER)
204 try:
205 target_names = flat_family_names(rule, variables, base)
206 except (TypeError, ValueError) as error:
207 return _refused_creation(
208 base_name, FamilyTopology.FLAT, MemberRole.FLAT_MEMBER, FamilyStatus.FAILED, str(error)
209 )
210 members = tuple(
211 ProspectiveMember(
212 source_name=base_name if offset == 0 else None,
213 target_name=target_name,
214 role=MemberRole.FLAT_MEMBER,
215 )
216 for offset, target_name in enumerate(target_names)
217 )
218 return ProspectiveFamilyPlan(
219 family_id=f"flat:{base_name}",
220 topology=FamilyTopology.FLAT,
221 base_name=base_name,
222 members=members,
223 )
226def _modelled_structural_plan(rule, variables, base_name, base): # pragma: no cover - see below
227 """Plan the channelized family for a NetBox release that can hold it."""
228 if base is None:
229 return _unclaimed(base_name, FamilyTopology.CHANNELIZED, MemberRole.PARENT)
230 try:
231 parent_name, channels = channelized_family_names(rule, base_name, base, variables)
232 except (TypeError, ValueError) as error:
233 return _refused_creation(
234 base_name, FamilyTopology.CHANNELIZED, MemberRole.PARENT, FamilyStatus.FAILED, str(error)
235 )
236 members = (
237 ProspectiveMember(source_name=base_name, target_name=parent_name, role=MemberRole.PARENT),
238 *(
239 ProspectiveMember(source_name=None, target_name=name, role=MemberRole.CHANNEL, channel_id=channel_id)
240 for channel_id, name in channels
241 ),
242 )
243 return ProspectiveFamilyPlan(
244 family_id=f"channelized:{base_name}",
245 topology=FamilyTopology.CHANNELIZED,
246 base_name=base_name,
247 members=members,
248 )
251def _structural_plan(rule, variables, base_name, base):
252 """Plan the channelized family a rule builds on one plain interface."""
253 if not supports_channelization():
254 return _refused_creation(
255 base_name, FamilyTopology.CHANNELIZED, MemberRole.PARENT, FamilyStatus.UNSUPPORTED, UNSUPPORTED_REASON
256 )
257 return _modelled_structural_plan(rule, variables, base_name, base) # pragma: no cover - see above
260def _plain_plan(rule, variables, base_name, base):
261 """Plan the family *rule* intends on the plain interface *base_name*, with *base* as ``{base}``."""
262 if rule.channel_count <= 0:
263 return _simple_plan(rule, variables, base_name, base)
264 if rule.breakout_mode == BreakoutModeChoices.CHANNELIZED:
265 return _structural_plan(rule, variables, base_name, base)
266 return _flat_creation_plan(rule, variables, base_name, base)
269@dataclass(frozen=True, slots=True)
270class _CreationContext:
271 """The module-wide facts every creation plan on one module shares."""
273 module: object
274 taken_names: frozenset
275 flat_expansion: bool
278def _creation_context(module, rule, interfaces, plain):
279 """Read the module-wide facts once, and only where a creation plan can use them."""
280 builds_channels = rule.channel_count > 0 and rule.breakout_mode == BreakoutModeChoices.CHANNELIZED
281 flat_expansion = bool(plain) and builds_channels and supports_channelization() and has_flat_expansion(module)
282 return _CreationContext(
283 module=module,
284 taken_names=frozenset(interface.name for interface in interfaces),
285 flat_expansion=flat_expansion,
286 )
289class _TemplateSuffixes:
290 """The module type's per-channel name suffixes, read only if a channel name needs recovering."""
292 def __init__(self, module):
293 self._module = module
294 self._suffixes = None
296 def get(self, channel_id, default=None): # pragma: no cover - requires channelization support
297 """Return the suffixes the templates give *channel_id*, reading them on first use."""
298 if self._suffixes is None:
299 self._suffixes = template_channel_suffixes(resolved_template_names(self._module))
300 return self._suffixes.get(channel_id, default)
303def _plain_base(rule, bases, name):
304 """Return ``{base}`` for *name*; a breakout rule builds only on a name that one template alone claims."""
305 return bases.builds_on(name) if breaks_out(rule) else bases.base_for(name)
308def plan_prospective_families(module, rule, variables, interfaces, bases) -> ProspectiveFamilyPlanSet:
309 """Return one plan for each family *rule* intends on the described *interfaces*.
311 *interfaces* are ``ProspectiveInterface`` values, so a name NetBox has not created yet is
312 planned exactly like one it has and no row is written. Collision checking covers the names
313 described here, because a prospective plan knows only the interfaces it was given. *bases*
314 gives each described name its ``{base}`` value.
315 """
316 interfaces = tuple(interfaces)
317 roots, children = _partition(interfaces)
318 families = [(root, children.get(root.name, ())) for root in roots]
319 context = _creation_context(
320 module, rule, interfaces, [root for root, family in families if not _is_family_root(rule, root, family)]
321 )
322 suffixes = _TemplateSuffixes(module)
323 plans = tuple(
324 _rename_plan(rule, variables, root, family, suffixes, bases)
325 if _is_family_root(rule, root, family)
326 else _plain_plan(rule, variables, root.name, _plain_base(rule, bases, root.name))
327 for root, family in families
328 )
329 return ProspectiveFamilyPlanSet(module_id=module.pk, plans=_settled_creations(plans, context, bases))
332def _is_structural_creation(plan) -> bool:
333 """Return whether *plan* builds a channelized family whose names were evaluated."""
334 return (
335 plan.topology == FamilyTopology.CHANNELIZED and plan.base_name is not None and plan.precondition_status is None
336 )
339def _settled_creations(plans, context, bases):
340 """Refuse the channelized families the module cannot hold, as installed execution refuses them.
342 The families are settled against each other as ``settle_module_families`` settles them: a name
343 they want more than once as a channel refuses each family that wants it, and the rows that they take over
344 (ADR 0017) are not flat siblings. A described interface with a channel name is one a family can
345 take over; only the rows the preview reads can say whether it does, so only the parent name is
346 checked against the described names here.
347 """
348 creations = [plan for plan in plans if _is_structural_creation(plan)]
349 if not creations:
350 return plans
351 repeated = repeated_names(plan.target_names[1:] for plan in creations)
352 families = ((plan.base_name, plan.target_names[1:], plan.precondition_status) for plan in creations)
353 names = module_taken_over_names(families, repeated, bases)
354 expanded = context.flat_expansion and has_flat_expansion(context.module, names)
355 return tuple(
356 _settled_creation(plan, context, repeated, expanded) if _is_structural_creation(plan) else plan
357 for plan in plans
358 )
361def _settled_creation(plan, context, repeated, expanded): # pragma: no cover - requires channelization support
362 """Return *plan*, or the refusal of the family it would build when the module cannot hold it."""
363 parent_name = plan.target_names[0]
364 refusal = repeated_name_refusal(plan.target_names[1:], repeated)
365 if refusal is not None:
366 reason = refusal[1]
367 elif expanded:
368 # Converting one sibling into a parent would strand the others beside the new family.
369 reason = FLAT_EXPANSION_REASON
370 elif parent_name != plan.base_name and parent_name in context.taken_names:
371 reason = f"{COLLISION_REASON}: {parent_name}"
372 else:
373 return plan
374 return _refused_creation(
375 plan.base_name, FamilyTopology.CHANNELIZED, MemberRole.PARENT, FamilyStatus.BLOCKED, reason
376 )