« prev ^ index » next coverage.py v7.14.3, created at 2026-07-01 19:09 +0000
1# SPDX-License-Identifier: Apache-2.0
2# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com>
3"""Django signals for automatic interface renaming on module install."""
5import functools
6import logging
8from django.db import transaction
9from django.db.models.signals import post_save, pre_save
10from django.dispatch import receiver
12logger = logging.getLogger("netbox_interface_name_rules")
14# NOTE: We intentionally do NOT hook into pre_save on dcim.Interface.
15#
16# NetBox's Module.save() creates interfaces via bulk_create() which bypasses
17# pre_save signals entirely. NetBox then manually dispatches post_save for
18# the Module object. The actual renaming path is therefore:
19#
20# Module.save()
21# → bulk_create() all interfaces (pre_save never fires)
22# → NetBox fires post_save(Module)
23# → on_module_saved → transaction.on_commit
24# → _apply_rules_deferred → apply_interface_name_rules()
25#
26# Adding a pre_save on Interface would be a no-op for the normal install
27# path and would create a false sense of security.
30@receiver(pre_save, sender="dcim.Module", dispatch_uid="interface_name_rules_pre_save_module")
31def on_module_pre_save(sender, instance, **kwargs):
32 """Capture old module_type_id before a Module save (for type-change detection)."""
33 if instance.pk is None:
34 instance._prev_module_type_id = None
35 return
36 try:
37 old = sender.objects.filter(pk=instance.pk).values("module_type_id").first()
38 instance._prev_module_type_id = old["module_type_id"] if old else None
39 except Exception:
40 logger.warning("Failed to capture previous module_type_id for Module pk=%s", instance.pk, exc_info=True)
41 instance._prev_module_type_id = None
44@receiver(post_save, sender="dcim.Module", dispatch_uid="interface_name_rules_post_save_module")
45def on_module_saved(sender, instance, created, **kwargs):
46 """Apply interface name rules after module install or module-type change.
48 Primary renaming path: NetBox's Module.save() creates interfaces via
49 bulk_create() and then manually dispatches post_save for the Module.
50 This handler defers the actual renaming to on_commit so that all
51 interfaces are visible in the DB before apply_interface_name_rules() runs.
53 Also handles module-type changes: when an existing module's type is changed
54 via the API (PATCH), force-reapply rules to rename existing interfaces
55 according to the new module type's rule.
56 """
57 module = instance
58 module_bay = getattr(module, "module_bay", None)
59 if not module_bay:
60 return
62 if created:
63 callback = functools.partial(_apply_rules_deferred, module.pk, module_bay.pk)
64 transaction.on_commit(callback)
65 return
67 # Module type change: re-apply rules with force_reapply=True
68 prev_module_type_id = getattr(instance, "_prev_module_type_id", None)
69 if prev_module_type_id is None:
70 return
71 if prev_module_type_id == instance.module_type_id:
72 return
74 logger.debug(
75 "Module %s type changed (%s → %s) — scheduling rule re-apply",
76 instance.pk,
77 prev_module_type_id,
78 instance.module_type_id,
79 )
80 callback = functools.partial(_apply_rules_deferred, module.pk, module_bay.pk, force_reapply=True)
81 transaction.on_commit(callback)
84def _apply_rules_deferred(module_pk, module_bay_pk, force_reapply=False):
85 """Apply interface name rules after transaction commit.
87 Called via transaction.on_commit so all interfaces created by bulk_create()
88 are visible in the DB before renaming begins.
90 Args:
91 module_pk: Primary key of the Module whose interfaces should be renamed.
92 module_bay_pk: Primary key of the ModuleBay containing the module.
93 force_reapply: When True, re-apply even if interfaces already have
94 the expected names (used for module-type changes).
96 """
97 from dcim.models import Module, ModuleBay
99 try:
100 module = Module.objects.get(pk=module_pk)
101 module_bay = ModuleBay.objects.get(pk=module_bay_pk)
102 except (Module.DoesNotExist, ModuleBay.DoesNotExist):
103 return
105 try:
106 from .engine import apply_interface_name_rules
108 renamed = apply_interface_name_rules(module, module_bay, force_reapply=force_reapply)
109 if renamed:
110 logger.info(
111 "Renamed %d interface(s) for %s in %s",
112 renamed,
113 module.module_type,
114 module_bay.name,
115 )
116 except Exception:
117 logger.exception(
118 "Failed to apply interface name rules for %s in %s",
119 module.module_type,
120 module_bay.name,
121 )
124# ---------------------------------------------------------------------------
125# Virtual Chassis membership change detection
126# ---------------------------------------------------------------------------
127# When a Device's vc_position or virtual_chassis changes, any module-attached
128# interfaces with rules using {vc_position} need to be renamed. We capture
129# the old values in pre_save (stored on the instance) and compare in post_save.
132@receiver(pre_save, sender="dcim.Device", dispatch_uid="interface_name_rules_pre_save_device")
133def on_device_pre_save(sender, instance, **kwargs):
134 """Capture old virtual_chassis_id and vc_position before a Device save."""
135 if instance.pk is None:
136 # New device — nothing to compare
137 instance._prev_virtual_chassis_id = None
138 instance._prev_vc_position = None
139 return
140 try:
141 old = sender.objects.filter(pk=instance.pk).values("virtual_chassis_id", "vc_position").first()
142 instance._prev_virtual_chassis_id = old["virtual_chassis_id"] if old else None
143 instance._prev_vc_position = old["vc_position"] if old else None
144 except Exception:
145 logger.warning("Failed to capture previous VC state for Device pk=%s", instance.pk, exc_info=True)
146 instance._prev_virtual_chassis_id = None
147 instance._prev_vc_position = None
150@receiver(post_save, sender="dcim.Device", dispatch_uid="interface_name_rules_post_save_device")
151def on_device_saved(sender, instance, created, **kwargs):
152 """Re-apply interface name rules when a device's VC membership or position changes."""
153 if created:
154 return
156 prev_vc_id = getattr(instance, "_prev_virtual_chassis_id", None)
157 prev_vc_pos = getattr(instance, "_prev_vc_position", None)
158 new_vc_id = instance.virtual_chassis_id
159 new_vc_pos = instance.vc_position
161 vc_id_changed = prev_vc_id != new_vc_id
162 vc_pos_changed = prev_vc_pos != new_vc_pos
164 if not (vc_id_changed or vc_pos_changed):
165 return
167 # Only rename when the device is (now) in a VC — if it was removed, no {vc_position} available.
168 if new_vc_id is None:
169 logger.debug(
170 "Device %s removed from VC — skipping interface rename (no vc_position available)",
171 instance.pk,
172 )
173 return
175 callback = functools.partial(_apply_rules_for_device_deferred, instance.pk)
176 transaction.on_commit(callback)
179def _apply_rules_for_device_deferred(device_pk):
180 """Re-apply all interface name rules for a device after a VC membership or position change.
182 Handles two categories of interfaces:
183 - Module-attached interfaces: re-applies module-level rules with force_reapply=True
184 so {vc_position} is updated even when the interface name already matches the template.
185 - Device-level interfaces (module=None): re-applies device-level rules which use
186 applies_to_device_interfaces=True rules matched by interface-name regex.
188 Args:
189 device_pk: Primary key of the Device to re-apply rules for.
191 """
192 from dcim.models import Device, Module
194 try:
195 device = Device.objects.select_related("virtual_chassis").get(pk=device_pk)
196 except Device.DoesNotExist:
197 return
199 total = 0
201 # Re-apply module interface rules
202 modules = Module.objects.filter(device=device).select_related(
203 "module_bay",
204 "module_type",
205 "device",
206 "device__device_type",
207 "device__platform",
208 "device__virtual_chassis",
209 )
210 try:
211 from .engine import apply_interface_name_rules, pinned_rule_cache
213 # Every module on this device matches against the same enabled-rule set, so pin it for the
214 # loop: the fingerprint is read once when the first lookup primes the cache instead of once
215 # per module. Safe because the loop only renames interfaces — it never edits rules.
216 with pinned_rule_cache():
217 for module in modules:
218 module_bay = module.module_bay
219 if not module_bay:
220 continue
221 try:
222 renamed = apply_interface_name_rules(module, module_bay, force_reapply=True)
223 total += renamed or 0
224 except Exception:
225 logger.exception(
226 "Failed to re-apply rules for %s in %s after VC change",
227 module.module_type,
228 module_bay.name,
229 )
230 except Exception:
231 logger.exception("Failed to re-apply module rules for device %s after VC change", device_pk)
233 # Re-apply device-level interface rules (interfaces with module=None)
234 try:
235 from .engine import apply_device_interface_rules
237 renamed = apply_device_interface_rules(device)
238 total += renamed or 0
239 except Exception:
240 logger.exception("Failed to re-apply device interface rules for device %s after VC change", device_pk)
242 if total:
243 logger.info(
244 "Re-renamed %d interface(s) for device %s after VC change",
245 total,
246 device,
247 )
250# ---------------------------------------------------------------------------
251# Cross-plugin integration: netbox-librenms-plugin name prediction
252# ---------------------------------------------------------------------------
253# When netbox-librenms-plugin is installed it exposes a Signal that asks
254# subscribers to rewrite the list of interface names it predicts a module's
255# templates will produce. Without this rewrite, its module-adoption lookup
256# would only see NetBox's raw template output and miss interfaces we renamed
257# after install. The import is guarded so INR works fine when librenms-plugin
258# is absent.
260try:
261 from netbox_librenms_plugin.signals import (
262 predict_module_interface_names as _librenms_predict_signal,
263 )
264except ImportError:
265 _librenms_predict_signal = None
268def on_librenms_predict_module_interface_names(sender, device, module, names, **kwargs):
269 """Rewrite librenms-plugin's predicted template names through our rule engine.
271 Always defined so it can be imported and tested in isolation even when
272 netbox-librenms-plugin is not installed. Signal registration is guarded
273 below so it only connects when the signal actually exists.
274 """
275 module_bay = getattr(module, "module_bay", None)
276 if module_bay is None: 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true
277 return None
278 try:
279 from .engine import predict_rule_output
281 return predict_rule_output(module, module_bay, names)
282 except Exception:
283 logger.exception(
284 "predict_rule_output failed for module pk=%s; leaving names unchanged",
285 getattr(module, "pk", "?"),
286 )
287 return None
290if _librenms_predict_signal is not None: 290 ↛ 291line 290 didn't jump to line 291 because the condition on line 290 was never true
291 receiver(
292 _librenms_predict_signal,
293 dispatch_uid="interface_name_rules_predict_module_interface_names",
294 )(on_librenms_predict_module_interface_names)