Coverage for netbox_interface_name_rules / engine.py: 100.00%

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com> 

3"""Core renaming engine — rule lookup and interface rename logic. 

4 

5This module is imported lazily by rename_triggers.py so that model imports happen 

6after Django is fully initialised. 

7""" 

8 

9import contextlib 

10import logging 

11from collections import defaultdict 

12from collections.abc import Iterator 

13from dataclasses import dataclass, replace 

14from typing import NamedTuple 

15 

16from django.core.exceptions import ValidationError 

17from django.db import DatabaseError 

18 

19from . import family as family_ops 

20from . import name_template, naming, rule_selection 

21from .family import template_names as family_template_names 

22from .regex_safety import compile_module_type_pattern 

23from .rename_outcomes import OutcomeKind, RenameOutcome, renamed_count 

24from .transactions import atomic_with_events 

25 

26logger = logging.getLogger(__name__) 

27 

28NO_RULE_REASON = "no rule matches the module after the change" 

29FLAT_REASON = "a flat breakout family is not renamed after a move, a bay edit or a parent module type change" 

30ELSEWHERE_REASON = "the interface is not on the device of its module" 

31STALE_BAY_REASON = "the module bay still has the parent bay it had before its module moved" 

32 

33 

34def pinned_rule_cache(): 

35 """Return the lower-level rule cache pinning context.""" 

36 return rule_selection.pinned_rule_cache() 

37 

38 

39def find_matching_rule(module_type, parent_module_type, device_type, platform=None): 

40 """Delegate rule selection while preserving the engine entry point.""" 

41 return rule_selection.find_matching_rule(module_type, parent_module_type, device_type, platform) 

42 

43 

44def build_variables(module_bay, device=None): 

45 """Delegate naming-variable construction while preserving the engine entry point.""" 

46 return naming.build_variables(module_bay, device=device) 

47 

48 

49def evaluate_name_template(template: str, variables: dict) -> str: 

50 """Delegate template evaluation while preserving the engine entry point.""" 

51 return name_template.evaluate_name_template(template, variables) 

52 

53 

54def _get_parent_module_type(module_bay): 

55 """Return the module type of the module installed in the parent bay, or None. 

56 

57 Used by ``apply_interface_name_rules`` to scope rules to a specific parent 

58 module type (e.g., SFP inside a CVR-X2-SFP converter). 

59 """ 

60 if module_bay.parent: 

61 parent_bay = module_bay.parent 

62 if hasattr(parent_bay, "installed_module") and parent_bay.installed_module: 

63 return parent_bay.installed_module.module_type 

64 return None 

65 

66 

67def _selected_rule(module, module_bay): 

68 """Return the rule that *module* in *module_bay* selects now, or None.""" 

69 device_type = module.device.device_type if module.device else None 

70 platform = module.device.platform if module.device else None 

71 return find_matching_rule(module.module_type, _get_parent_module_type(module_bay), device_type, platform) 

72 

73 

74def supports_channelization(): 

75 """Delegate the channelization capability check while preserving the engine entry point.""" 

76 return family_ops.supports_channelization() 

77 

78 

79def _vc_position_re(): 

80 """Delegate virtual-chassis token detection to template-name resolution.""" 

81 return family_template_names.vc_position_re() 

82 

83 

84def supports_vc_position_token(): 

85 """Return True when this NetBox resolves ``{vc_position}`` in component template names (4.6+). 

86 

87 Probed from the constant that carries the token rather than a version comparison, so a backport 

88 or an upstream removal is detected by what NetBox actually provides. 

89 """ 

90 return _vc_position_re() is not None 

91 

92 

93def _touches_a_family(plan) -> bool: 

94 """Return whether *plan* acts on a family rather than one standalone interface.""" 

95 if isinstance(plan, family_ops.InstalledFamilyPlan): 

96 return plan.parent_pk is not None or len(plan.members) > 1 

97 return True 

98 

99 

100def _run_scope(force_reapply, previous_forms): 

101 """Return the scope of an automatic run; after a move the claim decides alone, without one.""" 

102 if previous_forms is not None: 

103 return None 

104 return family_ops.RunScope.FORCED if force_reapply else family_ops.RunScope.INSTALL 

105 

106 

107_MEMBER_OUTCOME_KINDS = { 

108 family_ops.FamilyStatus.CHANGED: OutcomeKind.RENAMED, 

109 family_ops.FamilyStatus.BLOCKED: OutcomeKind.BLOCKED, 

110 family_ops.FamilyStatus.STALE: OutcomeKind.BLOCKED, 

111 family_ops.FamilyStatus.UNSUPPORTED: OutcomeKind.BLOCKED, 

112 family_ops.FamilyStatus.FAILED: OutcomeKind.FAILED, 

113} 

114 

115 

116def _rename_outcome(member) -> RenameOutcome: 

117 """Return the outcome fact of one executed family member.""" 

118 kind = _MEMBER_OUTCOME_KINDS[member.status] 

119 if kind == OutcomeKind.BLOCKED and member.reason == family_ops.UNCLAIMED_BASE_REASON: 

120 kind = OutcomeKind.UNCLAIMED 

121 return RenameOutcome(kind, member.current_name, member.reason, member.target_name) 

122 

123 

124def _rename_outcomes(family_outcomes) -> tuple[RenameOutcome, ...]: 

125 """Return the outcome facts of executed families; a member that kept its correct name has none.""" 

126 return tuple( 

127 _rename_outcome(member) 

128 for family in family_outcomes 

129 for member in family.members 

130 if member.status != family_ops.FamilyStatus.UNCHANGED 

131 ) 

132 

133 

134def _unresolved_outcomes(missing, interface_names) -> tuple[RenameOutcome, ...]: 

135 """Return one unresolved-variable fact per interface that keeps its name for lack of *missing*.""" 

136 reason = name_template.unavailable_reason(missing) 

137 return tuple(RenameOutcome(OutcomeKind.UNRESOLVED_VARIABLE, name, reason) for name in interface_names) 

138 

139 

140class NamingPoint(NamedTuple): 

141 """A point in the transaction at which NetBox gave a module's templates raw names. 

142 

143 ``naming`` is the ``ModuleNaming`` that holds the bay chain of the module then, or None for the 

144 committed module; ``vc_position`` is its device's virtual-chassis position then, and ``move`` is 

145 set when a move gave the names, which NetBox resolves with the resolver of its move planner. 

146 """ 

147 

148 naming: "ModuleNaming | None" 

149 vc_position: int | None 

150 move: bool 

151 

152 

153def _raw_names_at(module, naming_points): 

154 """Return, by template, the raw names that NetBox gave *module*'s templates at *naming_points*.""" 

155 names = defaultdict(set) 

156 for point in naming_points: 

157 templates = family_ops.resolved_template_names(module) if point.naming is None else point.naming.templates 

158 for template in templates: 

159 names[template.pk].add(template.at_chassis_position(point.vc_position, move=point.move).resolved) 

160 return names 

161 

162 

163def apply_interface_name_rules(module, module_bay, force_reapply=False): 

164 """Apply InterfaceNameRule rename after module installation. 

165 

166 Looks up a matching rule for (module_type, parent_module_type, device_type, platform) 

167 and renames interfaces created by NetBox's template instantiation. 

168 

169 Only processes interfaces whose name still matches the raw bay position 

170 (i.e., haven't been renamed yet), ensuring idempotency. Pass 

171 ``force_reapply=True`` to skip this check and re-apply rules to ALL 

172 module interfaces (used when vc_position or other variables change). 

173 

174 Every rename and every creation goes through the family package, so this path builds and names 

175 exactly what retroactive apply and the preview describe. 

176 

177 Returns: 

178 Number of interfaces renamed/created, or 0 if no rule matched. 

179 

180 """ 

181 return renamed_count(module_rule_outcomes(module, module_bay, force_reapply)) 

182 

183 

184def module_rule_outcomes( 

185 module, module_bay, force_reapply=False, report_only=False, naming=None, naming_points=() 

186) -> Iterator[RenameOutcome]: 

187 """Apply the module's rule as ``apply_interface_name_rules`` does, and yield its outcome facts. 

188 

189 Each family's facts are yielded before the next family runs, so a caller keeps them when a later 

190 family fails. With *report_only*, nothing is renamed: only a rule that needs a variable the 

191 device lacks gives facts. 

192 

193 *naming* is the module's ``ModuleNaming`` read before a move, a bay edit or a type change of its 

194 parent module. The rule then renames every name one template claims through its current or 

195 previous forms, and reports every other interface. Without a rule now, each interface the previous 

196 rule named keeps its name and is reported. When the previous rule is a flat breakout rule, nothing 

197 on the module is renamed and every interface is reported: NetBox keeps no link to a family, so it 

198 could be recognised by name only (ADR 0015). When an interface of the module is on another device, 

199 or the module's bay has a parent bay that does not hold the module that owns the bay, nothing is 

200 renamed and every interface is reported. 

201 

202 *naming_points* are the ``NamingPoint`` values at which NetBox gave the module's templates raw 

203 names in the transaction. The claim recognises the names of each point too. 

204 """ 

205 rule = _selected_rule(module, module_bay) 

206 previous_forms = None if naming is None else naming.previous_forms() 

207 

208 if ( 

209 previous_forms is not None 

210 and previous_forms.rule is not None 

211 and family_ops.builds_flat_family(previous_forms.rule) 

212 ): 

213 yield from _kept_flat_family(module) 

214 return 

215 if not rule: 

216 if previous_forms is not None and previous_forms.rule is not None: 

217 yield from _left_without_a_rule(module, previous_forms) 

218 return 

219 # One pin for the module: the claim and the family planner resolve its templates once. 

220 with family_ops.pinned_template_cache(): 

221 yield from _apply_rule_to_module( 

222 rule, module, module_bay, force_reapply, report_only, previous_forms, naming_points 

223 ) 

224 

225 

226def _kept_flat_family(module) -> Iterator[RenameOutcome]: 

227 """Yield a blocked fact for every interface of *module*, which a flat breakout rule before the save leaves alone.""" 

228 from dcim.models import Interface 

229 

230 for name in Interface.objects.filter(module_id=module.pk).order_by("pk").values_list("name", flat=True): 

231 yield RenameOutcome(OutcomeKind.BLOCKED, name, FLAT_REASON) 

232 

233 

234def _left_without_a_rule(module, previous_forms) -> Iterator[RenameOutcome]: 

235 """Yield a blocked fact for each interface the previous rule named, now that no rule matches the module.""" 

236 from dcim.models import Interface 

237 

238 interfaces = list(Interface.objects.filter(module_id=module.pk).order_by("pk")) 

239 for name in family_ops.names_the_previous_rule_gave(previous_forms, interfaces): 

240 yield RenameOutcome(OutcomeKind.BLOCKED, name, NO_RULE_REASON) 

241 

242 

243def _has_stale_parent_bay(module_bay) -> bool: 

244 """Return whether *module_bay* belongs to a module but its parent bay does not hold that module. 

245 

246 NetBox sets a bay's parent to the bay of the module that owns it; rule selection read both already. 

247 """ 

248 if module_bay.module_id is None: 

249 return False 

250 installed = getattr(module_bay.parent, "installed_module", None) 

251 return installed is None or installed.pk != module_bay.module_id 

252 

253 

254def _acted_on_names(rule, plans): 

255 """Return the name of every interface the admitted *plans* act on, in plan order.""" 

256 names = [] 

257 for plan in plans: 

258 if isinstance(plan, family_ops.InstalledFamilyPlan): 

259 keeps_parent = plan.parent_pk is not None and not family_ops.names_installed_parent(rule) 

260 names.extend( 

261 member.snapshot.name 

262 for member in plan.members 

263 if not (keeps_parent and member.role == family_ops.MemberRole.PARENT) 

264 ) 

265 continue 

266 names.append(plan.base.name) 

267 if isinstance(plan, family_ops.FlatCreationPlan): 

268 names.extend(member.name for member in plan.members) 

269 return tuple(dict.fromkeys(names)) 

270 

271 

272def _apply_rule_to_module( 

273 rule, module, module_bay, force_reapply, report_only=False, previous_forms=None, naming_points=() 

274): 

275 """Plan and execute every family *rule* intends on *module*; see ``module_rule_outcomes``.""" 

276 from dcim.models import Interface 

277 

278 variables = build_variables(module_bay, device=module.device) 

279 missing = name_template.unavailable_variables(variables, rule.name_template, rule.parent_name_template) 

280 if report_only and not missing: 

281 return 

282 interfaces = list(Interface.objects.filter(module_id=module.pk).order_by("pk")) 

283 # NetBox before 4.7 moves only the module row: its interfaces and nested bays keep the old placement. 

284 if any(interface.device_id != module.device_id for interface in interfaces): 

285 yield from (RenameOutcome(OutcomeKind.BLOCKED, i.name, ELSEWHERE_REASON) for i in interfaces) 

286 return 

287 if _has_stale_parent_bay(module_bay): 

288 yield from (RenameOutcome(OutcomeKind.BLOCKED, i.name, STALE_BAY_REASON) for i in interfaces) 

289 return 

290 bases = family_ops.module_raw_bases( 

291 module, rule, variables, interfaces, previous_forms, earlier_raw_names=_raw_names_at(module, naming_points) 

292 ) 

293 planned = family_ops.plan_module_families( 

294 module, rule, variables, interfaces, bases, _run_scope(force_reapply, previous_forms) 

295 ) 

296 installed, leftover = planned.installed, planned.leftover 

297 plans = [*installed, *leftover] 

298 

299 if missing: 

300 yield from _unresolved_outcomes(missing, _acted_on_names(rule, plans)) 

301 return 

302 

303 any_outcome = False 

304 for family in family_ops.execute_module_families(plans): 

305 for outcome in _rename_outcomes((family,)): 

306 any_outcome = True 

307 yield outcome 

308 families_seen = bool(installed) or any(_touches_a_family(plan) for plan in leftover) 

309 

310 if not force_reapply and previous_forms is None and leftover and not any_outcome and not families_seen: 

311 # Nothing renamed, skipped or built as a family: NetBox may already give these names. 

312 _flag_rule_potentially_deprecated(rule) 

313 

314 

315def predict_rule_output(module, module_bay, raw_names): 

316 """Predict the names apply_interface_name_rules would produce for raw_names. 

317 

318 Read-only: saves and mutates nothing. Used by external integrations (e.g. 

319 netbox-librenms-plugin) that need to know the post-rename names without applying any rule. 

320 

321 The names are planned by the family module from the module type's templates, so prediction 

322 describes the same families installed execution builds: a breakout rule expands one plain name 

323 into the family it creates, a name the templates describe as part of a channelized family 

324 follows that family instead, and a family the apply path refuses to touch predicts unchanged. 

325 Returns raw_names unchanged when no rule matches. 

326 

327 Precondition: *raw_names* are resolved by the caller at call time. A name captured before the 

328 device's virtual-chassis position changed is predicted from itself, not corrected to the name 

329 the templates resolve to now — this function maps the names it is given. 

330 """ 

331 rule = _selected_rule(module, module_bay) 

332 if not rule: 

333 return list(raw_names) 

334 

335 variables = build_variables(module_bay, device=module.device) 

336 described = family_ops.describe_module_interfaces(module, raw_names) 

337 channel_names = {interface.name for interface in described if interface.channel_id is not None} 

338 base_names = [name for name in raw_names if name not in channel_names] 

339 plan_set = family_ops.plan_prospective_families( 

340 module, rule, variables, described, family_ops.given_raw_names(module, rule, variables, base_names) 

341 ) 

342 return [name for raw_name in raw_names for name in plan_set.predicted_names(raw_name)] 

343 

344 

345def reapply_module_rules(device): 

346 """Re-apply module rules to every module on *device* after its virtual-chassis position changed. 

347 

348 The whole device is one batch: its modules match against one enabled-rule snapshot, and one 

349 module type's interface templates are read once however many modules carry it. An unexpected 

350 failure stops the module batch and propagates to the deferred callback boundary. 

351 

352 Returns the number of interfaces renamed across the device's modules. 

353 """ 

354 return renamed_count(device_module_rule_outcomes(device)) 

355 

356 

357def device_module_rule_outcomes( 

358 device, report_only=False, excluded_pks=(), naming_points=() 

359) -> Iterator[RenameOutcome]: 

360 """Reapply the rules of every module on *device* as ``reapply_module_rules`` does, and yield the outcome facts. 

361 

362 Each module's facts are yielded before the next module runs, so a caller keeps them when a later 

363 module fails. *report_only* is passed to ``module_rule_outcomes``. The modules whose primary keys 

364 are in *excluded_pks* are left out. *naming_points* are passed to ``module_rule_outcomes``. 

365 """ 

366 from dcim.models import Module 

367 

368 modules = list( 

369 Module.objects.filter(device=device) 

370 .exclude(pk__in=excluded_pks) 

371 .select_related( 

372 "module_type", 

373 "device__device_type", 

374 "device__platform", 

375 *family_template_names.BAY_CHAIN_RELATIONS, 

376 ) 

377 ) 

378 with pinned_rule_cache(), family_ops.pinned_template_cache(modules): 

379 for module in modules: 

380 yield from module_rule_outcomes( 

381 module, 

382 module.module_bay, 

383 force_reapply=True, 

384 report_only=report_only, 

385 naming_points=naming_points, 

386 ) 

387 

388 

389_NAMING_RELATIONS = ( 

390 "module_type", 

391 "device__device_type", 

392 "device__platform", 

393 "module_bay__parent__installed_module__module_type", 

394 *family_template_names.BAY_CHAIN_RELATIONS, 

395) 

396 

397 

398@dataclass(frozen=True, eq=False) 

399class ModuleNaming: 

400 """What named one module's interfaces at the time it was read: a move, a bay edit or a type change reads it. 

401 

402 The module type and the scope select the rule that state gave the module. The template variables 

403 and the templates as they resolved then rebuild the names that rule gave. The variables come from 

404 ``bay_chain`` and ``vc_position``, the virtual-chassis position of the device then, and 

405 ``previous_forms`` resolves the raw template names at ``vc_position`` too. ``device_pk`` is that 

406 device. ``bay_values`` are the ``naming.bay_naming_values`` of the module's bay then. 

407 ``raw_only`` is set when no rule has named the module's interfaces yet, because the module was 

408 installed in the same transaction: they carry raw template names only. 

409 """ 

410 

411 module_pk: int 

412 device_pk: int 

413 module_type: object 

414 parent_module_type: object | None 

415 device_type: object | None 

416 platform: object | None 

417 bay_chain: naming.BayChain 

418 vc_position: int | None 

419 templates: tuple 

420 bay_values: tuple 

421 raw_only: bool = False 

422 

423 @classmethod 

424 def of(cls, module): 

425 """Return the naming of *module*, which carries ``_NAMING_RELATIONS``, as it is now.""" 

426 device = module.device 

427 module_bay = module.module_bay 

428 return cls( 

429 module_pk=module.pk, 

430 device_pk=device.pk, 

431 module_type=module.module_type, 

432 parent_module_type=_get_parent_module_type(module_bay), 

433 device_type=device.device_type, 

434 platform=device.platform, 

435 bay_chain=naming.bay_chain(module_bay), 

436 vc_position=naming.chassis_position(device), 

437 templates=family_ops.resolved_template_names(module), 

438 bay_values=naming.bay_naming_values(module_bay.position, module_bay.name), 

439 ) 

440 

441 @property 

442 def variables(self): 

443 """Return the template variables of this naming.""" 

444 return naming.bay_chain_variables(self.bay_chain, self.vc_position) 

445 

446 def at_chassis_position(self, vc_position): 

447 """Return this naming at the virtual-chassis position *vc_position*; None means off a chassis.""" 

448 return replace(self, vc_position=vc_position) 

449 

450 def rule(self): 

451 """Return the rule that the module type and the scope of this naming select now, or None.""" 

452 return find_matching_rule(self.module_type, self.parent_module_type, self.device_type, self.platform) 

453 

454 def selects_another_rule(self, module) -> bool: 

455 """Return whether *module* now selects another rule than this naming; it reads the rule cache. 

456 

457 A module that ``committed_modules`` returned needs no other query. 

458 """ 

459 return _selected_rule(module, module.module_bay) != self.rule() 

460 

461 def previous_forms(self) -> family_ops.PreviousForms: 

462 """Return what rebuilds the names this naming gave, under the rule it selects now; no rule when ``raw_only``.""" 

463 rule = None if self.raw_only else self.rule() 

464 templates = {template.pk: template.at_chassis_position(self.vc_position) for template in self.templates} 

465 return family_ops.PreviousForms(rule, self.variables, templates) 

466 

467 

468def read_subtree_naming(module_pk) -> tuple[ModuleNaming, ...]: 

469 """Return the naming of the module and of every module nested in it, the module itself first. 

470 

471 A move reads this before its save: in the same save NetBox can re-resolve the position and name 

472 of each bay the module holds, so the nested modules' naming is gone afterwards. A bay edit reads 

473 it before its save, because the save changes what the names in the bay are built from. A type 

474 change reads it before its save, because the nested modules can then select another rule. 

475 """ 

476 from dcim.models import Module 

477 

478 modules = [] 

479 level = list(Module.objects.filter(pk=module_pk).select_related(*_NAMING_RELATIONS)) 

480 while level: 

481 modules.extend(level) 

482 level = list( 

483 Module.objects.filter(module_bay__module__in=level) 

484 .exclude(pk__in=[module.pk for module in modules]) 

485 .select_related(*_NAMING_RELATIONS) 

486 .order_by("pk") 

487 ) 

488 with family_ops.pinned_template_cache(modules): 

489 return tuple(ModuleNaming.of(module) for module in modules) 

490 

491 

492def committed_modules(module_pks) -> dict: 

493 """Return the committed modules that *module_pks* name, by primary key, with every relation a reapply reads.""" 

494 from dcim.models import Module 

495 

496 return Module.objects.select_related(*_NAMING_RELATIONS).in_bulk(module_pks) 

497 

498 

499@contextlib.contextmanager 

500def pinned_reapply(modules): 

501 """Share one enabled-rule snapshot, and the resolved templates of *modules*, across one reapply.""" 

502 with pinned_rule_cache(), family_ops.pinned_template_cache(modules): 

503 yield 

504 

505 

506def _device_interface_rules(device): 

507 """Return enabled device-interface rules in matching priority order.""" 

508 from django.db.models import Q 

509 

510 from .models import InterfaceNameRule 

511 

512 device_type = getattr(device, "device_type", None) 

513 platform = getattr(device, "platform", None) 

514 rules = list( 

515 InterfaceNameRule.objects.filter( 

516 applies_to_device_interfaces=True, 

517 enabled=True, 

518 ) 

519 .filter(Q(device_type=device_type) | Q(device_type__isnull=True)) 

520 .filter(Q(platform=platform) | Q(platform__isnull=True)) 

521 ) 

522 # Sort Python-side: specificity_score descending, then module_type_pattern length 

523 # descending (for device-interface rules with ties), then pk ascending for stability. 

524 # (InterfaceNameRule has no DB 'priority' field; specificity_score is a property.) 

525 rules.sort( 

526 key=lambda r: ( 

527 -r.specificity_score, 

528 -len(r.module_type_pattern or ""), 

529 r.pk, 

530 ) 

531 ) 

532 return rules 

533 

534 

535def _matches_device_interface(rule, interface): 

536 """Return whether one device-interface rule matches one device interface.""" 

537 if not rule.module_type_pattern: 

538 return True 

539 try: 

540 compiled = compile_module_type_pattern(rule.module_type_pattern) 

541 except ValidationError: 

542 return False 

543 return compiled.fullmatch(interface.name) is not None 

544 

545 

546def _apply_device_rule_to_families(device, vc_position, rule, families, claimed_pks, report_only, by_family): 

547 """Apply one rule to each eligible device-interface family and record its outcome facts in *by_family*. 

548 

549 A rule that renames a family, or finds it correct, replaces the skips earlier rules left on it; 

550 a failure stays reported. 

551 """ 

552 for interface, children in families: 

553 if interface.pk in claimed_pks or not _matches_device_interface(rule, interface): 

554 continue 

555 variables = naming.build_device_interface_variables(interface.name, vc_position) 

556 missing = name_template.unavailable_variables(variables, rule.name_template) 

557 if missing: 

558 names = (interface.name, *(child.name for child in children)) 

559 by_family.setdefault(interface.pk, []).extend(_unresolved_outcomes(missing, names)) 

560 claimed_pks.update((interface.pk, *(child.pk for child in children))) 

561 continue 

562 if report_only: 

563 claimed_pks.update((interface.pk, *(child.pk for child in children))) 

564 continue 

565 plan = family_ops.plan_device_interface_rename(device, rule, variables, interface, children) 

566 outcome = family_ops.execute_installed_plan(plan) 

567 if outcome.status in {family_ops.FamilyStatus.CHANGED, family_ops.FamilyStatus.UNCHANGED}: 

568 claimed_pks.update(plan.member_pks) 

569 failures = [earlier for earlier in by_family.get(interface.pk, ()) if earlier.kind == OutcomeKind.FAILED] 

570 by_family[interface.pk] = [*failures, *_rename_outcomes((outcome,))] 

571 else: 

572 by_family.setdefault(interface.pk, []).extend(_rename_outcomes((outcome,))) 

573 

574 

575def apply_device_interface_rules(device): 

576 """Rename device-level interfaces (module=None) when a device joins/changes position in a VC. 

577 

578 Finds all enabled rules with ``applies_to_device_interfaces=True`` that match the device's 

579 type and platform, then renames any matching interfaces using the name_template. 

580 

581 Template variables available: ``{vc_position}``, ``{base}`` (full current name), 

582 ``{port}`` (segment after the last ``/``, or the full name if no ``/`` present). 

583 

584 Channel subinterfaces are not matched independently. They follow the parent whose family 

585 a rule wins, so a template like ``eth{vc_position}`` cannot collapse a whole family onto 

586 one name. 

587 

588 Returns the number of interfaces renamed. 

589 """ 

590 if not getattr(device, "virtual_chassis_id", None): 

591 return 0 # Only rename for VC members (vc_position must be set) 

592 

593 if device.vc_position is None: 

594 return 0 # vc_position unset (e.g. VC master before position assigned) 

595 

596 return renamed_count(device_interface_rule_outcomes(device)) 

597 

598 

599def device_interface_rule_outcomes(device, report_only=False) -> Iterator[RenameOutcome]: 

600 """Apply the device-interface rules and yield the outcome facts. 

601 

602 Unlike ``apply_device_interface_rules``, this also runs for a device without a virtual-chassis 

603 position: a rule that reads ``{vc_position}`` then renames nothing and reports each interface it 

604 matches as unresolved. With *report_only*, no rule renames anything. 

605 

606 The facts are yielded after the last rule, because a later rule can replace a family's facts. 

607 When a rule fails, the facts recorded before the failure are yielded before the error propagates. 

608 """ 

609 from dcim.models import Interface 

610 

611 vc_position = naming.chassis_position(device) 

612 rules = _device_interface_rules(device) 

613 if not rules: 

614 return 

615 

616 interfaces = list(Interface.objects.filter(device=device, module=None).order_by("pk")) 

617 if not interfaces: 

618 return 

619 

620 families = family_ops.device_interface_families(interfaces) 

621 claimed_pks: set[int] = set() 

622 by_family: dict[int, list] = {} 

623 try: 

624 for rule in rules: 

625 _apply_device_rule_to_families(device, vc_position, rule, families, claimed_pks, report_only, by_family) 

626 finally: 

627 # A caller that extends a list keeps these facts when a later family raises. 

628 for outcomes in by_family.values(): 

629 yield from outcomes 

630 

631 

632def _flag_rule_potentially_deprecated(rule): 

633 """Tag a rule as 'potentially-deprecated' when its rename is a no-op. 

634 

635 Called from apply_interface_name_rules when a matching rule produces no 

636 renames because NetBox already generates the correct interface names. This 

637 may indicate the rule is no longer needed (e.g. after a NetBox upgrade that 

638 improved template resolution), or only needed for a subset of module types. 

639 

640 Adds a NetBox Tag 'potentially-deprecated' so the rule is visually flagged 

641 in the UI for operator review. A database failure is logged but never 

642 re-raised so the install path is not disrupted. 

643 """ 

644 from extras.models import Tag 

645 

646 from .models import InterfaceNameRule 

647 

648 try: 

649 with atomic_with_events(): 

650 # The caller read the rule before the reapply, so the stored row can be newer. 

651 stored = InterfaceNameRule.objects.select_for_update().filter(pk=rule.pk).order_by("pk").first() 

652 if stored is None: 

653 return 

654 tag, _ = Tag.objects.get_or_create( 

655 slug="potentially-deprecated", 

656 defaults={"name": "potentially-deprecated", "color": "ffc107"}, 

657 ) 

658 stored.snapshot() 

659 stored.tags.add(tag) 

660 except DatabaseError: 

661 logger.exception("Failed to flag rule '%s' as potentially-deprecated.", rule) 

662 return 

663 logger.info( 

664 "Rule '%s' flagged as potentially-deprecated: NetBox already generates the correct interface names.", 

665 stored, 

666 ) 

667 

668 

669def _matching_moduletype_pks(module_type_pattern): 

670 """Return PKs of ModuleTypes whose model name matches the given RE2 pattern. 

671 

672 Raises ValueError for invalid patterns, mirroring evaluate_name_template's 

673 error-handling convention so callers can treat both as ValueError. 

674 """ 

675 from dcim.models import ModuleType 

676 

677 try: 

678 compiled = compile_module_type_pattern(module_type_pattern) 

679 except ValidationError as exc: 

680 raise ValueError(exc.messages[0]) from exc 

681 return [mt.pk for mt in ModuleType.objects.only("pk", "model") if compiled.fullmatch(mt.model)] 

682 

683 

684def has_applicable_interfaces(rule) -> bool: 

685 """Check whether applying this rule right now would rename at least one interface. 

686 

687 Calls find_interfaces_for_rule(limit=1) to determine if any currently installed 

688 interface would receive a new name. Returns False when: 

689 - no matching modules/interfaces are installed, OR 

690 - all matching interfaces are already correctly named. 

691 

692 This is more expensive than a plain EXISTS query but ensures the Applicable 

693 column in the Apply Rules list accurately reflects "would something change?" 

694 rather than the misleading "do interfaces exist?". 

695 """ 

696 try: 

697 results, _ = find_interfaces_for_rule(rule, limit=1) 

698 return len(results) > 0 

699 except ValueError: 

700 return False 

701 

702 

703def _build_module_qs(rule): 

704 """Return a Module queryset filtered to the rule's scope (module type, parent, device, platform). 

705 

706 Shared by ``find_interfaces_for_rule`` and ``apply_rule_to_existing`` to avoid 

707 duplicating the filtering logic. 

708 """ 

709 from dcim.models import Module 

710 

711 if rule.module_type_is_regex: 

712 qs = Module.objects.filter(module_type__in=_matching_moduletype_pks(rule.module_type_pattern)) 

713 else: 

714 qs = Module.objects.filter(module_type=rule.module_type) 

715 if rule.parent_module_type: 

716 qs = qs.filter(module_bay__parent__installed_module__module_type=rule.parent_module_type) 

717 if rule.device_type: 

718 qs = qs.filter(device__device_type=rule.device_type) 

719 if rule.platform: 

720 qs = qs.filter(device__platform=rule.platform) 

721 return qs 

722 

723 

724_PREVIEW_ROLES = { 

725 family_ops.MemberRole.PARENT: "parent", 

726 family_ops.MemberRole.CHANNEL: "channel", 

727} 

728 

729 

730def _member_detail(plan, member) -> family_ops.PlannedName: 

731 """Describe one planned member so the UI can render a family as a family. 

732 

733 A flat family's members are the channels a breakout rule spells out; a plan that holds only 

734 one of them is a plain rename, not a family. 

735 """ 

736 role = _PREVIEW_ROLES.get(member.role) or ("channel" if len(plan.members) > 1 else "interface") 

737 return family_ops.PlannedName(member.target_name, role, member.channel_id) 

738 

739 

740def _plan_details(plan) -> list: 

741 """Describe every name the plan intends, or the error that stopped it from naming them.""" 

742 if plan.precondition_status != family_ops.FamilyStatus.FAILED: 

743 return [_member_detail(plan, member) for member in plan.members] 

744 root = _member_detail(plan, plan.members[0]) 

745 return [family_ops.PlannedName(f"<error: {plan.precondition_reason}>", root.role, root.channel_id)] 

746 

747 

748def _plan_changes_names(plan, existing_names) -> bool: 

749 """Return whether the planned family would rename, create or restructure anything. 

750 

751 A plan that renames members compares intent with the names they carry now. A plan that builds 

752 a channelized family always channelizes its base, also when it only takes over rows that carry 

753 the names already (ADR 0017). A plan that builds a flat family compares intent with the names 

754 the module already holds, so a family an earlier apply already installed previews as no change. 

755 """ 

756 if plan.base_name is None: # pragma: no cover - requires channelization support 

757 return plan.target_names != plan.source_names 

758 if plan.topology == family_ops.FamilyTopology.CHANNELIZED: 

759 return True 

760 return plan.target_names[0] != plan.base_name or any( 

761 target_name not in existing_names for target_name in plan.target_names 

762 ) 

763 

764 

765def _plan_entry(module, plan, interface, existing_names) -> dict | None: 

766 """Build the preview entry for one family plan, or None when it would change nothing. 

767 

768 The entry stays keyed on the interface the Apply view submits, and lists the family's names in 

769 ``new_names``, so the existing template loop keeps working unchanged. A plan the live topology 

770 blocks previews nothing, because the apply path would build nothing either. 

771 """ 

772 failed = plan.precondition_status == family_ops.FamilyStatus.FAILED 

773 if plan.precondition_status is not None and not failed: 

774 return None 

775 if not failed and not _plan_changes_names(plan, existing_names): 

776 return None 

777 details = _plan_details(plan) 

778 return { 

779 "module": module, 

780 "interface": interface, 

781 "current_name": interface.name, 

782 "new_names": [detail.name for detail in details], 

783 "name_details": details, 

784 } 

785 

786 

787def _plan_root_name(plan) -> str: 

788 """Return the name of the interface a plan is submitted through.""" 

789 return plan.base_name if plan.base_name is not None else plan.members[0].source_name 

790 

791 

792def _preview_plans(rule, plan_set) -> list: 

793 """Return the plans this preview reports. 

794 

795 A breakout rule on a module that already models channelized families renames those families and 

796 adds none beside them. Anywhere else it builds one family per base, and two bases that intend 

797 the same names are the one family an earlier apply already started, so it is offered once: the 

798 same family the apply path would build. 

799 """ 

800 if rule.channel_count <= 0: 

801 return list(plan_set.plans) 

802 installed = [plan for plan in plan_set.plans if plan.base_name is None] 

803 if installed: # pragma: no cover - requires a NetBox that models channelization 

804 return installed 

805 creations = [plan for plan in plan_set.plans if plan.base_name is not None] 

806 kept = family_ops.one_family_per_name_set([(plan.base_name, plan.target_names) for plan in creations]) 

807 return [creations[index] for index in kept] 

808 

809 

810def _installed_flat_entry(module, plan, interface) -> dict | None: 

811 """Build the preview entry for an installed flat family, or None when it keeps its names.""" 

812 if all(member.target_name == member.snapshot.name for member in plan.members): 

813 return None 

814 role = "channel" if len(plan.members) > 1 else "interface" 

815 details = [family_ops.PlannedName(member.target_name, role) for member in plan.members] 

816 return { 

817 "module": module, 

818 "interface": interface, 

819 "current_name": interface.name, 

820 "new_names": [detail.name for detail in details], 

821 "name_details": details, 

822 } 

823 

824 

825def _process_module(rule, module, ifaces, variables, limit, results, module_qs, processed_pks): 

826 """Preview one module from its family plans. Returns (checked_count, should_stop). 

827 

828 An installed flat family is previewed from the plan Apply Rules executes, because its members 

829 carry names no template describes. 

830 """ 

831 bases = family_ops.module_raw_bases(module, rule, variables, ifaces) 

832 installed = family_ops.plan_installed_flat_families(module, rule, variables, ifaces, bases)