Coverage for netbox_interface_name_rules / family / installed.py: 98.57%
110 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"""Discover and plan installed interface families."""
5from dcim.models import Interface
7from ..name_template import evaluate_name_template
8from .domain import (
9 FamilyStatus,
10 FamilyTopology,
11 InstalledFamilyPlan,
12 InstalledFamilyPlanSet,
13 InterfaceSnapshot,
14 MemberRole,
15 PlannedMember,
16)
17from .raw_bases import GivenRawNames, RawBases
18from .targets import (
19 UNCLAIMED_BASE_REASON,
20 builds_flat_family,
21 channelized_family_targets,
22 flat_family_names,
23 lockstep_family_targets,
24 template_channel_suffixes,
25)
26from .template_names import TemplateNames
29def is_plain_interface(interface) -> bool:
30 """Return whether *interface* can be a flat-family member."""
31 return (
32 getattr(interface, "parent_id", None) is None
33 and getattr(interface, "channel_id", None) is None
34 and getattr(interface, "channels", None) is None
35 )
38def is_channelized_parent(interface) -> bool:
39 """Return whether *interface* declares a channelized family."""
40 return getattr(interface, "channels", None) is not None
43def _is_channel(interface) -> bool:
44 """Return whether *interface* is bound to a parent channel."""
45 return getattr(interface, "channel_id", None) is not None
48def _flat_plan(module, target_names, interfaces, status=None, reason=""):
49 """Build one immutable plan for the flat family whose rows are *interfaces*."""
50 members = tuple(
51 PlannedMember(
52 snapshot=InterfaceSnapshot.from_interface(interface),
53 target_name=target_name,
54 role=MemberRole.FLAT_MEMBER,
55 )
56 for interface, target_name in zip(interfaces, target_names, strict=True)
57 )
58 return InstalledFamilyPlan(
59 family_id=f"flat:{members[0].snapshot.pk}",
60 topology=FamilyTopology.FLAT,
61 device_id=module.device_id,
62 module_id=module.pk,
63 members=members,
64 precondition_status=status,
65 precondition_reason=reason,
66 )
69def _claimed_flat_plan(module, rule, variables, family, interfaces):
70 """Return the plan for a claimed flat family, or None when its first row builds it again (ADR 0013)."""
71 kept = tuple(interface.name for interface in interfaces)
72 try:
73 target_names = flat_family_names(rule, variables, family.base_name)
74 except (TypeError, ValueError) as error:
75 return _flat_plan(
76 module, kept, interfaces, FamilyStatus.FAILED, f"failed to evaluate the family names: {error}"
77 )
78 if len(interfaces) == len(family.names):
79 return _flat_plan(module, target_names, interfaces)
80 if target_names == family.names:
81 return None
82 missing = len(family.names) - len(interfaces)
83 reason = f"the flat family is missing {missing} of its {len(family.names)} interfaces"
84 return _flat_plan(module, kept, interfaces, FamilyStatus.BLOCKED, reason)
87def _channelized_plan(device_id, module_id, parent, children, targets):
88 """Build one plan for an existing channelized family from the names *targets* intends."""
89 members = [
90 PlannedMember(
91 snapshot=InterfaceSnapshot.from_interface(parent),
92 target_name=targets.parent_name,
93 role=MemberRole.PARENT,
94 )
95 ]
96 members.extend(
97 PlannedMember(
98 snapshot=InterfaceSnapshot.from_interface(child),
99 target_name=target,
100 role=MemberRole.CHANNEL,
101 reason=reason,
102 )
103 for child, (target, reason) in zip(children, targets.channels, strict=True)
104 )
105 return InstalledFamilyPlan(
106 family_id=f"channelized:{parent.pk}",
107 topology=FamilyTopology.CHANNELIZED,
108 device_id=device_id,
109 module_id=module_id,
110 members=tuple(members),
111 parent_pk=parent.pk,
112 precondition_status=targets.status,
113 precondition_reason=targets.reason,
114 )
117def _module_family_targets(rule, variables, parent, base_name, children, suffixes):
118 """Return the names *rule* intends for a channelized family a module carries."""
119 return channelized_family_targets(
120 rule,
121 variables,
122 parent.name,
123 base_name,
124 parent.channels,
125 tuple((child.name, child.channel_id) for child in children),
126 suffixes,
127 )
130def _channelized_plans(module, rule, variables, interfaces, bases):
131 """Return one plan for every structurally discovered channelized family."""
132 parents = [interface for interface in interfaces if is_channelized_parent(interface)]
133 if not parents:
134 return []
135 children_by_parent: dict[int, list] = {}
136 for interface in interfaces:
137 if _is_channel(interface) and interface.parent_id is not None:
138 children_by_parent.setdefault(interface.parent_id, []).append(interface)
139 suffixes = template_channel_suffixes(bases.catalog.get())
140 plans = []
141 for parent in parents:
142 children = children_by_parent.get(parent.pk, [])
143 children.sort(key=lambda child: (child.channel_id, child.pk))
144 targets = _module_family_targets(rule, variables, parent, bases.base_for(parent.name), children, suffixes)
145 plans.append(_channelized_plan(module.device_id, module.pk, parent, children, targets))
146 return plans
149def interfaces_by_module(modules):
150 """Load every interface of a module batch in one query, in stable per-module order."""
151 by_module: dict[int, list] = {module.pk: [] for module in modules}
152 if not by_module: 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true
153 return by_module
154 rows = Interface.objects.filter(module_id__in=list(by_module)).order_by("module_id", "name")
155 for interface in rows:
156 by_module[interface.module_id].append(interface)
157 return by_module
160def device_interface_families(interfaces):
161 """Return each base interface outside a channel with its channel children, in channel order."""
162 children_by_parent: dict[int, list] = {}
163 for interface in interfaces:
164 if _is_channel(interface) and interface.parent_id is not None:
165 children_by_parent.setdefault(interface.parent_id, []).append(interface)
166 for children in children_by_parent.values():
167 children.sort(key=lambda child: (child.channel_id, child.pk))
168 return tuple(
169 (interface, tuple(children_by_parent.get(interface.pk, ())))
170 for interface in interfaces
171 if not _is_channel(interface)
172 )
175def plan_kept_interface(module, interface, reason) -> InstalledFamilyPlan:
176 """Return a plan that keeps the name of one interface of *module* and reports *reason*."""
177 return InstalledFamilyPlan(
178 family_id=f"flat:{interface.pk}",
179 topology=FamilyTopology.FLAT,
180 device_id=module.device_id,
181 module_id=module.pk,
182 members=(
183 PlannedMember(
184 snapshot=InterfaceSnapshot.from_interface(interface),
185 target_name=interface.name,
186 role=MemberRole.FLAT_MEMBER,
187 ),
188 ),
189 precondition_status=FamilyStatus.BLOCKED,
190 precondition_reason=reason,
191 )
194def _interface_rename_plan(device_id, module_id, rule, variables, interface, base_name) -> InstalledFamilyPlan:
195 """Return a plan that renames one interface which belongs to no family, from *base_name* as ``{base}``."""
196 status, reason, target_name = None, "", interface.name
197 if base_name is None:
198 status, reason = FamilyStatus.BLOCKED, UNCLAIMED_BASE_REASON
199 else:
200 try:
201 target_name = evaluate_name_template(rule.name_template, {**variables, "base": base_name})
202 except (TypeError, ValueError) as error:
203 status, reason = FamilyStatus.FAILED, f"failed to evaluate the interface name: {error}"
204 return InstalledFamilyPlan(
205 family_id=f"flat:{interface.pk}",
206 topology=FamilyTopology.FLAT,
207 device_id=device_id,
208 module_id=module_id,
209 members=(
210 PlannedMember(
211 snapshot=InterfaceSnapshot.from_interface(interface),
212 target_name=target_name,
213 role=MemberRole.FLAT_MEMBER,
214 ),
215 ),
216 precondition_status=status,
217 precondition_reason=reason,
218 )
221def plan_interface_rename(module, rule, variables, interface, bases) -> InstalledFamilyPlan:
222 """Return the plan that renames one module interface which belongs to no family."""
223 return _interface_rename_plan(
224 module.device_id,
225 module.pk,
226 rule,
227 variables,
228 interface,
229 bases.base_for(interface.name),
230 )
233def plan_device_interface_rename(device, rule, variables, interface, children=()) -> InstalledFamilyPlan:
234 """Return the plan that renames one device-level interface family."""
235 if not children and not is_channelized_parent(interface):
236 return _interface_rename_plan(device.pk, None, rule, variables, interface, interface.name)
237 # A device rule never builds a family, so its channel count says nothing about this one: the
238 # members keep the suffixes they carry under whatever name the parent takes. A device-level
239 # interface has no module template family, so there is no suffix to recover from one either.
240 targets = lockstep_family_targets(
241 rule, variables, interface.name, interface.name, tuple((child.name, child.channel_id) for child in children), {}
242 )
243 return _channelized_plan(device.pk, None, interface, children, targets)
246def module_raw_bases(
247 module, rule, variables, interfaces, previous_forms=None, catalog=None, earlier_raw_names=None
248) -> RawBases:
249 """Return the claim over the names of *module*'s interfaces outside a channel.
251 *previous_forms* holds what named the templates before a move, and *earlier_raw_names* the raw
252 names the templates gave earlier; see ``RawBases``. *catalog* has ``get()`` for the resolved
253 templates, and reads the module's templates when it is not given.
254 """
255 families = {
256 interface.name: tuple((child.name, child.channel_id) for child in children)
257 for interface, children in device_interface_families(interfaces)
258 }
259 plain = [interface.name for interface in interfaces if is_plain_interface(interface)]
260 catalog = TemplateNames(module) if catalog is None else catalog
261 return RawBases(module, rule, variables, families, plain, catalog, previous_forms, earlier_raw_names)
264def given_raw_names(module, rule, variables, names) -> GivenRawNames:
265 """Return bases for *names* a caller gives as the module's raw template names, without channels."""
266 return GivenRawNames(RawBases(module, rule, variables, dict.fromkeys(names, ()), names, TemplateNames(module)))
269def plan_installed_flat_families(module, rule, variables, interfaces, bases) -> list[InstalledFamilyPlan]:
270 """Return a plan for each flat family that one template alone claims, when *rule* is a flat breakout rule.
272 A move recognises no flat family (ADR 0015).
273 """
274 if not builds_flat_family(rule):
275 return []
276 plain = {interface.name: interface for interface in interfaces if is_plain_interface(interface)}
277 plans = (
278 _claimed_flat_plan(
279 module, rule, variables, family, tuple(plain[name] for name in family.names if name in plain)
280 )
281 for family in bases.flat_families()
282 )
283 return [plan for plan in plans if plan is not None]
286def half_built_members(rule, interfaces, bases) -> dict:
287 """Return, by the name of its first row, the other rows of each half-built flat family the claim accepted."""
288 if not builds_flat_family(rule):
289 return {}
290 rows = {interface.name: interface for interface in interfaces if is_plain_interface(interface)}
291 return {
292 family.names[0]: tuple(rows[name] for name in family.names[1:] if name in rows)
293 for family in bases.flat_families()
294 }
297def plan_installed_families(module, rule, variables) -> InstalledFamilyPlanSet:
298 """Return immutable plans for the installed families owned by *module*, reading its interfaces."""
299 interfaces = list(Interface.objects.filter(module_id=module.pk).order_by("pk"))
300 return plan_installed_families_from(
301 module, rule, variables, interfaces, module_raw_bases(module, rule, variables, interfaces)
302 )
305def plan_installed_families_from(module, rule, variables, interfaces, bases) -> InstalledFamilyPlanSet:
306 """Return the installed family plans for *module*, reading ``{base}`` and templates from *bases*.
308 A batch that already holds the module's interface rows passes them in, so planning a fleet
309 reads them once rather than once per module.
310 """
311 plans = _channelized_plans(module, rule, variables, interfaces, bases)
312 plans.extend(plan_installed_flat_families(module, rule, variables, interfaces, bases))
313 plans.sort(key=lambda plan: plan.member_pks[0])
314 return InstalledFamilyPlanSet(module_id=module.pk, plans=tuple(plans))