feat: add bulk module installation

2026-07-30 16:00:31 +02:00
parent 35e06ee004
commit 8a86ca28f7
12 changed files with 432 additions and 8 deletions
@@ -4,8 +4,8 @@ from netbox.plugins import PluginConfig
 class NetBoxUtilitiesConfig(PluginConfig):
    name = "netbox_utilities"
    verbose_name = "NetBox Utilities"
-    description = "Personal navigation and a global tenant filter"
-    version = "0.2.0"
+    description = "Personal navigation, global tenant filtering, and bulk module installation"
+    version = "0.3.0"
    author = "LKE"
    base_url = "utilities"
    min_version = "4.6.5"
@@ -1,8 +1,84 @@
+from dcim.choices import ModuleStatusChoices
+from dcim.models import Device, ModuleBay, ModuleType
 from django import forms
+from utilities.forms.fields import DynamicModelChoiceField, DynamicModelMultipleChoiceField

 from .models import UtilitiesSettings


+class BulkModuleInstallForm(forms.Form):
+    device = DynamicModelChoiceField(
+        label="Gerät",
+        queryset=Device.objects.all(),
+        selector=True,
+    )
+    module_type = DynamicModelChoiceField(
+        label="Modultyp",
+        queryset=ModuleType.objects.all(),
+        context={"parent": "manufacturer"},
+        selector=True,
+    )
+    module_bays = DynamicModelMultipleChoiceField(
+        label="Modulschächte",
+        queryset=ModuleBay.objects.all(),
+        query_params={"device_id": "$device"},
+        context={"disabled": "_occupied"},
+        help_text="Alle ausgewählten Modulschächte erhalten ein Modul des gewählten Typs.",
+    )
+    status = forms.ChoiceField(
+        label="Status",
+        choices=ModuleStatusChoices,
+        initial=ModuleStatusChoices.STATUS_ACTIVE,
+    )
+    replicate_components = forms.BooleanField(
+        label="Komponenten replizieren",
+        required=False,
+        initial=True,
+        help_text="Komponenten aus den Vorlagen des Modultyps automatisch anlegen.",
+    )
+    description = forms.CharField(
+        label="Beschreibung",
+        required=False,
+        widget=forms.Textarea(attrs={"rows": 3}),
+    )
+
+    def __init__(self, *args, user, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.fields["device"].queryset = Device.objects.restrict(user, "view")
+        self.fields["module_type"].queryset = ModuleType.objects.restrict(user, "view")
+
+        module_bays = ModuleBay.objects.restrict(user, "view").filter(
+            enabled=True,
+            installed_module__isnull=True,
+        )
+        device = self.data.get("device") or self.initial.get("device")
+        if isinstance(device, Device):
+            device = device.pk
+        if device:
+            module_bays = module_bays.filter(device_id=device)
+        self.fields["module_bays"].queryset = module_bays
+
+    def clean(self):
+        cleaned_data = super().clean()
+        device = cleaned_data.get("device")
+        module_bays = cleaned_data.get("module_bays")
+        if device and module_bays:
+            invalid_bays = [
+                module_bay.name
+                for module_bay in module_bays
+                if module_bay.device_id != device.pk
+                or not module_bay.enabled
+                or hasattr(module_bay, "installed_module")
+            ]
+            if invalid_bays:
+                self.add_error(
+                    "module_bays",
+                    "Diese Modulschächte sind nicht frei oder gehören nicht zum gewählten Gerät: "
+                    + ", ".join(invalid_bays),
+                )
+        return cleaned_data
+
+
 class UtilitiesSettingsForm(forms.ModelForm):
    def __init__(self, *args, tenant_filter_locked=False, tenant_required_locked=False, **kwargs):
        super().__init__(*args, **kwargs)
@@ -0,0 +1,85 @@
+from dcim.forms import ModuleForm
+from dcim.models import ModuleBay
+from django.db import IntegrityError, transaction
+
+
+class BulkModuleInstallError(Exception):
+    pass
+
+
+def build_module_form_data(device, module_bay, module_type, status, replicate_components, description=""):
+    return {
+        "device": device.pk,
+        "module_bay": module_bay.pk,
+        "module_type": module_type.pk,
+        "status": status,
+        "replicate_components": replicate_components,
+        "adopt_components": False,
+        "description": description,
+    }
+
+
+def _format_form_errors(form):
+    messages = []
+    for field_name, errors in form.errors.items():
+        field = form.fields.get(field_name)
+        label = field.label if field else "Allgemeiner Fehler"
+        messages.append(f"{label}: {', '.join(str(error) for error in errors)}")
+    return "; ".join(messages)
+
+
+def install_modules(
+    *,
+    device,
+    module_type,
+    module_bays,
+    status,
+    replicate_components=True,
+    description="",
+    permitted_modules=None,
+):
+    module_bay_ids = {module_bay.pk for module_bay in module_bays}
+    if not module_bay_ids:
+        return []
+
+    try:
+        with transaction.atomic():
+            locked_bays = list(ModuleBay.objects.select_for_update().filter(pk__in=module_bay_ids))
+            if {module_bay.pk for module_bay in locked_bays} != module_bay_ids:
+                raise BulkModuleInstallError("Mindestens ein ausgewählter Modulschacht existiert nicht mehr.")
+
+            installed_modules = []
+            for module_bay in locked_bays:
+                if (
+                    module_bay.device_id != device.pk
+                    or not module_bay.enabled
+                    or hasattr(module_bay, "installed_module")
+                ):
+                    raise BulkModuleInstallError(f'Der Modulschacht "{module_bay.name}" ist nicht mehr verfügbar.')
+
+                module_form = ModuleForm(
+                    data=build_module_form_data(
+                        device,
+                        module_bay,
+                        module_type,
+                        status,
+                        replicate_components,
+                        description,
+                    )
+                )
+                if not module_form.is_valid():
+                    raise BulkModuleInstallError(
+                        f'Fehler bei Modulschacht "{module_bay.name}": {_format_form_errors(module_form)}'
+                    )
+                module = module_form.save()
+                if permitted_modules is not None and not permitted_modules.filter(pk=module.pk).exists():
+                    raise BulkModuleInstallError(
+                        f'Keine Berechtigung zum Einbau eines Moduls in Modulschacht "{module_bay.name}".'
+                    )
+                installed_modules.append(module)
+
+            return installed_modules
+    except IntegrityError as error:
+        raise BulkModuleInstallError(
+            "Der Masseneinbau konnte wegen eines zwischenzeitlichen Datenkonflikts nicht gespeichert werden."
+        ) from error
@@ -1,6 +1,12 @@
 from netbox.plugins import PluginMenuItem

 menu_items = (
+    PluginMenuItem(
+        link="plugins:netbox_utilities:bulk_module_install",
+        link_text="Module mehrfach einbauen",
+        auth_required=True,
+        permissions=("dcim.add_module",),
+    ),
    PluginMenuItem(
        link="plugins:netbox_utilities:navigation_preferences",
        link_text="Navigation personalisieren",
@@ -1,3 +1,4 @@
+from dcim.models import ModuleBay
 from netbox.plugins import PluginTemplateExtension
 from tenancy.models import Tenant, TenantGroup

@@ -50,4 +51,25 @@ class UtilitiesGlobalContent(PluginTemplateExtension):
        )


-template_extensions = [UtilitiesGlobalContent]
+class DeviceUtilitiesContent(PluginTemplateExtension):
+    models = ["dcim.device"]
+
+    def buttons(self):
+        request = self.context["request"]
+        device = self.context["object"]
+        if not request.user.has_perm("dcim.add_module"):
+            return ""
+        if (
+            not ModuleBay.objects.restrict(request.user, "view")
+            .filter(
+                device=device,
+                enabled=True,
+                installed_module__isnull=True,
+            )
+            .exists()
+        ):
+            return ""
+        return self.render("netbox_utilities/device_bulk_module_button.html")
+
+
+template_extensions = [UtilitiesGlobalContent, DeviceUtilitiesContent]
@@ -0,0 +1,36 @@
+{% extends 'base/layout.html' %}
+{% load form_helpers %}
+
+{% block title %}Module mehrfach einbauen{% endblock %}
+
+{% render_errors form %}
+
+{% block content %}
+  <form method="post">
+    {% csrf_token %}
+    <div class="row mb-3">
+      <div class="col col-md-8 col-xl-6 offset-md-2 offset-xl-3">
+        <div class="card">
+          <h2 class="card-header">Module mehrfach einbauen</h2>
+          <div class="card-body">
+            <div class="alert alert-info">
+              In alle ausgewählten Modulschächte wird derselbe Modultyp eingebaut.
+              Seriennummern und Asset-Tags können anschließend je Modul ergänzt werden.
+              Tritt bei einem Modul ein Fehler auf, wird keines der ausgewählten Module gespeichert.
+            </div>
+            {% render_form form %}
+          </div>
+        </div>
+      </div>
+    </div>
+    <div class="row">
+      <div class="col col-md-8 col-xl-6 offset-md-2 offset-xl-3 text-end">
+        <a href="{{ return_url }}" class="btn btn-outline-secondary">Abbrechen</a>
+        <button type="submit" class="btn btn-primary">
+          <i class="mdi mdi-package-variant-plus" aria-hidden="true"></i>
+          Module einbauen
+        </button>
+      </div>
+    </div>
+  </form>
+{% endblock content %}
@@ -0,0 +1,5 @@
+<a href="{% url 'plugins:netbox_utilities:bulk_module_install' %}?device={{ object.pk }}"
+   class="btn btn-primary">
+  <i class="mdi mdi-package-variant-plus" aria-hidden="true"></i>
+  Module mehrfach einbauen
+</a>
@@ -0,0 +1,103 @@
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from django.test import SimpleTestCase
+
+from netbox_utilities.module_installation import (
+    BulkModuleInstallError,
+    build_module_form_data,
+    install_modules,
+)
+
+
+class BulkModuleInstallationTest(SimpleTestCase):
+    def setUp(self):
+        self.device = SimpleNamespace(pk=10)
+        self.module_type = SimpleNamespace(pk=20)
+        self.module_bay = SimpleNamespace(pk=30, device_id=10, enabled=True, name="A1")
+
+    def test_builds_core_module_form_data(self):
+        data = build_module_form_data(
+            self.device,
+            self.module_bay,
+            self.module_type,
+            "active",
+            True,
+            "Patchmodul",
+        )
+
+        self.assertEqual(
+            data,
+            {
+                "device": 10,
+                "module_bay": 30,
+                "module_type": 20,
+                "status": "active",
+                "replicate_components": True,
+                "adopt_components": False,
+                "description": "Patchmodul",
+            },
+        )
+
+    @patch("netbox_utilities.module_installation.transaction.atomic")
+    @patch("netbox_utilities.module_installation.ModuleForm")
+    @patch("netbox_utilities.module_installation.ModuleBay")
+    def test_installs_module_with_core_form(self, module_bay_model, module_form_class, atomic):
+        module_bay_model.objects.select_for_update.return_value.filter.return_value = [self.module_bay]
+        form = module_form_class.return_value
+        form.is_valid.return_value = True
+        installed_module = SimpleNamespace(pk=40)
+        form.save.return_value = installed_module
+
+        result = install_modules(
+            device=self.device,
+            module_type=self.module_type,
+            module_bays=[self.module_bay],
+            status="active",
+            replicate_components=True,
+            description="Patchmodul",
+        )
+
+        self.assertEqual(result, [installed_module])
+        form.save.assert_called_once_with()
+        atomic.assert_called_once_with()
+
+    @patch("netbox_utilities.module_installation.transaction.atomic")
+    @patch("netbox_utilities.module_installation.ModuleForm")
+    @patch("netbox_utilities.module_installation.ModuleBay")
+    def test_reports_bay_and_rolls_back_on_validation_error(self, module_bay_model, module_form_class, _atomic):
+        module_bay_model.objects.select_for_update.return_value.filter.return_value = [self.module_bay]
+        form = module_form_class.return_value
+        form.is_valid.return_value = False
+        form.errors = {"__all__": ["Eine Komponente existiert bereits."]}
+        form.fields = {}
+
+        with self.assertRaisesMessage(BulkModuleInstallError, 'Fehler bei Modulschacht "A1"'):
+            install_modules(
+                device=self.device,
+                module_type=self.module_type,
+                module_bays=[self.module_bay],
+                status="active",
+            )
+
+        form.save.assert_not_called()
+
+    @patch("netbox_utilities.module_installation.transaction.atomic")
+    @patch("netbox_utilities.module_installation.ModuleForm")
+    @patch("netbox_utilities.module_installation.ModuleBay")
+    def test_rejects_module_outside_add_permission(self, module_bay_model, module_form_class, _atomic):
+        module_bay_model.objects.select_for_update.return_value.filter.return_value = [self.module_bay]
+        form = module_form_class.return_value
+        form.is_valid.return_value = True
+        form.save.return_value = SimpleNamespace(pk=40)
+        permitted_modules = MagicMock()
+        permitted_modules.filter.return_value.exists.return_value = False
+
+        with self.assertRaisesMessage(BulkModuleInstallError, "Keine Berechtigung"):
+            install_modules(
+                device=self.device,
+                module_type=self.module_type,
+                module_bays=[self.module_bay],
+                status="active",
+                permitted_modules=permitted_modules,
+            )
@@ -5,6 +5,7 @@ from . import views
 app_name = "netbox_utilities"

 urlpatterns = [
+    path("modules/bulk-install/", views.BulkModuleInstallView.as_view(), name="bulk_module_install"),
    path("navigation/", views.NavigationPreferencesView.as_view(), name="navigation_preferences"),
    path("settings/", views.UtilitiesSettingsView.as_view(), name="settings"),
    path("tenant/select/", views.SelectTenantView.as_view(), name="select_tenant"),
@@ -1,3 +1,4 @@
+from dcim.models import Device, Module
 from django.contrib import messages
 from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin
 from django.http import HttpResponseBadRequest
@@ -7,14 +8,16 @@ from django.utils.http import url_has_allowed_host_and_scheme
 from django.views import View
 from netbox.plugins import get_plugin_config
 from tenancy.models import Tenant, TenantGroup
+from utilities.views import ContentTypePermissionRequiredMixin

-from .forms import UtilitiesSettingsForm
+from .forms import BulkModuleInstallForm, UtilitiesSettingsForm
 from .middleware import (
    SESSION_TENANT_GROUP_KEY,
    SESSION_TENANT_KEY,
    GlobalTenantFilterMiddleware,
 )
 from .models import NavigationPreference, UtilitiesSettings
+from .module_installation import BulkModuleInstallError, install_modules
 from .navigation_helpers import get_visible_menus, normalize_preferences
 from .runtime import (
    clear_runtime_settings_cache,
@@ -34,6 +37,71 @@ def _safe_return_url(request, default_name="home"):
    return reverse(default_name)


+class BulkModuleInstallView(ContentTypePermissionRequiredMixin, View):
+    template_name = "netbox_utilities/bulk_module_install.html"
+
+    def get_required_permission(self):
+        return "dcim.add_module"
+
+    def get(self, request):
+        initial_device = self._get_device(request.GET.get("device"), request.user)
+        form = BulkModuleInstallForm(
+            user=request.user,
+            initial={"device": initial_device} if initial_device else None,
+        )
+        return self._render(request, form, initial_device)
+
+    def post(self, request):
+        form = BulkModuleInstallForm(request.POST, user=request.user)
+        if form.is_valid():
+            try:
+                installed_modules = install_modules(
+                    device=form.cleaned_data["device"],
+                    module_type=form.cleaned_data["module_type"],
+                    module_bays=form.cleaned_data["module_bays"],
+                    status=form.cleaned_data["status"],
+                    replicate_components=form.cleaned_data["replicate_components"],
+                    description=form.cleaned_data["description"],
+                    permitted_modules=Module.objects.restrict(request.user, "add"),
+                )
+            except BulkModuleInstallError as error:
+                form.add_error(None, str(error))
+            else:
+                device = form.cleaned_data["device"]
+                count = len(installed_modules)
+                if count == 1:
+                    success_message = f"Ein Modul des Typs {form.cleaned_data['module_type']} wurde eingebaut."
+                else:
+                    success_message = f"{count} Module des Typs {form.cleaned_data['module_type']} wurden eingebaut."
+                messages.success(request, success_message)
+                return redirect("dcim:device_modulebays", pk=device.pk)
+
+        selected_device = form.cleaned_data.get("device") if hasattr(form, "cleaned_data") else None
+        if selected_device is None:
+            selected_device = self._get_device(request.POST.get("device"), request.user)
+        return self._render(request, form, selected_device)
+
+    @staticmethod
+    def _get_device(device_id, user):
+        if not device_id:
+            return None
+        return Device.objects.restrict(user, "view").filter(pk=device_id).first()
+
+    def _render(self, request, form, device):
+        if device:
+            return_url = reverse("dcim:device_modulebays", kwargs={"pk": device.pk})
+        else:
+            return_url = reverse("dcim:module_list")
+        return render(
+            request,
+            self.template_name,
+            {
+                "form": form,
+                "return_url": return_url,
+            },
+        )
+
+
 class NavigationPreferencesView(LoginRequiredMixin, View):
    template_name = "netbox_utilities/navigation_preferences.html"