Coverage for netbox_interface_name_rules / family / execution.py: 100.00%
80 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"""Execute installed interface-family plans."""
5import logging
7from dcim.models import Interface
8from django.core.exceptions import ValidationError
9from django.db import IntegrityError
10from django.db.models import Q
12from ..transactions import atomic_with_events
13from .domain import (
14 FamilyOutcome,
15 FamilyStatus,
16 InstalledFamilyPlan,
17 InstalledFamilyPlanSet,
18 InstalledPlanSetOutcome,
19 InterfaceSnapshot,
20 MemberOutcome,
21 MemberRole,
22)
23from .names import COLLISION_REASON, is_name_collision, name_is_taken, reconcile_after_parent_cascade
25logger = logging.getLogger(__name__)
27STALE_REASON = "installed family changed after planning"
28PARENT_BLOCKED_REASON = "family parent was blocked"
31def _lock_family(plan: InstalledFamilyPlan):
32 """Lock and return the current family rows in stable primary-key order."""
33 queryset = Interface.objects
34 if plan.parent_pk is None:
35 queryset = queryset.filter(pk__in=plan.member_pks)
36 else: # pragma: no cover - requires channelization support
37 # Only channels are family members; a virtual subinterface may also parent here.
38 queryset = queryset.filter(Q(pk=plan.parent_pk) | Q(parent_id=plan.parent_pk, channel_id__isnull=False))
39 return list(queryset.select_for_update().order_by("pk"))
42def _is_stale(plan: InstalledFamilyPlan, interfaces) -> bool:
43 """Return whether live identity, names, membership, or topology changed."""
44 planned = {member.snapshot.pk: member.snapshot for member in plan.members}
45 live = {interface.pk: InterfaceSnapshot.from_interface(interface) for interface in interfaces}
46 return planned != live
49def _member_outcome(member, status, reason=""):
50 """Build one immutable member outcome from its plan facts."""
51 return MemberOutcome(
52 interface_pk=member.snapshot.pk,
53 current_name=member.snapshot.name,
54 target_name=member.target_name,
55 status=status,
56 reason=reason,
57 )
60def _blocked_member(member, reason):
61 """Return a blocked member outcome."""
62 return _member_outcome(member, FamilyStatus.BLOCKED, reason)
65def _rename_member(member, interface):
66 """Apply one planned name inside a savepoint and return its explicit outcome."""
67 target_name = member.target_name
68 if target_name is None:
69 logger.warning(
70 "Cannot derive a name for channel interface %r; leaving it unchanged.",
71 member.snapshot.name,
72 )
73 return _blocked_member(member, member.reason)
74 if target_name == interface.name:
75 return _member_outcome(member, FamilyStatus.UNCHANGED)
76 if name_is_taken(interface.device_id, target_name, exclude_pk=interface.pk):
77 logger.warning(
78 "Interface name %r already exists on device %s; skipping rename of %r to %r.",
79 target_name,
80 interface.device_id,
81 interface.name,
82 target_name,
83 )
84 return _blocked_member(member, COLLISION_REASON)
86 previous_name = interface.name
87 interface.snapshot()
88 try:
89 with atomic_with_events():
90 interface.name = target_name
91 interface.full_clean()
92 interface.save()
93 except ValidationError as error:
94 interface.name = previous_name
95 logger.warning("NetBox rejected rename of %r to %r: %s", previous_name, target_name, error)
96 return _blocked_member(member, " ".join(error.messages))
97 except IntegrityError as error:
98 interface.name = previous_name
99 if is_name_collision(error):
100 logger.warning(
101 "Interface name %r became occupied while renaming %r; skipping.",
102 target_name,
103 previous_name,
104 )
105 return _blocked_member(member, COLLISION_REASON)
106 raise
107 return _member_outcome(member, FamilyStatus.CHANGED)
110def _family_status(members):
111 """Summarize member outcomes without hiding partial success."""
112 statuses = {member.status for member in members}
113 if FamilyStatus.CHANGED in statuses:
114 return FamilyStatus.CHANGED
115 if FamilyStatus.BLOCKED in statuses:
116 return FamilyStatus.BLOCKED
117 return FamilyStatus.UNCHANGED
120def _preserve_names_across_parent_cascade(plan, member_outcomes): # pragma: no cover - channelization only
121 """Schedule restoration of blocked conventional channel names after a parent cascade."""
122 if plan.parent_pk is None:
123 return
124 parent_member = next(member for member in plan.members if member.role == MemberRole.PARENT)
125 parent_outcome = next(outcome for outcome in member_outcomes if outcome.interface_pk == plan.parent_pk)
126 if parent_outcome.status != FamilyStatus.CHANGED:
127 return
128 outcomes_by_pk = {outcome.interface_pk: outcome for outcome in member_outcomes}
129 channels = tuple(
130 (
131 member.snapshot.pk,
132 member.snapshot.channel_id,
133 member.target_name
134 if outcomes_by_pk[member.snapshot.pk].status in (FamilyStatus.CHANGED, FamilyStatus.UNCHANGED)
135 else member.snapshot.name,
136 )
137 for member in plan.members
138 if member.role == MemberRole.CHANNEL
139 )
140 reconcile_after_parent_cascade(parent_member.snapshot.name, parent_member.target_name, channels)
143def _execute_channelized_members(plan, live_by_pk): # pragma: no cover - requires channelization support
144 """Execute a channelized family after its parent succeeds."""
145 parent_member = next(member for member in plan.members if member.role == MemberRole.PARENT)
146 parent_outcome = _rename_member(parent_member, live_by_pk[parent_member.snapshot.pk])
147 if parent_outcome.status == FamilyStatus.BLOCKED:
148 return (
149 parent_outcome,
150 *(
151 _blocked_member(member, PARENT_BLOCKED_REASON)
152 for member in plan.members
153 if member.role != MemberRole.PARENT
154 ),
155 )
156 return (
157 parent_outcome,
158 *(
159 _rename_member(member, live_by_pk[member.snapshot.pk])
160 for member in plan.members
161 if member.role != MemberRole.PARENT
162 ),
163 )
166def _execute_members(plan, live_by_pk):
167 """Execute members while enforcing parent-first family semantics."""
168 if plan.parent_pk is None:
169 return tuple(_rename_member(member, live_by_pk[member.snapshot.pk]) for member in plan.members)
170 return _execute_channelized_members(plan, live_by_pk) # pragma: no cover - channelization only
173def _stale_outcome(plan: InstalledFamilyPlan) -> FamilyOutcome:
174 """Return a stale result without applying any planned member."""
175 return FamilyOutcome(
176 family_id=plan.family_id,
177 topology=plan.topology,
178 status=FamilyStatus.STALE,
179 members=tuple(_member_outcome(member, FamilyStatus.STALE, STALE_REASON) for member in plan.members),
180 reason=STALE_REASON,
181 )
184def execute_installed_plan(plan: InstalledFamilyPlan) -> FamilyOutcome:
185 """Execute one family plan in its own transaction."""
186 with atomic_with_events():
187 interfaces = _lock_family(plan)
188 if _is_stale(plan, interfaces):
189 return _stale_outcome(plan)
190 if plan.precondition_status is not None:
191 logger.warning(
192 "Installed family of interface %r is %s: %s.",
193 plan.members[0].snapshot.name,
194 plan.precondition_status,
195 plan.precondition_reason,
196 )
197 return FamilyOutcome(
198 family_id=plan.family_id,
199 topology=plan.topology,
200 status=plan.precondition_status,
201 members=tuple(
202 _member_outcome(member, plan.precondition_status, plan.precondition_reason)
203 for member in plan.members
204 ),
205 reason=plan.precondition_reason,
206 )
207 live_by_pk = {interface.pk: interface for interface in interfaces}
208 member_outcomes = _execute_members(plan, live_by_pk)
209 _preserve_names_across_parent_cascade(plan, member_outcomes)
210 return FamilyOutcome(
211 family_id=plan.family_id,
212 topology=plan.topology,
213 status=_family_status(member_outcomes),
214 members=member_outcomes,
215 )
218def execute_installed_plan_set(plan_set: InstalledFamilyPlanSet) -> InstalledPlanSetOutcome:
219 """Execute each installed family in a separate transaction.
221 Only an installed plan set carries the row snapshots execution revalidates, so anything else
222 (a prospective plan set above all) is refused before a single row is locked.
223 """
224 if not isinstance(plan_set, InstalledFamilyPlanSet):
225 raise TypeError(f"{type(plan_set).__name__} is not an executable plan set")
226 return InstalledPlanSetOutcome(families=tuple(execute_installed_plan(plan) for plan in plan_set.plans))