« prev ^ index » next coverage.py v7.15.4, created at 2026-08-22 21:16 +0000
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.
5This module is imported lazily by signals.py so that model imports happen
6after Django is fully initialised.
7"""
9import ast
10import contextlib
11import copy
12import logging
13import re
14import threading
15from collections import defaultdict, namedtuple
17from django.core.exceptions import ValidationError
18from django.db import IntegrityError, transaction
19from django.db.models import Aggregate, F, TextField, Value
20from django.db.models.functions import Cast, Coalesce, Concat, Length
22from .choices import BreakoutModeChoices
24logger = logging.getLogger(__name__)
26# In-process cache of the enabled InterfaceNameRule set, keyed by a cheap fingerprint
27# of that set. find_matching_rule() is called once per module row when a device's
28# module-sync table is rendered, and each call used to run a DB query per scope
29# candidate per tier; loading the (small) rule set once and matching in memory removes
30# that per-row query storm. The fingerprint (see _enabled_rules_version) is re-read once
31# per call and the set reloaded only when it changes, so it self-invalidates on any
32# create/delete/edit — including raw bulk ``.update()`` of any matching field and SET_NULL
33# cascades that bypass auto_now, and across test transactions — with no signal wiring.
34# A reload publishes the new set by rebinding this name to a fresh dict in one atomic assignment
35# (see _get_enabled_rules), so a concurrent reader on another worker thread sees either the whole
36# old set or the whole new one — never exact/regex/memo torn across two versions.
37_RULE_CACHE = {"version": None, "exact": (), "regex": (), "memo": {}}
39# Per-version cap on the find_matching_rule memo. A long-lived worker that sees many distinct
40# (module_type, scope) contexts under a stable rule set would otherwise grow it without bound;
41# on overflow the memo is cleared wholesale and rebuilt lazily. The cap is far above the number
42# of contexts in any single module-sync render, so it never churns mid-render.
43_MEMO_MAX = 4096
45# Sentinel for the memo read. A single ``memo.get(sig, _MEMO_MISS)`` is one atomic dict lookup, so a
46# concurrent ``memo.clear()`` at the cap can't wedge between a membership test and the subscript the
47# way ``if sig in memo: return memo[sig]`` could — that compound read can raise a sporadic KeyError
48# under threaded workers sharing one per-version memo. A real "no rule matched" is memoized as None,
49# so the miss sentinel must be a distinct object, not None.
50_MEMO_MISS = object()
52# Per-thread pin depth (see pinned_rule_cache). While > 0 on the current thread, _get_enabled_rules
53# trusts the loaded set without re-reading the fingerprint, so an internal batch that wraps its loop
54# turns N per-call fingerprint queries into one. Thread-local so concurrent requests don't affect
55# each other; the depth defaults to 0, so any path that does not pin (the per-row signal handler,
56# single applies) re-reads the fingerprint on every call exactly as before.
57_pin = threading.local()
60@contextlib.contextmanager
61def pinned_rule_cache():
62 """Pin the enabled-rule set for the duration of the block, skipping the per-call fingerprint query.
64 find_matching_rule() normally re-reads a cheap fingerprint on every call to detect rule-set
65 changes. An internal batch that makes many lookups against an unchanging rule set — e.g.
66 _apply_rules_for_device_deferred(), which re-applies rules to every module on a device after a
67 virtual-chassis change — can wrap its loop in this context manager so the set is loaded and
68 fingerprinted once and reused for every call inside the block, turning N fingerprint queries
69 into one::
71 with pinned_rule_cache():
72 for module in modules:
73 apply_interface_name_rules(module, module.module_bay, force_reapply=True)
75 This is an INR-internal helper: only code that owns the batch loop pins, so it never becomes a
76 cross-plugin dependency. Other plugins integrate through the ``predict_module_interface_names``
77 signal, which is dispatched one row at a time — so an external caller neither can nor needs to
78 pin, and INR exposes no batch API across the plugin boundary.
80 Scope is explicit and per-thread: outside the block (and in any other thread/request) the normal
81 self-invalidating behaviour is unchanged, so there is no global staleness window. Nesting is
82 safe — only the outermost block manages the pin. Priming is lazy: the first find_matching_rule
83 inside the block does the one real fingerprint read + reload and captures that rule set into
84 thread-local state; the rest reuse that *snapshot* — so a block that makes no lookups does no
85 query at all, and a concurrent request reloading the shared cache on another thread cannot switch
86 this batch's rule set mid-loop. The loop may rename interfaces, but it must not edit rules: edits
87 to InterfaceNameRule made *inside* the block are not observed until it exits.
88 """
89 depth = getattr(_pin, "depth", 0)
90 _pin.depth = depth + 1
91 if depth == 0: 91 ↛ 93line 91 didn't jump to line 93 because the condition on line 91 was always true
92 _pin.primed = False # the first lookup inside primes the cache (lazily — an empty block stays query-free)
93 try:
94 yield
95 finally:
96 _pin.depth -= 1
97 if _pin.depth == 0: 97 ↛ exitline 97 didn't return from function 'pinned_rule_cache' because the condition on line 97 was always true
98 # Release the per-thread snapshot so the next block re-primes from the live cache.
99 _pin.primed = False
100 for attr in ("exact", "regex", "memo"):
101 _pin.__dict__.pop(attr, None)
104def _compile_pattern(pattern):
105 """Compile a regex *pattern* once, returning the compiled object or None for an invalid pattern.
107 A None result is skipped at match time, mirroring the previous per-match ``try/except re.error``
108 without recompiling the pattern on every lookup.
109 """
110 try:
111 return re.compile(pattern)
112 except re.error:
113 return None
116# Rule columns that affect matching or output, in a fixed order. The enabled-set fingerprint
117# (see _enabled_rules_version) is an md5 over these for every enabled rule, so it changes on ANY
118# edit that could change a lookup result. ``description`` is excluded — it is operator notes that
119# never affect matching. ``id`` anchors each row to its identity, so a compensating swap between
120# two rules (which leaves count + column sums unchanged) still changes the hash.
121_VERSION_COLUMNS = (
122 "id",
123 "module_type_id",
124 "module_type_is_regex",
125 "module_type_pattern",
126 "parent_module_type_id",
127 "device_type_id",
128 "platform_id",
129 "name_template",
130 "parent_name_template",
131 "breakout_mode",
132 "channel_count",
133 "channel_start",
134 "applies_to_device_interfaces",
135)
138class _Md5OrderedStringAgg(Aggregate):
139 """``md5(string_agg(<row>, <delim> ORDER BY id))`` expressed as one ORM aggregate.
141 Built through the ORM rather than a hand-formatted SQL string, so the table/column identifiers are
142 quoted by Django's compiler and the delimiter is a bound parameter — there is no string-interpolated
143 SQL to audit for injection. The template is ours, so it does not depend on ``StringAgg``'s Python
144 signature, which differs across the Django 5.x–6.x versions in CI. ``ORDER BY id`` makes string_agg
145 deterministic — a stable row order yields a stable hash for an unchanged set.
146 """
148 function = "STRING_AGG"
149 template = "MD5(%(function)s(%(expressions)s ORDER BY id))"
150 output_field = TextField()
153def _version_row_signature():
154 """Build the per-rule text signature expression for the fingerprint.
156 Each matching/output column is cast to text and emitted length-prefixed as ``<char length>:<value>``.
157 That makes the row encoding self-delimiting: distinct column tuples can never serialize to the same
158 string, even if a text column (name_template / module_type_pattern) contains digits, a colon, or the
159 control characters a plain separator scheme would rely on being absent. A nullable FK is coalesced to
160 '' (length 0), so null stays distinct from any value while keeping the column's slot.
161 """
162 empty = Value("", output_field=TextField())
163 colon = Value(":", output_field=TextField())
164 parts = []
165 for column in _VERSION_COLUMNS:
166 cast = Cast(F(column), output_field=TextField())
167 value = Coalesce(cast, empty, output_field=TextField()) if column.endswith("_id") else cast
168 parts.append(Cast(Length(value), output_field=TextField()))
169 parts.append(colon)
170 parts.append(value)
171 return Concat(*parts, output_field=TextField())
174# The signature expression is constant, so build it once and reuse it across calls.
175_ROW_SIGNATURE = _version_row_signature()
178def _enabled_rules_version():
179 """Return a deterministic content fingerprint of the enabled-rule set.
181 Computed server-side as an md5 over every enabled rule's matching/output columns, row-ordered
182 by pk, so the fingerprint changes on ANY edit that could change a lookup — including a raw bulk
183 ``.update()`` of a text field (name_template / module_type_pattern) or a boolean
184 (module_type_is_regex), which an aggregate of counts/sums cannot see, and a compensating edit
185 that keeps the column sums constant. It is one query returning a single 32-char hash, so it
186 stays cheap enough to re-read on every call.
188 Each column is length-prefixed (see _version_row_signature), so the per-row encoding is
189 self-delimiting and rows concatenate unambiguously — distinct rule sets cannot collide even if a
190 text column contains arbitrary bytes. A nullable FK renders as the empty string (a real id never
191 does), keeping null distinct from any value. The empty set aggregates to NULL → coalesced to a
192 stable empty fingerprint, so "no enabled rules" is fixed.
193 """
194 from .models import InterfaceNameRule
196 return InterfaceNameRule.objects.filter(enabled=True).aggregate(
197 fingerprint=Coalesce(
198 _Md5OrderedStringAgg(_ROW_SIGNATURE, Value("", output_field=TextField())),
199 Value("", output_field=TextField()),
200 )
201 )["fingerprint"]
204def _get_enabled_rules():
205 """Return ``(exact_rules, regex_rules, memo)``, reloading only when the rule set changes.
207 ``exact_rules`` are ordered by ``(module_type__model, pk)`` to mirror the model's default
208 ordering (so an in-memory ``first match`` equals the previous ``.first()``). ``regex_rules``
209 is a tuple of ``(compiled_pattern, rule)`` pairs, pre-sorted once by ``(-pattern length, pk)``
210 and with each pattern compiled once, so the regex tier neither re-sorts nor recompiles per
211 call. ``memo`` caches find_matching_rule results for the current rule-set version.
213 Inside a ``pinned_rule_cache()`` block on this thread, once the set has been primed (by the first
214 lookup in the block) the fingerprint query is skipped and the snapshot captured at prime time is
215 returned — never the live ``_RULE_CACHE``, which another thread may reload mid-block.
216 """
217 global _RULE_CACHE
219 pinned = getattr(_pin, "depth", 0) > 0
220 if pinned and getattr(_pin, "primed", False):
221 # Serve the snapshot captured when this block primed, not the shared cache: a concurrent
222 # request on another thread may reload _RULE_CACHE to a different version while we iterate, and
223 # a pinned batch must match every item against one consistent rule set.
224 return _pin.exact, _pin.regex, _pin.memo
226 from .models import InterfaceNameRule
228 # Read the module global exactly once. A reload below publishes the new set by rebinding
229 # _RULE_CACHE to a brand-new dict (a single atomic name assignment) rather than mutating this
230 # one in place, so this local is a consistent snapshot: exact/regex/memo can never be torn
231 # across two rule-set versions even if another thread reloads between the reads at the end.
232 cache = _RULE_CACHE
233 version = _enabled_rules_version()
234 if cache["version"] != version:
235 rules = list(InterfaceNameRule.objects.filter(enabled=True).order_by("module_type__model", "pk"))
236 exact = tuple(r for r in rules if not r.module_type_is_regex)
237 regex_rules = sorted(
238 (r for r in rules if r.module_type_is_regex),
239 key=lambda r: (-len(r.module_type_pattern or ""), r.pk),
240 )
241 regex = tuple((_compile_pattern(r.module_type_pattern), r) for r in regex_rules)
242 # Publish the whole new version atomically: a reader either sees the old dict or this one,
243 # never a mix of the two. Last writer wins; a concurrent reload to the same version just
244 # rebuilds redundantly, never corrupts.
245 cache = {"version": version, "exact": exact, "regex": regex, "memo": {}}
246 _RULE_CACHE = cache
247 if pinned:
248 # Pin this thread to the freshly-resolved set for the rest of the block. The tuples are
249 # immutable and the memo is COPIED into thread-local state — not aliased — so the pinned batch
250 # neither shares nor races the global memo: another thread clearing the shared memo at the cap
251 # can't evict our warmed entries (or wedge a KeyError) mid-loop, and entries we add stay private.
252 _pin.exact = cache["exact"]
253 _pin.regex = cache["regex"]
254 _pin.memo = dict(cache["memo"])
255 _pin.primed = True
256 return _pin.exact, _pin.regex, _pin.memo
257 return cache["exact"], cache["regex"], cache["memo"]
260def _get_parent_module_type(module_bay):
261 """Return the module type of the module installed in the parent bay, or None.
263 Used by ``apply_interface_name_rules`` to scope rules to a specific parent
264 module type (e.g., SFP inside a CVR-X2-SFP converter).
265 """
266 if module_bay.parent:
267 parent_bay = module_bay.parent
268 if hasattr(parent_bay, "installed_module") and parent_bay.installed_module: 268 ↛ 270line 268 didn't jump to line 270 because the condition on line 268 was always true
269 return parent_bay.installed_module.module_type
270 return None
273def supports_channelization():
274 """Return True when this NetBox models channelized subinterfaces (NetBox 4.7+).
276 Probed from the Interface model rather than a version comparison, so a backport or a
277 development build is detected by what it actually provides.
278 """
279 from dcim.models import Interface
280 from django.core.exceptions import FieldDoesNotExist
282 try:
283 Interface._meta.get_field("channel_id")
284 except FieldDoesNotExist:
285 return False
286 return True # pragma: no cover - only reachable on a NetBox that models channelization
289def _vc_position_re():
290 """Return NetBox's ``{vc_position}`` template-token regex, or None on a release without the token.
292 Imported inside the function so the probe reads the module as it stands at call time, matching
293 ``supports_channelization()``'s style.
294 """
295 try:
296 from dcim.constants import VC_POSITION_RE
297 except ImportError:
298 return None
299 return VC_POSITION_RE # pragma: no cover - only reachable on a NetBox that resolves the token
302def supports_vc_position_token():
303 """Return True when this NetBox resolves ``{vc_position}`` in component template names (4.6+).
305 Probed from the constant that carries the token rather than a version comparison, so a backport
306 or an upstream removal is detected by what NetBox actually provides.
307 """
308 return _vc_position_re() is not None
311def _is_channel_child(iface):
312 """Return True when *iface* is a channel subinterface bound to a parent's channel.
314 Structural, not name-based: a row imported without ``full_clean()`` may carry a ``channel_id``
315 without the channel type. ``channel_id`` does not exist before NetBox 4.7, so this is False
316 on every older release and the family paths below stay dormant there.
317 """
318 return getattr(iface, "channel_id", None) is not None
321def _is_channelized_parent(iface):
322 """Return True when *iface* declares a channel count.
324 A channelized parent owns a family even when no subinterface is bound yet — this, not the
325 presence of children, is what disables flat channel creation.
326 """
327 return getattr(iface, "channels", None) is not None
330def _partition_families(interfaces):
331 """Split *interfaces* into ``(bases, children_by_parent_pk)``.
333 Bases are the interfaces a rule may match on its own: standalone interfaces and channelized
334 parents. Channel subinterfaces are never independent candidates — they are renamed only by
335 following their parent, so they are grouped under it instead.
336 """
337 bases = []
338 children = defaultdict(list)
339 for iface in interfaces:
340 if not _is_channel_child(iface):
341 bases.append(iface)
342 continue
343 children[iface.parent_id].append(iface) # pragma: no cover - requires channelization support
344 for group in children.values(): # pragma: no cover - no channel children exist without support
345 group.sort(key=lambda child: child.channel_id)
346 return bases, children
349def _child_name_suffix(child_name, parent_name): # pragma: no cover - requires channelization support
350 """Return the suffix *child_name* adds to *parent_name*, or None when it adds none.
352 The first character must be non-alphanumeric so ``et0``/``et01`` is never mistaken for a
353 family; the punctuation itself is free-form (``:``, ``-``, ``_`` and ``@`` all occur in the
354 wild), so it is not restricted to a fixed separator.
355 """
356 if not parent_name or not child_name.startswith(parent_name):
357 return None
358 suffix = child_name[len(parent_name) :]
359 if not suffix or suffix[0].isalnum():
360 return None
361 return suffix
364def _unambiguous_claims(candidates, matchers, module): # pragma: no cover - requires vc_position token support
365 """Return the labels of *candidates* that exactly one drifted ``{vc_position}`` template claims.
367 *candidates* pairs a label with the name forms it is compared under. Both sides of the claim
368 have to be unique: a template matching two labels, or a label matched by two templates,
369 disqualifies everything involved with a warning rather than renaming a guess.
370 """
371 claims = defaultdict(list)
372 claimants = defaultdict(list)
373 for index, matcher in enumerate(matchers):
374 for label, forms in candidates:
375 if any(matcher.pattern.fullmatch(form) for form in forms):
376 claims[index].append(label)
377 claimants[label].append(index)
379 ambiguous = {index for index, claimed in claims.items() if len(claimed) > 1}
380 for index in sorted(ambiguous):
381 logger.warning(
382 "Interface template %r of %s could name any of %s since this device's virtual-chassis "
383 "position changed; skipping them all rather than renaming a guess.",
384 matchers[index].template_name,
385 module,
386 sorted(claims[index]),
387 )
388 for label, indexes in claimants.items():
389 if len(indexes) > 1:
390 logger.warning(
391 "Interface %r on %s could be the drifted name of any of the templates %s; "
392 "skipping it rather than renaming a guess.",
393 label,
394 module,
395 sorted(matchers[index].template_name for index in indexes),
396 )
397 ambiguous.update(indexes)
398 return [claims[index][0] for index in sorted(claims) if index not in ambiguous]
401def _drifted_candidates(interfaces, matchers, module): # pragma: no cover - requires vc_position token support
402 """Return the interfaces a single drifted ``{vc_position}`` template unambiguously claims.
404 *interfaces* and *matchers* are what the exact pass left unclaimed.
405 """
406 by_name = {iface.name: iface for iface in interfaces}
407 claimed = _unambiguous_claims([(iface.name, (iface.name,)) for iface in interfaces], matchers, module)
408 return [by_name[label] for label in claimed]
411def _forced_channel_bases(interfaces, raw_names, matchers, module):
412 """Return one interface per base a forced breakout rule should process, preferring the ":0" one.
414 A base is claimed exactly when either comparison form — the full base or its last path segment,
415 the latter covering already-renamed bases — is a raw name now; otherwise a token template's
416 matcher may claim it, under the same one-to-one policy ``_drifted_candidates`` applies. Two
417 bases with distinct rule outputs never collide downstream, so an ambiguous claim has to be
418 stopped here or it is not stopped at all.
419 """
420 seen_bases: dict = {}
421 forms_by_base: dict = {}
422 for i in interfaces:
423 # A channelized parent is its own base: its channels are separate rows, so the name needs no
424 # ":"-splitting to find them.
425 base = i.name if _is_channelized_parent(i) else i.name.rsplit(":", 1)[0]
426 forms = (base, base.rsplit("/", 1)[-1])
427 if not any(form in raw_names for form in forms) and not any( 427 ↛ 430line 427 didn't jump to line 430 because the condition on line 427 was never true
428 matcher.pattern.fullmatch(form) for matcher in matchers for form in forms
429 ):
430 continue
431 forms_by_base[base] = forms
432 if base not in seen_bases or i.name.endswith(":0"):
433 seen_bases[base] = i
435 exact_forms = {form for forms in forms_by_base.values() for form in forms if form in raw_names}
436 drifted = {base: forms for base, forms in forms_by_base.items() if not exact_forms & set(forms)}
437 if not drifted:
438 return list(seen_bases.values())
439 kept = set( # pragma: no cover - requires vc_position token support
440 _unambiguous_claims(drifted.items(), [m for m in matchers if m.resolved not in exact_forms], module)
441 )
442 return [i for base, i in seen_bases.items() if base not in drifted or base in kept] # pragma: no cover - see above
445def _collect_unrenamed(interfaces, rule, raw_names, force_reapply, matchers=(), module=None):
446 """Return the subset of *interfaces* that should be processed by the rule.
448 Normal (non-force) mode: only interfaces whose current name is still in the
449 raw template names (idempotency guard), plus the ones a *matchers* entry
450 claims as its own drifted name (see ``_drifted_candidates``).
452 force_reapply, non-channel: all interfaces (e.g. vc_position changed).
454 force_reapply, channel rule: one interface per base name (see
455 ``_forced_channel_bases``).
456 """
457 if not force_reapply:
458 exact = [i for i in interfaces if i.name in raw_names]
459 if not matchers:
460 return exact
461 claimed = {i.name for i in exact} # pragma: no cover - requires vc_position token support
462 return exact + _drifted_candidates( # pragma: no cover - see above
463 [i for i in interfaces if i.name not in claimed],
464 [m for m in matchers if m.resolved not in claimed],
465 module,
466 )
467 if rule.channel_count == 0:
468 return interfaces
469 return _forced_channel_bases(interfaces, raw_names, matchers, module)
472def apply_interface_name_rules(module, module_bay, force_reapply=False):
473 """Apply InterfaceNameRule rename after module installation.
475 Looks up a matching rule for (module_type, parent_module_type, device_type, platform)
476 and renames interfaces created by NetBox's template instantiation.
478 Only processes interfaces whose name still matches the raw bay position
479 (i.e., haven't been renamed yet), ensuring idempotency. Pass
480 ``force_reapply=True`` to skip this check and re-apply rules to ALL
481 module interfaces (used when vc_position or other variables change).
483 A channelized parent and its channel subinterfaces are processed as one family: the parent
484 decides, the children follow it (see ``_apply_rule_with_family``).
486 Returns:
487 Number of interfaces renamed/created, or 0 if no rule matched.
489 """
490 from dcim.models import Interface
492 device_type = module.device.device_type if module.device else None
493 platform = module.device.platform if module.device else None
494 rule = find_matching_rule(module.module_type, _get_parent_module_type(module_bay), device_type, platform)
496 if not rule:
497 return 0
499 variables = build_variables(module_bay, device=module.device)
500 interfaces = list(Interface.objects.filter(module=module))
502 if not interfaces:
503 return 0
505 # Only bases are rule candidates; the idempotency guard therefore looks at them alone.
506 bases, children_by_parent = _partition_families(interfaces)
507 # Determine raw names NetBox assigned from templates; fall back to bay_position.
508 raw = _raw_name_matchers(module)
509 raw_names = raw.names or {variables["bay_position"]}
510 unrenamed = _collect_unrenamed(bases, rule, raw_names, force_reapply, raw.matchers, module)
512 if not unrenamed:
513 return 0 # Already renamed (idempotent guard)
515 # A breakout rule on a module that has channelized families processes only those families —
516 # the same rule the preview and bulk-apply paths follow.
517 families_only = rule.channel_count > 0 and any(_is_channelized_parent(base) for base in bases)
519 renamed = 0
520 families_seen = families_only
521 conflicts: list = []
522 for iface in unrenamed:
523 children = children_by_parent.get(iface.pk, ())
524 if families_only and not _is_channelized_parent(iface): # pragma: no cover - see families_only above
525 logger.debug(
526 "Interface %r is not channelized; skipping it while rule '%s' breaks out this module's families.",
527 iface.name,
528 rule,
529 )
530 continue
531 families_seen = families_seen or bool(children) or _is_channelized_parent(iface)
532 try:
533 count = _apply_rule_with_family(rule, iface, children, variables, module, conflicts)
534 except (ValueError, ValidationError, IntegrityError):
535 # The collision pre-check closes the common case, but a concurrent
536 # insert can still win between that check and the save — surfacing
537 # here as IntegrityError/ValidationError out of the per-interface
538 # atomic block (which has already rolled back cleanly). Log and keep
539 # going so one racing interface never aborts the whole install batch,
540 # mirroring apply_rule_to_existing().
541 logger.exception(
542 "Failed to apply rule '%s' to interface '%s' (id=%s); skipping.",
543 rule,
544 iface.name,
545 iface.pk,
546 )
547 continue
548 if count is None:
549 # A structural skip (unsupported topology, channel-count mismatch) says nothing about the rule.
550 families_seen = True
551 continue
552 renamed += count
554 if unrenamed and renamed == 0 and not conflicts and not families_seen:
555 # All interfaces already have the names the rule would produce — flag as
556 # potentially obsolete (e.g., newer NetBox generates correct names natively).
557 # Skipped when the 0-count was caused by name collisions (a different reason
558 # than a no-op rule), so a collision never mislabels the rule as deprecated.
559 # Skipped for channelized families too: a structural skip, or a family whose parent
560 # deliberately keeps its raw name, says nothing about the rule being obsolete.
561 _flag_rule_potentially_deprecated(rule)
563 return renamed
566def _predicted_channel_name(rule, raw_name, variables, parents, children): # pragma: no cover - channelized only
567 """Return the name the channel template named *raw_name* takes under *rule*."""
568 parent_name, channel_id = children[raw_name]
569 if rule.channel_count > 0:
570 if parents.get(parent_name) != rule.channel_count:
571 return raw_name # channel-count mismatch: the apply path skips the whole family
572 channel = str(rule.channel_start + channel_id - 1)
573 return evaluate_name_template(rule.name_template, {**variables, "base": parent_name, "channel": channel})
574 # Simple rule: the channel follows its parent, keeping the suffix it adds to the parent's name.
575 parent_target = evaluate_name_template(rule.name_template, {**variables, "base": parent_name})
576 suffix = _child_name_suffix(raw_name, parent_name)
577 return raw_name if suffix is None else parent_target + suffix
580def _predicted_family_parent_name(rule, raw_name, variables, parents): # pragma: no cover - channelized only
581 """Return the name the parent template named *raw_name* takes under *rule*.
583 Only a channelized rule that names its parent renames it, and only when the family's channel
584 count is the one the rule describes — the same two conditions the apply path applies.
585 """
586 if not (_is_channelized_rule(rule) and rule.parent_name_template):
587 return raw_name
588 if parents.get(raw_name) != rule.channel_count:
589 return raw_name # channel-count mismatch: the apply path skips the whole family
590 return evaluate_name_template(rule.parent_name_template, {**variables, "base": raw_name})
593def _predicted_names(rule, raw_name, variables, parents, children, family_blocked=False):
594 """Return the names *raw_name* predicts to under *rule*.
596 A name the module type's templates describe as a channelized parent or channel follows its
597 family; a channelized rule on a plain name predicts the family it would build there, unless
598 *family_blocked* says the apply path refuses to build it; anything else keeps the per-name
599 prediction, expanding once per channel for a breakout rule and once for a simple one.
600 """
601 if raw_name in parents: # pragma: no cover - requires a NetBox that models channelization
602 if rule.channel_count > 0:
603 # The rule renames the family's existing channels; only a parent template moves the parent.
604 return [_predicted_family_parent_name(rule, raw_name, variables, parents)]
605 return [evaluate_name_template(rule.name_template, {**variables, "base": raw_name})]
606 if raw_name in children: # pragma: no cover - requires a NetBox that models channelization
607 return [_predicted_channel_name(rule, raw_name, variables, parents, children)]
608 if rule.channel_count > 0 and _is_channelized_rule(rule):
609 if family_blocked or not supports_channelization():
610 return [raw_name] # the apply path builds nothing here
611 parent_name, channels = _channelized_family_names(rule, raw_name, variables) # pragma: no cover - see above
612 return [parent_name, *(name for _, name in channels)] # pragma: no cover - see above
613 vars_copy = {**variables, "base": raw_name}
614 if rule.channel_count > 0:
615 return [
616 evaluate_name_template(rule.name_template, {**vars_copy, "channel": str(rule.channel_start + ch)})
617 for ch in range(rule.channel_count)
618 ]
619 return [evaluate_name_template(rule.name_template, vars_copy)]
622def predict_rule_output(module, module_bay, raw_names):
623 """Predict the names apply_interface_name_rules would produce for raw_names.
625 Read-only — saves and mutates nothing. A channelized rule additionally counts the module's
626 interfaces, because the apply path refuses to convert a module that already carries a flat
627 breakout family and the prediction has to say the same. Used by external integrations (e.g.,
628 netbox-librenms-plugin) that need to know the post-rename names without applying any rule.
630 For breakout rules (channel_count > 0), each raw name expands to
631 channel_count predicted names. For simple renames, one name in → one name
632 out. Returns raw_names unchanged when no rule matches or evaluation fails.
634 A name the module type's interface templates describe as part of a channelized family is
635 predicted as the apply path treats it instead: the family's channels are renamed in place, so a
636 breakout rule leaves the parent's name alone and maps each channel through its ``channel_id``
637 rather than expanding one name into a flat set. Names no template claims keep the per-name
638 prediction, so a module type without channelized templates is unaffected.
640 Precondition: *raw_names* are resolved by the caller at call time. A name captured before the
641 device's virtual-chassis position changed is predicted from itself, not corrected to the name
642 the templates resolve to now — this function maps the names it is given.
643 """
644 device_type = module.device.device_type if module.device else None
645 platform = module.device.platform if module.device else None
646 rule = find_matching_rule(module.module_type, _get_parent_module_type(module_bay), device_type, platform)
647 if not rule:
648 return list(raw_names)
650 variables = build_variables(module_bay, device=module.device)
651 parents, children = _template_families(module)
652 # Costs one count pair, and only where a channelized rule could otherwise predict a family.
653 family_blocked = (
654 rule.channel_count > 0
655 and _is_channelized_rule(rule)
656 and supports_channelization()
657 and _has_flat_expansion(module)
658 )
660 output = []
661 for raw_name in raw_names:
662 try:
663 output.extend(_predicted_names(rule, raw_name, variables, parents, children, family_blocked))
664 except (ValueError, TypeError, re.error):
665 # Template eval failed; apply path would also fail and leave the
666 # interface alone, so the predicted name is the raw name.
667 output.append(raw_name)
669 return output
672def _try_rename_device_interface(rule, iface, vc_position, device, renamed_pks, conflicts=None):
673 """Attempt to rename a single device-level interface using *rule*.
675 Returns ``True`` if the interface was successfully renamed, ``False`` otherwise.
676 Mutates ``renamed_pks`` on success.
678 A computed name already taken by another interface on the device is skipped
679 with a tidy WARNING (no traceback), mirroring the module-install path; pass a
680 list as *conflicts* to also collect them. ``full_clean()`` remains the
681 backstop for the rarer cross-member (VC) uniqueness violation.
682 """
683 if iface.pk in renamed_pks:
684 return False # Already renamed by a higher-priority rule
686 if rule.module_type_pattern:
687 try:
688 if not re.fullmatch(rule.module_type_pattern, iface.name):
689 return False
690 except re.error:
691 return False
693 port = iface.name.rsplit("/", 1)[-1] if "/" in iface.name else iface.name
694 variables = {"vc_position": vc_position, "base": iface.name, "port": port}
696 try:
697 new_name = evaluate_name_template(rule.name_template, variables)
698 except (ValueError, TypeError, re.error):
699 logger.exception(
700 "Failed to evaluate template %r for interface %s (rule %s)",
701 rule.name_template,
702 iface.name,
703 rule.pk,
704 )
705 return False
707 if new_name == iface.name:
708 return False
710 # Pre-check device-scope name uniqueness so an expected collision is a clean
711 # WARNING + skip instead of an ERROR traceback out of full_clean().
712 if _name_exists_on_device(device, new_name, exclude_pk=iface.pk):
713 _record_conflict(conflicts, device, iface.name, new_name, iface.pk)
714 return False
716 old_name = iface.name
717 iface.name = new_name
718 try:
719 iface.full_clean()
720 except ValidationError as exc:
721 logger.warning(
722 "Validation failed renaming device interface %r → %r (rule %s, device %s); skipping: %s",
723 old_name,
724 new_name,
725 rule.pk,
726 device.pk,
727 exc,
728 )
729 iface.name = old_name
730 return False
731 try:
732 iface.save()
733 except (IntegrityError, ValidationError):
734 logger.exception(
735 "DB save failed for device interface %s → %s (rule %s, device %s)",
736 old_name,
737 new_name,
738 rule.pk,
739 device.pk,
740 )
741 iface.name = old_name
742 return False
744 renamed_pks.add(iface.pk)
745 logger.debug("Renamed device interface %s → %s (rule %s, device %s)", old_name, new_name, rule.pk, device.pk)
746 return True
749def _try_rename_device_family(rule, iface, children, vc_position, device, renamed_pks, conflicts=None):
750 """Rename a device-level interface with *rule* and carry its channel subinterfaces along.
752 Returns the number of interfaces renamed. The whole family is claimed in *renamed_pks* the
753 moment its parent is renamed, so a lower-priority rule can never rename the leftovers of a
754 family a higher-priority rule already took.
756 Healing is best-effort here: device-level interfaces have no module template family to recover
757 a suffix from, so a child that lost its parent's prefix in an earlier run is left alone.
758 """
759 parent_before = iface.name
760 if not _try_rename_device_interface(rule, iface, vc_position, device, renamed_pks, conflicts):
761 return 0
762 count = 1
763 for child, target in _child_target_names( # pragma: no cover - requires channelization support
764 children, parent_before, iface.name, module=None
765 ):
766 renamed_pks.add(child.pk)
767 if target is None:
768 logger.warning(
769 "Cannot derive a name for channel interface %r from parent %r; leaving it unchanged.",
770 child.name,
771 iface.name,
772 )
773 continue
774 count += _rename_for_family(child, target, device, conflicts).count
775 return count
778def apply_device_interface_rules(device):
779 """Rename device-level interfaces (module=None) when a device joins/changes position in a VC.
781 Finds all enabled rules with ``applies_to_device_interfaces=True`` that match the device's
782 type and platform, then renames any matching interfaces using the name_template.
784 Template variables available: ``{vc_position}``, ``{base}`` (full current name),
785 ``{port}`` (segment after the last ``/``, or the full name if no ``/`` present).
787 Channel subinterfaces are not matched independently — they follow the parent whose family
788 a rule wins, so a template like ``eth{vc_position}`` cannot collapse a whole family onto
789 one name.
791 Returns the number of interfaces renamed.
792 """
793 from dcim.models import Interface
795 from .models import InterfaceNameRule
797 if not getattr(device, "virtual_chassis_id", None):
798 return 0 # Only rename for VC members (vc_position must be set)
800 if device.vc_position is None:
801 return 0 # vc_position unset (e.g. VC master before position assigned)
803 vc_position = str(device.vc_position)
804 device_type = getattr(device, "device_type", None)
805 platform = getattr(device, "platform", None)
807 from django.db.models import Q
809 rules = list(
810 InterfaceNameRule.objects.filter(
811 applies_to_device_interfaces=True,
812 enabled=True,
813 )
814 .filter(Q(device_type=device_type) | Q(device_type__isnull=True))
815 .filter(Q(platform=platform) | Q(platform__isnull=True))
816 )
817 # Sort Python-side: specificity_score descending, then module_type_pattern length
818 # descending (for device-interface rules with ties), then pk ascending for stability.
819 # (InterfaceNameRule has no DB 'priority' field; specificity_score is a property.)
820 rules.sort(
821 key=lambda r: (
822 -r.specificity_score,
823 -(len(r.module_type_pattern or "") if r.applies_to_device_interfaces else 0),
824 r.pk,
825 )
826 )
828 if not rules:
829 return 0
831 interfaces = list(Interface.objects.filter(device=device, module=None))
832 if not interfaces:
833 return 0
835 bases, children_by_parent = _partition_families(interfaces)
836 total = 0
837 renamed_pks: set[int] = set()
838 for rule in rules:
839 for iface in bases:
840 total += _try_rename_device_family(
841 rule, iface, children_by_parent.get(iface.pk, ()), vc_position, device, renamed_pks
842 )
844 return total
847# The bay chain InterfaceTemplate.resolve_name() dereferences while resolving {module}.
848_BAY_CHAIN_RELATIONS = (
849 "module_bay",
850 "module_bay__parent",
851 "module_bay__module",
852 "module_bay__module__module_bay",
853 "module_bay__module__module_bay__parent",
854 "module_bay__module__module_bay__module",
855)
858def _module_with_bay_chain(module):
859 """Re-fetch *module* with the bay chain InterfaceTemplate.resolve_name() dereferences.
861 Prefetches the module relationships to avoid a per-template query when resolving names.
862 """
863 from dcim.models import Module
865 return Module.objects.select_related(*_BAY_CHAIN_RELATIONS).get(pk=module.pk)
868# NetBox resolves {vc_position} once, at instantiation, so a raw name records the device's VC state
869# at that moment while its template keeps resolving to the current one: hence names *and* matchers.
870_RawMatcher = namedtuple("_RawMatcher", ("template_name", "resolved", "pattern"))
871_RawNames = namedtuple("_RawNames", ("names", "matchers"))
873# Brace-free stand-ins, so NetBox's placeholder pass, this plugin's and re.escape() all leave them be.
874_VC_SENTINEL = "InrVcPositionSentinel{}End"
875_BASE_SENTINEL = "InrBaseSentinelEnd"
878def _vc_position_alternatives(fallback): # pragma: no cover - requires vc_position token support
879 """Return the regex branch covering every value one ``{vc_position}`` occurrence resolves to.
881 Any member position and the implicit ``'0'`` are digits; an explicit ``{vc_position:X}`` fallback
882 adds a branch of its own, since NetBox does not require it to be numeric.
883 """
884 if fallback is None:
885 return r"\d+"
886 return f"(?:\\d+|{re.escape(fallback)})"
889def _raw_name_pattern(tmpl, module, token_re): # pragma: no cover - requires vc_position token support
890 """Return the matcher for every name *tmpl* has ever resolved to, or None without the token.
892 Each token occurrence becomes a sentinel; ``{module}`` is then resolved by NetBox's own code on a
893 shallow copy carrying that name (its VC pass no-ops on a token-free string), so no placeholder
894 resolution is reimplemented here.
895 """
896 fallbacks = []
898 def _mark(match):
899 fallbacks.append(match.group(1))
900 return _VC_SENTINEL.format(len(fallbacks) - 1)
902 marked = token_re.sub(_mark, tmpl.name)
903 if not fallbacks:
904 return None
905 stub = copy.copy(tmpl)
906 stub.name = marked
907 pattern = re.escape(stub.resolve_name(module))
908 for index, fallback in enumerate(fallbacks):
909 pattern = pattern.replace(_VC_SENTINEL.format(index), _vc_position_alternatives(fallback))
910 return _compile_pattern(pattern)
913def _raw_matchers(templates, module):
914 """Resolve *templates* against *module*: their names now, plus a matcher per token template."""
915 token_re = _vc_position_re()
916 names = set()
917 matchers = []
918 for tmpl in templates:
919 resolved = tmpl.resolve_name(module)
920 names.add(resolved)
921 pattern = None if token_re is None else _raw_name_pattern(tmpl, module, token_re)
922 if pattern is not None:
923 matchers.append(_RawMatcher(tmpl.name, resolved, pattern)) # pragma: no cover - token templates only
924 return _RawNames(names, matchers)
927def _raw_name_matchers(module):
928 """Return *module*'s raw template names and the drift matchers of its token templates."""
929 from dcim.models import InterfaceTemplate
931 module_fresh = _module_with_bay_chain(module)
932 templates = InterfaceTemplate.objects.filter(module_type=module_fresh.module_type)
933 return _raw_matchers(templates, module_fresh)
936def _get_raw_interface_names(module):
937 """Return the original interface names NetBox assigned from templates."""
938 return _raw_name_matchers(module).names
941def _raw_name_patterns(module):
942 """Return one compiled matcher per interface template of *module* whose name carries the token.
944 Empty for a module type no template of which uses ``{vc_position}``, and on every NetBox release
945 that does not resolve the token at all.
946 """
947 return [matcher.pattern for matcher in _raw_name_matchers(module).matchers]
950def _raw_names_by_module(modules): # pragma: no cover - only the conversion scan batches names
951 """Return ``{module pk: _RawNames}`` for *modules*, in one template query for all of them.
953 Raw names are a property of the module type, but ``_raw_name_matchers`` costs a module refetch
954 and a template query each — a scan over a fleet would pay that per module. *modules* must
955 already carry ``_BAY_CHAIN_RELATIONS``, since the names are resolved against them in memory.
956 """
957 from dcim.models import InterfaceTemplate
959 by_module_type = defaultdict(list)
960 for tmpl in InterfaceTemplate.objects.filter(module_type__in={module.module_type_id for module in modules}):
961 by_module_type[tmpl.module_type_id].append(tmpl)
962 return {module.pk: _raw_matchers(by_module_type[module.module_type_id], module) for module in modules}
965def _template_families(module):
966 """Return ``(parents, children)`` describing *module*'s channelized interface templates.
968 *parents* maps a channelized parent template's resolved name to its channel count; *children*
969 maps each channel template's resolved name to ``(parent_name, channel_id)``. Both are empty
970 where nothing can be channelized, so callers keep their pre-channelization behaviour without
971 paying for a template scan.
972 """
973 if not supports_channelization():
974 return {}, {}
975 return _resolve_template_families(module) # pragma: no cover - requires channelization support
978def _resolve_template_families(module): # pragma: no cover - requires a NetBox that models channelization
979 """Resolve *module*'s interface templates into the channelized families they describe.
981 Pairing through ``InterfaceTemplate.parent`` (rather than matching against the flat set of raw
982 names) keeps ambiguous prefixes like ``xe``/``xe-0`` apart, and a channel template whose parent
983 declares no channel count is not a family at all.
984 """
985 from dcim.models import InterfaceTemplate
987 module_fresh = _module_with_bay_chain(module)
988 templates = list(InterfaceTemplate.objects.filter(module_type=module_fresh.module_type))
989 resolved = {tmpl.pk: tmpl.resolve_name(module_fresh) for tmpl in templates}
990 parents_by_pk = {
991 tmpl.pk: (resolved[tmpl.pk], tmpl.channels) for tmpl in templates if getattr(tmpl, "channels", None) is not None
992 }
993 children = {}
994 for tmpl in templates:
995 channel_id = getattr(tmpl, "channel_id", None)
996 parent = parents_by_pk.get(getattr(tmpl, "parent_id", None))
997 if channel_id is None or parent is None:
998 continue
999 parent_name, _channels = parent
1000 children[resolved[tmpl.pk]] = (parent_name, channel_id)
1001 return dict(parents_by_pk.values()), children
1004def _template_channel_suffixes(module): # pragma: no cover - requires a NetBox that models channelization
1005 """Map ``channel_id`` → the set of name suffixes *module*'s interface templates give that channel.
1007 The suffix comes from the template family itself — each channel template's resolved name minus
1008 its parent template's resolved name — so a child that lost its parent's prefix in an earlier
1009 partial rename can still be repaired. A module type with several families may spell the same
1010 channel differently in each (``et0:2`` vs ``sw0.2``), so the suffixes are collected per channel
1011 rather than overwritten: the recovery only uses one when every family agrees on it.
1012 """
1013 suffixes = defaultdict(set)
1014 for child_name, (parent_name, channel_id) in _template_families(module)[1].items():
1015 suffix = _child_name_suffix(child_name, parent_name)
1016 if suffix is not None:
1017 suffixes[channel_id].add(suffix)
1018 return suffixes
1021def _recovered_suffix(child, suffixes): # pragma: no cover - requires channelization support
1022 """Return the template suffix for *child*'s channel, or None when it is not unambiguous.
1024 Once a parent has been renamed there is no reliable way back from a stranded child to the family
1025 it belongs to, so a channel spelled differently by two families is left alone rather than guessed.
1026 """
1027 candidates = suffixes.get(child.channel_id) or set()
1028 if len(candidates) == 1:
1029 return next(iter(candidates))
1030 if candidates:
1031 logger.warning(
1032 "Channel %s is spelled %s by different families of this module type; "
1033 "cannot recover a name for interface %r.",
1034 child.channel_id,
1035 sorted(candidates),
1036 child.name,
1037 )
1038 return None
1041def _child_target_names(children, parent_before, parent_after, module):
1042 """Pair every child with the name it takes when its parent is renamed to *parent_after*.
1044 The suffix is read from the child's own name against *parent_before* (the parent's name before
1045 this run's rename); when the child no longer carries that prefix the suffix is recovered from
1046 the module's template family instead. A child that neither shares the prefix nor has an
1047 unambiguous template pairing is returned with a None target — the engine leaves it alone rather
1048 than guessing at a free-form name.
1049 """
1050 suffixes = None
1051 targets = []
1052 for child in children: # pragma: no cover - requires channelization support
1053 suffix = _child_name_suffix(child.name, parent_before)
1054 if suffix is None and module is not None:
1055 if suffixes is None:
1056 suffixes = _template_channel_suffixes(module)
1057 suffix = _recovered_suffix(child, suffixes)
1058 targets.append((child, None if suffix is None else parent_after + suffix))
1059 return targets
1062def _flag_rule_potentially_deprecated(rule):
1063 """Tag a rule as 'potentially-deprecated' when its rename is a no-op.
1065 Called from apply_interface_name_rules when a matching rule produces no
1066 renames because NetBox already generates the correct interface names. This
1067 may indicate the rule is no longer needed (e.g. after a NetBox upgrade that
1068 improved template resolution), or only needed for a subset of module types.
1070 Adds a NetBox Tag 'potentially-deprecated' so the rule is visually flagged
1071 in the UI for operator review. Failures are logged but never re-raised so
1072 the install path is not disrupted.
1073 """
1074 try:
1075 from extras.models import Tag
1077 tag, _ = Tag.objects.get_or_create(
1078 slug="potentially-deprecated",
1079 defaults={"name": "potentially-deprecated", "color": "ffc107"},
1080 )
1081 rule.tags.add(tag)
1082 logger.info(
1083 "Rule '%s' flagged as potentially-deprecated: NetBox already generates the correct interface names.",
1084 rule,
1085 )
1086 except Exception:
1087 logger.exception("Failed to flag rule '%s' as potentially-deprecated.", rule)
1090def _scope_ids(parent_module_type, device_type, platform):
1091 """Map the (parent_module_type, device_type, platform) scope objects to their FK ids.
1093 ``None`` (no constraint) maps to ``None`` so it compares equal to a rule's unset scope FK.
1094 Centralises the ``x.pk if x is not None else None`` coalescing used by both match tiers and
1095 the memo key.
1096 """
1097 return (
1098 parent_module_type.pk if parent_module_type is not None else None,
1099 device_type.pk if device_type is not None else None,
1100 platform.pk if platform is not None else None,
1101 )
1104def _rule_scope_matches(rule, scope_ids):
1105 """Return True when *rule*'s (parent_module_type, device_type, platform) FKs equal *scope_ids*."""
1106 pmt_id, dt_id, pl_id = scope_ids
1107 return rule.parent_module_type_id == pmt_id and rule.device_type_id == dt_id and rule.platform_id == pl_id
1110def _build_candidates(parent_module_type, device_type, platform) -> list:
1111 """Build ordered list of (pmt, dt, pl) tuples from most to least specific.
1113 Each argument expands to ``[value, None]`` when provided, or ``[None]``
1114 when already absent. Deduplication ensures no key appears twice (which
1115 would happen when multiple inputs are None).
1116 """
1117 seen: set = set()
1118 candidates = []
1119 pmt_opts = [parent_module_type, None] if parent_module_type else [None]
1120 dt_opts = [device_type, None] if device_type else [None]
1121 pl_opts = [platform, None] if platform else [None]
1122 for pmt in pmt_opts:
1123 for dt in dt_opts:
1124 for pl in pl_opts:
1125 key = (pmt, dt, pl)
1126 if key not in seen: 1126 ↛ 1124line 1126 didn't jump to line 1124 because the condition on line 1126 was always true
1127 seen.add(key)
1128 candidates.append(key)
1129 return candidates
1132def _find_exact_match(module_type, candidates, exact_rules=None):
1133 """Tier 1: return the first enabled exact-FK rule in specificity order, or None.
1135 ``exact_rules`` is the preloaded, ``(module_type__model, pk)``-ordered enabled
1136 exact-rule set (the hot path passes it to avoid a DB query per call); when omitted
1137 it is loaded on demand so direct callers keep working.
1138 """
1139 if exact_rules is None:
1140 exact_rules, _, _ = _get_enabled_rules()
1142 # module_type fixed below → the (module_type__model, pk) ordering reduces to pk, so the
1143 # first matching rule equals the previous ``.filter(...).first()``.
1144 scoped = [r for r in exact_rules if r.module_type_id == module_type.pk]
1145 for candidate in candidates:
1146 scope_ids = _scope_ids(*candidate)
1147 for rule in scoped:
1148 if _rule_scope_matches(rule, scope_ids):
1149 return rule
1150 return None
1153def _find_regex_match(model_name: str, candidates, regex_rules=None):
1154 """Tier 2: return the first enabled regex rule whose pattern fullmatches *model_name*, or None.
1156 Tries candidates in specificity order; within each level longer patterns are tried first
1157 (more specific). ``regex_rules`` is the preloaded ``(compiled_pattern, rule)`` set — pre-sorted
1158 by ``(-pattern length, pk)`` with each pattern compiled once (a None compile is an invalid
1159 pattern, silently skipped); loaded on demand when omitted.
1160 """
1161 if regex_rules is None:
1162 _, regex_rules, _ = _get_enabled_rules()
1164 for candidate in candidates:
1165 scope_ids = _scope_ids(*candidate)
1166 for compiled, rule in regex_rules:
1167 if compiled is not None and _rule_scope_matches(rule, scope_ids) and compiled.fullmatch(model_name):
1168 return rule
1169 return None
1172def find_matching_rule(module_type, parent_module_type, device_type, platform=None):
1173 """Find the most specific InterfaceNameRule matching the context.
1175 Uses a two-tier strategy:
1176 Tier 1 — Exact FK match (priority order, most specific first):
1177 Iterates all combinations of (parent_module_type, device_type, platform)
1178 from fully-constrained to fully-unconstrained (None = any).
1179 Tier 2 — Regex pattern match (same priority order, longer patterns first):
1180 Same specificity cascade, but module_type_pattern is matched via
1181 re.fullmatch() against module_type.model. Patterns are iterated
1182 from longest to shortest to prefer more specific patterns.
1184 The enabled rule set is loaded once and matched in memory, and the per-context
1185 result is memoized for the current rule-set version, so repeated calls (e.g. one
1186 per module row in a module-sync render) don't re-query the database.
1188 Returns the first matching rule, or None if no rule matches.
1189 """
1190 if module_type is None:
1191 # Module rules are always keyed on a module type; both tiers dereference it
1192 # (module_type.pk / .model), so there is nothing to match without one.
1193 return None
1195 exact_rules, regex_rules, memo = _get_enabled_rules()
1196 # The regex tier matches against module_type.model (a live string), so the memo must key on
1197 # it too — otherwise a ModuleType.model rename (same pk) would return a stale regex result.
1198 sig = (module_type.pk, module_type.model, *_scope_ids(parent_module_type, device_type, platform))
1199 # One atomic lookup, not `if sig in memo: return memo[sig]`: another thread sharing this per-version
1200 # memo can clear it at the cap between a membership test and the subscript, raising KeyError.
1201 cached = memo.get(sig, _MEMO_MISS)
1202 if cached is not _MEMO_MISS:
1203 return cached
1205 candidates = _build_candidates(parent_module_type, device_type, platform)
1206 result = _find_exact_match(module_type, candidates, exact_rules) or _find_regex_match(
1207 module_type.model, candidates, regex_rules
1208 )
1209 if len(memo) >= _MEMO_MAX:
1210 memo.clear() # bound per-version memory; entries are rebuilt lazily on the next miss
1211 memo[sig] = result
1212 return result
1215def _extract_trailing_digits(s: str) -> str:
1216 r"""Return the trailing digit run of *s* without regex backtracking.
1218 Pure O(n) string scan — eliminates the polynomial backtracking risk that
1219 arises from using ``re.search(r"(\d+)$", ...)`` on strings ending in a
1220 non-digit character (e.g. ``"1" * n + "x"`` would cause O(n²) steps).
1222 Returns an empty string when *s* has no trailing digits.
1223 """
1224 i = len(s)
1225 while i > 0 and s[i - 1].isdigit():
1226 i -= 1
1227 return s[i:]
1230def _resolve_bay_position(module_bay):
1231 """Return (bay_position, bay_position_num) from a module bay's position field.
1233 Handles template expressions like ``{module}`` by extracting the trailing
1234 digit from the bay name. Falls back to ``"0"`` if no digit is found.
1235 """
1236 bay_position = module_bay.position or "0"
1237 if bay_position.startswith("{"):
1238 digits = _extract_trailing_digits(module_bay.name)
1239 bay_position = digits if digits else "0"
1240 digits = _extract_trailing_digits(bay_position)
1241 bay_position_num = digits if digits else "0"
1242 return bay_position, bay_position_num
1245def _resolve_slot(module_bay, bay_position_num, parent_bay_position):
1246 """Return the ``slot`` variable from the module bay hierarchy.
1248 When the bay has a parent bay, slot comes from the parent (or grandparent
1249 when two levels of nesting exist). When the bay belongs to an installed
1250 module with its own bay, slot comes from that module's bay position.
1251 Falls back to ``bay_position_num``.
1252 """
1253 if module_bay.parent:
1254 parent_bay = module_bay.parent
1255 if parent_bay.parent and hasattr(parent_bay.parent, "installed_module"):
1256 return parent_bay.parent.position or parent_bay_position
1257 return parent_bay_position
1258 if hasattr(module_bay, "module") and module_bay.module: 1258 ↛ 1259line 1258 didn't jump to line 1259 because the condition on line 1258 was never true
1259 owner_module = module_bay.module
1260 if hasattr(owner_module, "module_bay") and owner_module.module_bay:
1261 return owner_module.module_bay.position or bay_position_num
1262 return bay_position_num
1265def build_variables(module_bay, device=None):
1266 """Build template variable dict from a module bay's position context.
1268 Extracts numeric and raw position values from the bay and its parent chain,
1269 producing the variables available for name_template substitution.
1271 Returns a dict with keys: slot, bay_position, bay_position_num,
1272 parent_bay_position, sfp_slot, and optionally vc_position.
1274 ``vc_position`` is only injected when *device* is a Virtual Chassis member
1275 (device.virtual_chassis_id is set). Templates using ``{vc_position}`` on a
1276 non-VC device will raise ValueError during evaluation — this is intentional.
1277 Note: Juniper VC positions start at 0, so 0 is a valid real-world value and
1278 cannot be used as a "not in VC" sentinel.
1279 """
1280 bay_position, bay_position_num = _resolve_bay_position(module_bay)
1282 parent_bay_position = "0"
1283 if module_bay.parent:
1284 parent_bay_position = module_bay.parent.position or "0"
1286 slot = _resolve_slot(module_bay, bay_position_num, parent_bay_position)
1288 result = {
1289 "slot": slot,
1290 "bay_position": bay_position,
1291 "bay_position_num": bay_position_num,
1292 "parent_bay_position": parent_bay_position,
1293 "sfp_slot": bay_position_num,
1294 }
1295 if (
1296 device is not None
1297 and getattr(device, "virtual_chassis_id", None) is not None
1298 and device.vc_position is not None
1299 ):
1300 result["vc_position"] = str(device.vc_position)
1301 return result
1304def _name_exists_on_device(device, name, exclude_pk=None):
1305 """Return True if another interface on *device* already uses *name*.
1307 Pre-checks the per-device interface-name uniqueness NetBox enforces so a
1308 rename/create that would collide is skipped cleanly instead of raising
1309 mid-transaction. (VC-wide uniqueness is not pre-checked here; full_clean()
1310 remains the authoritative validator for that rarer cross-member case.)
1311 """
1312 from dcim.models import Interface
1314 qs = Interface.objects.filter(device=device, name=name)
1315 if exclude_pk is not None:
1316 qs = qs.exclude(pk=exclude_pk)
1317 return qs.exists()
1320def _record_skip(conflicts, device, current_name, attempted_name, interface_pk=None):
1321 """Append a skipped rename to *conflicts* when the caller is collecting them.
1323 The caller has already logged why it skipped; this only lets the interactive Apply view report
1324 how many renames were dropped.
1325 """
1326 if conflicts is not None:
1327 conflicts.append(
1328 {
1329 "device": str(device),
1330 "current_name": current_name,
1331 "attempted_name": attempted_name,
1332 "interface_pk": interface_pk,
1333 }
1334 )
1337def _record_conflict(conflicts, device, current_name, attempted_name, interface_pk=None):
1338 """Log a name collision at WARNING and record it as a skipped rename.
1340 Collisions are expected during automatic renaming (module install, type
1341 change, VC change) when the computed name is already taken on the device;
1342 they must never abort the batch, so callers skip the rename and carry on.
1343 """
1344 logger.warning(
1345 "Interface name %r already exists on device %s — skipping rename of %r → %r",
1346 attempted_name,
1347 device,
1348 current_name,
1349 attempted_name,
1350 )
1351 _record_skip(conflicts, device, current_name, attempted_name, interface_pk)
1354# What a family-aware rename did, so the children can act on their parent's outcome rather than on
1355# its computed target name (which says nothing about whether the parent actually took it).
1356_RENAMED = "renamed"
1357_UNCHANGED = "unchanged"
1358_COLLISION = "collision"
1359_ERROR = "error"
1361_RenameResult = namedtuple("_RenameResult", ("target_name", "outcome", "count"))
1364def _rename_in_place(iface, new_name, device, conflicts):
1365 """Rename *iface* to *new_name*; return 1 if renamed, 0 if no-op or collision."""
1366 if new_name == iface.name:
1367 return 0
1368 if _name_exists_on_device(device, new_name, exclude_pk=iface.pk):
1369 _record_conflict(conflicts, device, iface.name, new_name, iface.pk)
1370 return 0
1371 iface.name = new_name
1372 iface.full_clean()
1373 iface.save()
1374 return 1
1377def _rename_for_family(iface, new_name, device, conflicts): # pragma: no cover - requires channelization support
1378 """Rename *iface* as part of a family walk, reporting the outcome instead of raising.
1380 The save runs in its own savepoint so an unexpected failure on one member leaves the
1381 surrounding transaction usable: family processing is best-effort per interface, and only a
1382 failed *parent* stops the rest of its family.
1383 """
1384 if new_name == iface.name:
1385 return _RenameResult(new_name, _UNCHANGED, 0)
1386 old_name = iface.name
1387 try:
1388 with transaction.atomic():
1389 renamed = _rename_in_place(iface, new_name, device, conflicts)
1390 except (ValueError, ValidationError, IntegrityError):
1391 logger.exception("Failed to rename interface %r → %r on device %s; skipping.", old_name, new_name, device)
1392 iface.name = old_name
1393 return _RenameResult(new_name, _ERROR, 0)
1394 return _RenameResult(new_name, _RENAMED, 1) if renamed else _RenameResult(new_name, _COLLISION, 0)
1397def _restore_deferred_channel_names(reconciliations, db_alias): # pragma: no cover - channelization only
1398 """Restore plugin-owned names that NetBox's parent cascade changed after commit."""
1399 from dcim.models import Interface
1401 child_pks = [child_pk for child_pk, _final_name, _cascade_name in reconciliations]
1402 with transaction.atomic(using=db_alias):
1403 children = Interface.objects.using(db_alias).select_for_update().select_related("device").in_bulk(child_pks)
1404 for child_pk, final_name, cascade_name in reconciliations:
1405 child = children.get(child_pk)
1406 if child is None or child.name == final_name:
1407 continue
1408 if child.name != cascade_name:
1409 logger.warning(
1410 "Channel interface %s changed to unexpected name %r before deferred reconciliation; "
1411 "leaving it unchanged.",
1412 child_pk,
1413 child.name,
1414 )
1415 continue
1416 previous_name = child.name
1417 try:
1418 with transaction.atomic(using=db_alias):
1419 child.name = final_name
1420 child.full_clean()
1421 child.save(using=db_alias)
1422 except (ValueError, ValidationError, IntegrityError):
1423 child.name = previous_name
1424 logger.exception(
1425 "Failed to restore channel interface %s from NetBox's deferred name %r to %r; skipping.",
1426 child_pk,
1427 cascade_name,
1428 final_name,
1429 )
1432def _preserve_names_across_parent_cascade(parent, parent_before, final_names): # pragma: no cover
1433 """Run after NetBox's deferred cascade when a rule intentionally keeps old-parent child names."""
1434 if parent.name == parent_before:
1435 return
1437 reconciliations = []
1438 for child, final_name in final_names:
1439 old_conventional_name = f"{parent_before}:{child.channel_id}"
1440 cascade_name = f"{parent.name}:{child.channel_id}"
1441 if final_name == old_conventional_name and final_name != cascade_name:
1442 reconciliations.append((child.pk, final_name, cascade_name))
1443 if not reconciliations:
1444 return
1446 reconciliations = tuple(reconciliations)
1447 db_alias = parent._state.db
1448 transaction.on_commit(
1449 lambda: _restore_deferred_channel_names(reconciliations, db_alias),
1450 using=db_alias,
1451 )
1454def _rename_channel_children(parent, parent_before, children, module, conflicts): # pragma: no cover - see above
1455 """Carry the parent's new name onto its channel subinterfaces; return how many were renamed."""
1456 count = 0
1457 for child, target in _child_target_names(children, parent_before, parent.name, module):
1458 if target is None:
1459 logger.warning(
1460 "Cannot derive a name for channel interface %r from parent %r; leaving it unchanged.",
1461 child.name,
1462 parent.name,
1463 )
1464 continue
1465 count += _rename_for_family(child, target, module.device, conflicts).count
1466 return count
1469def _apply_simple_rule_to_family(rule, parent, children, variables, module, conflicts): # pragma: no cover
1470 """Rename a channelized family in lockstep with its parent; return the count renamed.
1472 The children act on the parent's *outcome*, never on its computed target: a parent that
1473 collided or failed to save leaves the whole family untouched, while a parent that already
1474 carries the right name still lets a stale child be repaired.
1475 """
1476 parent_before = parent.name
1477 new_name = evaluate_name_template(rule.name_template, {**variables, "base": parent.name})
1478 result = _rename_for_family(parent, new_name, module.device, conflicts)
1479 if result.outcome in (_COLLISION, _ERROR):
1480 logger.debug("Family of %r left unchanged: the parent could not be renamed to %r.", parent.name, new_name)
1481 return result.count
1482 return result.count + _rename_channel_children(parent, parent_before, children, module, conflicts)
1485def _rename_family_parent(rule, parent, variables, module, conflicts): # pragma: no cover - see above
1486 """Rename an existing family's parent per the rule's parent template; return its rename outcome.
1488 Only a channelized rule that names its parent touches it — a flat rule, or a blank parent
1489 template, leaves the parent the name it already has.
1490 """
1491 if not (_is_channelized_rule(rule) and rule.parent_name_template):
1492 return _RenameResult(parent.name, _UNCHANGED, 0)
1493 target = evaluate_name_template(rule.parent_name_template, {**variables, "base": parent.name})
1494 return _rename_for_family(parent, target, module.device, conflicts)
1497def _apply_breakout_rule_to_family(rule, parent, children, variables, module, conflicts): # pragma: no cover
1498 """Rename an already-channelized family; return the count renamed, or None when skipped.
1500 Nothing is ever created here: the channels the rule describes are rows NetBox already models,
1501 so a breakout rule renames them in place. The parent is renamed only when the rule builds
1502 channelized families and names their parent; the channels' ``{base}`` stays the parent's name
1503 as it was before that rename. A rule whose channel count disagrees with the hardware is a
1504 modelling mismatch — the family is skipped whole rather than renamed into a shape it does not
1505 have, and a parent that could not take its name stops the family the same way a simple rule's
1506 does.
1508 Every child's name is computed before the first save, so a template that only fails on a later
1509 channel (channel-dependent arithmetic) aborts the family untouched instead of half renaming it.
1510 """
1511 if getattr(parent, "channels", None) != rule.channel_count:
1512 logger.warning(
1513 "Interface %r provides %s channels but rule '%s' defines %s; skipping the family.",
1514 parent.name,
1515 getattr(parent, "channels", None),
1516 rule,
1517 rule.channel_count,
1518 )
1519 return None
1520 base_name = parent.name
1521 targets = [
1522 (
1523 child,
1524 evaluate_name_template(
1525 rule.name_template,
1526 {**variables, "base": base_name, "channel": str(rule.channel_start + child.channel_id - 1)},
1527 ),
1528 )
1529 for child in children
1530 ]
1531 result = _rename_family_parent(rule, parent, variables, module, conflicts)
1532 if result.outcome in (_COLLISION, _ERROR):
1533 logger.debug(
1534 "Family of %r left unchanged: the parent could not be renamed to %r.", base_name, result.target_name
1535 )
1536 return result.count
1537 count = result.count
1538 final_names = []
1539 for child, new_name in targets:
1540 previous_name = child.name
1541 child_result = _rename_for_family(child, new_name, module.device, conflicts)
1542 count += child_result.count
1543 final_name = new_name if child_result.outcome in (_RENAMED, _UNCHANGED) else previous_name
1544 final_names.append((child, final_name))
1545 _preserve_names_across_parent_cascade(parent, base_name, final_names)
1546 return count
1549def _is_channelized_rule(rule):
1550 """Return True when *rule* asks for the channelized topology instead of flat sibling interfaces."""
1551 return rule.breakout_mode == BreakoutModeChoices.CHANNELIZED
1554def _channelized_family_names(rule, base_name, variables): # pragma: no cover - requires channelization support
1555 """Return ``(parent_name, [(channel_id, name), ...])`` for the family *rule* builds on *base_name*.
1557 ``{base}`` is the base interface's current name for the parent and every channel; ``{channel}``
1558 is ``channel_start + channel_id - 1``. A blank parent template leaves the base's name alone.
1559 Takes the name rather than the interface so prediction can reuse it without a row to point at.
1560 """
1561 family_vars = {**variables, "base": base_name}
1562 parent_name = base_name
1563 if rule.parent_name_template:
1564 parent_name = evaluate_name_template(rule.parent_name_template, family_vars)
1565 channels = [
1566 (
1567 channel_id,
1568 evaluate_name_template(
1569 rule.name_template, {**family_vars, "channel": str(rule.channel_start + channel_id - 1)}
1570 ),
1571 )
1572 for channel_id in range(1, rule.channel_count + 1)
1573 ]
1574 return parent_name, channels
1577def _has_flat_expansion(module): # pragma: no cover - requires channelization support
1578 """Return True when *module* carries more interfaces than its module type's templates describe.
1580 A flat breakout leaves N-1 rows beyond the templates, so the surplus is the structural mark of a
1581 family an earlier apply installed. Counting templates rather than their resolved names keeps
1582 two templates that resolve to the same string from reading as one.
1583 """
1584 from dcim.models import Interface, InterfaceTemplate
1586 templates = InterfaceTemplate.objects.filter(module_type_id=module.module_type_id).count()
1587 return Interface.objects.filter(module=module).count() > templates
1590def _first_taken_name(device, names, exclude_pk): # pragma: no cover - requires channelization support
1591 """Return the first of *names* already used by another interface on *device*, or None."""
1592 for name in names:
1593 if _name_exists_on_device(device, name, exclude_pk=exclude_pk):
1594 return name
1595 return None
1598def _build_channelized_family(rule, base, variables, module, conflicts): # pragma: no cover - see above
1599 """Turn a plain base interface into a channelized family; return how many rows it changed.
1601 The whole family is preflighted before anything is written: a module that already carries a flat
1602 family, or a single occupied name — the parent's or any channel's — leaves the base exactly as
1603 it was instead of half converting it.
1604 """
1605 from dcim.choices import InterfaceTypeChoices
1606 from dcim.models import Interface
1608 device = module.device
1609 base_name = base.name
1610 parent_name, channels = _channelized_family_names(rule, base_name, variables)
1611 if _has_flat_expansion(module):
1612 # Converting one sibling into a parent would strand the others beside the new family.
1613 logger.warning(
1614 "Module %s already carries a flat breakout family; rule '%s' will not convert interface "
1615 "%r into the channelized parent %r — converting an installed family is a separate, "
1616 "explicit operation. Skipping.",
1617 module,
1618 rule,
1619 base.name,
1620 parent_name,
1621 )
1622 _record_skip(conflicts, device, base.name, parent_name, base.pk)
1623 return 0
1624 blocker = _first_taken_name(device, [parent_name, *(name for _, name in channels)], base.pk)
1625 if blocker is not None:
1626 logger.warning(
1627 "Cannot build the channelized family for interface %r on device %s: %r is already taken; skipping.",
1628 base.name,
1629 device,
1630 blocker,
1631 )
1632 _record_skip(conflicts, device, base.name, blocker, base.pk)
1633 return 0
1635 count = 0
1636 with transaction.atomic():
1637 created_channels = []
1638 base.channels = rule.channel_count
1639 if parent_name != base.name:
1640 base.name = parent_name
1641 count += 1
1642 base.full_clean()
1643 base.save()
1644 for channel_id, name in channels:
1645 channel = Interface(
1646 device=device,
1647 module=module,
1648 name=name,
1649 type=InterfaceTypeChoices.TYPE_CHANNEL,
1650 parent=base,
1651 channel_id=channel_id,
1652 enabled=base.enabled,
1653 )
1654 channel.full_clean()
1655 channel.save()
1656 created_channels.append((channel, name))
1657 count += 1
1658 _preserve_names_across_parent_cascade(base, base_name, created_channels)
1659 return count
1662def _apply_channelized_rule(rule, base, variables, module, conflicts):
1663 """Build the channelized family *rule* describes on a plain base interface.
1665 Returns None where NetBox cannot model channels: the rule describes a topology this release has
1666 no rows for, and building a flat family instead would silently give the operator another one.
1667 """
1668 if not supports_channelization():
1669 logger.warning(
1670 "Rule '%s' builds a channelized family, which this NetBox release cannot model; "
1671 "leaving interface %r unchanged.",
1672 rule,
1673 base.name,
1674 )
1675 return None
1676 return _build_channelized_family(rule, base, variables, module, conflicts) # pragma: no cover - see above
1679def _apply_rule_with_family(rule, iface, children, variables, module, conflicts):
1680 """Apply *rule* to *iface*, carrying its channel subinterfaces along.
1682 Returns the number of interfaces renamed/created, or None when a channelized family was
1683 skipped for a structural reason. Interfaces that own no family take the plain path unchanged.
1684 """
1685 if rule.channel_count > 0 and (_is_channelized_parent(iface) or children): # pragma: no cover
1686 return _apply_breakout_rule_to_family(rule, iface, children, variables, module, conflicts)
1687 if children: # pragma: no cover - requires channelization support
1688 return _apply_simple_rule_to_family(rule, iface, children, variables, module, conflicts)
1689 if rule.channel_count > 0 and _is_channelized_rule(rule):
1690 return _apply_channelized_rule(rule, iface, variables, module, conflicts)
1691 return _apply_rule_to_interface(rule, iface, {**variables, "base": iface.name}, module, conflicts=conflicts)
1694def _create_channel(iface, module, new_name, device, conflicts):
1695 """Create a breakout channel interface *new_name*; return 1 if created, else 0.
1697 Silently skips when this module already has the channel (idempotent
1698 re-apply); records a conflict when *new_name* is taken by a different
1699 interface on the device.
1700 """
1701 from dcim.models import Interface
1703 if Interface.objects.filter(module=module, name=new_name).exists():
1704 return 0 # idempotent: channel already created on this module
1705 if _name_exists_on_device(device, new_name):
1706 _record_conflict(conflicts, device, iface.name, new_name, iface.pk)
1707 return 0
1708 breakout_iface = Interface(
1709 device=device,
1710 module=module,
1711 name=new_name,
1712 type=iface.type,
1713 enabled=iface.enabled,
1714 )
1715 breakout_iface.full_clean()
1716 breakout_iface.save()
1717 return 1
1720def _apply_rule_to_interface(rule, iface, variables, module, conflicts=None):
1721 """Apply a single rule to an interface, handling breakout channels.
1723 All saves are wrapped in a transaction so a failure mid-breakout rolls
1724 back any partially created interfaces. A computed name that already exists
1725 on the device is skipped (logged, and recorded in *conflicts* when a list
1726 is passed) instead of raising — so automatic renaming (module install,
1727 module-type change, VC change) never aborts the rest of the batch on a
1728 name collision.
1730 Returns the number of interfaces renamed/created.
1731 """
1732 count = 0
1733 device = module.device
1735 with transaction.atomic():
1736 if rule.channel_count > 0:
1737 # Breakout: rename base interface and create additional channel interfaces
1738 for ch in range(rule.channel_count):
1739 variables["channel"] = str(rule.channel_start + ch)
1740 new_name = evaluate_name_template(rule.name_template, variables)
1741 if ch == 0:
1742 count += _rename_in_place(iface, new_name, device, conflicts)
1743 else:
1744 count += _create_channel(iface, module, new_name, device, conflicts)
1745 else:
1746 # Simple rename (converter offset, platform naming, etc.)
1747 new_name = evaluate_name_template(rule.name_template, variables)
1748 count += _rename_in_place(iface, new_name, device, conflicts)
1750 return count
1753def _find_channel_base(rule, ifaces, variables):
1754 """Find the best 'base' interface for a channel rule on a single module.
1756 Prefers an interface whose current name already equals the expected ch=0 name
1757 (i.e. it has already been renamed to channel 0 and is safe to re-process).
1758 Falls back to the first interface (alphabetically) so that on first apply,
1759 the template-created base interface becomes channel 0.
1761 This ensures apply_rule_to_existing / find_interfaces_for_rule call
1762 _apply_rule_to_interface exactly ONCE per module for channel rules, preventing
1763 duplicate-name IntegrityErrors when channels already exist.
1764 """
1765 if not ifaces:
1766 return None
1767 for iface in ifaces:
1768 vars_copy = dict(variables)
1769 vars_copy["base"] = iface.name
1770 vars_copy["channel"] = str(rule.channel_start) # ch=0
1771 try:
1772 ch0_name = evaluate_name_template(rule.name_template, vars_copy)
1773 if iface.name == ch0_name:
1774 return iface
1775 except ValueError:
1776 pass
1777 return ifaces[0]
1780def _matching_moduletype_pks(module_type_pattern):
1781 """Return PKs of ModuleTypes whose model name matches the given regex pattern.
1783 Raises ValueError for invalid regex patterns, mirroring evaluate_name_template's
1784 error-handling convention so callers can treat both as ValueError.
1785 """
1786 from dcim.models import ModuleType
1788 try:
1789 compiled = re.compile(module_type_pattern)
1790 except re.error as exc:
1791 raise ValueError(f"Invalid module_type_pattern regex '{module_type_pattern}': {exc}") from exc
1792 return [mt.pk for mt in ModuleType.objects.only("pk", "model") if compiled.fullmatch(mt.model)]
1795def has_applicable_interfaces(rule) -> bool:
1796 """Check whether applying this rule right now would rename at least one interface.
1798 Calls find_interfaces_for_rule(limit=1) to determine if any currently installed
1799 interface would receive a new name. Returns False when:
1800 - no matching modules/interfaces are installed, OR
1801 - all matching interfaces are already correctly named.
1803 This is more expensive than a plain EXISTS query but ensures the Applicable
1804 column in the Apply Rules list accurately reflects "would something change?"
1805 rather than the misleading "do interfaces exist?".
1806 """
1807 try:
1808 results, _ = find_interfaces_for_rule(rule, limit=1)
1809 return len(results) > 0
1810 except (ValueError, re.error):
1811 return False
1814def _build_module_qs(rule):
1815 """Return a Module queryset filtered to the rule's scope (module type, parent, device, platform).
1817 Shared by ``find_interfaces_for_rule`` and ``apply_rule_to_existing`` to avoid
1818 duplicating the filtering logic.
1819 """
1820 from dcim.models import Module
1822 if rule.module_type_is_regex:
1823 qs = Module.objects.filter(module_type__in=_matching_moduletype_pks(rule.module_type_pattern))
1824 else:
1825 qs = Module.objects.filter(module_type=rule.module_type)
1826 if rule.parent_module_type:
1827 qs = qs.filter(module_bay__parent__installed_module__module_type=rule.parent_module_type)
1828 if rule.device_type:
1829 qs = qs.filter(device__device_type=rule.device_type)
1830 if rule.platform:
1831 qs = qs.filter(device__platform=rule.platform)
1832 return qs
1835def _name_detail(name, role, channel_id=None) -> dict:
1836 """Describe one previewed name so the UI can render a family as a family.
1838 *role* is ``interface`` (a plain rename), ``parent`` (the family's physical interface) or
1839 ``channel``; *channel_id* is the parent channel a channel name is bound to, when known.
1840 """
1841 return {"name": name, "role": role, "channel_id": channel_id}
1844def _family_entry(module, parent, details, children) -> dict | None:
1845 """Build a family preview entry from per-name *details*, or None when nothing would change.
1847 The entry stays keyed on the parent — the PK the Apply view submits — and lists the family's
1848 names in ``new_names``, so the existing template loop keeps working unchanged.
1849 """
1850 if [detail["name"] for detail in details] == [parent.name, *(child.name for child in children)]:
1851 return None
1852 return {
1853 "module": module,
1854 "interface": parent,
1855 "current_name": parent.name,
1856 "new_names": [detail["name"] for detail in details],
1857 "name_details": details,
1858 }
1861def _evaluate_plain_interface(rule, module, iface, variables, children=()) -> dict | None:
1862 """Return a result dict if *iface* or one of its channels would be renamed by *rule*, else None.
1864 A channelized parent is previewed together with its channels, so the Apply page shows the whole
1865 family behind the one PK it submits.
1866 """
1867 vars_copy = {**variables, "base": iface.name}
1868 try:
1869 new_name = evaluate_name_template(rule.name_template, vars_copy)
1870 except ValueError as exc:
1871 new_name = f"<error: {exc}>"
1872 if children: # pragma: no cover - requires channelization support
1873 return _family_entry(module, iface, _lockstep_details(new_name, iface, children, module), children)
1874 return _family_entry(module, iface, [_name_detail(new_name, "interface")], ())
1877def _lockstep_details(new_name, parent, children, module) -> list: # pragma: no cover - see above
1878 """Per-name preview details for a family renamed in lockstep with its parent.
1880 A channel whose suffix cannot be derived previews as unchanged — the same thing the apply path
1881 does with it.
1882 """
1883 details = [_name_detail(new_name, "parent")]
1884 for child, target in _child_target_names(children, parent.name, new_name, module):
1885 details.append(_name_detail(child.name if target is None else target, "channel", child.channel_id))
1886 return details
1889def _channelized_family_entry(rule, module, parent, children, variables) -> dict | None: # pragma: no cover
1890 """Preview a breakout rule against an already-channelized family.
1892 Nothing is created — only the existing channels are renamed, plus the parent when the rule
1893 names one — so a family whose channel count disagrees with the rule previews as no change at
1894 all.
1895 """
1896 if getattr(parent, "channels", None) != rule.channel_count:
1897 return None
1898 parent_name = parent.name
1899 if _is_channelized_rule(rule) and rule.parent_name_template:
1900 try:
1901 parent_name = evaluate_name_template(rule.parent_name_template, {**variables, "base": parent.name})
1902 except ValueError as exc:
1903 parent_name = f"<error: {exc}>"
1904 details = [_name_detail(parent_name, "parent")]
1905 for child in children:
1906 channel = str(rule.channel_start + child.channel_id - 1)
1907 try:
1908 new_name = evaluate_name_template(
1909 rule.name_template, {**variables, "base": parent.name, "channel": channel}
1910 )
1911 except ValueError as exc:
1912 new_name = f"<error: {exc}>"
1913 details.append(_name_detail(new_name, "channel", child.channel_id))
1914 return _family_entry(module, parent, details, children)
1917def _channelized_family_preview(rule, module, base, variables) -> dict | None: # pragma: no cover - see below
1918 """Describe the channelized family a rule would build on a plain base interface."""
1919 try:
1920 parent_name, channels = _channelized_family_names(rule, base.name, variables)
1921 except ValueError as exc:
1922 return _family_entry(module, base, [_name_detail(f"<error: {exc}>", "parent")], ())
1923 details = [_name_detail(parent_name, "parent")]
1924 details.extend(_name_detail(name, "channel", channel_id) for channel_id, name in channels)
1925 return _family_entry(module, base, details, ())
1928def _channelized_creation_entry(rule, module, bases, variables) -> dict | None:
1929 """Return the preview entry for the family a channelized rule would build, or None for none.
1931 A release that cannot model channels previews nothing, because the apply path builds nothing
1932 there either; neither does a module whose flat family the apply path refuses to convert.
1933 """
1934 if not supports_channelization():
1935 return None
1936 if _has_flat_expansion(module): # pragma: no cover - requires channelization support
1937 return None
1938 return _channelized_family_preview( # pragma: no cover - requires channelization support
1939 rule, module, _find_channel_base(rule, bases, variables), variables
1940 )
1943def _channel_rule_entries(rule, module, bases, children_by_parent, variables) -> list:
1944 """Return the preview entries a channel rule produces for one module.
1946 A module whose base is already channelized previews per family (renames only); a channelized
1947 rule on a plain base previews the family it would build; anything else keeps the flat breakout
1948 preview of one entry per module.
1949 """
1950 families = [base for base in bases if _is_channelized_parent(base)]
1951 if families: # pragma: no cover - requires a NetBox that models channelization
1952 entries = [
1953 _channelized_family_entry(rule, module, parent, children_by_parent.get(parent.pk, ()), variables)
1954 for parent in families
1955 ]
1956 return [entry for entry in entries if entry]
1957 if _is_channelized_rule(rule):
1958 entry = _channelized_creation_entry(rule, module, bases, variables)
1959 return [entry] if entry else []
1960 entry = _channel_rule_entry(rule, module, bases, variables)
1961 return [entry] if entry else []
1964def _channel_rule_entry(rule, module, ifaces, variables) -> dict | None:
1965 """Return a result dict if the channel rule would change any name for this module, else None."""
1966 base_iface = _find_channel_base(rule, ifaces, variables)
1967 if base_iface is None: 1967 ↛ 1968line 1967 didn't jump to line 1968 because the condition on line 1967 was never true
1968 return None
1969 vars_copy = {**variables, "base": base_iface.name}
1970 expected_names = []
1971 try:
1972 for ch in range(rule.channel_count):
1973 expected_names.append(
1974 evaluate_name_template(rule.name_template, {**vars_copy, "channel": str(rule.channel_start + ch)})
1975 )
1976 except ValueError as exc:
1977 expected_names = [f"<error: {exc}>"]
1978 existing_names = {i.name for i in ifaces}
1979 # Report if any channel name is missing or the base itself needs renaming
1980 if any(n not in existing_names for n in expected_names) or (
1981 expected_names and expected_names[0] != base_iface.name
1982 ):
1983 return {
1984 "module": module,
1985 "interface": base_iface,
1986 "current_name": base_iface.name,
1987 "new_names": expected_names,
1988 "name_details": [_name_detail(name, "channel") for name in expected_names],
1989 }
1990 return None
1993def _count_remaining_interfaces(module_qs, processed_pks) -> int:
1994 """Count the rule candidates in modules not yet visited during a find_interfaces_for_rule scan."""
1995 from dcim.models import Interface
1997 qs = Interface.objects.filter(module__in=module_qs.exclude(pk__in=processed_pks))
1998 if supports_channelization(): # pragma: no cover - the column exists only on NetBox 4.7+
1999 qs = qs.filter(channel_id__isnull=True) # a family counts once, through its parent
2000 return qs.count()
2003def _process_channel_module(rule, module, ifaces, variables, limit, results, module_qs, processed_pks):
2004 """Process one module for a channel rule. Returns (checked_count, should_stop)."""
2005 bases, children_by_parent = _partition_families(ifaces)
2006 checked = len(bases)
2007 if not bases:
2008 return checked, False
2009 for entry in _channel_rule_entries(rule, module, bases, children_by_parent, variables):
2010 results.append(entry)
2011 if limit is not None and len(results) >= limit:
2012 return checked + _count_remaining_interfaces(module_qs, processed_pks), True
2013 return checked, False
2016def _process_plain_module(rule, module, ifaces, variables, limit, results, module_qs, processed_pks):
2017 """Process one module for a plain (non-channel) rule. Returns (checked_count, should_stop)."""
2018 bases, children_by_parent = _partition_families(ifaces)
2019 checked = 0
2020 for iface_idx, iface in enumerate(bases):
2021 checked += 1
2022 entry = _evaluate_plain_interface(rule, module, iface, variables, children_by_parent.get(iface.pk, ()))
2023 if entry:
2024 results.append(entry)
2025 if limit is not None and len(results) >= limit:
2026 checked += len(bases) - (iface_idx + 1)
2027 checked += _count_remaining_interfaces(module_qs, processed_pks)
2028 return checked, True
2029 return checked, False
2032def find_interfaces_for_rule(rule, limit=None):
2033 """Find interfaces that would be renamed by applying the given rule retroactively.
2035 Searches for all Module instances matching the rule's criteria and computes
2036 what their interfaces would be renamed to.
2038 Returns a tuple ``(results, total_checked)`` where *results* is a list of dicts::
2040 {
2041 "module": Module instance,
2042 "interface": Interface instance,
2043 "current_name": str,
2044 "new_names": list[str], # one entry per channel, or single-element
2045 "name_details": list[dict], # {"name", "role", "channel_id"} per new_names entry
2046 }
2048 Only includes entries where at least one new_name differs from current_name.
2049 A channelized parent is reported once, with its channels' names in the same entry, so the
2050 caller can act on the family through the parent PK it already submits. If *limit* is set the
2051 list is truncated after that many changed entries, but *total_checked* always reflects the full
2052 count of families examined (a family counts once, however many channels it has).
2053 """
2054 from dcim.models import Interface
2056 module_qs = _build_module_qs(rule).select_related(
2057 "module_type",
2058 "device",
2059 "device__device_type",
2060 "device__platform",
2061 "device__virtual_chassis",
2062 "module_bay",
2063 "module_bay__parent",
2064 )
2065 process_fn = _process_channel_module if rule.channel_count > 0 else _process_plain_module
2067 # Batch-load all interfaces for matching modules to avoid N+1 queries.
2068 ifaces_by_module = defaultdict(list)
2069 for iface in Interface.objects.filter(module__in=module_qs).order_by("module_id", "name"):
2070 ifaces_by_module[iface.module_id].append(iface)
2072 processed_pks = set()
2073 results = []
2074 total_checked = 0
2075 for module in module_qs:
2076 processed_pks.add(module.pk)
2077 variables = build_variables(module.module_bay, device=module.device)
2078 ifaces = ifaces_by_module.get(module.pk, [])
2079 checked, stop = process_fn(rule, module, ifaces, variables, limit, results, module_qs, processed_pks)
2080 total_checked += checked
2081 if stop:
2082 return results, total_checked
2084 return results, total_checked
2087def _apply_channel_rule_to_module(rule, module, ifaces, variables, id_set, conflicts):
2088 """Apply a channel rule to one module via its base interface; return the rename count.
2090 On a module whose base is already channelized the rule renames the existing channels, once per
2091 family. Otherwise the rule is processed ONCE per module (not per interface) so existing
2092 channel names are not re-created. An unexpected failure (e.g. a save race) is logged and
2093 skipped so it never aborts the surrounding batch.
2094 """
2095 bases, children_by_parent = _partition_families(ifaces)
2096 if not bases:
2097 return 0
2098 families = [base for base in bases if _is_channelized_parent(base)]
2099 if families: # pragma: no cover - requires a NetBox that models channelization
2100 count = 0
2101 for parent in families:
2102 if id_set is not None and parent.pk not in id_set:
2103 continue
2104 count += _apply_family(rule, parent, children_by_parent.get(parent.pk, ()), variables, module, conflicts)
2105 return count
2106 base_iface = _find_channel_base(rule, bases, variables)
2107 if id_set is not None and base_iface.pk not in id_set:
2108 return 0
2109 vars_copy = dict(variables)
2110 vars_copy["base"] = base_iface.name
2111 try:
2112 if _is_channelized_rule(rule):
2113 return _apply_channelized_rule(rule, base_iface, variables, module, conflicts) or 0
2114 return _apply_rule_to_interface(rule, base_iface, vars_copy, module, conflicts=conflicts)
2115 except (ValueError, ValidationError, IntegrityError):
2116 logger.exception(
2117 "Failed to apply channel rule '%s' to module '%s' (id=%s); skipping.",
2118 rule,
2119 module,
2120 module.pk,
2121 )
2122 return 0
2125def _apply_family(rule, iface, children, variables, module, conflicts):
2126 """Apply *rule* to one base interface and its channels, logging (never raising) on failure."""
2127 try:
2128 return _apply_rule_with_family(rule, iface, children, variables, module, conflicts) or 0
2129 except (ValueError, ValidationError, IntegrityError):
2130 logger.exception(
2131 "Failed to apply rule '%s' to interface '%s' (id=%s); skipping.",
2132 rule,
2133 iface.name,
2134 iface.pk,
2135 )
2136 return 0
2139def _apply_plain_rule_to_module(rule, module, ifaces, variables, id_set, conflicts):
2140 """Apply a non-channel rule to each selected interface on one module; return the rename count.
2142 Each base interface is independent: an unexpected failure on one is logged and skipped so the
2143 rest of the module (and batch) still process. Channel subinterfaces are not selectable on
2144 their own — they are renamed only as part of the family whose parent was selected.
2145 """
2146 bases, children_by_parent = _partition_families(ifaces)
2147 count = 0
2148 for iface in bases:
2149 if id_set is not None and iface.pk not in id_set:
2150 continue
2151 count += _apply_family(rule, iface, children_by_parent.get(iface.pk, ()), variables, module, conflicts)
2152 return count
2155def apply_rule_to_existing(rule, limit=None, interface_ids=None, conflicts=None):
2156 """Apply a rule retroactively to all matching installed modules.
2158 Unlike apply_interface_name_rules(), this does not skip already-renamed
2159 interfaces — it re-evaluates every interface on each matching module.
2161 For channel rules (channel_count > 0), each module is processed as a single
2162 unit using _find_channel_base() to pick the base interface. Calling
2163 _apply_rule_to_interface for every interface in the module would produce
2164 duplicate-name IntegrityErrors when channel interfaces already exist.
2166 If *interface_ids* is provided (list/set of Interface PKs), only those
2167 interfaces are processed; all others are skipped. For channel rules the
2168 base interface PK is used as the selector. An empty *interface_ids*
2169 collection returns 0 immediately without touching the database. Selecting a
2170 channelized parent brings its channel subinterfaces along; selecting a channel
2171 subinterface on its own does nothing, because it is not an independent candidate.
2173 If *conflicts* is a list, each interface skipped because its target name is
2174 already taken on the device is appended to it (and logged) — letting the
2175 caller report how many renames were dropped. Collisions never raise.
2177 Returns the number of interfaces renamed/created.
2178 """
2179 from dcim.models import Interface
2181 id_set = frozenset(interface_ids) if interface_ids is not None else None
2182 if id_set is not None and not id_set:
2183 return 0
2185 if not rule.enabled:
2186 return 0
2188 module_qs = _build_module_qs(rule)
2190 # Batch-load interfaces to avoid N+1 queries in the module loop.
2191 ifaces_by_module = defaultdict(list)
2192 for iface in Interface.objects.filter(module__in=module_qs).order_by("module_id", "name"):
2193 ifaces_by_module[iface.module_id].append(iface)
2195 count = 0
2196 for module in module_qs.select_related("module_bay", "module_type", "device", "device__virtual_chassis"):
2197 variables = build_variables(module.module_bay, device=module.device)
2198 ifaces = ifaces_by_module.get(module.pk, [])
2200 if rule.channel_count > 0:
2201 count += _apply_channel_rule_to_module(rule, module, ifaces, variables, id_set, conflicts)
2202 else:
2203 count += _apply_plain_rule_to_module(rule, module, ifaces, variables, id_set, conflicts)
2205 if limit is not None and count >= limit:
2206 return count
2208 return count
2211# ---------------------------------------------------------------------------
2212# Assisted flat → channelized conversion
2213# ---------------------------------------------------------------------------
2214# An earlier flat apply leaves N sibling interfaces where NetBox 4.7+ models a channelized parent
2215# with N channel subinterfaces. Converting one rewrites rows an operator owns — cables, addresses,
2216# tags — so it is never a side effect of applying a rule: the operator confirms it per family.
2218# The ch-0 row, the names its family carries now, and the names it would carry once converted.
2219_ConversionFamily = namedtuple("_ConversionFamily", ("module", "base", "current_names", "parent_name", "channel_names"))
2222def _conversion_offered(rule):
2223 """Return True when *rule* describes a topology an installed flat family could be converted into.
2225 A disabled rule renames nothing on any apply path, so it converts nothing either. A flat family
2226 has no parent row — its ch-0 interface *is* the base — so without a parent name there is nowhere
2227 for that base to go, and the conversion is not offered at all.
2228 """
2229 return rule.enabled and _is_channelized_rule(rule) and rule.channel_count > 0 and bool(rule.parent_name_template)
2232def _base_marked_ch0_name(rule, variables): # pragma: no cover - requires channelization support
2233 """Return *rule*'s escaped ch-0 output with ``{base}`` left as a sentinel, or None when it cannot be.
2235 Evaluated once per rule: the sentinel stands in for ``{base}`` so a raw matcher can be spliced
2236 over it afterwards.
2237 """
2238 try:
2239 evaluated = evaluate_name_template(
2240 rule.name_template, {**variables, "base": _BASE_SENTINEL, "channel": str(rule.channel_start)}
2241 )
2242 except (ValueError, TypeError):
2243 # A {base} inside an arithmetic expression cannot take a non-numeric stand-in — see the docs.
2244 logger.debug(
2245 "Rule '%s' evaluates {base} arithmetically, so a base predating a virtual-chassis "
2246 "position change cannot be recovered from its output names; not offering its families.",
2247 rule,
2248 )
2249 return None
2250 if _BASE_SENTINEL not in evaluated:
2251 return None # the rule's output does not carry the base, so no drift reached it
2252 return re.escape(evaluated)
2255def _recovered_bases(rule, interfaces, variables, matchers): # pragma: no cover - channelization only
2256 """Return the historical ``{base}`` values *rule*'s installed families still spell on this module.
2258 A flat family carries rule-*output* names, so a raw matcher cannot be run against them directly:
2259 it is spliced into the rule's own ch-0 output as a capture instead — a repeated ``{base}`` becomes
2260 a backreference rather than a second group — and the capture yields the base the family was named
2261 with. Conversion rewrites rows an operator owns, so anything ambiguous (one matcher over two
2262 bases, or two templates recovering the same one) yields nothing at all.
2263 """
2264 marked = _base_marked_ch0_name(rule, variables)
2265 if marked is None:
2266 return []
2267 head, _, tail = marked.partition(_BASE_SENTINEL)
2268 tail = tail.replace(_BASE_SENTINEL, "(?P=base)")
2269 recovered = defaultdict(int)
2270 for matcher in matchers:
2271 family_pattern = _compile_pattern(f"{head}(?P<base>{matcher.pattern.pattern}){tail}")
2272 if family_pattern is None:
2273 continue
2274 matches = (family_pattern.fullmatch(iface.name) for iface in interfaces)
2275 bases = {match.group("base") for match in matches if match}
2276 if len(bases) == 1:
2277 recovered[bases.pop()] += 1
2278 return [base for base, claims in recovered.items() if claims == 1]
2281def _family_on(rule, module, by_name, variables, base_name): # pragma: no cover - channelization only
2282 """Return the family *rule* describes on *base_name*, or None when this module carries none."""
2283 family_vars = {**variables, "base": base_name}
2284 parent_name = evaluate_name_template(rule.parent_name_template, family_vars)
2285 channel_names = [
2286 evaluate_name_template(rule.name_template, {**family_vars, "channel": str(rule.channel_start + offset)})
2287 for offset in range(rule.channel_count)
2288 ]
2289 base = by_name.get(channel_names[0])
2290 if base is None or _is_channel_child(base) or _is_channelized_parent(base):
2291 return None
2292 return _ConversionFamily(
2293 module=module,
2294 base=base,
2295 current_names=[name for name in channel_names if name in by_name],
2296 parent_name=parent_name,
2297 channel_names=channel_names,
2298 )
2301def _conversion_family(rule, module, interfaces, variables, raw): # pragma: no cover - channelization only
2302 """Return the flat family *rule* would convert on *module*, or None when it carries none.
2304 Identification is by name: ``name_template`` is evaluated over the rule's channel range against
2305 each raw template name, and the ch-0 name has to still be a plain interface — a family that was
2306 already converted (its ch-0 name now belongs to a channel row) is therefore never offered twice.
2307 A family named before this device's virtual-chassis position changed spells a base no template
2308 resolves to any more, so those bases are recovered from the family's own names and then
2309 identified exactly the same way.
2310 """
2311 by_name = {iface.name: iface for iface in interfaces}
2312 for base_name in sorted(raw.names):
2313 family = _family_on(rule, module, by_name, variables, base_name)
2314 if family is not None:
2315 return family
2316 for base_name in _recovered_bases(rule, interfaces, variables, raw.matchers):
2317 family = _family_on(rule, module, by_name, variables, base_name)
2318 if family is not None:
2319 return family
2320 return None
2323def _conversion_families(rule): # pragma: no cover - requires channelization support
2324 """Yield the flat family each module in *rule*'s scope still carries.
2326 A flat breakout is applied once per module (see ``_apply_channel_rule_to_module``), so a module
2327 carries at most one such family and the conversion mirrors that.
2328 """
2329 from dcim.models import Interface
2331 modules = list(_build_module_qs(rule).select_related("module_type", "device", *_BAY_CHAIN_RELATIONS))
2332 raw_by_module = _raw_names_by_module(modules)
2333 ifaces_by_module = defaultdict(list)
2334 for iface in Interface.objects.filter(module__in=[module.pk for module in modules]).order_by("module_id", "name"):
2335 ifaces_by_module[iface.module_id].append(iface)
2336 for module in modules:
2337 variables = build_variables(module.module_bay, device=module.device)
2338 ifaces = ifaces_by_module.get(module.pk, [])
2339 family = _conversion_family(rule, module, ifaces, variables, raw_by_module[module.pk])
2340 if family is not None:
2341 yield family
2344def _validate_or_block(iface, role): # pragma: no cover - requires channelization support
2345 """Run NetBox's own validation on *iface*, restating a rejection as this family's blocking reason."""
2346 try:
2347 iface.full_clean()
2348 except ValidationError as exc:
2349 raise ValidationError(f"{role} {iface.name!r}: {' '.join(exc.messages)}") from exc
2352def _split_ch0_row(rule, family, base): # pragma: no cover - requires channelization support
2353 """Make *base* the family's parent and move its logical identity onto a new channel-1 child.
2355 Everything an operator configured on the ch-0 row described a channel, not the cage carrying it,
2356 so addresses, VLANs, MTU, description and tags move; custom fields can mean either thing and are
2357 copied. The physical row keeps its pk, cable, type, module link and mark_connected.
2358 """
2359 from dcim.choices import InterfaceTypeChoices
2360 from dcim.models import Interface
2362 carried = {
2363 "description": base.description,
2364 "mtu": base.mtu,
2365 "mode": base.mode,
2366 "untagged_vlan_id": base.untagged_vlan_id,
2367 }
2368 tagged_vlans = list(base.tagged_vlans.all())
2369 tags = list(base.tags.all())
2371 base.name = family.parent_name
2372 base.channels = rule.channel_count
2373 base.description = ""
2374 base.mtu = None
2375 base.mode = ""
2376 base.untagged_vlan = None
2377 _validate_or_block(base, "parent")
2378 base.save() # BaseInterface.save() drops the tagged VLANs of an interface that no longer tags
2379 base.tags.clear()
2381 channel = Interface(
2382 device=family.module.device,
2383 module=family.module,
2384 name=family.channel_names[0],
2385 type=InterfaceTypeChoices.TYPE_CHANNEL,
2386 parent=base,
2387 channel_id=1,
2388 enabled=base.enabled,
2389 custom_field_data=dict(base.custom_field_data or {}),
2390 **carried,
2391 )
2392 _validate_or_block(channel, "channel")
2393 channel.save()
2394 channel.tagged_vlans.set(tagged_vlans)
2395 channel.tags.set(tags)
2396 base.ip_addresses.all().update(assigned_object_id=channel.pk)
2397 base.fhrp_group_assignments.all().update(interface_id=channel.pk)
2400def _rewrite_family(rule, family): # pragma: no cover - requires channelization support
2401 """Convert *family* in place, raising ValidationError with the reason when it cannot be converted.
2403 Only what upstream cannot decide for us is checked here: the parent's name has to be free, every
2404 sibling has to be present, a sibling already bound to another parent's channel is not ours to
2405 take, and a cabled sibling cannot become a channel — TYPE_CHANNEL is nonconnectable but not
2406 virtual, so ``Interface.clean()`` accepts a cable on one. Everything else is left to
2407 ``full_clean()`` on each prospective row, which inherits upstream's rules as they grow.
2408 """
2409 from dcim.choices import InterfaceTypeChoices
2410 from dcim.models import Interface
2412 device = family.module.device
2413 # Locked for the transaction: the checks below act on this snapshot, and the saves write it back.
2414 by_name = {iface.name: iface for iface in Interface.objects.select_for_update().filter(module=family.module)}
2415 base = by_name.get(family.channel_names[0])
2416 if base is None or base.pk != family.base.pk:
2417 raise ValidationError(
2418 f"{family.channel_names[0]!r} is gone or replaced: the family changed since it was scanned"
2419 )
2421 if _name_exists_on_device(device, family.parent_name, exclude_pk=base.pk):
2422 raise ValidationError(f"the parent name {family.parent_name!r} is already taken on {device}")
2424 siblings = []
2425 for channel_id, name in enumerate(family.channel_names[1:], start=2):
2426 sibling = by_name.get(name)
2427 if sibling is None or sibling.pk == base.pk:
2428 raise ValidationError(f"{name!r} is missing: this module carries no complete flat family")
2429 # Rebinding it validates cleanly, so only this check keeps the other family whole.
2430 if _is_channel_child(sibling):
2431 owner = sibling.parent.name if sibling.parent_id else "another parent"
2432 raise ValidationError(
2433 f"{name!r} is already channel {sibling.channel_id} of {owner}; "
2434 f"converting would take it out of that family"
2435 )
2436 if sibling.cable_id:
2437 raise ValidationError(f"{name!r} has a cable attached; a channel takes its cable from the parent")
2438 siblings.append((channel_id, sibling))
2440 _split_ch0_row(rule, family, base)
2441 for channel_id, sibling in siblings:
2442 sibling.type = InterfaceTypeChoices.TYPE_CHANNEL
2443 sibling.parent = base
2444 sibling.channel_id = channel_id
2445 _validate_or_block(sibling, "channel")
2446 sibling.save()
2449def _convert_family(rule, family, commit): # pragma: no cover - requires channelization support
2450 """Convert *family*; return an empty string on success, or the reason it was refused.
2452 The whole conversion runs inside one savepoint, so a dry run (*commit* False) and a family that
2453 turns out to be unconvertible both leave every row exactly as it was — the rows are re-read here
2454 too, so a rolled-back dry run cannot hand mutated objects back to the caller.
2455 """
2456 try:
2457 with transaction.atomic():
2458 _rewrite_family(rule, family)
2459 if not commit:
2460 transaction.set_rollback(True)
2461 except (ValidationError, IntegrityError, ValueError) as exc:
2462 return "; ".join(getattr(exc, "messages", [str(exc)]))
2463 return ""
2466def _conversion_metadata_note(family): # pragma: no cover - requires channelization support
2467 """Return the sentence the Apply page shows about where the ch-0 row's configuration ends up."""
2468 return (
2469 f"The addresses, VLANs, MTU, description and tags on {family.base.name} move to the new "
2470 f"channel 1 interface that takes over that name; custom field values are copied. The physical "
2471 f"row keeps its ID and becomes the parent {family.parent_name}, so automation keyed on that "
2472 f"interface ID will address the parent afterwards."
2473 )
2476def _conversion_verdict(family, reason): # pragma: no cover - requires channelization support
2477 """Describe what converting *family* would do, and why it cannot be done when it cannot."""
2478 details = [_name_detail(family.parent_name, "parent")]
2479 details.extend(
2480 _name_detail(name, "channel", channel_id) for channel_id, name in enumerate(family.channel_names, start=1)
2481 )
2482 return {
2483 "module": family.module,
2484 "interface": family.base,
2485 "current_name": family.base.name,
2486 "current_names": family.current_names,
2487 "new_names": [family.parent_name, *family.channel_names],
2488 "name_details": details,
2489 "convertible": not reason,
2490 "reason": reason,
2491 "metadata_note": _conversion_metadata_note(family),
2492 }
2495def find_convertible_families(rule, limit=None) -> tuple:
2496 """Return ``(verdicts, has_more)`` for the flat families *rule* could convert, convertible or not.
2498 Nothing is written: every family is converted inside a savepoint that is rolled back again, so
2499 each verdict carries the reason NetBox itself would refuse the family rather than a guess at its
2500 rules. Each verdict names the ch-0 row the confirm form submits, the family's current names,
2501 the names it would carry, and where the ch-0 row's configuration lands.
2503 That dry run is what the scan costs, and a blocked family costs it too, so *limit* caps the
2504 families examined — one verdict each — rather than the convertible ones among them. A family
2505 beyond the limit is never dry-run; *has_more* reports that one was left unexamined.
2506 """
2507 if not (_conversion_offered(rule) and supports_channelization()):
2508 return [], False
2509 return _find_convertible_families(rule, limit) # pragma: no cover - requires channelization support
2512def _find_convertible_families(rule, limit): # pragma: no cover - requires channelization support
2513 """Dry-run at most *limit* of *rule*'s flat families; see ``find_convertible_families``."""
2514 verdicts = []
2515 for family in _conversion_families(rule):
2516 if limit is not None and len(verdicts) >= limit:
2517 return verdicts, True
2518 verdicts.append(_conversion_verdict(family, _convert_family(rule, family, commit=False)))
2519 return verdicts, False
2522def convert_flat_families(rule, base_pks=None, conflicts=None) -> int:
2523 """Convert *rule*'s installed flat families to the channelized topology; return how many.
2525 *base_pks* is the set of ch-0 interface pks the operator confirmed: ``None`` converts every
2526 convertible family (the batch the background job runs), an empty collection converts none. A
2527 family that cannot be converted is logged, appended to *conflicts* in the usual skipped-rename
2528 shape and passed over — it is never half converted, and never costs the rest of the batch.
2529 """
2530 if not supports_channelization():
2531 logger.warning(
2532 "Rule '%s' converts flat families into the channelized topology, which this NetBox release "
2533 "cannot model; nothing was converted.",
2534 rule,
2535 )
2536 return 0
2537 return _convert_flat_families(rule, base_pks, conflicts) # pragma: no cover - see above
2540def _convert_flat_families(rule, base_pks, conflicts): # pragma: no cover - requires channelization support
2541 """Convert the confirmed flat families of *rule*; see ``convert_flat_families``."""
2542 if not _conversion_offered(rule):
2543 return 0
2544 selected = None if base_pks is None else frozenset(base_pks)
2545 if selected is not None and not selected:
2546 return 0
2548 converted = 0
2549 for family in _conversion_families(rule):
2550 if selected is not None and family.base.pk not in selected:
2551 continue
2552 current_name = family.base.name
2553 reason = _convert_family(rule, family, commit=True)
2554 if reason:
2555 logger.warning(
2556 "Cannot convert the flat family of interface %r on %s into the channelized parent %r: %s. Skipping.",
2557 current_name,
2558 family.module,
2559 family.parent_name,
2560 reason,
2561 )
2562 _record_skip(conflicts, family.module.device, current_name, family.parent_name, family.base.pk)
2563 continue
2564 converted += 1
2565 return converted
2568def evaluate_name_template(template: str, variables: dict) -> str:
2569 """Evaluate a name template with variable substitution and safe arithmetic.
2571 Supports templates like:
2572 "GigabitEthernet{slot}/{8 + ({parent_bay_position} - 1) * 2 + {sfp_slot}}"
2574 Variables are substituted first, then any brace-enclosed expression
2575 containing arithmetic operators is safely evaluated via AST. True division
2576 (/) is not allowed — use floor division (//) instead. Results are cast to
2577 int to ensure interface names are always whole numbers.
2578 """
2579 # First pass: substitute all simple variables
2580 result = template
2581 for key, value in variables.items():
2582 result = result.replace(f"{{{key}}}", str(value))
2584 # Second pass: evaluate any remaining brace-enclosed arithmetic expressions
2585 def _eval_expr(match):
2586 expr = match.group(1).strip()
2587 # Allow digits, arithmetic operators (excluding lone /), parens, whitespace.
2588 # Negative lookahead disallows a single / that is not part of //.
2589 if not re.match(r"^(?!.*(?<!/)/(?!/))[\d\s\+\-\*\(\/\)]+$", expr):
2590 raise ValueError(f"Unsafe expression in name template: {expr}")
2591 try:
2592 node = ast.parse(expr, mode="eval")
2593 for child in ast.walk(node):
2594 if not isinstance(
2595 child,
2596 (
2597 ast.Expression,
2598 ast.BinOp,
2599 ast.UnaryOp,
2600 ast.Constant,
2601 ast.Add,
2602 ast.Sub,
2603 ast.Mult,
2604 ast.FloorDiv,
2605 ast.USub,
2606 ast.UAdd,
2607 ),
2608 ):
2609 raise ValueError(f"Unsafe AST node in expression: {type(child).__name__}")
2610 return str(int(eval(compile(node, "<template>", "eval")))) # noqa: S307
2611 except (SyntaxError, TypeError, ZeroDivisionError) as e:
2612 raise ValueError(f"Invalid arithmetic expression '{expr}': {e}") from e
2614 return re.sub(r"\{([^}]+)\}", _eval_expr, result)