Coverage for netbox_interface_name_rules / family / batch.py: 100.00%
152 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"""Apply one rule to a batch of installed modules through family plans.
5Every module in the batch is planned family by family and executed family by family, so a module
6that cannot take its names costs the batch only that family. The module rows, their interfaces and
7each module type's templates are read once for the whole batch.
8"""
10import logging
11from dataclasses import dataclass
13from ..naming import build_variables
14from .domain import (
15 FamilyOutcome,
16 FamilyStatus,
17 FamilyTopology,
18 FlatCreationPlan,
19 InstalledFamilyPlan,
20 MemberOutcome,
21 MemberRole,
22 RunScope,
23 StructuralFamilyPlan,
24)
25from .execution import execute_installed_plan
26from .installed import (
27 half_built_members,
28 interfaces_by_module,
29 module_raw_bases,
30 plan_installed_families_from,
31 plan_interface_rename,
32 plan_kept_interface,
33)
34from .names import COLLISION_REASON, channelized_name_refusal, first_taken_name, name_holders
35from .prospective import FLAT_EXPANSION_REASON
36from .structural import (
37 execute_flat_family,
38 execute_module_structural_families,
39 execute_structural_family,
40 has_flat_expansion,
41 module_taken_over_names,
42 plan_flat_family,
43 plan_structural_family,
44 settle_module_families,
45)
46from .targets import (
47 UNCLAIMED_BASE_REASON,
48 breaks_out,
49 builds_channelized_family,
50)
51from .template_names import pinned_template_cache
53logger = logging.getLogger(__name__)
55NOT_RENAMED_REASON = "the module is not renamed while one of its interfaces is unclaimed"
56CHANNELIZED_MODULE_REASON = "the module already models channelized families, so no family is added beside them"
58# A member left with the name it had for a reason the operator can act on. An unsupported topology
59# is not one of them: the release cannot hold the family, so nothing was dropped by this batch.
60_SKIPPED_STATUSES = (FamilyStatus.BLOCKED, FamilyStatus.STALE, FamilyStatus.FAILED)
62_EXECUTORS = {
63 InstalledFamilyPlan: execute_installed_plan,
64 StructuralFamilyPlan: execute_structural_family,
65 FlatCreationPlan: execute_flat_family,
66}
69@dataclass(frozen=True, slots=True)
70class ModuleFamilyPlans:
71 """The families a rule intends on one module, split by how each was found.
73 *installed* are the families the module already carries; *leftover* are the plans for the
74 interfaces no installed family claimed, whether the rule renames one or builds a family on it.
75 """
77 installed: tuple
78 leftover: tuple
80 @property
81 def plans(self) -> tuple:
82 """Return every plan, installed families first."""
83 return (*self.installed, *self.leftover)
86@dataclass(frozen=True, slots=True)
87class BatchOutcome:
88 """Every family one batch operation planned, and what happened to it."""
90 families: tuple[FamilyOutcome, ...]
92 @property
93 def changed_count(self) -> int:
94 """Return the number of interfaces the batch renamed or created."""
95 return sum(family.changed_count for family in self.families)
97 @property
98 def skipped_members(self) -> tuple[MemberOutcome, ...]:
99 """Return every member a collision, a stale plan or a failure left as it was."""
100 return tuple(
101 member for family in self.families for member in family.members if member.status in _SKIPPED_STATUSES
102 )
104 @property
105 def changed_families(self) -> tuple[FamilyOutcome, ...]:
106 """Return every family this batch actually rewrote."""
107 return tuple(family for family in self.families if family.status == FamilyStatus.CHANGED)
109 @property
110 def blocked_families(self) -> tuple[FamilyOutcome, ...]:
111 """Return every family a collision, a stale plan or a failure left as it was."""
112 return tuple(family for family in self.families if family.status in _SKIPPED_STATUSES)
115def execute_family_plan(plan) -> FamilyOutcome:
116 """Execute one planned family through the executor its plan kind owns.
118 A plan carrying no live rows (a prospective plan above all) has no executor, so a preview
119 object is refused here rather than locking anything.
120 """
121 executor = _EXECUTORS.get(type(plan))
122 if executor is None:
123 raise TypeError(f"{type(plan).__name__} is not an executable family plan")
124 return executor(plan)
127def _is_channel(interface) -> bool:
128 """Return whether *interface* is bound to a parent channel."""
129 return getattr(interface, "channel_id", None) is not None
132def _is_top_level(interface) -> bool:
133 """Return whether *interface* is neither a channel nor a subinterface of another interface."""
134 return not _is_channel(interface) and getattr(interface, "parent_id", None) is None
137def _creation_plan(module, rule, variables, base, base_name, family_bases, members):
138 """Return the plan that builds the family *rule* describes on one plain interface."""
139 if builds_channelized_family(rule):
140 return plan_structural_family(module, rule, variables, base, base_name, family_bases)
141 return plan_flat_family(module, rule, variables, base, base_name, members)
144def _family_bases(interfaces, bases) -> frozenset:
145 """Return the primary key of each interface the claim gives a template, which no other family takes over."""
146 return frozenset(interface.pk for interface in interfaces if bases.claims_alone(interface.name))
149def _creation_plans(module, rule, variables, interfaces, plain, bases, selected_pks, members):
150 """Return a plan for each selected interface; a row that a family keeps or takes over is in that family's plan.
152 The module's families are settled against each other before the selection, so a repeated name does not
153 depend on it. Only a family that the selection reaches takes over its rows.
154 """
155 family_bases = _family_bases(interfaces, bases)
156 plans = [
157 _creation_plan(
158 module, rule, variables, base, bases.builds_on(base.name), family_bases, members.get(base.name, ())
159 )
160 for base in plain
161 ]
162 if builds_channelized_family(rule):
163 applied = frozenset(plan.base.pk for plan in plans if _reaches(plan, selected_pks))
164 plans = settle_module_families(module, plans, interfaces, bases.catalog.get(), bases, applied)
165 plans = [plan for plan in plans if _reaches(plan, selected_pks)]
166 kept = {member.pk for plan in plans if isinstance(plan, FlatCreationPlan) for member in plan.members}
167 kept.update(holder.pk for plan in plans if isinstance(plan, StructuralFamilyPlan) for holder in plan.holders)
168 return [plan for plan in plans if plan.base.pk not in kept]
171def _is_candidate(interface, rule, bases, previous_forms):
172 """Return whether a leftover *interface* is a candidate of its own; a subinterface is not always one."""
173 if _is_top_level(interface):
174 return True
175 # After a move, or under a breakout rule unless a template claims it, a subinterface belongs to its parent.
176 return previous_forms is None and not (breaks_out(rule) and not bases.claim(interface.name).claimed)
179def _in_scope_installed(plans, bases, scope):
180 """Return the installed families *scope* reaches: a channelized one always, a flat one on install by a raw name."""
181 if scope == RunScope.FORCED:
182 return plans
183 return tuple(
184 plan
185 for plan in plans
186 if plan.topology == FamilyTopology.CHANNELIZED
187 or any(bases.claim(member.snapshot.name).raw for member in plan.members)
188 )
191def _is_unclaimed_top_level(interface, bases) -> bool:
192 """Return whether *interface* is a top-level interface that no single template claims."""
193 return _is_top_level(interface) and bases.base_for(interface.name) is None
196def _kept_module_plans(module, rule, variables, interfaces, bases) -> ModuleFamilyPlans:
197 """Report each unclaimed top-level interface, and keep every other interface of *module* (ADR 0015)."""
198 return ModuleFamilyPlans(
199 installed=(),
200 leftover=tuple(
201 plan_interface_rename(module, rule, variables, interface, bases)
202 if _is_unclaimed_top_level(interface, bases)
203 else plan_kept_interface(module, interface, NOT_RENAMED_REASON)
204 for interface in interfaces
205 ),
206 )
209def plan_module_families(
210 module, rule, variables, interfaces, bases, scope=None, selected_pks=None
211) -> ModuleFamilyPlans:
212 """Return one plan for every family *rule* intends on *module* that *selected_pks* reaches.
214 A *scope* limits an automatic run before interfaces that intend one family collapse (ADR 0013).
215 """
216 previous_forms = bases.previous_forms
217 if previous_forms is not None and any(_is_unclaimed_top_level(interface, bases) for interface in interfaces):
218 # An unclaimed interface may belong to a family, so its module keeps every name.
219 return _kept_module_plans(module, rule, variables, interfaces, bases)
220 installed = plan_installed_families_from(module, rule, variables, interfaces, bases)
221 claimed = installed.member_pks
222 plain = [
223 interface
224 for interface in interfaces
225 if interface.pk not in claimed
226 and not _is_channel(interface)
227 and _is_candidate(interface, rule, bases, previous_forms)
228 and (scope != RunScope.INSTALL or bases.claim(interface.name).raw)
229 ]
230 if not breaks_out(rule):
231 leftover = _selected(
232 [plan_interface_rename(module, rule, variables, interface, bases) for interface in plain], selected_pks
233 )
234 elif any(plan.topology == FamilyTopology.CHANNELIZED for plan in installed.plans):
235 # A breakout rule renames these families and adds none beside them; it reports a claimed one when selected.
236 leftover = _selected(
237 [
238 plan_kept_interface(
239 module,
240 interface,
241 UNCLAIMED_BASE_REASON if bases.builds_on(interface.name) is None else CHANNELIZED_MODULE_REASON,
242 )
243 for interface in plain
244 if selected_pks is not None or bases.builds_on(interface.name) is None
245 ],
246 selected_pks,
247 )
248 for interface in plain:
249 logger.debug(
250 "Interface %r is not channelized; skipping it while rule '%s' breaks out this module's families.",
251 interface.name,
252 rule,
253 )
254 else:
255 members = half_built_members(rule, interfaces, bases)
256 leftover = _creation_plans(module, rule, variables, interfaces, plain, bases, selected_pks, members)
257 installed_plans = installed.plans if scope is None else _in_scope_installed(installed.plans, bases, scope)
258 return ModuleFamilyPlans(installed=tuple(_selected(installed_plans, selected_pks)), leftover=tuple(leftover))
261def _creation_refusal(module, plan, holders, own_pks, family_bases):
262 """Return the reason the apply gives *plan* when one of its names is in use, or None."""
263 if plan.topology == FamilyTopology.CHANNELIZED:
264 channel_names = [member.target_name for member in plan.members if member.role == MemberRole.CHANNEL]
265 parent_name = plan.members[0].target_name
266 refusal = channelized_name_refusal(parent_name, channel_names, holders, own_pks, module.pk, family_bases)
267 return None if refusal is None else refusal[1]
268 owners = {name: holder.pk for name, holder in holders.items()}
269 taken = first_taken_name(plan.target_names, owners, own_pks)
270 return None if taken is None else f"{COLLISION_REASON}: {taken}"
273def creation_name_refusals(module, rule, interfaces, bases, plans) -> dict[str, str]:
274 """Return, by base name, why each creation in *plans* cannot take its names on the device.
276 *plans* are prospective plans. Each creation is checked as its executor checks it at execution,
277 with the rows it keeps or takes over, so a preview offers no family that the apply refuses because
278 one of its names is in use, and gives the reason the apply gives.
279 """
280 creations = [plan for plan in plans if plan.base_name is not None and plan.precondition_status is None]
281 if not breaks_out(rule) or not creations:
282 return {}
283 pks = {interface.name: interface.pk for interface in interfaces}
284 members = half_built_members(rule, interfaces, bases)
285 holders = name_holders(module.device_id, {name for plan in creations for name in plan.target_names})
286 family_bases = _family_bases(interfaces, bases)
287 refusals = {}
288 for plan in creations:
289 own_pks = {pks[plan.base_name], *(member.pk for member in members.get(plan.base_name, ()))}
290 refusal = _creation_refusal(module, plan, holders, own_pks, family_bases)
291 if refusal is not None:
292 refusals[plan.base_name] = refusal
293 refusals.update(_flat_expansion_refusals(module, creations, refusals, bases))
294 return refusals
297def _flat_expansion_refusals(module, creations, refusals, bases): # pragma: no cover - channelization only
298 """Return the refusal of each channelized creation that the rows of the *refusals* families block.
300 A refused family takes over no row, so its rows can be flat siblings, as ``settle_module_families``
301 decides at execution. *creations* are settled already, so no channel name in them repeats.
302 """
303 channelized = [plan for plan in creations if plan.topology == FamilyTopology.CHANNELIZED]
304 if not any(plan.base_name in refusals for plan in channelized):
305 return {}
306 built = [plan for plan in channelized if plan.base_name not in refusals]
307 families = ((plan.base_name, plan.target_names[1:], plan.precondition_status) for plan in built)
308 if not built or not has_flat_expansion(module, module_taken_over_names(families, frozenset(), bases)):
309 return {}
310 return dict.fromkeys((plan.base_name for plan in built), FLAT_EXPANSION_REASON)
313def _selection_pks(plan):
314 """Return the interface primary keys a selection reaches this family through.
316 A channelized family is submitted through its parent alone, because a channel is not an
317 independent candidate on any path.
318 """
319 if isinstance(plan, InstalledFamilyPlan):
320 if plan.parent_pk is None:
321 return plan.member_pks
322 return (plan.parent_pk,)
323 if isinstance(plan, FlatCreationPlan):
324 return plan.member_pks
325 return (plan.base.pk,)
328def _reaches(plan, selected_pks):
329 """Return whether the operator's interface selection reaches *plan*; no selection reaches every plan."""
330 return selected_pks is None or bool(selected_pks.intersection(_selection_pks(plan)))
333def _selected(plans, selected_pks):
334 """Return the plans the operator's interface selection reaches."""
335 return [plan for plan in plans if _reaches(plan, selected_pks)]
338def execute_module_families(plans):
339 """Execute each planned family in order, and yield its outcome before the next family runs.
341 The channelized families that the module builds run together, in one transaction, in the place of the first one.
342 """
343 channelized = [plan for plan in plans if isinstance(plan, StructuralFamilyPlan)]
344 for plan in plans:
345 if not isinstance(plan, StructuralFamilyPlan):
346 yield execute_family_plan(plan)
347 elif plan is channelized[0]:
348 yield from execute_module_structural_families(channelized)
351def _apply_module(rule, module, interfaces, selected_pks):
352 """Plan and execute every selected family on one module."""
353 variables = build_variables(module.module_bay, device=module.device)
354 bases = module_raw_bases(module, rule, variables, interfaces)
355 plans = plan_module_families(module, rule, variables, interfaces, bases, selected_pks=selected_pks).plans
356 return list(execute_module_families(plans))
359def apply_rule_to_modules(rule, modules, selected_pks=None, limit=None) -> BatchOutcome:
360 """Apply *rule* to every module in *modules*, one family at a time.
362 *selected_pks* limits the batch to the families those interfaces reach; *limit* stops it after
363 the module that reached that many changed interfaces.
364 """
365 if not modules:
366 return BatchOutcome(families=())
367 by_module = interfaces_by_module(modules)
368 families: list[FamilyOutcome] = []
369 changed = 0
370 with pinned_template_cache(modules):
371 for module in modules:
372 outcomes = _apply_module(rule, module, by_module[module.pk], selected_pks)
373 families.extend(outcomes)
374 changed += sum(outcome.changed_count for outcome in outcomes)
375 if limit is not None and changed >= limit:
376 break
377 return BatchOutcome(families=tuple(families))