« 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>
3import dataclasses
4import logging
5import re
7import yaml
8from django.conf import settings
9from django.contrib import messages
10from django.core.exceptions import PermissionDenied
11from django.http import JsonResponse
12from django.shortcuts import redirect, render
13from django.urls import reverse
14from django.utils.http import url_has_allowed_host_and_scheme
15from netbox.views import generic
16from netbox.views.generic.base import BaseMultiObjectView
17from utilities.views import register_model_view
19from .choices import BreakoutModeChoices
20from .filters import InterfaceNameRuleFilterSet
21from .forms import (
22 InterfaceNameRuleBulkEditForm,
23 InterfaceNameRuleFilterForm,
24 InterfaceNameRuleForm,
25 InterfaceNameRuleImportForm,
26 RuleTestForm,
27)
28from .models import InterfaceNameRule
29from .tables import InterfaceNameRuleTable
31logger = logging.getLogger(__name__)
33try:
34 _plugins_config = getattr(settings, "PLUGINS_CONFIG", {})
35 APPLY_BATCH_LIMIT = max(1, int(_plugins_config.get("netbox_interface_name_rules", {}).get("apply_batch_limit", 50)))
36except (ValueError, TypeError):
37 APPLY_BATCH_LIMIT = 50
40@dataclasses.dataclass
41class RulePreview:
42 """Lightweight stand-in for InterfaceNameRule used in the test/preview view."""
44 module_type_is_regex: bool
45 module_type_pattern: str
46 module_type: object
47 parent_module_type: object
48 device_type: object
49 platform: object
50 name_template: str
51 channel_count: int
52 channel_start: int
53 parent_name_template: str = ""
54 breakout_mode: str = BreakoutModeChoices.FLAT
57try:
58 from netbox.object_actions import AddObject, BulkDelete, BulkEdit, BulkExport, BulkImport
60 class _YAMLOnlyExport(BulkExport):
61 """Export action that only offers YAML (no CSV "Current View" option)."""
63 template_name = "netbox_interface_name_rules/buttons/export_yaml_only.html"
65 # No BulkRename: rules have no name field, and its button template renders nothing anyway.
66 _LIST_VIEW_ACTIONS: tuple = (AddObject, BulkImport, _YAMLOnlyExport, BulkEdit, BulkDelete)
67except ImportError:
68 _LIST_VIEW_ACTIONS = None
71class InterfaceNameRuleListView(generic.ObjectListView):
72 """List view for InterfaceNameRule."""
74 queryset = InterfaceNameRule.objects.all()
75 table = InterfaceNameRuleTable
76 filterset = InterfaceNameRuleFilterSet
77 filterset_form = InterfaceNameRuleFilterForm
78 template_name = "netbox_interface_name_rules/interfacenamerule_list.html"
79 if _LIST_VIEW_ACTIONS is not None: 79 ↛ 82line 79 didn't jump to line 82 because the condition on line 79 was always true
80 actions = _LIST_VIEW_ACTIONS
82 def export_yaml(self):
83 """Export all rules as a single YAML list (overrides NetBox's per-object concatenation)."""
84 data = []
85 for rule in self.queryset.order_by("pk").select_related(
86 "module_type", "parent_module_type", "device_type", "platform"
87 ):
88 entry = {}
89 for header, value in zip(InterfaceNameRule.csv_headers, rule.to_csv()):
90 if (value != "" and value is not None) or header in {"name_template"}:
91 entry[header] = value
92 data.append(entry)
93 return yaml.dump(data, default_flow_style=False, allow_unicode=True, sort_keys=False)
96class InterfaceNameRuleCreateView(generic.ObjectEditView):
97 """Create view for InterfaceNameRule."""
99 queryset = InterfaceNameRule.objects.all()
100 form = InterfaceNameRuleForm
103@register_model_view(InterfaceNameRule, "bulk_import", path="import", detail=False)
104class InterfaceNameRuleBulkImportView(generic.BulkImportView):
105 """Bulk import view for InterfaceNameRule."""
107 queryset = InterfaceNameRule.objects.all()
108 model_form = InterfaceNameRuleImportForm
111class InterfaceNameRuleView(generic.ObjectView):
112 """Detail view for InterfaceNameRule."""
114 queryset = InterfaceNameRule.objects.all()
117class InterfaceNameRuleEditView(generic.ObjectEditView):
118 """Edit view for InterfaceNameRule."""
120 queryset = InterfaceNameRule.objects.all()
121 form = InterfaceNameRuleForm
124class InterfaceNameRuleDeleteView(generic.ObjectDeleteView):
125 """Delete view for InterfaceNameRule."""
127 queryset = InterfaceNameRule.objects.all()
130class InterfaceNameRuleBulkEditView(generic.BulkEditView):
131 """Bulk edit view for InterfaceNameRule."""
133 queryset = InterfaceNameRule.objects.all()
134 filterset = InterfaceNameRuleFilterSet
135 table = InterfaceNameRuleTable
136 form = InterfaceNameRuleBulkEditForm
139class InterfaceNameRuleBulkDeleteView(generic.BulkDeleteView):
140 """Bulk delete view for InterfaceNameRule."""
142 queryset = InterfaceNameRule.objects.all()
143 table = InterfaceNameRuleTable
146class InterfaceNameRuleChangeLogView(generic.ObjectChangeLogView):
147 """Change-log view for InterfaceNameRule."""
149 queryset = InterfaceNameRule.objects.all()
152class InterfaceNameRuleDuplicateView(generic.ObjectView):
153 """Redirect to the add view pre-populated with a clone of the given rule."""
155 queryset = InterfaceNameRule.objects.all()
157 def get(self, request, **kwargs):
158 """Redirect to the add view pre-populated with fields cloned from the given rule."""
159 from utilities.querydict import prepare_cloned_fields
161 rule = self.get_object(**kwargs)
162 params = prepare_cloned_fields(rule)
163 url = reverse("plugins:netbox_interface_name_rules:interfacenamerule_add")
164 return redirect(f"{url}?{params.urlencode()}")
167class RuleTestView(BaseMultiObjectView):
168 """Live-preview a name template with user-supplied variable values and optional DB lookup."""
170 queryset = InterfaceNameRule.objects.all()
171 template_name = "netbox_interface_name_rules/rule_test.html"
173 def get_required_permission(self):
174 """Return the permission required to access the rule test tool."""
175 return "netbox_interface_name_rules.add_interfacenamerule"
177 def get(self, request):
178 """Render the test form, pre-populated from rule_id query param if given."""
179 initial = {}
180 loaded_rule = None
181 rule_id = request.GET.get("rule_id")
182 can_view = request.user.has_perm("netbox_interface_name_rules.view_interfacenamerule")
183 if rule_id and not can_view:
184 messages.warning(request, "You do not have permission to load an existing rule.")
185 if rule_id and can_view:
186 try:
187 loaded_rule = (
188 InterfaceNameRule.objects.restrict(request.user, "view")
189 .select_related("module_type", "parent_module_type", "device_type", "platform")
190 .get(pk=int(rule_id))
191 )
192 initial = {
193 "name_template": loaded_rule.name_template,
194 "parent_name_template": loaded_rule.parent_name_template,
195 "breakout_mode": loaded_rule.breakout_mode,
196 "module_type_is_regex": loaded_rule.module_type_is_regex,
197 "module_type": loaded_rule.module_type,
198 "module_type_pattern": loaded_rule.module_type_pattern,
199 "parent_module_type": loaded_rule.parent_module_type,
200 "device_type": loaded_rule.device_type,
201 "platform": loaded_rule.platform,
202 "channel_count": loaded_rule.channel_count,
203 "channel_start": loaded_rule.channel_start,
204 }
205 except (InterfaceNameRule.DoesNotExist, ValueError):
206 pass
207 return render(request, self.template_name, {"form": RuleTestForm(initial=initial), "loaded_rule": loaded_rule})
209 def post(self, request):
210 """Evaluate the submitted template and return a preview or redirect to save."""
211 form = RuleTestForm(request.POST)
212 preview_results = None
213 db_preview = None
214 db_total = 0
215 error = None
216 action = request.POST.get("action", "check")
218 if form.is_valid():
219 cd = form.cleaned_data
220 if action == "save_rule":
221 return self._handle_save_rule(request, cd)
222 preview_results, error = self._evaluate_template_preview(cd)
223 db_preview, db_total, db_error = self._fetch_db_preview(cd)
224 if db_error and not error:
225 error = db_error
227 return render(
228 request,
229 self.template_name,
230 {
231 "form": form,
232 "preview_results": preview_results,
233 "db_preview": db_preview,
234 "db_total": db_total,
235 "error": error,
236 },
237 )
239 def _find_existing_rule(self, cd, user=None):
240 """Return the first existing rule matching the form data, or None."""
241 module_type_is_regex = cd.get("module_type_is_regex", False)
242 qs = InterfaceNameRule.objects.restrict(user, "view") if user else InterfaceNameRule.objects.all()
243 if module_type_is_regex:
244 qs = qs.filter(module_type_is_regex=True, module_type_pattern=cd.get("module_type_pattern", ""))
245 else:
246 qs = qs.filter(module_type_is_regex=False, module_type=cd.get("module_type"))
247 for field in ("parent_module_type", "device_type", "platform"):
248 val = cd.get(field)
249 if val:
250 qs = qs.filter(**{field: val})
251 else:
252 qs = qs.filter(**{f"{field}__isnull": True})
253 return qs.first()
255 def _handle_save_rule(self, request, cd):
256 """Find an existing matching rule or redirect to the add-rule form with pre-filled params."""
257 from urllib.parse import urlencode
259 name_template = cd["name_template"]
260 channel_count = cd.get("channel_count") or 0
261 channel_start = cd.get("channel_start") or 0
262 module_type_is_regex = cd.get("module_type_is_regex", False)
263 module_type = cd.get("module_type")
265 # Skip duplicate detection when the user lacks view permission — we cannot
266 # query existing rules without it, so add-only users always land on the
267 # create form (potentially allowing duplicates).
268 if request.user.has_perm("netbox_interface_name_rules.view_interfacenamerule"):
269 existing = self._find_existing_rule(cd, request.user)
270 if existing:
271 messages.info(
272 request,
273 f"A matching rule already exists (#{existing.pk}). Redirecting to edit it.",
274 )
275 return redirect(
276 reverse("plugins:netbox_interface_name_rules:interfacenamerule_edit", args=[existing.pk])
277 )
279 params = {
280 "name_template": name_template,
281 "parent_name_template": cd.get("parent_name_template", ""),
282 "breakout_mode": cd.get("breakout_mode") or BreakoutModeChoices.FLAT,
283 "module_type_is_regex": "on" if module_type_is_regex else "",
284 "channel_count": channel_count,
285 "channel_start": channel_start,
286 }
287 if module_type_is_regex:
288 params["module_type_pattern"] = cd.get("module_type_pattern", "")
289 elif module_type: 289 ↛ 291line 289 didn't jump to line 291 because the condition on line 289 was always true
290 params["module_type"] = module_type.pk
291 for field in ("parent_module_type", "device_type", "platform"):
292 val = cd.get(field)
293 if val:
294 params[field] = val.pk
295 add_url = reverse("plugins:netbox_interface_name_rules:interfacenamerule_add")
296 return redirect(f"{add_url}?{urlencode(params)}")
298 def _evaluate_template_preview(self, cd):
299 """Evaluate the rule's templates against form variables; return (preview_results, error).
301 Each row carries the role the DB preview uses — ``parent``, ``channel`` or ``interface`` —
302 so a channelized rule shows the parent it creates alongside the channels under it.
303 """
304 from .engine import evaluate_name_template
306 name_template = cd["name_template"]
307 channel_count = cd.get("channel_count") or 0
308 channel_start = cd.get("channel_start") or 0
309 variables = {
310 "slot": cd.get("var_slot") or "1",
311 "bay_position": cd.get("var_bay_position") or "1",
312 "bay_position_num": cd.get("var_bay_position_num") or "1",
313 "parent_bay_position": cd.get("var_parent_bay_position") or "1",
314 "sfp_slot": cd.get("var_sfp_slot") or "1",
315 "base": cd.get("var_base") or "Ethernet1",
316 }
318 def row(result, role, channel=None):
319 """Describe one previewed name."""
320 return {"source": variables["base"], "channel": channel, "result": result, "role": role}
322 try:
323 if channel_count > 0:
324 preview_results = []
325 if cd.get("breakout_mode") == BreakoutModeChoices.CHANNELIZED:
326 parent_template = cd.get("parent_name_template") or ""
327 parent_name = (
328 evaluate_name_template(parent_template, variables) if parent_template else variables["base"]
329 )
330 preview_results.append(row(parent_name, "parent"))
331 for ch in range(channel_count):
332 channel = str(channel_start + ch)
333 preview_results.append(
334 row(
335 evaluate_name_template(name_template, {**variables, "channel": channel}), "channel", channel
336 )
337 )
338 else:
339 preview_results = [row(evaluate_name_template(name_template, variables), "interface")]
340 return preview_results, None
341 except Exception as exc:
342 logger.exception("Template evaluation error: %s", exc)
343 return None, type(exc).__name__
345 def _fetch_db_preview(self, cd):
346 """Run find_interfaces_for_rule against the DB; return (db_preview, db_total, error)."""
347 from .engine import find_interfaces_for_rule
349 name_template = cd["name_template"]
350 channel_count = cd.get("channel_count") or 0
351 channel_start = cd.get("channel_start") or 0
352 module_type_is_regex = cd.get("module_type_is_regex", False)
353 module_type = cd.get("module_type")
354 module_type_pattern = cd.get("module_type_pattern", "")
356 if not ((module_type_is_regex and module_type_pattern) or (not module_type_is_regex and module_type)):
357 return None, 0, None
359 fake = RulePreview(
360 module_type_is_regex=module_type_is_regex,
361 module_type_pattern=module_type_pattern,
362 module_type=module_type,
363 parent_module_type=cd.get("parent_module_type"),
364 device_type=cd.get("device_type"),
365 platform=cd.get("platform"),
366 name_template=name_template,
367 parent_name_template=cd.get("parent_name_template", ""),
368 breakout_mode=cd.get("breakout_mode") or BreakoutModeChoices.FLAT,
369 channel_count=channel_count,
370 channel_start=channel_start,
371 )
372 try:
373 db_preview, db_total = find_interfaces_for_rule(fake, limit=100)
374 return db_preview, db_total, None
375 except (re.error, ValueError) as exc:
376 return [], 0, f"Invalid module type regex: {exc}"
377 except Exception as exc:
378 logger.exception("Unexpected error in find_interfaces_for_rule: %s", exc)
379 return [], 0, f"Unexpected error: {type(exc).__name__}"
382class RuleApplyListView(BaseMultiObjectView):
383 """Display all rules with buttons to preview/apply each one."""
385 queryset = InterfaceNameRule.objects.all()
386 template_name = "netbox_interface_name_rules/rule_apply.html"
388 def get_required_permission(self):
389 """Return the permission required to access the apply-rules page."""
390 return "netbox_interface_name_rules.view_interfacenamerule"
392 def get(self, request):
393 """Render the list of all rules with apply/preview buttons."""
394 rules = self.queryset.select_related("module_type", "parent_module_type", "device_type", "platform").order_by(
395 "pk"
396 )
397 return render(request, self.template_name, {"rules": rules, "batch_limit": APPLY_BATCH_LIMIT})
400class RuleApplicableView(generic.ObjectView):
401 """Return JSON indicating whether a rule would rename at least one interface.
403 Called on demand from the Apply Rules page — NOT at page load — to avoid
404 expensive full-scan queries blocking the initial render.
405 """
407 queryset = InterfaceNameRule.objects.all()
409 def get(self, request, **kwargs):
410 """Return JSON {"applicable": bool} for the rule identified by pk."""
411 from .engine import has_applicable_interfaces
413 rule = self.get_object(**kwargs)
414 try:
415 applicable = has_applicable_interfaces(rule)
416 except Exception as exc:
417 logger.exception("applicability scan failed for rule %s", kwargs.get("pk"))
418 return JsonResponse(
419 {"applicable": None, "error": f"scan failed: {type(exc).__name__}"},
420 status=500,
421 )
422 return JsonResponse({"applicable": applicable})
425class RuleApplyDetailView(generic.ObjectView):
426 """Show a preview of changes for a specific rule and allow applying them."""
428 queryset = InterfaceNameRule.objects.all()
429 template_name = "netbox_interface_name_rules/rule_apply_detail.html"
430 additional_permissions = ["dcim.change_interface"]
432 def get(self, request, **kwargs):
433 """Render a preview of all interfaces that would be renamed by this rule."""
434 from .engine import find_convertible_families, find_interfaces_for_rule, supports_channelization
436 rule = self.get_object(**kwargs)
437 try:
438 preview, total_checked = find_interfaces_for_rule(rule, limit=APPLY_BATCH_LIMIT)
439 except (re.error, ValueError) as exc:
440 logger.exception("Failed to compute preview for rule %s: %s", rule, exc)
441 messages.error(request, f"Failed to compute preview: {exc}")
442 preview, total_checked = [], 0
443 except Exception as exc:
444 logger.exception("Unexpected error computing preview for rule %s: %s", rule, exc)
445 messages.error(request, f"Failed to compute preview: {type(exc).__name__}")
446 preview, total_checked = [], 0
447 # A conversion-scan failure must not blank the unrelated apply preview above.
448 try:
449 # Each family scanned costs a dry-run conversion, so the scan takes the same batch cap.
450 conversions, conversions_have_more = find_convertible_families(rule, limit=APPLY_BATCH_LIMIT)
451 except (re.error, ValueError) as exc:
452 logger.exception("Failed to compute the conversion preview for rule %s: %s", rule, exc)
453 messages.error(request, f"Failed to compute the conversion preview: {exc}")
454 conversions, conversions_have_more = [], False
455 except Exception as exc:
456 logger.exception("Unexpected error computing the conversion preview for rule %s: %s", rule, exc)
457 messages.error(request, f"Failed to compute the conversion preview: {type(exc).__name__}")
458 conversions, conversions_have_more = [], False
459 return render(
460 request,
461 self.template_name,
462 {
463 "rule": rule,
464 "preview": preview,
465 "conversions": conversions,
466 "conversions_have_more": conversions_have_more,
467 "total_checked": total_checked,
468 "batch_limit": APPLY_BATCH_LIMIT,
469 "has_more": len(preview) >= APPLY_BATCH_LIMIT,
470 "can_apply": request.user.has_perm("dcim.change_interface"),
471 # Families exist only where NetBox models channelization; older releases count plain interfaces.
472 "families_supported": supports_channelization(),
473 "checked_unit": "families" if supports_channelization() else "interfaces",
474 },
475 )
477 def _enqueue(self, request, rule, job_class, name):
478 """Enqueue *job_class* against *rule* and report the outcome to the operator."""
479 try:
480 job = job_class.enqueue(
481 # instance is intentionally omitted: InterfaceNameRule does not
482 # inherit JobsMixin, so passing instance= would fail full_clean().
483 # The job is still named and findable in Core → Jobs.
484 name=name,
485 user=request.user,
486 rule_id=rule.pk,
487 )
488 messages.success(request, f"Background job enqueued (job #{job.pk}). Check Core → Jobs for status.")
489 except Exception as e:
490 logger.exception("Failed to enqueue background job for rule %s: %s", rule, e)
491 messages.error(request, f"Failed to enqueue background job: {type(e).__name__}")
493 def _convert(self, request, rule):
494 """Convert the flat families the operator confirmed on this page."""
495 from .engine import convert_flat_families
497 convert_ids = [int(i) for i in request.POST.getlist("convert_ids") if i.isdigit()]
498 if not convert_ids:
499 messages.warning(request, "No families selected; nothing was converted.")
500 return
501 # Converting the first APPLY_BATCH_LIMIT of them would rewrite a set the operator never chose.
502 if len(convert_ids) > APPLY_BATCH_LIMIT:
503 messages.warning(
504 request,
505 f"{len(convert_ids)} families selected; synchronous conversion is limited to "
506 f"{APPLY_BATCH_LIMIT} family(ies) per run, so nothing was converted. "
507 f"Use Convert as Background Job to convert all of them.",
508 )
509 return
510 conflicts = []
511 try:
512 count = convert_flat_families(rule, convert_ids, conflicts=conflicts)
513 except Exception as e:
514 logger.exception("Failed to convert families for rule %s: %s", rule, e)
515 messages.error(request, f"Failed to convert families: {type(e).__name__}")
516 return
517 messages.success(request, f"Converted {count} interface family(ies) to the channelized topology.")
518 if conflicts: 518 ↛ 519line 518 didn't jump to line 519 because the condition on line 518 was never true
519 messages.warning(request, f"{len(conflicts)} family(ies) skipped — the plugin log names each one.")
521 def post(self, request, **kwargs):
522 """Apply or convert (foreground batch or background job) and redirect back."""
523 from .engine import apply_rule_to_existing
525 rule = self.get_object(**kwargs)
526 action = request.POST.get("action", "apply")
528 if action == "convert":
529 self._convert(request, rule)
530 elif action == "convert_background":
531 from .jobs import ConvertFlatFamiliesJob
533 self._enqueue(request, rule, ConvertFlatFamiliesJob, f"Convert flat families: {rule}")
534 elif action == "background":
535 from .jobs import ApplyRuleJob
537 self._enqueue(request, rule, ApplyRuleJob, f"Apply rule: {rule}")
538 else:
539 try:
540 raw_ids = request.POST.getlist("interface_ids")
541 interface_ids = [int(i) for i in raw_ids if i.isdigit()]
542 if not interface_ids:
543 messages.warning(request, "No interfaces selected; nothing was applied.")
544 else:
545 conflicts = []
546 count = apply_rule_to_existing(
547 rule, limit=APPLY_BATCH_LIMIT, interface_ids=interface_ids, conflicts=conflicts
548 )
549 messages.success(request, f"Applied rule: {count} interface(s) renamed.")
550 if conflicts: 550 ↛ 551line 550 didn't jump to line 551 because the condition on line 550 was never true
551 messages.warning(
552 request,
553 f"{len(conflicts)} interface(s) skipped — the plugin log names each one.",
554 )
555 except Exception as e:
556 logger.exception("Failed to apply rule %s: %s", rule, e)
557 messages.error(request, f"Failed to apply rule {rule}: {type(e).__name__}")
559 return redirect("plugins:netbox_interface_name_rules:interfacenamerule_apply_detail", pk=rule.pk)
562@register_model_view(InterfaceNameRule, name="toggle", path="toggle")
563class RuleToggleView(generic.ObjectView):
564 """POST /rules/<pk>/toggle/ — flip the enabled flag on a rule."""
566 queryset = InterfaceNameRule.objects.all()
568 def post(self, request, pk):
569 """Toggle the enabled flag on the rule and return JSON or redirect."""
570 rule = self.get_object(pk=pk)
571 if not request.user.has_perm("netbox_interface_name_rules.change_interfacenamerule"): 571 ↛ 572line 571 didn't jump to line 572 because the condition on line 571 was never true
572 if request.headers.get("X-Requested-With") == "XMLHttpRequest":
573 return JsonResponse({"error": "Permission denied"}, status=403)
574 raise PermissionDenied
575 rule.enabled = not rule.enabled
576 rule.save(update_fields=["enabled"])
577 if request.headers.get("X-Requested-With") == "XMLHttpRequest":
578 return JsonResponse({"enabled": rule.enabled, "pk": pk})
579 state = "enabled" if rule.enabled else "disabled"
580 messages.success(request, f"Rule '{rule}' {state}.")
581 referer = request.META.get("HTTP_REFERER", "")
582 if referer and url_has_allowed_host_and_scheme(
583 referer, allowed_hosts={request.get_host()}, require_https=request.is_secure()
584 ):
585 return redirect(referer)
586 return redirect("plugins:netbox_interface_name_rules:interfacenamerule_list")