Coverage for netbox_interface_name_rules / rule_selection.py: 98.31%
134 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-10-16 10:13 +0000
1# SPDX-License-Identifier: Apache-2.0
2# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com>
3"""Select an enabled interface-name rule for a module context."""
5import contextlib
6import threading
7from dataclasses import dataclass
9from django.core.exceptions import ValidationError
10from django.db.models import Aggregate, F, TextField, Value
11from django.db.models.functions import Cast, Coalesce, Concat, Length
13from .regex_safety import compile_module_type_pattern
16@dataclass(frozen=True, slots=True)
17class _RuleSnapshot:
18 """One enabled-rule set read from one alias; only the memo changes after publication."""
20 alias: str | None
21 version: str | None
22 exact: tuple
23 regex: tuple
24 memo: dict
27def _empty_rule_cache():
28 """Return a new snapshot that matches no alias, so the next read reloads."""
29 return _RuleSnapshot(alias=None, version=None, exact=(), regex=(), memo={})
32# A reload rebinds this to a new snapshot, so a reader never mixes entries from two versions.
33_RULE_CACHE = _empty_rule_cache()
35# Bound the number of module and scope contexts retained for one rule-set version.
36_MEMO_MAX = 4096
37_MEMO_MISS = object()
39# A pinned batch owns a thread-local rule-set snapshot and a private memo.
40_pin = threading.local()
43@contextlib.contextmanager
44def pinned_rule_cache():
45 """Pin one enabled-rule snapshot for all selections inside the block.
47 The first selection loads and fingerprints the rule set. Later selections in
48 the same thread and from the same read alias skip the fingerprint query.
49 Nested blocks share the snapshot. The pin is thread-local, and an empty block
50 does not load rules.
51 """
52 depth = getattr(_pin, "depth", 0)
53 _pin.depth = depth + 1
54 if depth == 0: 54 ↛ 56line 54 didn't jump to line 56 because the condition on line 54 was always true
55 _pin.primed = False
56 try:
57 yield
58 finally:
59 _pin.depth -= 1
60 if _pin.depth == 0: 60 ↛ exitline 60 didn't return from function 'pinned_rule_cache' because the condition on line 60 was always true
61 _pin.primed = False
62 for attr in ("alias", "exact", "regex", "memo"):
63 _pin.__dict__.pop(attr, None)
66def module_types_matching_pattern(pattern):
67 """Return the model names *pattern* matches among the module types that exist now.
69 Compiles and full-matches exactly as selection does, so a pattern that returns nothing here
70 also selects nothing when a module is installed.
71 """
72 from dcim.models import ModuleType
74 compiled = compile_module_type_pattern(pattern)
75 return tuple(model for model in ModuleType.objects.values_list("model", flat=True) if compiled.fullmatch(model))
78def _enabled_module_rules():
79 """Return the queryset of the enabled rules that module selection can pick.
81 Device-interface rows reuse module_type_pattern as an interface-name filter, so they are never module rules.
82 """
83 from .models import InterfaceNameRule
85 return InterfaceNameRule.objects.filter(enabled=True, applies_to_device_interfaces=False)
88def parent_type_scopes_a_rule(module_type_ids) -> bool:
89 """Return whether an enabled module rule has one of *module_type_ids* as its parent module type.
91 One EXISTS query: the rule cache would load every rule during the save when it is not current.
92 """
93 return _enabled_module_rules().filter(parent_module_type_id__in=module_type_ids).exists()
96def compile_stored_pattern(pattern):
97 """Compile a stored pattern once, or return None when RE2 rejects it."""
98 try:
99 return compile_module_type_pattern(pattern)
100 except ValidationError:
101 return None
104# These fields can change either matching or the selected rule's output. The row
105# identity prevents compensating edits across two rules from preserving the hash.
106_VERSION_COLUMNS = (
107 "id",
108 "module_type_id",
109 "module_type_is_regex",
110 "module_type_pattern",
111 "parent_module_type_id",
112 "device_type_id",
113 "platform_id",
114 "name_template",
115 "parent_name_template",
116 "breakout_mode",
117 "channel_count",
118 "channel_start",
119 "applies_to_device_interfaces",
120)
123class _Md5OrderedStringAgg(Aggregate):
124 """Build ``md5(string_agg(<row>, <delimiter> ORDER BY id))``."""
126 function = "STRING_AGG"
127 template = "MD5(%(function)s(%(expressions)s ORDER BY id))"
128 output_field = TextField()
131def _version_row_signature():
132 """Build an unambiguous, length-prefixed signature for one rule row."""
133 empty = Value("", output_field=TextField())
134 colon = Value(":", output_field=TextField())
135 parts = []
136 for column in _VERSION_COLUMNS:
137 cast = Cast(F(column), output_field=TextField())
138 value = Coalesce(cast, empty, output_field=TextField()) if column.endswith("_id") else cast
139 parts.extend((Cast(Length(value), output_field=TextField()), colon, value))
140 return Concat(*parts, output_field=TextField())
143_ROW_SIGNATURE = _version_row_signature()
146def _enabled_rules_version(alias):
147 """Return a deterministic content fingerprint of all enabled rules on *alias*.
149 PostgreSQL hashes the matching and output columns in primary-key order. Each
150 value is length-prefixed, so arbitrary text cannot create field or row boundary
151 collisions. The empty rule set has a stable empty fingerprint.
152 """
153 from .models import InterfaceNameRule
155 return (
156 InterfaceNameRule.objects.using(alias)
157 .filter(enabled=True)
158 .aggregate(
159 fingerprint=Coalesce(
160 _Md5OrderedStringAgg(_ROW_SIGNATURE, Value("", output_field=TextField())),
161 Value("", output_field=TextField()),
162 )
163 )["fingerprint"]
164 )
167def _get_enabled_rules():
168 """Return the exact rules, regex rules, and memo for the current version on the read alias.
170 Exact rules retain model ordering, which reduces to primary-key order for one
171 module type. Regex rules are compiled once and ordered by decreasing pattern
172 length, then primary key. A reload publishes one new cache snapshot so a
173 concurrent reader cannot combine values from two versions.
174 """
175 # One module-level cache, replaced atomically.
176 global _RULE_CACHE # noqa: PLW0603
178 # A branch can hold other rules than main, so a snapshot belongs to the alias it was read from.
179 alias = _enabled_module_rules().db
180 pinned = getattr(_pin, "depth", 0) > 0
181 if pinned and getattr(_pin, "primed", False) and _pin.alias == alias:
182 # Return the thread's snapshot. Another thread can replace the shared cache.
183 return _pin.exact, _pin.regex, _pin.memo
185 cache = _RULE_CACHE
186 version = _enabled_rules_version(alias)
187 if (cache.alias, cache.version) != (alias, version):
188 rules = list(_enabled_module_rules().using(alias).order_by("module_type__model", "pk"))
189 exact = tuple(rule for rule in rules if not rule.module_type_is_regex)
190 regex_rules = sorted(
191 (rule for rule in rules if rule.module_type_is_regex),
192 key=lambda rule: (-len(rule.module_type_pattern or ""), rule.pk),
193 )
194 regex = tuple((compile_stored_pattern(rule.module_type_pattern), rule) for rule in regex_rules)
195 cache = _RuleSnapshot(alias=alias, version=version, exact=exact, regex=regex, memo={})
196 _RULE_CACHE = cache
198 if pinned:
199 # Keep a private memo so another thread cannot clear this batch's entries.
200 _pin.alias = alias
201 _pin.exact = cache.exact
202 _pin.regex = cache.regex
203 _pin.memo = dict(cache.memo)
204 _pin.primed = True
205 return _pin.exact, _pin.regex, _pin.memo
207 return cache.exact, cache.regex, cache.memo
210def _scope_ids(parent_module_type, device_type, platform):
211 """Map optional scope objects to their foreign-key values."""
212 return (
213 parent_module_type.pk if parent_module_type is not None else None,
214 device_type.pk if device_type is not None else None,
215 platform.pk if platform is not None else None,
216 )
219def _rule_scope_matches(rule, scope_ids):
220 """Return whether a rule has exactly the requested scope values."""
221 parent_module_type_id, device_type_id, platform_id = scope_ids
222 return (
223 rule.parent_module_type_id == parent_module_type_id
224 and rule.device_type_id == device_type_id
225 and rule.platform_id == platform_id
226 )
229def _build_candidates(parent_module_type, device_type, platform) -> list:
230 """Build scope combinations from most specific to least specific."""
231 seen: set = set()
232 candidates = []
233 parent_options = [parent_module_type, None] if parent_module_type else [None]
234 device_options = [device_type, None] if device_type else [None]
235 platform_options = [platform, None] if platform else [None]
236 for parent in parent_options:
237 for device in device_options:
238 for candidate_platform in platform_options:
239 candidate = (parent, device, candidate_platform)
240 if candidate not in seen: 240 ↛ 238line 240 didn't jump to line 238 because the condition on line 240 was always true
241 seen.add(candidate)
242 candidates.append(candidate)
243 return candidates
246def _find_exact_match(module_type, candidates, exact_rules=None):
247 """Return the first enabled exact rule in scope-precedence order."""
248 if exact_rules is None:
249 exact_rules, _, _ = _get_enabled_rules()
251 scoped_rules = [rule for rule in exact_rules if rule.module_type_id == module_type.pk]
252 for candidate in candidates:
253 scope_ids = _scope_ids(*candidate)
254 for rule in scoped_rules:
255 if _rule_scope_matches(rule, scope_ids):
256 return rule
257 return None
260def _find_regex_match(model_name: str, candidates, regex_rules=None):
261 """Return the first enabled regex rule in scope-precedence order."""
262 if regex_rules is None:
263 _, regex_rules, _ = _get_enabled_rules()
265 for candidate in candidates:
266 scope_ids = _scope_ids(*candidate)
267 for compiled, rule in regex_rules:
268 if compiled is not None and _rule_scope_matches(rule, scope_ids) and compiled.fullmatch(model_name):
269 return rule
270 return None
273def find_matching_rule(module_type, parent_module_type, device_type, platform=None):
274 """Return the most specific enabled rule for a module and scope context.
276 Exact module-type rules take priority over regular-expression rules. Within
277 each tier, parent module type, device type, and platform scopes are tried from
278 most specific to least specific.
279 """
280 if module_type is None:
281 return None
283 exact_rules, regex_rules, memo = _get_enabled_rules()
284 signature = (
285 module_type.pk,
286 # Regex matching reads the live model name, so the memo key must include it.
287 module_type.model,
288 *_scope_ids(parent_module_type, device_type, platform),
289 )
290 # A single dictionary read cannot race with another thread's memo clear between
291 # a membership check and a later subscript.
292 cached = memo.get(signature, _MEMO_MISS)
293 if cached is not _MEMO_MISS:
294 return cached
296 candidates = _build_candidates(parent_module_type, device_type, platform)
297 result = _find_exact_match(module_type, candidates, exact_rules) or _find_regex_match(
298 module_type.model,
299 candidates,
300 regex_rules,
301 )
302 if len(memo) >= _MEMO_MAX:
303 memo.clear()
304 memo[signature] = result
305 return result