Coverage for netbox_interface_name_rules / models.py: 100.00%
132 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>
3import inspect
5from dcim.models import DeviceType, ModuleType, Platform
6from django.core.exceptions import ValidationError
7from django.db import models, router
8from django.urls import reverse
9from netbox.models import NetBoxModel
10from taggit.managers import TaggableManager
12from .branching import replay_in_progress
13from .choices import BreakoutModeChoices
14from .name_template import validate_rule
15from .regex_safety import compile_module_type_pattern
16from .transactions import atomic_with_events
19def csv_export_entry(headers, values):
20 """Pair CSV headers with values, dropping empties but always keeping ``name_template``."""
21 return {
22 header: value
23 for header, value in zip(headers, values, strict=True)
24 if (value != "" and value is not None) or header == "name_template"
25 }
28DEVICE_RULE_PARENT_MODULE_TYPE_ERROR = "Parent module type must be empty for device-level interface rules."
29_RULE_VALIDATION_FIELDS = frozenset(inspect.signature(validate_rule).parameters)
32class InterfaceNameRule(NetBoxModel):
33 """Post-install interface rename rule for module types.
35 Handles cases where NetBox's position-based naming can't produce
36 the correct interface name, such as converter offset (CVR-X2-SFP)
37 or breakout transceivers (QSFP+ 4x10G).
39 ``name_template.TEMPLATE_VARIABLES`` is the catalogue for the variables available in each
40 naming context. The language substitutes those variables, then evaluates supported integer
41 arithmetic in each remaining brace group.
43 A device type may compose the parent into a bay position, so a position can be
44 path-shaped, such as "TenGigabitEthernet3/2/1". Arithmetic takes the _num form.
46 Module type matching supports two modes:
47 - Exact: FK reference to a specific ModuleType (default)
48 - Regex: RE2 pattern matched against the complete ModuleType.model value
50 Scoping fields (all optional):
51 - parent_module_type: match only when installed inside this module type
52 - device_type: match only devices of this hardware model
53 - platform: match only devices running this software platform/OS
54 """
56 module_type = models.ForeignKey(
57 ModuleType,
58 on_delete=models.CASCADE,
59 null=True,
60 blank=True,
61 related_name="+",
62 verbose_name="Module Type",
63 help_text="The module type whose installation triggers this rename rule (exact match)",
64 )
65 module_type_pattern = models.CharField(
66 max_length=255,
67 blank=True,
68 default="",
69 verbose_name="Module Type Pattern",
70 help_text=(
71 "RE2 pattern matched against the complete module type model name (e.g. 'QSFP-DD-400G-.*'). "
72 "With Applies to Device Interfaces enabled it filters the complete interface name instead."
73 ),
74 )
75 module_type_is_regex = models.BooleanField(
76 default=False,
77 verbose_name="Use Regex Pattern",
78 help_text="When enabled, use regex pattern instead of exact module type FK",
79 )
80 parent_module_type = models.ForeignKey(
81 ModuleType,
82 on_delete=models.SET_NULL,
83 null=True,
84 blank=True,
85 related_name="+",
86 verbose_name="Parent Module Type",
87 help_text=(
88 "If set, rule only applies when installed inside this parent module type. "
89 "Must be empty for device-level interface rules."
90 ),
91 )
92 device_type = models.ForeignKey(
93 DeviceType,
94 on_delete=models.SET_NULL,
95 null=True,
96 blank=True,
97 related_name="+",
98 verbose_name="Device Type",
99 help_text="If set, rule only applies to devices of this device type",
100 )
101 platform = models.ForeignKey(
102 Platform,
103 on_delete=models.SET_NULL,
104 null=True,
105 blank=True,
106 related_name="+",
107 verbose_name="Platform",
108 help_text="If set, rule only applies to devices running this software platform/OS",
109 )
110 name_template = models.CharField(
111 max_length=255,
112 help_text=(
113 "Interface name template expression, e.g. "
114 "'GigabitEthernet{slot_num}/{8 + ({parent_bay_position_num} - 1) * 2 + {sfp_slot}}'"
115 ),
116 )
117 parent_name_template = models.CharField(
118 max_length=255,
119 blank=True,
120 default="",
121 verbose_name="Parent Name Template",
122 help_text=(
123 "Optional name template for the channelized parent interface, e.g. 'et-0/0/{bay_position}'. "
124 "Same variables as the name template, minus {channel}. Blank leaves the parent's current name."
125 ),
126 )
127 breakout_mode = models.CharField(
128 max_length=20,
129 choices=BreakoutModeChoices,
130 default=BreakoutModeChoices.FLAT,
131 verbose_name="Breakout Mode",
132 help_text=(
133 "Topology a breakout rule produces: 'flat' renames the base to the first channel and creates "
134 "the remaining channels as sibling interfaces; 'channelized' turns the base into a channelized "
135 "parent with one channel subinterface per channel (requires a NetBox that models channels)."
136 ),
137 )
138 channel_count = models.PositiveSmallIntegerField(
139 default=0,
140 help_text="Number of breakout channels (0 = no breakout). Creates this many interfaces per template.",
141 )
142 channel_start = models.PositiveSmallIntegerField(
143 default=0,
144 help_text="Starting channel number for breakout interfaces (e.g., 0 for Juniper; Cisco varies by model—check device docs)",
145 )
146 description = models.TextField(
147 blank=True,
148 help_text="Optional description or notes about this rule",
149 )
150 enabled = models.BooleanField(
151 default=True,
152 help_text="When disabled, this rule is ignored during module installation and Apply Rules operations.",
153 )
154 applies_to_device_interfaces = models.BooleanField(
155 default=False,
156 verbose_name="Applies to Device Interfaces",
157 help_text=(
158 "When enabled, this rule renames device-level interfaces (module=None) when the device "
159 "joins or changes position in a Virtual Chassis. "
160 "The Module Type and Parent Module Type fields must be empty; the Module Type Pattern (if set) "
161 "is used as a regex to filter which interface names to rename."
162 ),
163 )
165 # Override inherited tags to avoid reverse accessor clash when co-installed
166 # with another plugin that has a model of the same name.
167 tags = TaggableManager(through="extras.TaggedItem", related_name="+")
169 def clean(self):
170 """Validate regex/FK mode exclusivity and required fields."""
171 super().clean()
172 self._normalise_mode()
173 if self.applies_to_device_interfaces:
174 # Device-level rules must not reference a module type
175 if self.module_type:
176 raise ValidationError({"module_type": "Module type must be empty for device-level interface rules."})
177 if self.parent_module_type_id:
178 raise ValidationError({"parent_module_type": DEVICE_RULE_PARENT_MODULE_TYPE_ERROR})
179 # module_type_pattern is an optional interface-name filter regex
180 if self.module_type_pattern:
181 compile_module_type_pattern(self.module_type_pattern)
182 elif self.module_type_is_regex:
183 if not self.module_type_pattern:
184 raise ValidationError({"module_type_pattern": "Regex pattern is required when regex mode is enabled."})
185 if self.module_type:
186 raise ValidationError({"module_type": "Cannot set both module type FK and regex pattern. Choose one."})
187 compile_module_type_pattern(self.module_type_pattern)
188 elif not self.module_type:
189 raise ValidationError({"module_type": "Module type is required when regex mode is disabled."})
190 validate_rule(**self._rule_values())
192 def _normalise_mode(self):
193 """Clear the mode fields that the rule's mode gives no meaning."""
194 if self.applies_to_device_interfaces:
195 self.module_type_is_regex = False
196 elif not self.module_type_is_regex:
197 self.module_type_pattern = ""
199 def _rule_values(self, fields=_RULE_VALIDATION_FIELDS):
200 """Return the in-memory values of the named validate_rule() fields."""
201 return {field: getattr(self, field) for field in fields}
203 def _loaded_fields(self):
204 """Return the loaded fields Model.save() writes when a field is deferred, else None."""
205 fields = self._meta.concrete_fields
206 deferred = self.get_deferred_fields() - {field.attname for field in fields if field.generated}
207 loaded = {field.attname for field in fields if not field.primary_key} - deferred
208 return frozenset(loaded) if deferred and loaded else None
210 def get_absolute_url(self):
211 """Return the detail URL for this rule."""
212 return reverse("plugins:netbox_interface_name_rules:interfacenamerule_detail", args=[self.pk])
214 clone_fields = [
215 "module_type",
216 "module_type_pattern",
217 "module_type_is_regex",
218 "parent_module_type",
219 "device_type",
220 "platform",
221 "name_template",
222 "parent_name_template",
223 "breakout_mode",
224 "channel_count",
225 "channel_start",
226 "description",
227 "enabled",
228 "applies_to_device_interfaces",
229 ]
231 def get_breakout_mode_color(self):
232 """Return the badge colour NetBox renders the breakout mode with."""
233 return BreakoutModeChoices.colors.get(self.breakout_mode)
235 @property
236 def specificity_score(self) -> int:
237 """Numeric priority score — higher beats lower in rule lookup.
239 The engine selects rules in this order (``find_matching_rule``):
241 1. **Exact FK match** always outranks regex at any scope.
242 2. **Scope specificity** (more constraints = higher priority):
243 parent_module_type contributes 4 pts, device_type 2 pts, platform 1 pt.
244 This mirrors the candidate-iteration order in the engine.
245 3. **Regex pattern length** (longer = more specific string match).
247 Score layout:
248 - Exact FK rules: 1000 + scope (1000–1007)
249 - Regex rules: scope × 100 + len(pattern)
250 e.g. device-scoped 15-char pattern → 2×100+15 = 215
251 platform-scoped 2-char pattern → 1×100+2 = 102
252 Exact rules always outrank regex (max regex score with scope=7,
253 max_length=255 would be 7×100+255 = 955, so 1000 safely exceeds
254 any possible regex score).
256 Scope bit weights: parent_module_type=4, device_type=2, platform=1.
257 A device-level rule takes no parent_module_type, so its scope is at most 3.
258 Two rules with the same score fall back to lowest pk (first created).
259 """
260 scope = (
261 (4 if self.parent_module_type_id else 0)
262 + (2 if self.device_type_id else 0)
263 + (1 if self.platform_id else 0)
264 )
265 if not self.module_type_is_regex:
266 return 1000 + scope
267 return scope * 100 + len(self.module_type_pattern)
269 @property
270 def specificity_label(self) -> str:
271 """Short human-readable description of what this rule matches."""
272 if self.applies_to_device_interfaces:
273 mode = f"iface-filter({len(self.module_type_pattern)})" if self.module_type_pattern else "iface-filter(*)"
274 else:
275 mode = "exact" if not self.module_type_is_regex else f"regex({len(self.module_type_pattern)})"
276 parts = []
277 if self.parent_module_type_id:
278 parts.append("parent")
279 if self.device_type_id:
280 parts.append("device")
281 if self.platform_id:
282 parts.append("platform")
283 scope = "+".join(parts) if parts else "global"
284 return f"{mode} / {scope}"
286 class Meta:
287 ordering = ["module_type__model", "pk"]
288 constraints = [
289 models.CheckConstraint(
290 condition=(
291 models.Q(
292 applies_to_device_interfaces=True,
293 module_type__isnull=True,
294 module_type_is_regex=False,
295 )
296 | models.Q(
297 applies_to_device_interfaces=False,
298 module_type_is_regex=True,
299 module_type__isnull=True,
300 module_type_pattern__gt="",
301 )
302 | models.Q(
303 applies_to_device_interfaces=False,
304 module_type_is_regex=False,
305 module_type__isnull=False,
306 )
307 ),
308 name="interfacenamerule_module_type_mode_check",
309 ),
310 models.CheckConstraint(
311 # Enforce the enum and integer implications of validate_rule() as ~P | Q.
312 condition=(
313 models.Q(breakout_mode__in=[BreakoutModeChoices.FLAT, BreakoutModeChoices.CHANNELIZED])
314 & (
315 models.Q(applies_to_device_interfaces=False)
316 | ~models.Q(breakout_mode=BreakoutModeChoices.CHANNELIZED)
317 )
318 & (models.Q(applies_to_device_interfaces=False) | models.Q(parent_name_template=""))
319 & (models.Q(parent_name_template="") | models.Q(breakout_mode=BreakoutModeChoices.CHANNELIZED))
320 & (~models.Q(breakout_mode=BreakoutModeChoices.CHANNELIZED) | ~models.Q(channel_count=0))
321 ),
322 name="interfacenamerule_breakout_topology_check",
323 ),
324 models.CheckConstraint(
325 condition=models.Q(applies_to_device_interfaces=False) | models.Q(parent_module_type__isnull=True),
326 name="interfacenamerule_device_rule_scope_check",
327 ),
328 models.UniqueConstraint(
329 fields=["module_type", "parent_module_type", "device_type", "platform"],
330 condition=models.Q(module_type_is_regex=False, applies_to_device_interfaces=False),
331 nulls_distinct=False,
332 name="interfacenamerule_unique_exact",
333 ),
334 models.UniqueConstraint(
335 fields=["module_type_pattern", "parent_module_type", "device_type", "platform"],
336 condition=models.Q(module_type_is_regex=True),
337 nulls_distinct=False,
338 name="interfacenamerule_unique_regex",
339 ),
340 models.UniqueConstraint(
341 fields=["module_type_pattern", "device_type", "platform"],
342 condition=models.Q(applies_to_device_interfaces=True),
343 nulls_distinct=False,
344 name="interfacenamerule_unique_device_iface",
345 ),
346 ]
348 def save(self, **kwargs):
349 """Normalise the mode fields and validate topology and templates before a plain ORM write."""
350 using = kwargs.get("using") or router.db_for_write(self.__class__, instance=self)
351 # A merge started in an active branch replays its changes on default.
352 if not replay_in_progress() and using != (routed := router.db_for_write(self.__class__)):
353 raise RuntimeError(f"A rule save writes to {using!r}, but the router gives {routed!r} for a rule.")
354 update_fields = kwargs.get("update_fields")
355 if update_fields is not None:
356 # Django accepts any iterable. Reading a generator here would leave Django an empty
357 # one, and it skips the write when update_fields is empty.
358 update_fields = frozenset(update_fields)
359 elif not kwargs.get("force_insert") and using == self._state.db and self.pk is not None:
360 # Name Model.save()'s implicit update_fields now: a validation read would load the deferred fields.
361 update_fields = self._loaded_fields()
362 if update_fields is None:
363 # The API saves its validated attrs, not the instance clean() normalised.
364 self._normalise_mode()
365 validate_rule(**self._rule_values())
366 return super().save(**kwargs)
367 kwargs["update_fields"] = update_fields
368 if written := _RULE_VALIDATION_FIELDS.intersection(update_fields):
369 # Validate the stored row on the alias Model.save() writes to, locked against a concurrent save.
370 kwargs["using"] = using
371 with atomic_with_events() as block:
372 # netbox-branching routes an exempted rule model to default, which is an alias of every scope.
373 if using not in block.aliases:
374 raise RuntimeError(f"A rule save writes to {using!r}, outside the write scope {block.aliases}.")
375 stored = self.__class__._base_manager.using(using).select_for_update().filter(pk=self.pk)
376 # No ordering: the default one joins a nullable module type, which FOR UPDATE refuses.
377 row = stored.order_by().values("pk", *(_RULE_VALIDATION_FIELDS - written)).first()
378 # A missing row is left to Django, which refuses the update itself.
379 if row is not None:
380 del row["pk"]
381 validate_rule(**(self._rule_values(written) | row))
382 return super().save(**kwargs)
383 return super().save(**kwargs)
385 def __str__(self):
386 if self.module_type_is_regex:
387 module = f"/{self.module_type_pattern}/"
388 else:
389 module = self.module_type.model if self.module_type else "?"
390 parent = f" in {self.parent_module_type.model}" if self.parent_module_type else ""
391 device = f" on {self.device_type.model}" if self.device_type else ""
392 platform = f" [{self.platform.name}]" if self.platform else ""
393 return f"{module}{parent}{device}{platform} → {self.name_template}"
395 csv_headers = [
396 "module_type",
397 "module_type_pattern",
398 "module_type_is_regex",
399 "parent_module_type",
400 "device_type",
401 "platform",
402 "name_template",
403 "parent_name_template",
404 "breakout_mode",
405 "channel_count",
406 "channel_start",
407 "description",
408 "enabled",
409 "applies_to_device_interfaces",
410 ]
412 def to_csv(self):
413 """Return a tuple of field values for CSV export (matches csv_headers order)."""
414 return (
415 self.module_type.model if self.module_type else "",
416 self.module_type_pattern,
417 self.module_type_is_regex,
418 self.parent_module_type.model if self.parent_module_type else "",
419 self.device_type.model if self.device_type else "",
420 self.platform.name if self.platform else "",
421 self.name_template,
422 self.parent_name_template,
423 self.breakout_mode,
424 self.channel_count,
425 self.channel_start,
426 self.description,
427 self.enabled,
428 self.applies_to_device_interfaces,
429 )
431 def to_yaml(self):
432 """Return a YAML document for this rule (used by NetBox's built-in Export)."""
433 import yaml
435 entry = csv_export_entry(self.csv_headers, self.to_csv())
436 return yaml.dump([entry], default_flow_style=False, allow_unicode=True, sort_keys=False)