Coverage for netbox_interface_name_rules / family / structural.py: 100.00%
119 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 and execute the creation of interface families."""
5import logging
6from dataclasses import replace
8from dcim.choices import InterfaceTypeChoices
9from dcim.models import Interface, InterfaceTemplate
10from django.core.exceptions import ValidationError
11from django.db import IntegrityError
13from ..transactions import atomic_with_events
14from .capabilities import supports_channelization
15from .domain import (
16 FamilyOutcome,
17 FamilyStatus,
18 FamilyTopology,
19 FlatCreationPlan,
20 InterfaceSnapshot,
21 MemberOutcome,
22 PlannedChannel,
23 StructuralFamilyPlan,
24)
25from .names import (
26 COLLISION_REASON,
27 belongs_elsewhere,
28 channelized_name_refusal,
29 first_taken_name,
30 is_name_collision,
31 name_holders,
32 name_owners,
33 reconcile_after_parent_cascade,
34 repeated_name_refusal,
35 repeated_names,
36 validate_or_block,
37)
38from .targets import UNCLAIMED_BASE_REASON, channelized_family_names, flat_family_names
40logger = logging.getLogger(__name__)
42UNSUPPORTED_REASON = "this NetBox release cannot model channelized interfaces"
43STALE_REASON = "the base interface changed after planning"
44MODULE_CHANGED_REASON = "the module's interfaces changed after planning"
45HOLDER_CHANGED_REASON = "an interface with one of the channel names changed after planning"
46SIBLING_LEFT_REASON = "another interface on the module would stay a flat sibling of this family"
49def _flat_expansion(module_type_id, module_id, channel_names=()) -> bool: # pragma: no cover - channelization only
50 """Return whether the module carries more plain interfaces outside *channel_names* than its templates describe."""
51 templates = InterfaceTemplate.objects.filter(module_type_id=module_type_id).count()
52 plain = Interface.objects.filter(module_id=module_id, channel_id__isnull=True).exclude(name__in=channel_names)
53 return plain.count() > templates
56def has_flat_expansion(module, channel_names=()) -> bool: # pragma: no cover - requires channelization support
57 """Return whether *module* carries more plain interfaces than its module type's templates describe.
59 A flat breakout leaves N-1 rows beyond the templates, so the surplus is the structural mark of a
60 family an earlier apply installed. A channel belongs to the parent that declares it, so it is
61 never part of that surplus: counting one would stop a second port from gaining its own family.
62 Counting templates rather than their resolved names keeps two templates that resolve to the same
63 string from reading as one. A row with one of *channel_names* is a row the family takes over
64 (ADR 0017), so it is not part of the surplus either.
65 """
66 return _flat_expansion(module.module_type_id, module.pk, channel_names)
69def taken_over_names(base_name, channel_names) -> tuple[str, ...]:
70 """Return the channel names whose rows a family on *base_name* takes over instead of stranding.
72 A flat family gives its first row one of its channel names, so the siblings of such a base stay siblings.
73 """
74 return () if base_name in channel_names else tuple(channel_names)
77def flat_surplus(interfaces, templates) -> int:
78 """Return how many more rows outside a channel *interfaces* hold than *templates* has, as ``has_flat_expansion``."""
79 return sum(1 for interface in interfaces if getattr(interface, "channel_id", None) is None) - len(templates)
82def _plan(module, base, parent_target_name, channels, family_bases, status=None, reason="", holders=None):
83 """Build one immutable structural plan for *base*; *holders* snapshots the rows with a channel name, by name."""
84 holders = holders or {}
85 channel_names = [name for _channel_id, name in channels]
86 return StructuralFamilyPlan(
87 family_id=f"structural:{base.pk}",
88 device_id=module.device_id,
89 module_id=module.pk,
90 module_type_id=module.module_type_id,
91 base=InterfaceSnapshot.from_interface(base),
92 parent_target_name=parent_target_name,
93 channel_count=len(channels),
94 channels=tuple(
95 PlannedChannel(channel_id=channel_id, name=name, holder=holders.get(name)) for channel_id, name in channels
96 ),
97 family_bases=frozenset(family_bases),
98 precondition_status=status,
99 precondition_reason=reason,
100 taken_over_names=taken_over_names(base.name, channel_names),
101 )
104def _modelled_plan(module, rule, variables, base, base_name, family_bases): # pragma: no cover - channelization only
105 """Return the plan for a NetBox release that can hold the family."""
106 if base_name is None:
107 return _plan(module, base, base.name, (), family_bases, FamilyStatus.BLOCKED, UNCLAIMED_BASE_REASON)
108 try:
109 parent_target_name, channels = channelized_family_names(rule, base.name, base_name, variables)
110 except (TypeError, ValueError) as error:
111 reason = f"failed to evaluate the family names: {error}"
112 return _plan(module, base, base.name, (), family_bases, FamilyStatus.FAILED, reason)
113 channel_names = [name for _channel_id, name in channels]
114 names = (parent_target_name, *channel_names)
115 rows = {name: row for name, row in name_holders(module.device_id, names).items() if row.pk != base.pk}
116 # A row of another module or family is reported by its own outcome, so this plan does not hold it.
117 holders = {
118 name: InterfaceSnapshot.from_interface(row)
119 for name, row in rows.items()
120 if name in channel_names and not belongs_elsewhere(row, module.pk, family_bases)
121 }
122 refusal = channelized_name_refusal(parent_target_name, channel_names, rows, (base.pk,), module.pk, family_bases)
123 if refusal is not None:
124 return _plan(
125 module, base, parent_target_name, channels, family_bases, FamilyStatus.BLOCKED, refusal[1], holders
126 )
127 return _plan(module, base, parent_target_name, channels, family_bases, holders=holders)
130def plan_structural_family(module, rule, variables, base, base_name, family_bases) -> StructuralFamilyPlan:
131 """Return the plan for the channelized family *rule* builds on plain interface *base*.
133 *base_name* is the value of ``{base}``, or None when no template claims *base*. *family_bases*
134 are the primary keys of the module's rows that the claim gives a template, which no family takes
135 over. ``settle_module_families`` then settles the families on the module against each other.
136 """
137 if not supports_channelization():
138 return _plan(module, base, base.name, (), family_bases, FamilyStatus.UNSUPPORTED, UNSUPPORTED_REASON)
139 return _modelled_plan(module, rule, variables, base, base_name, family_bases) # pragma: no cover - see above
142def module_taken_over_names(families, repeated, bases) -> frozenset:
143 """Return the names of the rows that the channelized families planned on one module take over.
145 *families* gives the base name, channel names and precondition status of each family whose names
146 were evaluated, and *repeated* the channel names that the families want more than once. A refused
147 family takes over no row, so its rows stay flat siblings. A row the claim gives a template is no
148 family's to take over, so it is not one of them.
149 """
150 return frozenset(
151 name
152 for base_name, channel_names, status in families
153 if status is None and repeated_name_refusal(channel_names, repeated) is None
154 for name in taken_over_names(base_name, channel_names)
155 if not bases.claims_alone(name)
156 )
159def _without_holders(plan, names=None):
160 """Return *plan* holding no row with one of *names*, or no row at all when *names* is None."""
161 channels = tuple(
162 replace(channel, holder=None) if names is None or channel.name in names else channel
163 for channel in plan.channels
164 )
165 return replace(plan, channels=channels)
168def _settled(plan, repeated, names, expanded, module): # pragma: no cover - requires channelization support
169 """Return *plan* as the module's families settle it; see ``settle_module_families``."""
170 refusal = repeated_name_refusal(plan.channel_names, repeated)
171 if refusal is not None:
172 plan = _without_holders(plan, repeated)
173 return replace(plan, precondition_status=FamilyStatus.BLOCKED, precondition_reason=refusal[1])
174 if plan.precondition_status is not None:
175 return plan
176 if expanded:
177 # Converting one sibling into a parent would strand the others beside the new family.
178 reason = f"module {module} already carries a flat breakout family"
179 return replace(_without_holders(plan), precondition_status=FamilyStatus.BLOCKED, precondition_reason=reason)
180 return replace(plan, taken_over_names=tuple(sorted(names)))
183def settle_module_families( # pragma: no cover - channelization only
184 module, plans, interfaces, templates, bases, applied
185):
186 """Return *plans* with the structural families on *module* settled against each other before execution.
188 A channel name that the families want more than once refuses every family that wants it, and no
189 family takes over its row. Planning refuses a family whose parent name is in use, and no family
190 takes a row over by its parent name. Then, when *module* carries a flat breakout family, every other family is
191 refused. *interfaces* are the module's rows and *templates* its resolved templates. The rows
192 that the module's families take over are not flat siblings (ADR 0017); any other row beyond the
193 templates is. *applied* holds the base of each family that the apply reaches: a family that it
194 does not reach takes over no row. Each family that is built keeps the names of the rows that the
195 module's families take over, for the check under lock.
196 """
197 evaluated = [plan for plan in plans if isinstance(plan, StructuralFamilyPlan) and plan.channels]
198 repeated = repeated_names(plan.channel_names for plan in evaluated)
199 reached = [plan for plan in evaluated if plan.base.pk in applied]
200 families = ((plan.base.name, plan.channel_names, plan.precondition_status) for plan in reached)
201 names = module_taken_over_names(families, repeated, bases)
202 expanded = flat_surplus([interface for interface in interfaces if interface.name not in names], templates) > 0
203 settled = {plan.base.pk: _settled(plan, repeated, names, expanded, module) for plan in evaluated}
204 return [settled.get(plan.base.pk, plan) if isinstance(plan, StructuralFamilyPlan) else plan for plan in plans]
207def _outcome(plan, status, members, reason=""):
208 """Build the immutable outcome of one structural family operation."""
209 return FamilyOutcome(
210 family_id=plan.family_id,
211 topology=FamilyTopology.CHANNELIZED,
212 status=status,
213 members=members,
214 reason=reason,
215 )
218def _refused(plan, status, reason, target_name):
219 """Log why the family was not built and return an outcome for the base and each held row of its module."""
220 logger.warning("Cannot build a channelized family on interface %r: %s.", plan.base.name, reason)
221 members = (
222 MemberOutcome(
223 interface_pk=plan.base.pk,
224 current_name=plan.base.name,
225 target_name=target_name,
226 status=status,
227 reason=reason,
228 ),
229 # A row with no module is reported only by the family that takes it over, so a batch reports it once.
230 *(
231 MemberOutcome(holder.pk, holder.name, holder.name, status, reason)
232 for holder in plan.holders
233 if holder.module_id == plan.module_id
234 ),
235 )
236 return _outcome(plan, status, members, reason)
239def _locked_rows(plan):
240 """Lock the base and every planned holder in primary-key order, and return the live rows by primary key."""
241 return (
242 Interface.objects.select_for_update(of=("self",))
243 .select_related("device", "module")
244 .filter(pk__in=(plan.base.pk, *(holder.pk for holder in plan.holders)))
245 .order_by("pk")
246 .in_bulk()
247 )
250def _stale_reason(plan, rows): # pragma: no cover - requires channelization support
251 """Return why the locked *rows* no longer match *plan*, or an empty string."""
252 base = rows.get(plan.base.pk)
253 if base is None or InterfaceSnapshot.from_interface(base) != plan.base:
254 return STALE_REASON
255 if any(
256 holder.pk not in rows or InterfaceSnapshot.from_interface(rows[holder.pk]) != holder for holder in plan.holders
257 ):
258 return HOLDER_CHANGED_REASON
259 # A sibling added since planning would be stranded beside the family this plan builds.
260 if _flat_expansion(plan.module_type_id, plan.module_id, plan.taken_over_names):
261 return MODULE_CHANGED_REASON
262 return ""
265def _name_refusal(plan): # pragma: no cover - requires channelization support
266 """Return ``(status, reason, name)`` when the family cannot take its names, or None.
268 One query, because the caller holds the plan's row locks while this runs.
269 """
270 repeated = repeated_name_refusal(plan.channel_names, repeated_names([plan.channel_names]))
271 if repeated is not None:
272 return FamilyStatus.BLOCKED, repeated[1], repeated[0]
273 channel_names = plan.channel_names
274 holders = name_holders(plan.device_id, plan.target_names)
275 refusal = channelized_name_refusal(
276 plan.parent_target_name, channel_names, holders, (plan.base.pk,), plan.module_id, plan.family_bases
277 )
278 if refusal is not None:
279 name, reason = refusal
280 return FamilyStatus.BLOCKED, reason, name
281 # Every row left with a channel name qualifies, so the plan must hold exactly those rows.
282 planned = {channel.name: channel.holder.pk for channel in plan.channels if channel.holder is not None}
283 live = {name: row.pk for name, row in holders.items() if name in channel_names and row.pk != plan.base.pk}
284 if live != planned:
285 return FamilyStatus.STALE, HOLDER_CHANGED_REASON, plan.parent_target_name
286 return None
289def _first_taken_name(plan, own_pks):
290 """Return the first planned name an interface outside *own_pks* already owns on the device, or None.
292 One query, because the caller holds the plan's row locks while this runs.
293 """
294 return first_taken_name(plan.target_names, name_owners(plan.device_id, plan.target_names), own_pks)
297def _channel_row(channel, parent, rows): # pragma: no cover - requires channelization support
298 """Return the row for *channel* under *parent*: its holder in *rows* bound to the family, else a new row."""
299 if channel.holder is None:
300 return Interface(
301 device=parent.device,
302 module=parent.module,
303 name=channel.name,
304 type=InterfaceTypeChoices.TYPE_CHANNEL,
305 parent=parent,
306 channel_id=channel.channel_id,
307 enabled=parent.enabled,
308 )
309 row = rows[channel.holder.pk]
310 row.snapshot()
311 row.module = parent.module
312 row.parent = parent
313 row.channel_id = channel.channel_id
314 row.type = InterfaceTypeChoices.TYPE_CHANNEL
315 return row
318def _create_channels(plan, parent, rows): # pragma: no cover - requires channelization support
319 """Create or take over every planned channel under *parent* and return their outcomes."""
320 members = []
321 for channel in plan.channels:
322 row = _channel_row(channel, parent, rows)
323 validate_or_block(row, "channel")
324 row.save()
325 members.append(
326 MemberOutcome(
327 interface_pk=row.pk,
328 current_name=channel.name,
329 target_name=channel.name,
330 status=FamilyStatus.CHANGED,
331 )
332 )
333 return members
336def _create_family(plan, rows): # pragma: no cover - requires channelization support
337 """Rewrite the base in *rows* into the family parent, create or take over its channels, and return every outcome."""
338 base = rows[plan.base.pk]
339 parent_status = FamilyStatus.CHANGED if plan.parent_target_name != base.name else FamilyStatus.UNCHANGED
340 base.snapshot()
341 base.channels = plan.channel_count
342 base.name = plan.parent_target_name
343 base.full_clean()
344 base.save()
345 parent_member = MemberOutcome(
346 interface_pk=base.pk,
347 current_name=plan.base.name,
348 target_name=plan.parent_target_name,
349 status=parent_status,
350 )
351 channel_members = _create_channels(plan, base, rows)
352 reconcile_after_parent_cascade(
353 plan.base.name,
354 plan.parent_target_name,
355 tuple(
356 (member.interface_pk, channel.channel_id, channel.name)
357 for member, channel in zip(channel_members, plan.channels, strict=True)
358 ),
359 )
360 return (parent_member, *channel_members)
363def _install_family(plan): # pragma: no cover - requires channelization support
364 """Create the whole family in one transaction, or write nothing at all."""
365 try:
366 with atomic_with_events():
367 rows = _locked_rows(plan)
368 stale = _stale_reason(plan, rows)
369 if stale:
370 return _refused(plan, FamilyStatus.STALE, stale, plan.parent_target_name)
371 refusal = _name_refusal(plan)
372 if refusal is not None:
373 return _refused(plan, *refusal)
374 members = _create_family(plan, rows)
375 except ValidationError as error:
376 return _refused(plan, FamilyStatus.BLOCKED, " ".join(error.messages), plan.parent_target_name)
377 except IntegrityError as error:
378 if not is_name_collision(error):
379 raise
380 return _refused(plan, FamilyStatus.BLOCKED, COLLISION_REASON, plan.parent_target_name)
381 return _outcome(plan, FamilyStatus.CHANGED, members)
384def execute_structural_family(plan: StructuralFamilyPlan) -> FamilyOutcome:
385 """Create the planned channelized family, or leave every row exactly as it was.
387 Only a structural plan names the base row to rewrite, so anything else (a prospective plan
388 above all) is refused before a single row is locked.
389 """
390 if not isinstance(plan, StructuralFamilyPlan):
391 raise TypeError(f"{type(plan).__name__} is not an executable family plan")
392 if plan.precondition_status is not None:
393 return _refused(plan, plan.precondition_status, plan.precondition_reason, plan.parent_target_name)
394 return _install_family(plan) # pragma: no cover - requires channelization support
397def execute_module_structural_families(plans) -> tuple[FamilyOutcome, ...]: # pragma: no cover - channelization only
398 """Build the channelized families planned on one module in one transaction, and keep none beside a flat sibling.
400 A family that is not built leaves its rows as flat siblings, so each family built beside them rolls back (ADR 0017).
401 """
402 if len({plan.module_id for plan in plans}) != 1:
403 raise ValueError("The channelized families of one transaction must be on one module.")
404 with atomic_with_events() as block:
405 outcomes = tuple(execute_structural_family(plan) for plan in plans)
406 built = [outcome.status == FamilyStatus.CHANGED for outcome in outcomes]
407 if not any(built) or not _flat_expansion(plans[0].module_type_id, plans[0].module_id):
408 return outcomes
409 block.set_rollback()
410 return tuple(
411 _refused(plan, FamilyStatus.BLOCKED, SIBLING_LEFT_REASON, plan.parent_target_name) if was_built else outcome
412 for plan, outcome, was_built in zip(plans, outcomes, built, strict=True)
413 )
416# ---------------------------------------------------------------------------
417# Flat breakout families
418# ---------------------------------------------------------------------------
419# A flat family is N sibling interfaces on one module, built whole or not at all (ADR 0001).
422def _flat_creation_plan(module, base, target_names, status=None, reason="", members=()):
423 """Build one immutable flat-creation plan for *base*, with the rows *members* it keeps."""
424 return FlatCreationPlan(
425 family_id=f"flat:{base.pk}",
426 device_id=module.device_id,
427 module_id=module.pk,
428 base=InterfaceSnapshot.from_interface(base),
429 target_names=target_names,
430 precondition_status=status,
431 precondition_reason=reason,
432 members=tuple(InterfaceSnapshot.from_interface(member) for member in members),
433 )
436def plan_flat_family(module, rule, variables, base, base_name, members=()) -> FlatCreationPlan:
437 """Return the plan for the flat breakout family *rule* builds on plain interface *base*.
439 *base_name* is the value of ``{base}``, or None when no template claims *base*. *members* are
440 the rows of a half-built family that the claim gave the same template; the family keeps them.
441 """
442 if base_name is None:
443 return _flat_creation_plan(module, base, (base.name,), FamilyStatus.BLOCKED, UNCLAIMED_BASE_REASON)
444 try:
445 target_names = flat_family_names(rule, variables, base_name)
446 except (TypeError, ValueError) as error:
447 reason = f"failed to evaluate the family names: {error}"
448 return _flat_creation_plan(module, base, (base.name,), FamilyStatus.FAILED, reason)
449 if not target_names:
450 reason = "flat family requires at least one target name"
451 return _flat_creation_plan(module, base, (base.name,), FamilyStatus.FAILED, reason)
452 if any(member.name not in target_names[1:] for member in members):
453 raise ValueError(f"a row the flat family on {base.name!r} keeps must carry one of its sibling names")
454 return _flat_creation_plan(module, base, target_names, members=members)
457def _flat_outcome(plan, status, members, reason=""):
458 """Build the immutable outcome of one flat family operation."""
459 return FamilyOutcome(
460 family_id=plan.family_id,
461 topology=FamilyTopology.FLAT,
462 status=status,
463 members=members,
464 reason=reason,
465 )
468def _flat_refused(plan, status, reason):
469 """Log why the family was not built and return an outcome for every planned row, which it left as it was."""
470 logger.warning("Cannot build a flat family on interface %r: %s.", plan.base.name, reason)
471 members = (
472 MemberOutcome(plan.base.pk, plan.base.name, plan.target_names[0], status, reason),
473 *(MemberOutcome(member.pk, member.name, member.name, status, reason) for member in plan.members),
474 )
475 return _flat_outcome(plan, status, members, reason)
478def _locked_flat_rows(plan):
479 """Lock the base and every planned member in primary-key order, and return the live rows by primary key."""
480 return (
481 Interface.objects.select_for_update(of=("self",))
482 .select_related("device", "module")
483 .filter(pk__in=plan.member_pks)
484 .order_by("pk")
485 .in_bulk()
486 )
489def _is_flat_stale(plan, rows):
490 """Return whether the base or a planned member is gone or changed since planning."""
491 planned = {plan.base.pk: plan.base, **{member.pk: member for member in plan.members}}
492 return planned != {pk: InterfaceSnapshot.from_interface(row) for pk, row in rows.items()}
495def _build_flat_family(plan, base):
496 """Name the base, keep each planned member, and create every other sibling."""
497 target_name = plan.target_names[0]
498 status = FamilyStatus.UNCHANGED
499 if target_name != base.name:
500 base.snapshot()
501 base.name = target_name
502 base.full_clean()
503 base.save()
504 status = FamilyStatus.CHANGED
505 outcomes = [MemberOutcome(base.pk, plan.base.name, target_name, status)]
506 kept = {member.name: member for member in plan.members}
507 for name in plan.target_names[1:]:
508 if name in kept:
509 outcomes.append(MemberOutcome(kept[name].pk, name, name, FamilyStatus.UNCHANGED))
510 continue
511 row = Interface(device=base.device, module=base.module, name=name, type=base.type, enabled=base.enabled)
512 row.full_clean()
513 row.save()
514 outcomes.append(MemberOutcome(row.pk, name, name, FamilyStatus.CHANGED))
515 return tuple(outcomes)
518def _install_flat_family(plan):
519 """Build the whole family in one transaction, or write nothing at all."""
520 try:
521 with atomic_with_events():
522 rows = _locked_flat_rows(plan)
523 if _is_flat_stale(plan, rows):
524 return _flat_refused(plan, FamilyStatus.STALE, STALE_REASON)
525 taken = _first_taken_name(plan, plan.member_pks)
526 if taken is not None:
527 return _flat_refused(plan, FamilyStatus.BLOCKED, f"{COLLISION_REASON}: {taken}")
528 members = _build_flat_family(plan, rows[plan.base.pk])
529 except ValidationError as error:
530 return _flat_refused(plan, FamilyStatus.BLOCKED, " ".join(error.messages))
531 except IntegrityError as error:
532 if not is_name_collision(error):
533 raise
534 return _flat_refused(plan, FamilyStatus.BLOCKED, COLLISION_REASON)
535 changed = any(member.status == FamilyStatus.CHANGED for member in members)
536 return _flat_outcome(plan, FamilyStatus.CHANGED if changed else FamilyStatus.UNCHANGED, members)
539def execute_flat_family(plan: FlatCreationPlan) -> FamilyOutcome:
540 """Build the planned flat breakout family, or leave every row exactly as it was."""
541 if not isinstance(plan, FlatCreationPlan):
542 raise TypeError(f"{type(plan).__name__} is not an executable family plan")
543 if plan.precondition_status is not None:
544 return _flat_refused(plan, plan.precondition_status, plan.precondition_reason)
545 return _install_flat_family(plan)