Coverage for netbox_interface_name_rules / views.py: 95.71%
371 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 dataclasses
4import logging
5import re
7import yaml
8from django.conf import settings
9from django.contrib import messages
10from django.core.exceptions import PermissionDenied, ValidationError
11from django.http import JsonResponse
12from django.shortcuts import redirect, render
13from django.urls import reverse
14from django.utils import timezone
15from django.utils.http import url_has_allowed_host_and_scheme
16from netbox.views import generic
17from netbox.views.generic.base import BaseMultiObjectView
18from utilities.views import register_model_view
20from .choices import BreakoutModeChoices
21from .filters import InterfaceNameRuleFilterSet
22from .forms import (
23 InterfaceNameRuleBulkEditForm,
24 InterfaceNameRuleFilterForm,
25 InterfaceNameRuleForm,
26 InterfaceNameRuleImportForm,
27 RuleTestForm,
28)
29from .models import InterfaceNameRule, csv_export_entry
30from .name_template import NamingContext, variables_for_context
31from .tables import InterfaceNameRuleTable
32from .template_variable_reference import naming_context_reference, rule_tester_variable_rows, variable_reference_rows
33from .transactions import atomic_with_events, write_scope
35logger = logging.getLogger(__name__)
37try:
38 _plugins_config = getattr(settings, "PLUGINS_CONFIG", {})
39 APPLY_BATCH_LIMIT = max(1, int(_plugins_config.get("netbox_interface_name_rules", {}).get("apply_batch_limit", 50)))
40except (ValueError, TypeError):
41 APPLY_BATCH_LIMIT = 50
44def _failure_text(error):
45 """Return what the operator reads about *error*: each channel to rename back, or else the error type."""
46 from .family import ChannelReconciliationError
48 return str(error) if isinstance(error, ChannelReconciliationError) else type(error).__name__
51@dataclasses.dataclass
52class RulePreview:
53 """Lightweight stand-in for InterfaceNameRule used in the test/preview view."""
55 module_type_is_regex: bool
56 module_type_pattern: str
57 module_type: object
58 parent_module_type: object
59 device_type: object
60 platform: object
61 name_template: str
62 channel_count: int
63 channel_start: int
64 parent_name_template: str = ""
65 breakout_mode: str = BreakoutModeChoices.FLAT
66 applies_to_device_interfaces: bool = False
69try:
70 from netbox.object_actions import AddObject, BulkDelete, BulkEdit, BulkExport, BulkImport
72 class _YAMLOnlyExport(BulkExport):
73 """Export action that only offers YAML (no CSV "Current View" option)."""
75 template_name = "netbox_interface_name_rules/buttons/export_yaml_only.html"
77 # No BulkRename: rules have no name field, and its button template renders nothing anyway.
78 _LIST_VIEW_ACTIONS: tuple = (AddObject, BulkImport, _YAMLOnlyExport, BulkEdit, BulkDelete)
79except ImportError:
80 _LIST_VIEW_ACTIONS = None
83class InterfaceNameRuleListView(generic.ObjectListView):
84 """List view for InterfaceNameRule."""
86 queryset = InterfaceNameRule.objects.all()
87 table = InterfaceNameRuleTable
88 filterset = InterfaceNameRuleFilterSet
89 filterset_form = InterfaceNameRuleFilterForm
90 template_name = "netbox_interface_name_rules/interfacenamerule_list.html"
91 if _LIST_VIEW_ACTIONS is not None: 91 ↛ 94line 91 didn't jump to line 94 because the condition on line 91 was always true
92 actions = _LIST_VIEW_ACTIONS
94 def get_extra_context(self, request):
95 """Add the catalogue-backed template-variable reference."""
96 return {
97 "template_naming_contexts": naming_context_reference(),
98 "template_variable_rows": variable_reference_rows(),
99 }
101 def export_yaml(self):
102 """Export all rules as a single YAML list (overrides NetBox's per-object concatenation)."""
103 data = [
104 csv_export_entry(InterfaceNameRule.csv_headers, rule.to_csv())
105 for rule in self.queryset.order_by("pk").select_related(
106 "module_type", "parent_module_type", "device_type", "platform"
107 )
108 ]
109 return yaml.dump(data, default_flow_style=False, allow_unicode=True, sort_keys=False)
112def _submitted(request, field):
113 """Return a submitted rule field, wherever the form put it.
115 NetBox 4.4 gave the quick-add modal a ``quickadd`` form prefix; 4.3 posts the fields under
116 their own names, so the prefix cannot be assumed from the quick-add marker alone.
117 """
118 return request.POST.get(f"quickadd-{field}") or request.POST.get(field)
121class ZeroMatchPatternWarningMixin:
122 """Warn when a saved regex rule matches no module type that exists now.
124 Saving still succeeds, because a rule may name a module type that is not in the database yet.
125 NetBox's edit view does its work in ``post`` and offers no post-save hook, so this wraps it.
126 A quick add answers with a modal fragment that renders no messages, so that warning reaches
127 the operator on the next page they load, and it answers a successful save with 200 rather
128 than a redirect, which is why a save is recognised from the database and not the response.
129 """
131 def post(self, request, *args, **kwargs):
132 """Save through the parent view, then report a pattern that selects nothing."""
133 pattern = (_submitted(request, "module_type_pattern") or "").strip()
134 regex_rule = pattern and _submitted(request, "module_type_is_regex")
135 # A device-interface rule filters interface names, so no module type is expected to match.
136 device_rule = _submitted(request, "applies_to_device_interfaces")
137 started = timezone.now()
138 response = super().post(request, *args, **kwargs)
139 if regex_rule and not device_rule:
140 self._warn_if_pattern_matches_nothing(request, pattern, started)
141 return response
143 def _warn_if_pattern_matches_nothing(self, request, pattern, started):
144 # Only a rule written by this request saved; a rejected form leaves nothing to report on.
145 saved = self.queryset.model.objects.filter(
146 module_type_pattern=pattern, module_type_is_regex=True, last_updated__gte=started
147 )
148 if not saved.exists():
149 return
150 from .rule_selection import module_types_matching_pattern
152 try:
153 matched = module_types_matching_pattern(pattern)
154 except ValidationError:
155 return
156 if not matched:
157 messages.warning(
158 request,
159 f"Pattern {pattern!r} matches no module type, so this rule will not fire. "
160 "The field takes a regular expression, not a glob: '*' repeats the character "
161 "before it, so 'GLC-T*' does not match 'GLC-TE'. Use 'GLC-T.*'.",
162 )
165class InterfaceNameRuleCreateView(ZeroMatchPatternWarningMixin, generic.ObjectEditView):
166 """Create view for InterfaceNameRule."""
168 queryset = InterfaceNameRule.objects.all()
169 form = InterfaceNameRuleForm
172@register_model_view(InterfaceNameRule, "bulk_import", path="import", detail=False)
173class InterfaceNameRuleBulkImportView(generic.BulkImportView):
174 """Bulk import view for InterfaceNameRule."""
176 queryset = InterfaceNameRule.objects.all()
177 model_form = InterfaceNameRuleImportForm
180class InterfaceNameRuleView(generic.ObjectView):
181 """Detail view for InterfaceNameRule."""
183 queryset = InterfaceNameRule.objects.all()
186class InterfaceNameRuleEditView(ZeroMatchPatternWarningMixin, generic.ObjectEditView):
187 """Edit view for InterfaceNameRule."""
189 queryset = InterfaceNameRule.objects.all()
190 form = InterfaceNameRuleForm
193class InterfaceNameRuleDeleteView(generic.ObjectDeleteView):
194 """Delete view for InterfaceNameRule."""
196 queryset = InterfaceNameRule.objects.all()
199class InterfaceNameRuleBulkEditView(generic.BulkEditView):
200 """Bulk edit view for InterfaceNameRule."""
202 queryset = InterfaceNameRule.objects.all()
203 filterset = InterfaceNameRuleFilterSet
204 table = InterfaceNameRuleTable
205 form = InterfaceNameRuleBulkEditForm
208class InterfaceNameRuleBulkDeleteView(generic.BulkDeleteView):
209 """Bulk delete view for InterfaceNameRule."""
211 queryset = InterfaceNameRule.objects.all()
212 table = InterfaceNameRuleTable
215class InterfaceNameRuleChangeLogView(generic.ObjectChangeLogView):
216 """Change-log view for InterfaceNameRule."""
218 queryset = InterfaceNameRule.objects.all()
221class InterfaceNameRuleDuplicateView(generic.ObjectView):
222 """Redirect to the add view pre-populated with a clone of the given rule."""
224 queryset = InterfaceNameRule.objects.all()
226 def get(self, request, **kwargs):
227 """Redirect to the add view pre-populated with fields cloned from the given rule."""
228 from utilities.querydict import prepare_cloned_fields
230 rule = self.get_object(**kwargs)
231 params = prepare_cloned_fields(rule)
232 url = reverse("plugins:netbox_interface_name_rules:interfacenamerule_add")
233 return redirect(f"{url}?{params.urlencode()}")
236class RuleTestView(BaseMultiObjectView):
237 """Live-preview a name template with user-supplied variable values and optional DB lookup."""
239 queryset = InterfaceNameRule.objects.all()
240 template_name = "netbox_interface_name_rules/rule_test.html"
242 def get_required_permission(self):
243 """Return the permission required to access the rule test tool."""
244 return "netbox_interface_name_rules.add_interfacenamerule"
246 def get(self, request):
247 """Render the test form, pre-populated from rule_id query param if given."""
248 initial = {}
249 loaded_rule = None
250 rule_id = request.GET.get("rule_id")
251 can_view = request.user.has_perm("netbox_interface_name_rules.view_interfacenamerule")
252 if rule_id and not can_view:
253 messages.warning(request, "You do not have permission to load an existing rule.")
254 if rule_id and can_view:
255 try:
256 loaded_rule = (
257 InterfaceNameRule.objects.restrict(request.user, "view")
258 .select_related("module_type", "parent_module_type", "device_type", "platform")
259 .get(pk=int(rule_id))
260 )
261 initial = {
262 "applies_to_device_interfaces": loaded_rule.applies_to_device_interfaces,
263 "name_template": loaded_rule.name_template,
264 "parent_name_template": loaded_rule.parent_name_template,
265 "breakout_mode": loaded_rule.breakout_mode,
266 "module_type_is_regex": loaded_rule.module_type_is_regex,
267 "module_type": loaded_rule.module_type,
268 "module_type_pattern": loaded_rule.module_type_pattern,
269 "parent_module_type": loaded_rule.parent_module_type,
270 "device_type": loaded_rule.device_type,
271 "platform": loaded_rule.platform,
272 "channel_count": loaded_rule.channel_count,
273 "channel_start": loaded_rule.channel_start,
274 }
275 except (InterfaceNameRule.DoesNotExist, ValueError):
276 pass
277 return render(
278 request,
279 self.template_name,
280 {
281 "form": RuleTestForm(initial=initial),
282 "loaded_rule": loaded_rule,
283 **self._variable_reference(),
284 },
285 )
287 def post(self, request):
288 """Evaluate the submitted template and return a preview or redirect to save."""
289 form = RuleTestForm(request.POST)
290 preview_results = None
291 db_preview = None
292 db_total = 0
293 error = None
294 action = request.POST.get("action", "check")
296 if form.is_valid():
297 cd = form.cleaned_data
298 if action == "save_rule":
299 return self._handle_save_rule(request, cd)
300 preview_results, error = self._evaluate_template_preview(cd)
301 db_preview, db_total, db_error = self._fetch_db_preview(cd)
302 if db_error and not error:
303 error = db_error
305 return render(
306 request,
307 self.template_name,
308 {
309 "form": form,
310 "preview_results": preview_results,
311 "db_preview": db_preview,
312 "db_total": db_total,
313 "error": error,
314 **self._variable_reference(),
315 },
316 )
318 @staticmethod
319 def _variable_reference():
320 device_variables = {variable.name for variable in variables_for_context(NamingContext.DEVICE_INTERFACE)}
321 return {
322 "module_only_variable_fields": {
323 f"var_{variable.name}"
324 for variable in variables_for_context(NamingContext.MODULE_MEMBER)
325 if variable.name not in device_variables
326 },
327 "module_variable_rows": rule_tester_variable_rows(NamingContext.MODULE_MEMBER),
328 "device_variable_rows": rule_tester_variable_rows(NamingContext.DEVICE_INTERFACE),
329 }
331 def _find_existing_rule(self, cd, user=None):
332 """Return the first existing rule matching the form data, or None."""
333 module_type_is_regex = cd.get("module_type_is_regex", False)
334 qs = InterfaceNameRule.objects.restrict(user, "view") if user else InterfaceNameRule.objects.all()
335 device_rule = cd.get("applies_to_device_interfaces", False)
336 qs = qs.filter(applies_to_device_interfaces=device_rule)
337 if device_rule:
338 qs = qs.filter(module_type_pattern=cd.get("module_type_pattern", ""))
339 elif module_type_is_regex:
340 qs = qs.filter(module_type_is_regex=True, module_type_pattern=cd.get("module_type_pattern", ""))
341 else:
342 qs = qs.filter(module_type_is_regex=False, module_type=cd.get("module_type"))
343 for field in ("parent_module_type", "device_type", "platform"):
344 val = cd.get(field)
345 qs = qs.filter(**{field: val}) if val else qs.filter(**{f"{field}__isnull": True})
346 return qs.first()
348 def _handle_save_rule(self, request, cd):
349 """Find an existing matching rule or redirect to the add-rule form with pre-filled params."""
350 from urllib.parse import urlencode
352 name_template = cd["name_template"]
353 channel_count = cd.get("channel_count") or 0
354 channel_start = cd.get("channel_start") or 0
355 module_type_is_regex = cd.get("module_type_is_regex", False)
356 module_type = cd.get("module_type")
358 # Skip duplicate detection when the user lacks view permission — we cannot
359 # query existing rules without it, so add-only users always land on the
360 # create form (potentially allowing duplicates).
361 if request.user.has_perm("netbox_interface_name_rules.view_interfacenamerule"):
362 existing = self._find_existing_rule(cd, request.user)
363 if existing:
364 messages.info(
365 request,
366 f"A matching rule already exists (#{existing.pk}). Redirecting to edit it.",
367 )
368 return redirect(
369 reverse("plugins:netbox_interface_name_rules:interfacenamerule_edit", args=[existing.pk])
370 )
372 params = {
373 "name_template": name_template,
374 "parent_name_template": cd.get("parent_name_template", ""),
375 "breakout_mode": cd.get("breakout_mode") or BreakoutModeChoices.FLAT,
376 "module_type_is_regex": "on" if module_type_is_regex else "",
377 "applies_to_device_interfaces": "on" if cd.get("applies_to_device_interfaces") else "",
378 "channel_count": channel_count,
379 "channel_start": channel_start,
380 }
381 if module_type_is_regex or cd.get("applies_to_device_interfaces"):
382 params["module_type_pattern"] = cd.get("module_type_pattern", "")
383 elif module_type:
384 params["module_type"] = module_type.pk
385 for field in ("parent_module_type", "device_type", "platform"):
386 val = cd.get(field)
387 if val:
388 params[field] = val.pk
389 add_url = reverse("plugins:netbox_interface_name_rules:interfacenamerule_add")
390 return redirect(f"{add_url}?{urlencode(params)}")
392 def _evaluate_template_preview(self, cd):
393 """Evaluate the rule's templates against form variables; return (preview_results, error).
395 Each row carries the role the DB preview uses — ``parent``, ``channel`` or ``interface`` —
396 so a channelized rule shows the parent it creates alongside the channels under it.
397 """
398 from .name_template import UnavailableVariableError, evaluate_name_template
400 name_template = cd["name_template"]
401 channel_count = cd.get("channel_count") or 0
402 channel_start = cd.get("channel_start") or 0
403 variables = self._template_preview_variables(cd)
405 def row(result, role, channel=None):
406 """Describe one previewed name."""
407 return {
408 "source": variables["base"],
409 "channel": channel,
410 "result": result,
411 "role": role,
412 "port": variables.get("port"),
413 }
415 try:
416 if channel_count > 0 and not cd.get("applies_to_device_interfaces"):
417 preview_results = []
418 if cd.get("breakout_mode") == BreakoutModeChoices.CHANNELIZED:
419 parent_template = cd.get("parent_name_template") or ""
420 parent_name = (
421 evaluate_name_template(parent_template, variables) if parent_template else variables["base"]
422 )
423 preview_results.append(row(parent_name, "parent"))
424 for ch in range(channel_count):
425 channel = str(channel_start + ch)
426 preview_results.append(
427 row(
428 evaluate_name_template(name_template, {**variables, "channel": channel}), "channel", channel
429 )
430 )
431 else:
432 preview_results = [row(evaluate_name_template(name_template, variables), "interface")]
433 except UnavailableVariableError as exc:
434 return None, str(exc)
435 except Exception as exc:
436 logger.exception("Template evaluation error")
437 return None, type(exc).__name__
438 else:
439 return preview_results, None
441 @staticmethod
442 def _template_preview_variables(cd):
443 """Build the template variables from the form's sample values."""
444 from .naming import build_bay_chain_variables, build_device_interface_variables
446 if cd.get("applies_to_device_interfaces"):
447 return build_device_interface_variables(cd.get("var_base") or "Ethernet1", cd.get("var_vc_position"))
448 slot = cd.get("var_slot") or "1"
449 bay_position = cd.get("var_bay_position") or "1"
450 parent_bay_position = cd.get("var_parent_bay_position") or "1"
451 variables = build_bay_chain_variables(slot, bay_position, parent_bay_position, cd.get("var_vc_position"))
452 variables["base"] = cd.get("var_base") or "Ethernet1"
453 return variables
455 def _fetch_db_preview(self, cd):
456 """Run find_interfaces_for_rule against the DB; return (db_preview, db_total, error)."""
457 from .engine import find_interfaces_for_rule
459 if cd.get("applies_to_device_interfaces"):
460 return None, 0, None
462 name_template = cd["name_template"]
463 channel_count = cd.get("channel_count") or 0
464 channel_start = cd.get("channel_start") or 0
465 module_type_is_regex = cd.get("module_type_is_regex", False)
466 module_type = cd.get("module_type")
467 module_type_pattern = cd.get("module_type_pattern", "")
469 if not ((module_type_is_regex and module_type_pattern) or (not module_type_is_regex and module_type)):
470 return None, 0, None
472 fake = RulePreview(
473 module_type_is_regex=module_type_is_regex,
474 module_type_pattern=module_type_pattern,
475 module_type=module_type,
476 parent_module_type=cd.get("parent_module_type"),
477 device_type=cd.get("device_type"),
478 platform=cd.get("platform"),
479 name_template=name_template,
480 parent_name_template=cd.get("parent_name_template", ""),
481 breakout_mode=cd.get("breakout_mode") or BreakoutModeChoices.FLAT,
482 channel_count=channel_count,
483 channel_start=channel_start,
484 )
485 try:
486 db_preview, db_total = find_interfaces_for_rule(fake, limit=100)
487 except (re.error, ValueError) as exc:
488 return [], 0, f"Invalid module type regex: {exc}"
489 except Exception as exc:
490 logger.exception("Unexpected error in find_interfaces_for_rule")
491 return [], 0, f"Unexpected error: {type(exc).__name__}"
492 else:
493 return db_preview, db_total, None
496class RuleApplyListView(BaseMultiObjectView):
497 """Display all rules with buttons to preview/apply each one."""
499 queryset = InterfaceNameRule.objects.all()
500 template_name = "netbox_interface_name_rules/rule_apply.html"
502 def get_required_permission(self):
503 """Return the permission required to access the apply-rules page."""
504 return "netbox_interface_name_rules.view_interfacenamerule"
506 def get(self, request):
507 """Render the list of all rules with apply/preview buttons."""
508 rules = self.queryset.select_related("module_type", "parent_module_type", "device_type", "platform").order_by(
509 "pk"
510 )
511 return render(request, self.template_name, {"rules": rules, "batch_limit": APPLY_BATCH_LIMIT})
514class RuleApplicableView(generic.ObjectView):
515 """Return JSON indicating whether a rule would rename at least one interface.
517 Called on demand from the Apply Rules page — NOT at page load — to avoid
518 expensive full-scan queries blocking the initial render.
519 """
521 queryset = InterfaceNameRule.objects.all()
523 def get(self, request, **kwargs):
524 """Return JSON {"applicable": bool} for the rule identified by pk."""
525 from .engine import has_applicable_interfaces
527 rule = self.get_object(**kwargs)
528 try:
529 applicable = has_applicable_interfaces(rule)
530 except Exception as exc:
531 logger.exception("applicability scan failed for rule %s", kwargs.get("pk"))
532 return JsonResponse(
533 {"applicable": None, "error": f"scan failed: {type(exc).__name__}"},
534 status=500,
535 )
536 return JsonResponse({"applicable": applicable})
539class RuleApplyDetailView(generic.ObjectView):
540 """Show a preview of changes for a specific rule and allow applying them."""
542 queryset = InterfaceNameRule.objects.all()
543 template_name = "netbox_interface_name_rules/rule_apply_detail.html"
544 additional_permissions = ["dcim.change_interface"]
546 def get(self, request, **kwargs):
547 """Render a preview of all interfaces that would be renamed by this rule."""
548 from .engine import find_convertible_families, find_interfaces_for_rule, supports_channelization
549 from .family import ConversionPreview
551 rule = self.get_object(**kwargs)
552 try:
553 preview, total_checked = find_interfaces_for_rule(rule, limit=APPLY_BATCH_LIMIT)
554 except (re.error, ValueError) as exc:
555 logger.exception("Failed to compute preview for rule %s", rule)
556 messages.error(request, f"Failed to compute preview: {exc}")
557 preview, total_checked = [], 0
558 except Exception as exc:
559 logger.exception("Unexpected error computing preview for rule %s", rule)
560 messages.error(request, f"Failed to compute preview: {type(exc).__name__}")
561 preview, total_checked = [], 0
562 # A conversion-scan failure must not blank the unrelated apply preview above.
563 try:
564 # Each family scanned costs a dry-run conversion, so the scan takes the same batch cap.
565 with write_scope():
566 preview_conversions = find_convertible_families(rule, limit=APPLY_BATCH_LIMIT)
567 except (re.error, ValueError) as exc:
568 logger.exception("Failed to compute the conversion preview for rule %s", rule)
569 messages.error(request, f"Failed to compute the conversion preview: {exc}")
570 preview_conversions = ConversionPreview(candidates=())
571 except Exception as exc:
572 logger.exception("Unexpected error computing the conversion preview for rule %s", rule)
573 messages.error(request, f"Failed to compute the conversion preview: {type(exc).__name__}")
574 preview_conversions = ConversionPreview(candidates=())
575 return render(
576 request,
577 self.template_name,
578 {
579 "rule": rule,
580 "preview": preview,
581 "conversions": preview_conversions.candidates,
582 "conversions_available": any(candidate.convertible for candidate in preview_conversions.candidates),
583 "conversions_have_more": preview_conversions.has_more,
584 "total_checked": total_checked,
585 "batch_limit": APPLY_BATCH_LIMIT,
586 "has_more": len(preview) >= APPLY_BATCH_LIMIT,
587 "can_apply": request.user.has_perm("dcim.change_interface"),
588 # Families exist only where NetBox models channelization; older releases count plain interfaces.
589 "families_supported": supports_channelization(),
590 "checked_unit": "families" if supports_channelization() else "interfaces",
591 },
592 )
594 def _enqueue(self, request, rule, job_class, name):
595 """Enqueue *job_class* against *rule* in the branch of *request*, and report the outcome to the operator."""
596 from .jobs import rule_job_kwargs
598 kwargs = rule_job_kwargs(rule.pk)
599 try:
600 # No instance=: the rule has no JobsMixin, so Job.full_clean() would refuse it.
601 job = job_class.enqueue(name=name, user=request.user, **kwargs)
602 messages.success(request, f"Background job enqueued (job #{job.pk}). Check Core → Jobs for status.")
603 except Exception as e:
604 logger.exception("Failed to enqueue background job for rule %s", rule)
605 messages.error(request, f"Failed to enqueue background job: {type(e).__name__}")
607 def _convert(self, request, rule):
608 """Convert the flat families the operator confirmed on this page."""
609 from .engine import convert_flat_families
611 convert_ids = [int(i) for i in request.POST.getlist("convert_ids") if i.isdigit()]
612 if not convert_ids:
613 messages.warning(request, "No families selected; nothing was converted.")
614 return
615 # Converting the first APPLY_BATCH_LIMIT of them would rewrite a set the operator never chose.
616 if len(convert_ids) > APPLY_BATCH_LIMIT:
617 messages.warning(
618 request,
619 f"{len(convert_ids)} families selected; synchronous conversion is limited to "
620 f"{APPLY_BATCH_LIMIT} family(ies) per run, so nothing was converted. "
621 f"Use Convert as Background Job to convert all of them.",
622 )
623 return
624 try:
625 with write_scope():
626 outcome = convert_flat_families(rule, convert_ids)
627 except Exception as e:
628 logger.exception("Failed to convert families for rule %s", rule)
629 messages.error(request, f"Failed to convert families: {_failure_text(e)}")
630 return
631 converted = len(outcome.changed_families)
632 messages.success(request, f"Converted {converted} interface family(ies) to the channelized topology.")
633 if outcome.blocked_families:
634 messages.warning(
635 request, f"{len(outcome.blocked_families)} family(ies) skipped. The plugin log names each one."
636 )
638 def post(self, request, **kwargs):
639 """Apply or convert (foreground batch or background job) and redirect back."""
640 from .engine import apply_rule_to_existing
642 rule = self.get_object(**kwargs)
643 action = request.POST.get("action", "apply")
645 if action == "convert":
646 self._convert(request, rule)
647 elif action == "convert_background":
648 from .jobs import ConvertFlatFamiliesJob
650 self._enqueue(request, rule, ConvertFlatFamiliesJob, f"Convert flat families: {rule}")
651 elif action == "background":
652 from .jobs import ApplyRuleJob
654 self._enqueue(request, rule, ApplyRuleJob, f"Apply rule: {rule}")
655 else:
656 try:
657 raw_ids = request.POST.getlist("interface_ids")
658 interface_ids = [int(i) for i in raw_ids if i.isdigit()]
659 if not interface_ids:
660 messages.warning(request, "No interfaces selected; nothing was applied.")
661 else:
662 with write_scope():
663 outcome = apply_rule_to_existing(rule, limit=APPLY_BATCH_LIMIT, interface_ids=interface_ids)
664 messages.success(request, f"Applied rule: {outcome.changed_count} interface(s) renamed.")
665 if outcome.skipped_members:
666 messages.warning(
667 request,
668 f"{len(outcome.skipped_members)} interface(s) skipped. The plugin log names each one.",
669 )
670 except Exception as e:
671 logger.exception("Failed to apply rule %s", rule)
672 messages.error(request, f"Failed to apply rule {rule}: {_failure_text(e)}")
674 return redirect("plugins:netbox_interface_name_rules:interfacenamerule_apply_detail", pk=rule.pk)
677@register_model_view(InterfaceNameRule, name="toggle", path="toggle")
678class RuleToggleView(generic.ObjectView):
679 """POST /rules/<pk>/toggle/ — flip the enabled flag on a rule."""
681 queryset = InterfaceNameRule.objects.all()
683 def post(self, request, pk):
684 """Toggle the enabled flag on the rule and return JSON or redirect."""
685 rule = self.get_object(pk=pk)
686 if not request.user.has_perm("netbox_interface_name_rules.change_interfacenamerule"): 686 ↛ 687line 686 didn't jump to line 687 because the condition on line 686 was never true
687 if request.headers.get("X-Requested-With") == "XMLHttpRequest":
688 return JsonResponse({"error": "Permission denied"}, status=403)
689 raise PermissionDenied
690 with atomic_with_events():
691 rule = InterfaceNameRule.objects.select_for_update().get(pk=rule.pk)
692 rule.snapshot()
693 rule.enabled = not rule.enabled
694 rule.save(update_fields=["enabled"])
695 if request.headers.get("X-Requested-With") == "XMLHttpRequest":
696 return JsonResponse({"enabled": rule.enabled, "pk": pk})
697 state = "enabled" if rule.enabled else "disabled"
698 messages.success(request, f"Rule '{rule}' {state}.")
699 referer = request.META.get("HTTP_REFERER", "")
700 if referer and url_has_allowed_host_and_scheme(
701 referer, allowed_hosts={request.get_host()}, require_https=request.is_secure()
702 ):
703 return redirect(referer)
704 return redirect("plugins:netbox_interface_name_rules:interfacenamerule_list")