inside is foster-parented outside the
+ # table by the parser, so both OOB elements are preserved.
+ row_html = format_html("
", mark_safe(row_html))
+ else:
+ row_html = mark_safe("")
+
+ return HttpResponse(oob_modal + row_html, content_type="text/html")
+
+
+class CreatePlatformFromImportView(
+ LibreNMSPermissionMixin, NetBoxObjectPermissionMixin, LibreNMSAPIMixin, DeviceImportHelperMixin, View
+):
+ """HTMX view to create a Platform (and optionally a mapping and device assignment) from the import page."""
+
+ def get(self, request, device_id):
+ """Render the shared create-platform form fragment for the import HTMX modal."""
+ from dcim.models import Manufacturer
+
+ post_server_key = (request.GET.get("server_key") or "").strip()
+ if post_server_key:
+ from netbox_librenms_plugin.librenms_api import LibreNMSAPI
+
+ self._librenms_api = LibreNMSAPI(server_key=post_server_key)
+
+ libre_device = fetch_device_with_cache(device_id, self.librenms_api)
+ if not libre_device:
+ return HttpResponse(
+ '
Device not found in LibreNMS.
',
+ status=404,
+ )
+
+ librenms_os = (libre_device.get("os") or "").strip().lower()
+ manufacturers = list(Manufacturer.objects.all().order_by("name"))
+
+ _, validation, _ = self.get_validated_device_with_selections(device_id, request)
+ device_pk = None
+ selected_manufacturer_pk = None
+ if validation:
+ existing = validation.get("existing_device")
+ if existing:
+ device_pk = existing.pk
+ device_type = getattr(existing, "device_type", None)
+ if device_type:
+ selected_manufacturer_pk = device_type.manufacturer_id
+
+ htmx_include = (
+ f"[name=role_{device_id}], [name=rack_{device_id}], "
+ f"[name=cluster_{device_id}], #use-sysname-toggle, #strip-domain-toggle"
+ )
+
+ return render(
+ request,
+ "netbox_librenms_plugin/htmx/create_platform_modal.html",
+ {
+ "librenms_os": librenms_os,
+ "platform_name": librenms_os,
+ "manufacturers": manufacturers,
+ "form_action": request.path,
+ "device_pk": device_pk,
+ "selected_manufacturer_pk": selected_manufacturer_pk,
+ "server_key": self.librenms_api.server_key,
+ "use_htmx": True,
+ "htmx_include": htmx_include,
+ },
+ )
+
+ def post(self, request, device_id):
+ """Create platform + optional mapping + optional device assignment, then return OOB swaps."""
+ from dcim.models import Manufacturer, Platform
+
+ from netbox_librenms_plugin.models import PlatformMapping
+
+ if error := self.require_write_permission():
+ return error
+
+ post_server_key = (request.POST.get("server_key") or "").strip()
+ if post_server_key:
+ from netbox_librenms_plugin.librenms_api import LibreNMSAPI
+
+ self._librenms_api = LibreNMSAPI(server_key=post_server_key)
+
+ create_mapping = _parse_boolish(request.POST.get("create_mapping")) is True
+ device_pk_str = (request.POST.get("device_pk") or "").strip()
+ device_pk = None
+ if device_pk_str:
+ try:
+ device_pk = int(device_pk_str)
+ except (ValueError, TypeError):
+ device_pk = None
+
+ # Re-resolve the matched NetBox object via current validation. This
+ # tells us which model (Device vs VirtualMachine) to assign to and
+ # protects against a stale/spoofed hidden device_pk: we only mutate an
+ # existing object when current validation unambiguously resolves it
+ # (and the supplied device_pk, if any, agrees with that resolution).
+ try:
+ _, _validation, _ = self.get_validated_device_with_selections(device_id, request)
+ except Exception:
+ logger.exception(
+ "CreatePlatformFromImportView: failed to resolve assignment target for device_id=%s",
+ device_id,
+ )
+ return _htmx_error_response("Unable to confirm the target object for platform assignment.")
+ existing_obj = _validation.get("existing_device") if _validation else None
+
+ if existing_obj is not None and (device_pk is None or device_pk == existing_obj.pk):
+ target_model = type(existing_obj)
+ target_pk = existing_obj.pk
+ else:
+ # Either validation could not confirm a target, or the hidden
+ # device_pk disagreed with the validated object. Don't guess a
+ # model β create the platform but skip assignment so we never
+ # mutate the wrong record.
+ target_model = None
+ target_pk = None
+
+ perms = [("add", Platform)]
+ if create_mapping:
+ perms.append(("add", PlatformMapping))
+ if target_model is not None:
+ perms.append(("change", target_model))
+ self.required_object_permissions = {"POST": perms}
+
+ if error := self.require_object_permissions("POST"):
+ return error
+
+ platform_name = (request.POST.get("platform_name") or "").strip()
+ manufacturer_id = (request.POST.get("manufacturer") or "").strip()
+ librenms_os = (request.POST.get("librenms_os") or "").strip().lower()
+
+ if not platform_name:
+ return _htmx_error_response("Platform name is required.")
+
+ if Platform.objects.filter(name__iexact=platform_name).exists():
+ return _htmx_error_response(f'Platform "{platform_name}" already exists.')
+
+ manufacturer = None
+ if manufacturer_id:
+ try:
+ manufacturer = Manufacturer.objects.get(pk=int(manufacturer_id))
+ except (Manufacturer.DoesNotExist, ValueError, TypeError):
+ pass
+
+ try:
+ with transaction.atomic():
+ platform = Platform(
+ name=platform_name,
+ slug=slugify(platform_name),
+ manufacturer=manufacturer,
+ )
+ platform.full_clean()
+ platform.save()
+
+ if target_model is not None and target_pk is not None:
+ try:
+ target = target_model.objects.select_for_update().get(pk=target_pk)
+ target.platform = platform
+ target.full_clean()
+ target.save()
+ logger.info(
+ "CreatePlatformFromImportView: assigned platform '%s' to %s pk=%s",
+ platform.name,
+ target_model.__name__,
+ target_pk,
+ )
+ except target_model.DoesNotExist:
+ logger.warning(
+ "CreatePlatformFromImportView: %s pk=%s not found; platform "
+ "'%s' created but not assigned to any object",
+ target_model.__name__,
+ target_pk,
+ platform.name,
+ )
+ else:
+ logger.info(
+ "CreatePlatformFromImportView: no existing NetBox object matched "
+ "for LibreNMS device_id=%s; platform '%s' created without assignment",
+ device_id,
+ platform.name,
+ )
+
+ if create_mapping and librenms_os:
+ if not PlatformMapping.objects.filter(librenms_os__iexact=librenms_os).exists():
+ try:
+ with transaction.atomic():
+ PlatformMapping.objects.create(
+ librenms_os=librenms_os.lower(),
+ netbox_platform=platform,
+ )
+ except IntegrityError:
+ # Concurrent request created the mapping; safe to ignore.
+ pass
+ except (ValidationError, IntegrityError) as exc:
+ logger.exception("CreatePlatformFromImportView: failed to create platform: %s", exc)
+ return _htmx_error_response("Error creating platform. Please try again.")
+
+ cache_key = get_import_device_cache_key(device_id, self.librenms_api.server_key)
+ cache.delete(cache_key)
+
+ # Re-render the modal content as an OOB swap so it updates in place.
+ # The inner views render via Django templates (auto-escaped), so the
+ # decoded content is already safe HTML; wrap with format_html + mark_safe
+ # to compose the OOB envelope without introducing new escape boundaries.
+ detail_view = DeviceValidationDetailsView()
+ detail_view._librenms_api = self._librenms_api
+ modal_html = detail_view.get(request, device_id).content.decode("utf-8")
+ oob_modal = format_html(
+ '
{}
',
+ mark_safe(modal_html),
+ )
+
+ # Re-validate and include the background table row as a second OOB swap so the
+ # row reflects the new platform/mapping immediately without a secondary request.
+ libre_device, validation, selections = self.get_validated_device_with_selections(device_id, request)
+ if libre_device is not None and validation is not None:
+ row_response = self.render_device_row(request, libre_device, validation, selections)
+ row_html = row_response.content.decode("utf-8")
+ row_html = format_html("
", mark_safe(row_html))
+ else:
+ row_html = mark_safe("")
+
+ return HttpResponse(oob_modal + row_html, content_type="text/html")
+
+
class SaveUserPrefView(LibreNMSPermissionMixin, View):
"""Save a user preference via POST. Used by JS toggle handlers."""
diff --git a/netbox_librenms_plugin/views/mapping_views.py b/netbox_librenms_plugin/views/mapping_views.py
index b1fcec9c77..60f792360e 100644
--- a/netbox_librenms_plugin/views/mapping_views.py
+++ b/netbox_librenms_plugin/views/mapping_views.py
@@ -1,15 +1,69 @@
+from django.http import HttpResponse, HttpResponseBadRequest
+from django.views import View
from netbox.views import generic
from utilities.views import register_model_view
-from netbox_librenms_plugin.filters import InterfaceTypeMappingFilterSet
+from netbox_librenms_plugin.filters import (
+ CarrierAutoInstallRuleFilterSet,
+ DeviceTypeMappingFilterSet,
+ InterfaceTypeMappingFilterSet,
+ InventoryIgnoreRuleFilterSet,
+ ModuleBayMappingFilterSet,
+ ModuleTypeMappingFilterSet,
+ NormalizationRuleFilterSet,
+ PlatformMappingFilterSet,
+)
from netbox_librenms_plugin.forms import (
+ CarrierAutoInstallRuleFilterForm,
+ CarrierAutoInstallRuleForm,
+ CarrierAutoInstallRuleImportForm,
+ DeviceTypeMappingFilterForm,
+ DeviceTypeMappingForm,
+ DeviceTypeMappingImportForm,
InterfaceTypeMappingFilterForm,
InterfaceTypeMappingForm,
InterfaceTypeMappingImportForm,
+ InventoryIgnoreRuleFilterForm,
+ InventoryIgnoreRuleForm,
+ InventoryIgnoreRuleImportForm,
+ ModuleBayMappingFilterForm,
+ ModuleBayMappingForm,
+ ModuleBayMappingImportForm,
+ ModuleTypeMappingFilterForm,
+ ModuleTypeMappingForm,
+ ModuleTypeMappingImportForm,
+ NormalizationRuleFilterForm,
+ NormalizationRuleForm,
+ NormalizationRuleImportForm,
+ PlatformMappingFilterForm,
+ PlatformMappingForm,
+ PlatformMappingImportForm,
+)
+from netbox_librenms_plugin.models import (
+ CarrierAutoInstallRule,
+ DeviceTypeMapping,
+ InterfaceTypeMapping,
+ InventoryIgnoreRule,
+ ModuleBayMapping,
+ ModuleTypeMapping,
+ NormalizationRule,
+ PlatformMapping,
+)
+from netbox_librenms_plugin.tables.mappings import (
+ CarrierAutoInstallRuleTable,
+ DeviceTypeMappingTable,
+ InterfaceTypeMappingTable,
+ InventoryIgnoreRuleTable,
+ ModuleBayMappingTable,
+ ModuleTypeMappingTable,
+ NormalizationRuleTable,
+ PlatformMappingTable,
+)
+from netbox_librenms_plugin.views.mixins import (
+ LibreNMSPermissionMixin,
+ LibreNMSWritePermissionMixin,
+ NetBoxObjectPermissionMixin,
)
-from netbox_librenms_plugin.models import InterfaceTypeMapping
-from netbox_librenms_plugin.tables.mappings import InterfaceTypeMappingTable
-from netbox_librenms_plugin.views.mixins import LibreNMSPermissionMixin
class InterfaceTypeMappingListView(LibreNMSPermissionMixin, generic.ObjectListView):
@@ -24,7 +78,7 @@ class InterfaceTypeMappingListView(LibreNMSPermissionMixin, generic.ObjectListVi
template_name = "netbox_librenms_plugin/interfacetypemapping_list.html"
-class InterfaceTypeMappingCreateView(LibreNMSPermissionMixin, generic.ObjectEditView):
+class InterfaceTypeMappingCreateView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
"""
Provides a view for creating a new `InterfaceTypeMapping` object.
"""
@@ -34,7 +88,7 @@ class InterfaceTypeMappingCreateView(LibreNMSPermissionMixin, generic.ObjectEdit
@register_model_view(InterfaceTypeMapping, "bulk_import", path="import", detail=False)
-class InterfaceTypeMappingBulkImportView(LibreNMSPermissionMixin, generic.BulkImportView):
+class InterfaceTypeMappingBulkImportView(LibreNMSWritePermissionMixin, generic.BulkImportView):
"""
Provides a view for bulk importing `InterfaceTypeMapping` objects from CSV, JSON, or YAML.
Supports three import methods: direct import, file upload, and data file.
@@ -52,7 +106,7 @@ class InterfaceTypeMappingView(LibreNMSPermissionMixin, generic.ObjectView):
queryset = InterfaceTypeMapping.objects.all()
-class InterfaceTypeMappingEditView(LibreNMSPermissionMixin, generic.ObjectEditView):
+class InterfaceTypeMappingEditView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
"""
Provides a view for editing a specific `InterfaceTypeMapping` object.
"""
@@ -61,7 +115,7 @@ class InterfaceTypeMappingEditView(LibreNMSPermissionMixin, generic.ObjectEditVi
form = InterfaceTypeMappingForm
-class InterfaceTypeMappingDeleteView(LibreNMSPermissionMixin, generic.ObjectDeleteView):
+class InterfaceTypeMappingDeleteView(LibreNMSWritePermissionMixin, generic.ObjectDeleteView):
"""
Provides a view for deleting a specific `InterfaceTypeMapping` object.
"""
@@ -69,7 +123,7 @@ class InterfaceTypeMappingDeleteView(LibreNMSPermissionMixin, generic.ObjectDele
queryset = InterfaceTypeMapping.objects.all()
-class InterfaceTypeMappingBulkDeleteView(LibreNMSPermissionMixin, generic.BulkDeleteView):
+class InterfaceTypeMappingBulkDeleteView(LibreNMSWritePermissionMixin, generic.BulkDeleteView):
"""
Provides a view for deleting multiple `InterfaceTypeMapping` objects.
"""
@@ -84,3 +138,487 @@ class InterfaceTypeMappingChangeLogView(LibreNMSPermissionMixin, generic.ObjectC
"""
queryset = InterfaceTypeMapping.objects.all()
+
+
+# --- DeviceTypeMapping views ---
+
+
+class DeviceTypeMappingListView(LibreNMSPermissionMixin, generic.ObjectListView):
+ """Provides a view for listing all DeviceTypeMapping objects."""
+
+ queryset = DeviceTypeMapping.objects.select_related("netbox_device_type")
+ table = DeviceTypeMappingTable
+ filterset = DeviceTypeMappingFilterSet
+ filterset_form = DeviceTypeMappingFilterForm
+ template_name = "netbox_librenms_plugin/devicetypemapping_list.html"
+
+
+class DeviceTypeMappingCreateView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
+ """Provides a view for creating a new DeviceTypeMapping object."""
+
+ queryset = DeviceTypeMapping.objects.all()
+ form = DeviceTypeMappingForm
+
+
+@register_model_view(DeviceTypeMapping, "bulk_import", path="import", detail=False)
+class DeviceTypeMappingBulkImportView(LibreNMSWritePermissionMixin, generic.BulkImportView):
+ """Provides a view for bulk importing DeviceTypeMapping objects."""
+
+ queryset = DeviceTypeMapping.objects.all()
+ model_form = DeviceTypeMappingImportForm
+
+
+class DeviceTypeMappingView(LibreNMSPermissionMixin, generic.ObjectView):
+ """Provides a view for displaying details of a specific DeviceTypeMapping object."""
+
+ queryset = DeviceTypeMapping.objects.all()
+
+
+class DeviceTypeMappingEditView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
+ """Provides a view for editing a specific DeviceTypeMapping object."""
+
+ queryset = DeviceTypeMapping.objects.all()
+ form = DeviceTypeMappingForm
+
+
+class DeviceTypeMappingDeleteView(LibreNMSWritePermissionMixin, generic.ObjectDeleteView):
+ """Provides a view for deleting a specific DeviceTypeMapping object."""
+
+ queryset = DeviceTypeMapping.objects.all()
+
+
+class DeviceTypeMappingBulkDeleteView(LibreNMSWritePermissionMixin, generic.BulkDeleteView):
+ """Provides a view for deleting multiple DeviceTypeMapping objects."""
+
+ queryset = DeviceTypeMapping.objects.all()
+ table = DeviceTypeMappingTable
+
+
+class DeviceTypeMappingChangeLogView(LibreNMSPermissionMixin, generic.ObjectChangeLogView):
+ """Provides a view for displaying the change log of a specific DeviceTypeMapping object."""
+
+ queryset = DeviceTypeMapping.objects.all()
+
+
+# --- ModuleTypeMapping views ---
+
+
+class ModuleTypeMappingListView(LibreNMSPermissionMixin, generic.ObjectListView):
+ """Provides a view for listing all ModuleTypeMapping objects."""
+
+ queryset = ModuleTypeMapping.objects.select_related("netbox_module_type")
+ table = ModuleTypeMappingTable
+ filterset = ModuleTypeMappingFilterSet
+ filterset_form = ModuleTypeMappingFilterForm
+ template_name = "netbox_librenms_plugin/moduletypemapping_list.html"
+
+
+class ModuleTypeMappingCreateView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
+ """Provides a view for creating a new ModuleTypeMapping object."""
+
+ queryset = ModuleTypeMapping.objects.all()
+ form = ModuleTypeMappingForm
+
+
+@register_model_view(ModuleTypeMapping, "bulk_import", path="import", detail=False)
+class ModuleTypeMappingBulkImportView(LibreNMSWritePermissionMixin, generic.BulkImportView):
+ """Provides a view for bulk importing ModuleTypeMapping objects."""
+
+ queryset = ModuleTypeMapping.objects.all()
+ model_form = ModuleTypeMappingImportForm
+
+
+class ModuleTypeMappingView(LibreNMSPermissionMixin, generic.ObjectView):
+ """Provides a view for displaying details of a specific ModuleTypeMapping object."""
+
+ queryset = ModuleTypeMapping.objects.all()
+
+
+class ModuleTypeMappingEditView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
+ """Provides a view for editing a specific ModuleTypeMapping object."""
+
+ queryset = ModuleTypeMapping.objects.all()
+ form = ModuleTypeMappingForm
+
+
+class ModuleTypeMappingDeleteView(LibreNMSWritePermissionMixin, generic.ObjectDeleteView):
+ """Provides a view for deleting a specific ModuleTypeMapping object."""
+
+ queryset = ModuleTypeMapping.objects.all()
+
+
+class ModuleTypeMappingBulkDeleteView(LibreNMSWritePermissionMixin, generic.BulkDeleteView):
+ """Provides a view for deleting multiple ModuleTypeMapping objects."""
+
+ queryset = ModuleTypeMapping.objects.all()
+ table = ModuleTypeMappingTable
+
+
+class ModuleTypeMappingChangeLogView(LibreNMSPermissionMixin, generic.ObjectChangeLogView):
+ """Provides a view for displaying the change log of a specific ModuleTypeMapping object."""
+
+ queryset = ModuleTypeMapping.objects.all()
+
+
+# --- ModuleBayMapping views ---
+
+
+class ModuleBayMappingListView(LibreNMSPermissionMixin, generic.ObjectListView):
+ """Provides a view for listing all ModuleBayMapping objects."""
+
+ queryset = ModuleBayMapping.objects.all()
+ table = ModuleBayMappingTable
+ filterset = ModuleBayMappingFilterSet
+ filterset_form = ModuleBayMappingFilterForm
+ template_name = "netbox_librenms_plugin/modulebaymapping_list.html"
+
+
+class ModuleBayMappingCreateView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
+ """Provides a view for creating a new ModuleBayMapping object."""
+
+ queryset = ModuleBayMapping.objects.all()
+ form = ModuleBayMappingForm
+
+
+@register_model_view(ModuleBayMapping, "bulk_import", path="import", detail=False)
+class ModuleBayMappingBulkImportView(LibreNMSWritePermissionMixin, generic.BulkImportView):
+ """Provides a view for bulk importing ModuleBayMapping objects."""
+
+ queryset = ModuleBayMapping.objects.all()
+ model_form = ModuleBayMappingImportForm
+
+
+class ModuleBayMappingView(LibreNMSPermissionMixin, generic.ObjectView):
+ """Provides a view for displaying details of a specific ModuleBayMapping object."""
+
+ queryset = ModuleBayMapping.objects.all()
+
+
+class ModuleBayMappingEditView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
+ """Provides a view for editing a specific ModuleBayMapping object."""
+
+ queryset = ModuleBayMapping.objects.all()
+ form = ModuleBayMappingForm
+
+
+class ModuleBayMappingDeleteView(LibreNMSWritePermissionMixin, generic.ObjectDeleteView):
+ """Provides a view for deleting a specific ModuleBayMapping object."""
+
+ queryset = ModuleBayMapping.objects.all()
+
+
+class ModuleBayMappingBulkDeleteView(LibreNMSWritePermissionMixin, generic.BulkDeleteView):
+ """Provides a view for deleting multiple ModuleBayMapping objects."""
+
+ queryset = ModuleBayMapping.objects.all()
+ table = ModuleBayMappingTable
+
+
+class ModuleBayMappingChangeLogView(LibreNMSPermissionMixin, generic.ObjectChangeLogView):
+ """Provides a view for displaying the change log of a specific ModuleBayMapping object."""
+
+ queryset = ModuleBayMapping.objects.all()
+
+
+# --- NormalizationRule views ---
+
+
+class NormalizationRuleListView(LibreNMSPermissionMixin, generic.ObjectListView):
+ """Provides a view for listing all NormalizationRule objects."""
+
+ queryset = NormalizationRule.objects.select_related("manufacturer")
+ table = NormalizationRuleTable
+ filterset = NormalizationRuleFilterSet
+ filterset_form = NormalizationRuleFilterForm
+ template_name = "netbox_librenms_plugin/normalizationrule_list.html"
+
+
+class NormalizationRuleCreateView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
+ """Provides a view for creating a new NormalizationRule object."""
+
+ queryset = NormalizationRule.objects.all()
+ form = NormalizationRuleForm
+
+
+@register_model_view(NormalizationRule, "bulk_import", path="import", detail=False)
+class NormalizationRuleBulkImportView(LibreNMSWritePermissionMixin, generic.BulkImportView):
+ """Provides a view for bulk importing NormalizationRule objects."""
+
+ queryset = NormalizationRule.objects.all()
+ model_form = NormalizationRuleImportForm
+
+
+class NormalizationRuleView(LibreNMSPermissionMixin, generic.ObjectView):
+ """Provides a view for displaying details of a specific NormalizationRule object."""
+
+ queryset = NormalizationRule.objects.all()
+
+
+class NormalizationRuleEditView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
+ """Provides a view for editing a specific NormalizationRule object."""
+
+ queryset = NormalizationRule.objects.all()
+ form = NormalizationRuleForm
+
+
+class NormalizationRuleDeleteView(LibreNMSWritePermissionMixin, generic.ObjectDeleteView):
+ """Provides a view for deleting a specific NormalizationRule object."""
+
+ queryset = NormalizationRule.objects.all()
+
+
+class NormalizationRuleBulkDeleteView(LibreNMSWritePermissionMixin, generic.BulkDeleteView):
+ """Provides a view for deleting multiple NormalizationRule objects."""
+
+ queryset = NormalizationRule.objects.all()
+ table = NormalizationRuleTable
+
+
+class NormalizationRuleChangeLogView(LibreNMSPermissionMixin, generic.ObjectChangeLogView):
+ """Provides a view for displaying the change log of a specific NormalizationRule object."""
+
+ queryset = NormalizationRule.objects.all()
+
+
+# --- InventoryIgnoreRule views ---
+
+
+class InventoryIgnoreRuleListView(LibreNMSPermissionMixin, generic.ObjectListView):
+ """Provides a view for listing all InventoryIgnoreRule objects."""
+
+ queryset = InventoryIgnoreRule.objects.all()
+ table = InventoryIgnoreRuleTable
+ filterset = InventoryIgnoreRuleFilterSet
+ filterset_form = InventoryIgnoreRuleFilterForm
+ template_name = "netbox_librenms_plugin/inventoryignorerule_list.html"
+
+
+class InventoryIgnoreRuleCreateView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
+ """Provides a view for creating a new InventoryIgnoreRule object."""
+
+ queryset = InventoryIgnoreRule.objects.all()
+ form = InventoryIgnoreRuleForm
+
+
+@register_model_view(InventoryIgnoreRule, "bulk_import", path="import", detail=False)
+class InventoryIgnoreRuleBulkImportView(LibreNMSWritePermissionMixin, generic.BulkImportView):
+ """Provides a view for bulk importing InventoryIgnoreRule objects."""
+
+ queryset = InventoryIgnoreRule.objects.all()
+ model_form = InventoryIgnoreRuleImportForm
+
+
+class InventoryIgnoreRuleView(LibreNMSPermissionMixin, generic.ObjectView):
+ """Provides a view for displaying details of a specific InventoryIgnoreRule object."""
+
+ queryset = InventoryIgnoreRule.objects.all()
+
+
+class InventoryIgnoreRuleEditView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
+ """Provides a view for editing a specific InventoryIgnoreRule object."""
+
+ queryset = InventoryIgnoreRule.objects.all()
+ form = InventoryIgnoreRuleForm
+
+
+class InventoryIgnoreRuleDeleteView(LibreNMSWritePermissionMixin, generic.ObjectDeleteView):
+ """Provides a view for deleting a specific InventoryIgnoreRule object."""
+
+ queryset = InventoryIgnoreRule.objects.all()
+
+
+class InventoryIgnoreRuleBulkDeleteView(LibreNMSWritePermissionMixin, generic.BulkDeleteView):
+ """Provides a view for deleting multiple InventoryIgnoreRule objects."""
+
+ queryset = InventoryIgnoreRule.objects.all()
+ table = InventoryIgnoreRuleTable
+
+
+class InventoryIgnoreRuleChangeLogView(LibreNMSPermissionMixin, generic.ObjectChangeLogView):
+ """Provides a view for displaying the change log of a specific InventoryIgnoreRule object."""
+
+ queryset = InventoryIgnoreRule.objects.all()
+
+
+# --- BulkExportYAML views ---
+
+
+class BulkExportYAMLView(LibreNMSPermissionMixin, NetBoxObjectPermissionMixin, View):
+ """Base view that exports selected mapping objects as YAML."""
+
+ queryset = None
+
+ @property
+ def required_object_permissions(self):
+ return {"POST": [("view", self.queryset.model)]}
+
+ def post(self, request):
+ if error := self.require_object_permissions("POST"):
+ return error
+ pks = request.POST.getlist("pk")
+ if not pks:
+ return HttpResponseBadRequest("No objects selected.")
+ try:
+ int_pks = [int(pk) for pk in pks]
+ except (ValueError, TypeError):
+ return HttpResponseBadRequest("Invalid pk value.")
+ objects = self.queryset.filter(pk__in=int_pks).order_by("pk")
+ if not objects:
+ return HttpResponseBadRequest("No matching objects found.")
+ yaml_parts = [obj.to_yaml() for obj in objects]
+ content = "---\n".join(yaml_parts)
+ response = HttpResponse(content, content_type="text/yaml; charset=utf-8")
+ response["Content-Disposition"] = 'attachment; filename="export.yaml"'
+ return response
+
+
+class InterfaceTypeMappingBulkExportYAMLView(BulkExportYAMLView):
+ queryset = InterfaceTypeMapping.objects.all()
+
+
+class DeviceTypeMappingBulkExportYAMLView(BulkExportYAMLView):
+ queryset = DeviceTypeMapping.objects.select_related("netbox_device_type")
+
+
+class ModuleTypeMappingBulkExportYAMLView(BulkExportYAMLView):
+ queryset = ModuleTypeMapping.objects.select_related("netbox_module_type")
+
+
+class ModuleBayMappingBulkExportYAMLView(BulkExportYAMLView):
+ queryset = ModuleBayMapping.objects.all()
+
+
+class NormalizationRuleBulkExportYAMLView(BulkExportYAMLView):
+ queryset = NormalizationRule.objects.select_related("manufacturer")
+
+
+class InventoryIgnoreRuleBulkExportYAMLView(BulkExportYAMLView):
+ queryset = InventoryIgnoreRule.objects.all()
+
+
+class PlatformMappingBulkExportYAMLView(BulkExportYAMLView):
+ queryset = PlatformMapping.objects.select_related("netbox_platform")
+
+
+# --- PlatformMapping views ---
+
+
+class PlatformMappingListView(LibreNMSPermissionMixin, generic.ObjectListView):
+ """Provides a view for listing all PlatformMapping objects."""
+
+ queryset = PlatformMapping.objects.select_related("netbox_platform")
+ table = PlatformMappingTable
+ filterset = PlatformMappingFilterSet
+ filterset_form = PlatformMappingFilterForm
+ template_name = "netbox_librenms_plugin/platformmapping_list.html"
+
+
+class PlatformMappingCreateView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
+ """Provides a view for creating a new PlatformMapping object."""
+
+ queryset = PlatformMapping.objects.all()
+ form = PlatformMappingForm
+
+
+@register_model_view(PlatformMapping, "bulk_import", path="import", detail=False)
+class PlatformMappingBulkImportView(LibreNMSWritePermissionMixin, generic.BulkImportView):
+ """Provides a view for bulk importing PlatformMapping objects."""
+
+ queryset = PlatformMapping.objects.all()
+ model_form = PlatformMappingImportForm
+
+
+class PlatformMappingView(LibreNMSPermissionMixin, generic.ObjectView):
+ """Provides a view for displaying details of a specific PlatformMapping object."""
+
+ queryset = PlatformMapping.objects.all()
+
+
+class PlatformMappingEditView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
+ """Provides a view for editing a specific PlatformMapping object."""
+
+ queryset = PlatformMapping.objects.all()
+ form = PlatformMappingForm
+
+
+class PlatformMappingDeleteView(LibreNMSWritePermissionMixin, generic.ObjectDeleteView):
+ """Provides a view for deleting a specific PlatformMapping object."""
+
+ queryset = PlatformMapping.objects.all()
+
+
+class PlatformMappingBulkDeleteView(LibreNMSWritePermissionMixin, generic.BulkDeleteView):
+ """Provides a view for deleting multiple PlatformMapping objects."""
+
+ queryset = PlatformMapping.objects.all()
+ table = PlatformMappingTable
+
+
+class PlatformMappingChangeLogView(LibreNMSPermissionMixin, generic.ObjectChangeLogView):
+ """Provides a view for displaying the change log of a specific PlatformMapping object."""
+
+ queryset = PlatformMapping.objects.all()
+
+
+# --- CarrierAutoInstallRule views ---
+
+
+class CarrierAutoInstallRuleListView(LibreNMSPermissionMixin, generic.ObjectListView):
+ """Provides a view for listing all CarrierAutoInstallRule objects."""
+
+ queryset = CarrierAutoInstallRule.objects.select_related("manufacturer", "carrier_module_type")
+ table = CarrierAutoInstallRuleTable
+ filterset = CarrierAutoInstallRuleFilterSet
+ filterset_form = CarrierAutoInstallRuleFilterForm
+ template_name = "netbox_librenms_plugin/carrierautoinstallrule_list.html"
+
+
+class CarrierAutoInstallRuleCreateView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
+ """Provides a view for creating a new CarrierAutoInstallRule object."""
+
+ queryset = CarrierAutoInstallRule.objects.all()
+ form = CarrierAutoInstallRuleForm
+
+
+@register_model_view(CarrierAutoInstallRule, "bulk_import", path="import", detail=False)
+class CarrierAutoInstallRuleBulkImportView(LibreNMSWritePermissionMixin, generic.BulkImportView):
+ """Provides a view for bulk importing CarrierAutoInstallRule objects."""
+
+ queryset = CarrierAutoInstallRule.objects.all()
+ model_form = CarrierAutoInstallRuleImportForm
+
+
+class CarrierAutoInstallRuleView(LibreNMSPermissionMixin, generic.ObjectView):
+ """Provides a view for displaying details of a specific CarrierAutoInstallRule object."""
+
+ queryset = CarrierAutoInstallRule.objects.select_related("manufacturer", "carrier_module_type")
+
+
+class CarrierAutoInstallRuleEditView(LibreNMSWritePermissionMixin, generic.ObjectEditView):
+ """Provides a view for editing a specific CarrierAutoInstallRule object."""
+
+ queryset = CarrierAutoInstallRule.objects.all()
+ form = CarrierAutoInstallRuleForm
+
+
+class CarrierAutoInstallRuleDeleteView(LibreNMSWritePermissionMixin, generic.ObjectDeleteView):
+ """Provides a view for deleting a specific CarrierAutoInstallRule object."""
+
+ queryset = CarrierAutoInstallRule.objects.all()
+
+
+class CarrierAutoInstallRuleBulkDeleteView(LibreNMSWritePermissionMixin, generic.BulkDeleteView):
+ """Provides a view for deleting multiple CarrierAutoInstallRule objects."""
+
+ queryset = CarrierAutoInstallRule.objects.all()
+ table = CarrierAutoInstallRuleTable
+
+
+class CarrierAutoInstallRuleChangeLogView(LibreNMSPermissionMixin, generic.ObjectChangeLogView):
+ """Provides a view for displaying the change log of a specific CarrierAutoInstallRule object."""
+
+ queryset = CarrierAutoInstallRule.objects.all()
+
+
+class CarrierAutoInstallRuleBulkExportYAMLView(BulkExportYAMLView):
+ queryset = CarrierAutoInstallRule.objects.select_related("manufacturer", "carrier_module_type")
diff --git a/netbox_librenms_plugin/views/mixins.py b/netbox_librenms_plugin/views/mixins.py
index 62136017a1..f327183223 100644
--- a/netbox_librenms_plugin/views/mixins.py
+++ b/netbox_librenms_plugin/views/mixins.py
@@ -1,6 +1,8 @@
+import json
+
from django.contrib import messages
from django.contrib.auth.mixins import PermissionRequiredMixin
-from django.http import HttpResponse
+from django.http import HttpResponse, JsonResponse
from django.shortcuts import redirect
from django.utils.http import url_has_allowed_host_and_scheme
from utilities.permissions import get_permission_for_model
@@ -9,6 +11,17 @@
from netbox_librenms_plugin.librenms_api import LibreNMSAPI
+def parse_request_json(request):
+ """Parse JSON from request.body, returning (data, error_response).
+
+ On success returns (dict, None). On malformed input returns (None, JsonResponse 400).
+ """
+ try:
+ return json.loads(request.body), None
+ except (TypeError, ValueError):
+ return None, JsonResponse({"status": "error", "message": "Invalid JSON payload"}, status=400)
+
+
def _get_safe_redirect_url(request):
"""
Return a validated redirect URL from the HTTP Referer header.
@@ -62,6 +75,7 @@ def require_write_permission(self, error_message=None):
if self.request.headers.get("HX-Request"):
return HttpResponse("", headers={"HX-Redirect": referrer})
+ # referrer is safe: validated by _get_safe_redirect_url via url_has_allowed_host_and_scheme
return redirect(referrer)
return None
@@ -83,6 +97,18 @@ def require_write_permission_json(self, error_message=None):
return None
+class LibreNMSWritePermissionMixin(LibreNMSPermissionMixin):
+ """
+ Mixin for mutation views requiring LibreNMS plugin write permission.
+
+ Sets permission_required to 'change_librenmssettings' so that only users
+ with write access can access Create, Edit, Delete, BulkImport, and
+ BulkDelete views.
+ """
+
+ permission_required = PERM_CHANGE_PLUGIN
+
+
class NetBoxObjectPermissionMixin:
"""
Mixin for views requiring specific NetBox object permissions.
@@ -144,6 +170,7 @@ def require_object_permissions(self, method):
if self.request.headers.get("HX-Request"):
return HttpResponse("", headers={"HX-Redirect": referrer})
+ # referrer is safe: validated by _get_safe_redirect_url via url_has_allowed_host_and_scheme
return redirect(referrer)
return None
diff --git a/netbox_librenms_plugin/views/object_sync/__init__.py b/netbox_librenms_plugin/views/object_sync/__init__.py
index e9893cf7d8..118b2ff944 100644
--- a/netbox_librenms_plugin/views/object_sync/__init__.py
+++ b/netbox_librenms_plugin/views/object_sync/__init__.py
@@ -5,9 +5,11 @@
DeviceInterfaceTableView,
DeviceIPAddressTableView,
DeviceLibreNMSSyncView,
+ DeviceModuleTableView,
DeviceVLANTableView,
SaveVlanGroupOverridesView,
SingleInterfaceVerifyView,
+ SingleModuleVerifyView,
SingleVlanGroupVerifyView,
VerifyVlanSyncGroupView,
)
diff --git a/netbox_librenms_plugin/views/object_sync/devices.py b/netbox_librenms_plugin/views/object_sync/devices.py
index b3777e4cfd..05606058f7 100644
--- a/netbox_librenms_plugin/views/object_sync/devices.py
+++ b/netbox_librenms_plugin/views/object_sync/devices.py
@@ -1,4 +1,4 @@
-import json
+import copy
from dcim.models import Device
from django.core.cache import cache
@@ -17,6 +17,7 @@
LibreNMSInterfaceTable,
VCInterfaceTable,
)
+from netbox_librenms_plugin.tables.modules import LibreNMSModuleTable, VCModuleTable
from netbox_librenms_plugin.utils import (
get_interface_name_field,
get_librenms_sync_device,
@@ -30,8 +31,15 @@
from ..base.interfaces_view import BaseInterfaceTableView
from ..base.ip_addresses_view import BaseIPAddressTableView
from ..base.librenms_sync_view import BaseLibreNMSSyncView
+from ..base.modules_view import BaseModuleTableView, _check_ignore_rules
from ..base.vlan_table_view import BaseVLANTableView
-from ..mixins import CacheMixin, LibreNMSAPIMixin, LibreNMSPermissionMixin
+from ..mixins import (
+ CacheMixin,
+ LibreNMSAPIMixin,
+ LibreNMSPermissionMixin,
+ NetBoxObjectPermissionMixin,
+ parse_request_json,
+)
@register_model_view(Device, name="librenms_sync", path="librenms-sync")
@@ -46,24 +54,32 @@ def get_interface_context(self, request, obj):
"""Return interface sync context for the device."""
interface_name_field = get_interface_name_field(request)
interface_table_view = DeviceInterfaceTableView()
- interface_table_view.request = request
+ interface_table_view.request = copy.copy(request)
return interface_table_view.get_context_data(request, obj, interface_name_field)
def get_cable_context(self, request, obj):
"""Return cable sync context for the device."""
cable_table_view = DeviceCableTableView()
+ cable_table_view.request = copy.copy(request)
return cable_table_view.get_context_data(request, obj)
def get_ip_context(self, request, obj):
"""Return IP address sync context for the device."""
ipaddress_table_view = DeviceIPAddressTableView()
+ ipaddress_table_view.request = copy.copy(request)
return ipaddress_table_view.get_context_data(request, obj)
def get_vlan_context(self, request, obj):
vlan_table_view = DeviceVLANTableView()
- vlan_table_view.request = request
+ vlan_table_view.request = copy.copy(request)
return vlan_table_view.get_vlan_context(request, obj)
+ def get_module_context(self, request, obj):
+ """Return module sync context for the device."""
+ module_table_view = DeviceModuleTableView()
+ module_table_view.request = copy.copy(request)
+ return module_table_view.get_context_data(request, obj)
+
class DeviceInterfaceTableView(BaseInterfaceTableView):
"""Interface synchronization table for Devices."""
@@ -106,14 +122,18 @@ class SingleInterfaceVerifyView(LibreNMSPermissionMixin, LibreNMSAPIMixin, Cache
def post(self, request):
"""Verify interface data against cached LibreNMS ports for a device."""
- data = json.loads(request.body)
+ data, err = parse_request_json(request)
+ if err:
+ return err
selected_device_id = data.get("device_id")
interface_name = data.get("interface_name")
interface_name_field = data.get("interface_name_field") or get_interface_name_field()
- server_key = data.get("server_key") or self.librenms_api.server_key
+ server_key = data.get("server_key")
if not selected_device_id:
return JsonResponse({"status": "error", "message": "No device ID provided"}, status=400)
+ if not server_key:
+ server_key = self.librenms_api.server_key
selected_device = get_object_or_404(Device, pk=selected_device_id)
@@ -149,6 +169,169 @@ def post(self, request):
return JsonResponse({"status": "error", "message": "Interface data not found"}, status=404)
+class SingleModuleVerifyView(
+ LibreNMSPermissionMixin,
+ NetBoxObjectPermissionMixin,
+ LibreNMSAPIMixin,
+ CacheMixin,
+ View,
+):
+ """Verify module row data against cached LibreNMS inventory for a selected VC member."""
+
+ # JSON endpoint that returns the rendered module row β only viewers of the
+ # underlying Device should be able to surface its inventory data.
+ required_object_permissions = {"POST": [("view", Device)]}
+
+ def post(self, request):
+ data, err = parse_request_json(request)
+ if err:
+ return err
+ selected_device_id = data.get("device_id")
+ ent_physical_index = data.get("ent_physical_index")
+ server_key = data.get("server_key") or self.librenms_api.server_key
+ row_depth = data.get("depth", 0)
+
+ if not selected_device_id:
+ return JsonResponse({"status": "error", "message": "No device ID provided"}, status=400)
+ if ent_physical_index in (None, ""):
+ return JsonResponse({"status": "error", "message": "No entPhysicalIndex provided"}, status=400)
+
+ try:
+ ent_physical_index = int(ent_physical_index)
+ except (TypeError, ValueError):
+ return JsonResponse({"status": "error", "message": "Invalid entPhysicalIndex"}, status=400)
+
+ try:
+ row_depth = int(row_depth)
+ except (TypeError, ValueError):
+ row_depth = 0
+
+ selected_device = get_object_or_404(Device, pk=selected_device_id)
+
+ # Read-only verify endpoint: only require object-view permission, not
+ # plugin write (which require_all_permissions_json would enforce).
+ if error := self.require_object_permissions_json("POST"):
+ return error
+
+ if selected_device.virtual_chassis:
+ sync_device = get_librenms_sync_device(selected_device, server_key=server_key)
+ if sync_device is None:
+ return JsonResponse({"status": "error", "message": "No sync device found for VC"}, status=404)
+ else:
+ sync_device = selected_device
+
+ cached_payload = cache.get(self.get_cache_key(sync_device, "inventory", server_key))
+ if not isinstance(cached_payload, dict):
+ return JsonResponse({"status": "error", "message": "No cached inventory data"}, status=404)
+
+ inventory_data = cached_payload.get("inventory") or []
+ index_map = {idx: item for item in inventory_data if (idx := item.get("entPhysicalIndex")) is not None}
+ item = index_map.get(ent_physical_index)
+ if not item:
+ return JsonResponse({"status": "error", "message": "Inventory row not found"}, status=404)
+
+ module_table_view = DeviceModuleTableView()
+ # Shallow-copy the request so the child view can mutate request.GET /
+ # request.POST without affecting this request object.
+ module_table_view.request = copy.copy(request)
+
+ from netbox_librenms_plugin.utils import (
+ get_enabled_ignore_rules,
+ load_bay_mappings,
+ preload_normalization_rules,
+ )
+
+ module_table_view._exact_bay_mappings, module_table_view._regex_bay_mappings = load_bay_mappings()
+ manufacturer = getattr(getattr(selected_device, "device_type", None), "manufacturer", None)
+ module_table_view._norm_rules_bay = preload_normalization_rules("module_bay")
+ module_table_view._norm_rules_type = preload_normalization_rules("module_type", manufacturer=manufacturer)
+
+ children_by_parent = {}
+ for inventory_item in inventory_data:
+ parent_idx = inventory_item.get("entPhysicalContainedIn")
+ if parent_idx is not None:
+ children_by_parent.setdefault(parent_idx, []).append(inventory_item)
+
+ ignore_rules = get_enabled_ignore_rules()
+ device_serial = (getattr(selected_device, "serial", None) or "").strip()
+ ignore_cache = {
+ inventory_item["entPhysicalIndex"]: _check_ignore_rules(
+ inventory_item,
+ index_map.get(inventory_item.get("entPhysicalContainedIn")),
+ ignore_rules,
+ index_map,
+ device_serial,
+ )
+ for inventory_item in inventory_data
+ if inventory_item.get("entPhysicalIndex") is not None
+ }
+
+ module_types = module_table_view._get_module_types()
+ transparent_indices = module_table_view._find_transparent_indices(inventory_data, ignore_cache)
+ top_items = module_table_view._collect_top_items(
+ inventory_data,
+ index_map,
+ ignore_rules,
+ device_serial,
+ transparent_indices,
+ ignore_cache,
+ )
+ table_data = module_table_view._build_table_rows_for_member(
+ selected_device,
+ top_items,
+ index_map,
+ children_by_parent,
+ ignore_rules,
+ device_serial,
+ module_types,
+ manufacturer=manufacturer,
+ )
+ module_table_view._detect_serial_conflicts(table_data)
+
+ # entPhysicalIndex should be unique, depth fallback handles malformed duplicates.
+ row = next(
+ (
+ candidate
+ for candidate in table_data
+ if candidate.get("ent_physical_index") == ent_physical_index and candidate.get("depth", 0) == row_depth
+ ),
+ None,
+ )
+ if row is None:
+ row = next(
+ (candidate for candidate in table_data if candidate.get("ent_physical_index") == ent_physical_index),
+ None,
+ )
+ if row is None:
+ return JsonResponse({"status": "error", "message": "Inventory row not found"}, status=404)
+
+ has_write_permission = self.has_write_permission()
+ table_class = VCModuleTable if selected_device.virtual_chassis else LibreNMSModuleTable
+ table = table_class(
+ [],
+ device=selected_device,
+ server_key=server_key,
+ has_write_permission=has_write_permission,
+ can_add_module=has_write_permission and request.user.has_perm("dcim.add_module"),
+ can_change_module=has_write_permission and request.user.has_perm("dcim.change_module"),
+ can_delete_module=has_write_permission and request.user.has_perm("dcim.delete_module"),
+ can_add_module_bay_template=(has_write_permission and request.user.has_perm("dcim.add_modulebaytemplate")),
+ can_add_module_type=(has_write_permission and request.user.has_perm("dcim.add_moduletype")),
+ can_add_carrier_rule=(
+ has_write_permission and request.user.has_perm("netbox_librenms_plugin.add_carrierautoinstallrule")
+ ),
+ can_add_module_bay_mapping=(
+ has_write_permission and request.user.has_perm("netbox_librenms_plugin.add_modulebaymapping")
+ ),
+ can_add_module_type_mapping=(
+ has_write_permission and request.user.has_perm("netbox_librenms_plugin.add_moduletypemapping")
+ ),
+ )
+ table.configure(request)
+ formatted_row = table.format_module_data(row)
+ return JsonResponse({"status": "success", "formatted_row": formatted_row})
+
+
class SingleVlanGroupVerifyView(LibreNMSPermissionMixin, CacheMixin, View):
"""
Verify VLAN assignments for an interface against a specific VLAN group.
@@ -161,7 +344,9 @@ class SingleVlanGroupVerifyView(LibreNMSPermissionMixin, CacheMixin, View):
def post(self, request):
from ipam.models import VLAN, VLANGroup
- data = json.loads(request.body)
+ data, err = parse_request_json(request)
+ if err:
+ return err
device_id = data.get("device_id")
interface_name = data.get("interface_name")
vlan_group_id = data.get("vlan_group_id")
@@ -287,7 +472,9 @@ class VerifyVlanSyncGroupView(LibreNMSPermissionMixin, View):
def post(self, request):
from ipam.models import VLAN, VLANGroup
- data = json.loads(request.body)
+ data, err = parse_request_json(request)
+ if err:
+ return err
vlan_group_id = data.get("vlan_group_id")
vid_str = data.get("vid", "")
librenms_name = data.get("name", "")
@@ -339,13 +526,17 @@ def post(self, request):
if error := self.require_write_permission_json():
return error
- data = json.loads(request.body)
+ data, err = parse_request_json(request)
+ if err:
+ return err
device_id = data.get("device_id")
vid_group_map = data.get("vid_group_map", {})
- server_key = data.get("server_key") or self.librenms_api.server_key
+ server_key = data.get("server_key")
if not device_id:
return JsonResponse({"status": "error", "message": "No device ID provided"}, status=400)
+ if not server_key:
+ server_key = self.librenms_api.server_key
device = get_object_or_404(Device, pk=device_id)
@@ -393,3 +584,38 @@ class DeviceVLANTableView(BaseVLANTableView):
"""VLAN synchronization table view for Devices."""
model = Device
+
+
+class DeviceModuleTableView(BaseModuleTableView):
+ """Module/inventory synchronization view for Devices."""
+
+ model = Device
+
+ def get_table(self, data, obj):
+ """Return the module sync table."""
+ user = self.request.user
+ has_write_permission = self.has_write_permission()
+ table_class = VCModuleTable if hasattr(obj, "virtual_chassis") and obj.virtual_chassis else LibreNMSModuleTable
+ table = table_class(
+ data,
+ device=obj,
+ server_key=self.librenms_api.server_key,
+ has_write_permission=has_write_permission,
+ can_add_module=has_write_permission and user.has_perm("dcim.add_module"),
+ can_change_module=has_write_permission and user.has_perm("dcim.change_module"),
+ can_delete_module=has_write_permission and user.has_perm("dcim.delete_module"),
+ can_add_module_bay_template=(has_write_permission and user.has_perm("dcim.add_modulebaytemplate")),
+ can_add_module_type=(has_write_permission and user.has_perm("dcim.add_moduletype")),
+ can_add_carrier_rule=(
+ has_write_permission and user.has_perm("netbox_librenms_plugin.add_carrierautoinstallrule")
+ ),
+ can_add_module_bay_mapping=(
+ has_write_permission and user.has_perm("netbox_librenms_plugin.add_modulebaymapping")
+ ),
+ can_add_module_type_mapping=(
+ has_write_permission and user.has_perm("netbox_librenms_plugin.add_moduletypemapping")
+ ),
+ )
+ server_key = self.librenms_api.server_key
+ table.htmx_url = f"{self.request.path}?tab=modules" + (f"&server_key={server_key}" if server_key else "")
+ return table
diff --git a/netbox_librenms_plugin/views/sync/device_fields.py b/netbox_librenms_plugin/views/sync/device_fields.py
index d92ab33ddf..68f6e855ad 100644
--- a/netbox_librenms_plugin/views/sync/device_fields.py
+++ b/netbox_librenms_plugin/views/sync/device_fields.py
@@ -8,13 +8,16 @@
from django.http import HttpResponse
from django.shortcuts import get_object_or_404, redirect
+from django.utils.html import escape
from django.views import View
from virtualization.models import VirtualMachine
from netbox_librenms_plugin.import_utils import _determine_device_name
from netbox_librenms_plugin.import_utils.virtual_chassis import _generate_vc_member_name
+from netbox_librenms_plugin.models import PlatformMapping
from netbox_librenms_plugin.utils import (
find_by_librenms_id,
+ find_matching_platform,
get_librenms_sync_device,
match_librenms_hardware_to_device_type,
migrate_legacy_librenms_id,
@@ -195,7 +198,14 @@ def post(self, request, pk):
match_result = match_librenms_hardware_to_device_type(hardware)
- if not match_result or not match_result["matched"]:
+ if match_result is None:
+ messages.error(
+ request,
+ f"Ambiguous hardware match for '{hardware}': multiple matching mappings/device types found.",
+ )
+ return redirect("plugins:netbox_librenms_plugin:device_librenms_sync", pk=pk)
+
+ if not match_result["matched"]:
messages.error(
request,
f"No matching DeviceType found for hardware '{hardware}'",
@@ -254,19 +264,24 @@ def post(self, request, pk):
messages.warning(request, "No OS information available in LibreNMS")
return redirect("plugins:netbox_librenms_plugin:device_librenms_sync", pk=pk)
- platform_name = os_name
-
- try:
- platform = Platform.objects.get(name__iexact=platform_name)
- except Platform.DoesNotExist:
+ result = find_matching_platform(os_name)
+ if result["match_type"] == "ambiguous":
+ messages.error(
+ request,
+ "Multiple platforms match '{}'. Please resolve the ambiguity via a Platform Mapping.".format(os_name),
+ )
+ return redirect("plugins:netbox_librenms_plugin:device_librenms_sync", pk=pk)
+ if not result["found"] or result["platform"] is None:
messages.error(
request,
"Platform '{}' does not exist in NetBox. Use 'Create & Sync' button to create it first.".format(
- platform_name
+ os_name
),
)
return redirect("plugins:netbox_librenms_plugin:device_librenms_sync", pk=pk)
+ platform = result["platform"]
+
old_platform = device.platform
device.platform = platform
try:
@@ -301,6 +316,17 @@ class CreateAndAssignPlatformView(LibreNMSPermissionMixin, NetBoxObjectPermissio
def post(self, request, pk):
"""Create a new platform and assign it to the device."""
+ # Read create_mapping before permission check so it can be included in the check.
+ create_mapping = bool(request.POST.get("create_mapping"))
+ if create_mapping:
+ self.required_object_permissions = {
+ "POST": [
+ ("change", Device),
+ ("add", Platform),
+ ("add", PlatformMapping),
+ ],
+ }
+
# Check both plugin write and NetBox object permissions
if error := self.require_all_permissions("POST"):
return error
@@ -309,6 +335,7 @@ def post(self, request, pk):
platform_name = request.POST.get("platform_name")
manufacturer_id = request.POST.get("manufacturer")
+ librenms_os = (request.POST.get("librenms_os") or "").strip().lower()
if not platform_name:
messages.error(request, "Platform name is required")
@@ -398,10 +425,46 @@ def post(self, request, pk):
)
return redirect("plugins:netbox_librenms_plugin:device_librenms_sync", pk=pk)
- messages.success(
- request,
- f"Created platform '{platform}' and assigned to device",
- )
+ mapping_created = False
+ mapping_error = None
+ mapping_existed = False
+ if create_mapping and librenms_os:
+ existing = PlatformMapping.objects.filter(librenms_os__iexact=librenms_os).first()
+ if existing is not None:
+ mapping_existed = True
+ else:
+ try:
+ with transaction.atomic():
+ mapping = PlatformMapping(librenms_os=librenms_os, netbox_platform=platform)
+ mapping.full_clean()
+ mapping.save()
+ mapping_created = True
+ except ValidationError as e:
+ mapping_error = e.message_dict if hasattr(e, "message_dict") else str(e)
+ logger.exception("Failed to create PlatformMapping '%s' -> '%s'", librenms_os, platform_name)
+ except IntegrityError:
+ # Concurrent insert: mapping was created by another request
+ mapping_existed = True
+ logger.warning(
+ "IntegrityError creating PlatformMapping '%s' -> '%s'; treating as already existing",
+ librenms_os,
+ platform_name,
+ )
+
+ msg = f"Created platform '{platform}' and assigned to device"
+ if mapping_created:
+ msg += f" β platform mapping '{librenms_os}' β '{platform}' added"
+ messages.success(request, msg)
+ if mapping_error:
+ messages.warning(
+ request,
+ f"Platform mapping '{librenms_os}' β '{platform}' could not be created: {mapping_error}",
+ )
+ elif mapping_existed:
+ messages.info(
+ request,
+ f"Platform mapping for '{librenms_os}' already exists; not modified.",
+ )
return redirect("plugins:netbox_librenms_plugin:device_librenms_sync", pk=pk)
@@ -514,7 +577,7 @@ def post(self, request, pk):
if object_type == "virtualmachine":
object_type = "vm"
if object_type not in ("device", "vm"):
- return HttpResponse(f"Invalid object_type: {object_type!r}", status=400)
+ return HttpResponse(f"Invalid object_type: {escape(object_type)}", status=400)
target_model = VirtualMachine if object_type == "vm" else Device
self.required_object_permissions = {"POST": [("change", target_model)]}
@@ -618,7 +681,7 @@ def post(self, request, pk):
if object_type == "virtualmachine":
object_type = "vm"
if object_type not in ("device", "vm"):
- return HttpResponse(f"Invalid object_type: {object_type!r}", status=400)
+ return HttpResponse(f"Invalid object_type: {escape(object_type)}", status=400)
target_model = VirtualMachine if object_type == "vm" else Device
self.required_object_permissions = {"POST": [("change", target_model)]}
diff --git a/netbox_librenms_plugin/views/sync/devices.py b/netbox_librenms_plugin/views/sync/devices.py
index 71add20d48..b42de9220e 100644
--- a/netbox_librenms_plugin/views/sync/devices.py
+++ b/netbox_librenms_plugin/views/sync/devices.py
@@ -1,14 +1,25 @@
from dcim.models import Device
from django.contrib import messages
+from django.http import HttpResponse
from django.shortcuts import get_object_or_404, redirect
+from django.utils.html import escape
from django.views import View
from virtualization.models import VirtualMachine
from netbox_librenms_plugin.forms import AddToLIbreSNMPV1V2, AddToLIbreSNMPV3
-from netbox_librenms_plugin.views.mixins import LibreNMSAPIMixin, LibreNMSPermissionMixin
-
-
-class AddDeviceToLibreNMSView(LibreNMSPermissionMixin, LibreNMSAPIMixin, View):
+from netbox_librenms_plugin.views.mixins import (
+ LibreNMSAPIMixin,
+ LibreNMSPermissionMixin,
+ NetBoxObjectPermissionMixin,
+)
+
+
+class AddDeviceToLibreNMSView(
+ LibreNMSPermissionMixin,
+ NetBoxObjectPermissionMixin,
+ LibreNMSAPIMixin,
+ View,
+):
"""Add a NetBox device or VM to LibreNMS via the API."""
def get_form_class(self):
@@ -25,23 +36,44 @@ def get_object(self, object_id, object_type=None):
"""
Return the Device or VirtualMachine for the given ID.
- Uses object_type hint when provided to avoid PK collision ambiguity
- (Device and VirtualMachine share independent PK sequences).
+ Uses *object_type* to pick the correct model, preventing a false
+ match when a Device and VirtualMachine share the same PK.
+
+ Returns ``None`` for invalid *object_type* β callers must short-circuit
+ with HTTP 400 before using the result (a missing/unknown object_type is
+ a client error, not a missing-resource error).
"""
if object_type == "virtualmachine":
return get_object_or_404(VirtualMachine, pk=object_id)
- try:
- return Device.objects.get(pk=object_id)
- except Device.DoesNotExist:
- return get_object_or_404(VirtualMachine, pk=object_id)
+ if object_type == "device":
+ return get_object_or_404(Device, pk=object_id)
+ return None
def post(self, request, object_id):
"""Add a device to LibreNMS using the submitted SNMP form."""
- # Check write permission before adding device to LibreNMS
- if error := self.require_write_permission():
+ # Resolve the target object first so we can apply object-level perms
+ # against the correct model (Device vs VirtualMachine).
+ object_type = request.POST.get("object_type")
+ self.object = self.get_object(object_id, object_type=object_type)
+ if self.object is None:
+ # Match the convention used in views/sync/device_fields.py β return
+ # 400 (Bad Request) with an escaped echo of the offending value
+ # rather than raising 404, which would mislead clients into
+ # thinking the object itself is missing.
+ return HttpResponse(
+ f"Invalid object_type: {escape(str(object_type))}",
+ status=400,
+ )
+
+ # Plugin write perm + NetBox change perm on the resolved model. The
+ # mapping is set per-request because object_type can be either
+ # device or virtualmachine; static class-level declaration cannot
+ # express that branch.
+ target_model = VirtualMachine if object_type == "virtualmachine" else Device
+ self.required_object_permissions = {"POST": [("change", target_model)]}
+ if error := self.require_all_permissions("POST"):
return error
- self.object = self.get_object(object_id, request.POST.get("object_type"))
form_class = self.get_form_class()
snmp_version = request.POST.get("v1v2-snmp_version") or request.POST.get("v3-snmp_version")
diff --git a/netbox_librenms_plugin/views/sync/interfaces.py b/netbox_librenms_plugin/views/sync/interfaces.py
index 2259ff4892..4a4bee29c1 100644
--- a/netbox_librenms_plugin/views/sync/interfaces.py
+++ b/netbox_librenms_plugin/views/sync/interfaces.py
@@ -1,5 +1,7 @@
from urllib.parse import quote_plus
+import logging
+
from dcim.models import Device, Interface, MACAddress
from django.contrib import messages
from django.core.cache import cache
@@ -26,6 +28,8 @@
VlanAssignmentMixin,
)
+logger = logging.getLogger(__name__)
+
class SyncInterfacesView(
LibreNMSPermissionMixin, NetBoxObjectPermissionMixin, LibreNMSAPIMixin, VlanAssignmentMixin, CacheMixin, View
@@ -382,8 +386,9 @@ def post(self, request, object_type, object_id):
errors.append(f"Error deleting interface {interface_name or interface_id}: {str(exc)}")
continue
- except Exception as exc: # pragma: no cover
- return JsonResponse({"error": f"Transaction failed: {str(exc)}"}, status=500)
+ except Exception: # pragma: no cover
+ logger.exception("DeleteNetBoxInterfacesView transaction failed")
+ return JsonResponse({"error": "Transaction failed. Please check server logs."}, status=500)
response_data = {
"status": "success",
diff --git a/netbox_librenms_plugin/views/sync/modules.py b/netbox_librenms_plugin/views/sync/modules.py
new file mode 100644
index 0000000000..342b1b160c
--- /dev/null
+++ b/netbox_librenms_plugin/views/sync/modules.py
@@ -0,0 +1,1440 @@
+"""Sync action views for module/inventory installation from LibreNMS."""
+
+import re
+
+from django.contrib import messages
+from django.core.cache import cache
+from django.core.exceptions import ValidationError
+from django.db import IntegrityError, models, transaction
+from django.http import HttpResponse
+from django.shortcuts import get_object_or_404, redirect, render
+from django.urls import reverse
+from django.views import View
+
+from netbox_librenms_plugin.utils import get_librenms_sync_device, get_module_types_indexed
+from netbox_librenms_plugin.views.base.modules_view import _PLACEHOLDER_VALUES
+from netbox_librenms_plugin.views.mixins import (
+ CacheMixin,
+ LibreNMSAPIMixin,
+ LibreNMSPermissionMixin,
+ NetBoxObjectPermissionMixin,
+)
+
+
+def _modules_redirect_response(request, sync_url):
+ """Return a redirect response that works for both classic and HTMX form posts.
+
+ For HTMX requests (those carrying ``HX-Request: true``) we return an empty
+ 200 response with an ``HX-Redirect`` header so the browser performs a full
+ navigation that picks up Django messages and refreshes the modules table.
+ For non-HTMX requests we return a normal Django redirect.
+ """
+ target = f"{sync_url}?tab=modules#librenms-module-table"
+ if request.headers.get("HX-Request") == "true":
+ response = HttpResponse(status=204)
+ response["HX-Redirect"] = target
+ return response
+ return redirect(target)
+
+
+def _extract_inventory_list(cached_payload):
+ """Extract the inventory row list from a cached payload.
+
+ The cache stores ``{"inventory": [...], "librenms_id": ...}``; anything
+ else is treated as a cache miss to match BaseModuleTableView.get_context_data.
+ """
+ if isinstance(cached_payload, dict):
+ return cached_payload.get("inventory") or []
+ return []
+
+
+def _report_install_results(request, installed, skipped, failed):
+ """Emit Django messages summarising an install run."""
+ if installed:
+ messages.success(request, f"Installed {len(installed)} module(s): {', '.join(installed)}")
+ if skipped:
+ messages.info(request, f"Skipped {len(skipped)}: {'; '.join(skipped)}")
+ if failed:
+ messages.warning(request, f"Failed {len(failed)}: {'; '.join(failed)}")
+
+
+def _resolve_target_device(page_device, selected_device_id):
+ """Resolve and validate a target device from row-level VC selection."""
+ if not selected_device_id:
+ return page_device
+
+ try:
+ selected_device_id = int(selected_device_id)
+ except (TypeError, ValueError):
+ return page_device
+
+ if not getattr(page_device, "virtual_chassis", None):
+ return page_device
+
+ member = page_device.virtual_chassis.members.filter(pk=selected_device_id).first()
+ return member or page_device
+
+
+class _SerialConflictAmbiguous(Exception):
+ """Raised inside ReplaceModuleView's transaction when more than one module
+ holds the incoming serial β used to abort the atomic block and surface a
+ user-friendly error after the rollback."""
+
+ def __init__(self, serial):
+ super().__init__(serial)
+ self.serial = serial
+
+
+def _get_sync_device_for_inventory(device, server_key):
+ """Return the VC sync device used for module inventory cache keys."""
+ return get_librenms_sync_device(device, server_key=server_key) or device
+
+
+class InstallModuleView(LibreNMSPermissionMixin, NetBoxObjectPermissionMixin, CacheMixin, View):
+ """Install a NetBox Module into a ModuleBay from LibreNMS inventory data."""
+
+ def post(self, request, pk):
+ from dcim.models import Device, Module, ModuleBay, ModuleType
+
+ self.required_object_permissions = {"POST": [("add", Module)]}
+ if error := self.require_all_permissions("POST"):
+ return error
+
+ page_device = get_object_or_404(Device, pk=pk)
+ target_device = _resolve_target_device(page_device, request.POST.get("selected_device_id"))
+ serial = request.POST.get("serial", "").strip()
+ if serial.lower() in _PLACEHOLDER_VALUES:
+ serial = ""
+ sync_url = reverse("plugins:netbox_librenms_plugin:device_librenms_sync", kwargs={"pk": pk})
+
+ try:
+ module_bay_id = int(request.POST.get("module_bay_id"))
+ module_type_id = int(request.POST.get("module_type_id"))
+ except (TypeError, ValueError):
+ messages.error(request, "Missing or invalid module bay/module type ID.")
+ return redirect(f"{sync_url}?tab=modules#librenms-module-table")
+
+ get_object_or_404(ModuleBay, pk=module_bay_id, device=target_device) # verify bay belongs to selected device
+ module_type = get_object_or_404(ModuleType, pk=module_type_id)
+
+ try:
+ with transaction.atomic():
+ # Re-fetch bay under lock to prevent TOCTOU race with concurrent installs.
+ locked_bay = ModuleBay.objects.select_for_update().get(pk=module_bay_id)
+ if hasattr(locked_bay, "installed_module") and locked_bay.installed_module:
+ messages.warning(request, f"Module bay '{locked_bay.name}' already has a module installed.")
+ return redirect(f"{sync_url}?tab=modules#librenms-module-table")
+ module = Module(
+ device=target_device,
+ module_bay=locked_bay,
+ module_type=module_type,
+ serial=serial,
+ status="active",
+ )
+ module.full_clean()
+ module.save()
+
+ messages.success(
+ request, f"Installed {module_type.model} in {locked_bay.name} (serial: {serial or 'N/A'})."
+ )
+ except (ValidationError, IntegrityError) as e:
+ messages.error(request, f"Failed to install module: {e}")
+
+ return redirect(f"{sync_url}?tab=modules#librenms-module-table")
+
+
+class InstallBranchView(LibreNMSPermissionMixin, NetBoxObjectPermissionMixin, LibreNMSAPIMixin, CacheMixin, View):
+ """Install a module and all its installable descendants from LibreNMS inventory."""
+
+ def post(self, request, pk):
+ from dcim.models import Device, Module, ModuleBay, ModuleType
+
+ self.required_object_permissions = {"POST": [("add", Module)]}
+ if error := self.require_all_permissions("POST"):
+ return error
+
+ page_device = get_object_or_404(Device, pk=pk)
+ target_device = _resolve_target_device(page_device, request.POST.get("selected_device_id"))
+ parent_index = request.POST.get("parent_index")
+ server_key = request.POST.get("server_key") or self.librenms_api.server_key
+ sync_url = reverse("plugins:netbox_librenms_plugin:device_librenms_sync", kwargs={"pk": pk})
+
+ if not parent_index:
+ messages.error(request, "Missing parent inventory index.")
+ return redirect(f"{sync_url}?tab=modules#librenms-module-table")
+
+ try:
+ parent_index = int(parent_index)
+ except ValueError:
+ messages.error(request, "Invalid parent inventory index.")
+ return redirect(f"{sync_url}?tab=modules#librenms-module-table")
+
+ # Get cached inventory data
+ sync_device = _get_sync_device_for_inventory(target_device, server_key)
+ cached_payload = cache.get(self.get_cache_key(sync_device, "inventory", server_key=server_key))
+ cached_data = _extract_inventory_list(cached_payload)
+ if not cached_data:
+ messages.error(request, "No cached inventory data. Please refresh modules first.")
+ return redirect(f"{sync_url}?tab=modules#librenms-module-table")
+
+ # Load ignore rules so the branch respects the same filters shown in the table
+ from netbox_librenms_plugin.utils import get_enabled_ignore_rules
+
+ ignore_rules = get_enabled_ignore_rules()
+ device_serial = (getattr(target_device, "serial", None) or "").strip()
+
+ # Build index map and collect the branch to install
+ index_map = {idx: item for item in cached_data if (idx := item.get("entPhysicalIndex")) is not None}
+ branch_items = self._collect_branch(parent_index, cached_data, ignore_rules, device_serial, index_map)
+
+ if not branch_items:
+ messages.warning(request, "No installable items found in this branch.")
+ return redirect(f"{sync_url}?tab=modules#librenms-module-table")
+
+ # Load module types (with mappings)
+ module_types = get_module_types_indexed()
+
+ # Preload all ModuleBayMappings once to avoid N+1 per-item queries.
+ # Filter by device manufacturer so vendor-scoped mappings only apply to
+ # matching vendors and mismatched ones are skipped.
+ from netbox_librenms_plugin.utils import load_bay_mappings
+ from netbox_librenms_plugin.views.base.modules_view import BaseModuleTableView
+
+ exact_mappings, regex_mappings = load_bay_mappings()
+ mfr_id = getattr(getattr(target_device, "device_type", None), "manufacturer_id", None)
+ exact_mappings = BaseModuleTableView._filter_mappings_by_manufacturer(exact_mappings, mfr_id)
+ regex_mappings = BaseModuleTableView._filter_mappings_by_manufacturer(regex_mappings, mfr_id)
+
+ # Install top-down: each install may create new child bays
+ installed = []
+ skipped = []
+ failed = []
+
+ try:
+ with transaction.atomic():
+ for item in branch_items:
+ result = self._install_single(
+ target_device,
+ item,
+ index_map,
+ module_types,
+ ModuleBay,
+ ModuleType,
+ Module,
+ exact_mappings=exact_mappings,
+ regex_mappings=regex_mappings,
+ )
+ if result["status"] == "installed":
+ installed.append(result["name"])
+ elif result["status"] == "skipped":
+ skipped.append(f"{result['name']}: {result['reason']}")
+ else:
+ failed.append(f"{result['name']}: {result['reason']}")
+ except (ValidationError, IntegrityError) as e:
+ messages.error(request, f"Branch install failed: {e}")
+ return redirect(f"{sync_url}?tab=modules#librenms-module-table")
+
+ _report_install_results(request, installed, skipped, failed)
+ return redirect(f"{sync_url}?tab=modules#librenms-module-table")
+
+ def _collect_branch(self, parent_index, inventory_data, ignore_rules=None, device_serial="", index_map=None):
+ """
+ Collect all items in a branch depth-first, parent first.
+
+ Returns items in install order (parent before children).
+ Optionally filters items matching 'skip' ignore rules; 'transparent' items
+ are excluded from installation but their children are still collected.
+ """
+ items = []
+ parent = next((i for i in inventory_data if i.get("entPhysicalIndex") == parent_index), None)
+ if parent:
+ if ignore_rules:
+ from netbox_librenms_plugin.views.base.modules_view import _check_ignore_rules
+
+ ancestor = index_map.get(parent.get("entPhysicalContainedIn")) if index_map else None
+ action = _check_ignore_rules(parent, ancestor, ignore_rules, index_map, device_serial)
+ if action == "skip":
+ return []
+ if action == "transparent":
+ self._collect_children(
+ parent_index,
+ inventory_data,
+ items,
+ visited={parent_index},
+ ignore_rules=ignore_rules,
+ device_serial=device_serial,
+ index_map=index_map,
+ )
+ return items
+ model = (parent.get("entPhysicalModelName") or "").strip()
+ if model:
+ items.append(parent)
+ self._collect_children(
+ parent_index,
+ inventory_data,
+ items,
+ visited={parent_index},
+ ignore_rules=ignore_rules,
+ device_serial=device_serial,
+ index_map=index_map,
+ )
+ return items
+
+ def _collect_children(
+ self, parent_idx, inventory_data, items, visited=None, ignore_rules=None, device_serial="", index_map=None
+ ):
+ """Recursively collect children with models, depth-first.
+
+ When ignore_rules are provided, items matching a 'skip' rule (and their
+ subtree) are excluded. Items matching 'transparent' are not installed but
+ their children are still collected at the same depth.
+ """
+ if visited is None:
+ visited = set()
+ children = [i for i in inventory_data if i.get("entPhysicalContainedIn") == parent_idx]
+ for child in children:
+ child_idx = child.get("entPhysicalIndex")
+ if child_idx is None:
+ continue
+ if child_idx in visited:
+ continue
+ visited.add(child_idx)
+ # Apply ignore rules when provided
+ if ignore_rules:
+ from netbox_librenms_plugin.views.base.modules_view import _check_ignore_rules
+
+ parent_item = index_map.get(child.get("entPhysicalContainedIn")) if index_map else None
+ action = _check_ignore_rules(child, parent_item, ignore_rules, index_map, device_serial)
+ if action == "skip":
+ continue
+ if action == "transparent":
+ # Don't install this item but still collect its children
+ self._collect_children(
+ child_idx, inventory_data, items, visited, ignore_rules, device_serial, index_map
+ )
+ continue
+ model = (child.get("entPhysicalModelName") or "").strip()
+ if model:
+ items.append(child)
+ # Always recurse to find deeper items (containers may lack models)
+ self._collect_children(child_idx, inventory_data, items, visited, ignore_rules, device_serial, index_map)
+
+ @staticmethod
+ def _install_single(
+ device,
+ item,
+ index_map,
+ module_types,
+ ModuleBay,
+ ModuleType,
+ Module,
+ exact_mappings=None,
+ regex_mappings=None,
+ ):
+ """
+ Try to install a single inventory item.
+
+ Re-fetches module bays each time since parent installs create new ones.
+ Scopes bay lookup to the correct parent module to handle duplicate bay names.
+ """
+ from netbox_librenms_plugin.utils import resolve_module_type
+
+ model_name = (item.get("entPhysicalModelName") or "").strip()
+ serial = (item.get("entPhysicalSerialNum") or "").strip()
+ if serial.lower() in _PLACEHOLDER_VALUES:
+ serial = ""
+ name = item.get("entPhysicalName", "") or model_name
+
+ # Match module type (direct, then normalization fallback)
+ manufacturer = getattr(getattr(device, "device_type", None), "manufacturer", None)
+ matched_type = resolve_module_type(model_name, module_types, manufacturer=manufacturer)
+ if not matched_type:
+ return {"status": "skipped", "name": name, "reason": "no matching type"}
+
+ # Re-fetch module bays (parent install creates new child bays)
+ bays = ModuleBay.objects.filter(device=device).select_related("installed_module__module_type")
+
+ # Use preloaded mappings if provided, otherwise load from DB
+ if exact_mappings is None or regex_mappings is None:
+ from netbox_librenms_plugin.utils import load_bay_mappings
+
+ exact_mappings, regex_mappings = load_bay_mappings()
+
+ # Determine if this item belongs under an installed module
+ # by tracing its LibreNMS parent hierarchy to an installed item
+ parent_module_id = InstallBranchView._find_parent_module_id(
+ item, index_map, bays, exact_mappings, regex_mappings
+ )
+
+ if parent_module_id:
+ bay_dict = {bay.name: bay for bay in bays if bay.module_id == parent_module_id}
+ else:
+ bay_dict = {bay.name: bay for bay in bays if not bay.module_id}
+
+ # Match module bay using preloaded mapping data
+ matched_bay = InstallBranchView._match_bay(item, index_map, bay_dict, exact_mappings, regex_mappings)
+ if not matched_bay:
+ return {"status": "skipped", "name": name, "reason": "no matching bay"}
+
+ # Install (lock bay to prevent concurrent installs)
+ try:
+ with transaction.atomic(): # savepoint: failure here won't abort parent tx
+ locked_bay = (
+ ModuleBay.objects.select_for_update(of=("self",))
+ .select_related("installed_module")
+ .get(pk=matched_bay.pk)
+ )
+ if hasattr(locked_bay, "installed_module") and locked_bay.installed_module:
+ return {"status": "skipped", "name": name, "reason": "bay already occupied"}
+
+ module = Module(
+ device=device,
+ module_bay=locked_bay,
+ module_type=matched_type,
+ serial=serial,
+ status="active",
+ )
+ module.full_clean()
+ module.save()
+ except (ValidationError, IntegrityError) as e:
+ error_msg = str(e)
+ if "dcim_interface_unique_device_name" in error_msg:
+ error_msg = (
+ "duplicate interface name β this module type's interface template "
+ "uses the '{module}' token which resolves to the same name for all siblings. "
+ "An interface naming plugin with a rewrite rule for this module type can fix this."
+ )
+ return {"status": "failed", "name": name, "reason": error_msg}
+
+ return {"status": "installed", "name": f"{matched_type.model} β {matched_bay.name}"}
+
+ @staticmethod
+ def _find_parent_module_id(item, index_map, device_bays, exact_mappings, regex_mappings):
+ """
+ Find the NetBox module ID for the installed parent of this inventory item.
+
+ Walks up the LibreNMS hierarchy to find an ancestor whose name matches
+ an installed module bay on the device.
+
+ Args:
+ item: The inventory item dict.
+ index_map: Dict mapping entPhysicalIndex to inventory item.
+ device_bays: Pre-fetched queryset/list of ModuleBay objects for the device.
+ exact_mappings: Pre-filtered list of exact ModuleBayMapping objects.
+ regex_mappings: Pre-filtered list of regex ModuleBayMapping objects.
+ """
+ current = item
+ # Build bay name β list of bays for duplicate-name disambiguation
+ bay_by_name: dict = {}
+ for bay in device_bays:
+ bay_by_name.setdefault(bay.name, []).append(bay)
+
+ # Build exact_mapping index: prefer class-specific over class-empty
+ exact_mapping_by_key: dict = {}
+ for m in exact_mappings:
+ key = (m.librenms_name, m.librenms_class)
+ if key not in exact_mapping_by_key:
+ exact_mapping_by_key[key] = m
+
+ visited = set()
+ while True:
+ parent_idx = current.get("entPhysicalContainedIn", 0)
+ if not parent_idx or parent_idx not in index_map:
+ return None
+ if parent_idx in visited:
+ return None
+ visited.add(parent_idx)
+ parent = index_map[parent_idx]
+ parent_name = parent.get("entPhysicalName", "")
+ parent_descr = parent.get("entPhysicalDescr", "")
+ parent_class = parent.get("entPhysicalClass", "")
+
+ # Check if this parent matches an installed module bay on the device
+ for bay in device_bays:
+ if hasattr(bay, "installed_module") and bay.installed_module:
+ if bay.name == parent_name or (parent_descr and bay.name == parent_descr):
+ return bay.installed_module.pk
+
+ # Also check ModuleBayMapping for indirect matches (exact then regex)
+ for name in [parent_name, parent_descr]:
+ if not name:
+ continue
+ # Exact-name mapping: prefer class-specific, fall back to class-empty
+ mapping = exact_mapping_by_key.get((name, parent_class))
+ if not mapping:
+ mapping = exact_mapping_by_key.get((name, ""))
+ if mapping:
+ candidates = bay_by_name.get(mapping.netbox_bay_name, [])
+ if len(candidates) == 1:
+ bay = candidates[0]
+ else:
+ occupied = [b for b in candidates if hasattr(b, "installed_module") and b.installed_module]
+ bay = occupied[0] if len(occupied) == 1 else None
+ if bay and hasattr(bay, "installed_module") and bay.installed_module:
+ return bay.installed_module.pk
+ # Regex mapping: class-specific first, then empty-class fallback
+ # (concatenate, don't use ``or``, so fallback is tried even when
+ # class-specific rules exist but none match β mirrors base view)
+ class_matches = [rm for rm in regex_mappings if rm.librenms_class == parent_class]
+ fallback_matches = [rm for rm in regex_mappings if rm.librenms_class == ""]
+ for rm in class_matches + fallback_matches:
+ compiled = rm._compiled_pattern
+ if compiled is None:
+ continue
+ try:
+ match = compiled.fullmatch(name)
+ except re.error:
+ continue
+ if not match:
+ continue
+ try:
+ bay_name = match.expand(rm.netbox_bay_name)
+ except (re.error, IndexError):
+ continue
+ candidates = bay_by_name.get(bay_name, [])
+ if len(candidates) == 1:
+ bay = candidates[0]
+ else:
+ occupied = [b for b in candidates if hasattr(b, "installed_module") and b.installed_module]
+ bay = occupied[0] if len(occupied) == 1 else None
+ if bay and hasattr(bay, "installed_module") and bay.installed_module:
+ return bay.installed_module.pk
+
+ current = parent
+
+ @staticmethod
+ def _match_bay(item, index_map, module_bays, exact_mappings, regex_mappings):
+ """Match an inventory item to a module bay (same logic as BaseModuleTableView)."""
+ from netbox_librenms_plugin.views.base.modules_view import BaseModuleTableView
+
+ # Resolve parent name by walking up the containment hierarchy
+ contained_in = item.get("entPhysicalContainedIn", 0)
+ parent_name = None
+ if contained_in:
+ visited_anc = set()
+ current_idx = contained_in
+ while current_idx and current_idx not in visited_anc:
+ visited_anc.add(current_idx)
+ ancestor = index_map.get(current_idx)
+ if not ancestor:
+ break
+ ancestor_name = ancestor.get("entPhysicalName", "")
+ if ancestor_name:
+ parent_name = ancestor_name
+ break
+ current_idx = ancestor.get("entPhysicalContainedIn", 0)
+
+ item_name = item.get("entPhysicalName", "")
+ item_descr = item.get("entPhysicalDescr", "")
+ phys_class = item.get("entPhysicalClass", "")
+
+ # Build candidate names: parent, item name, item description
+ candidate_names = [n for n in [parent_name, item_name, item_descr] if n]
+
+ # Check mapping for each candidate (exact match)
+ for name in candidate_names:
+ bay = BaseModuleTableView._lookup_exact_bay_mapping(name, phys_class, module_bays, exact_mappings)
+ if bay:
+ return bay
+
+ # Regex pattern matching using preloaded list
+ for name in candidate_names:
+ bay = BaseModuleTableView._lookup_regex_bay_mapping(name, phys_class, module_bays, regex_mappings)
+ if bay:
+ return bay
+
+ # Fallback: exact match on candidate names against bay dict
+ for name in candidate_names:
+ if name in module_bays:
+ return module_bays[name]
+
+ # Positional fallback for items inside converters
+ return BaseModuleTableView._match_bay_by_position(item, index_map, module_bays)
+
+
+class InstallSelectedView(LibreNMSPermissionMixin, NetBoxObjectPermissionMixin, LibreNMSAPIMixin, CacheMixin, View):
+ """
+ Install a user-selected set of inventory items by their entPhysicalIndex values.
+
+ Reuses InstallBranchView._install_single for each selected item so every item
+ goes through the same type/bay/serial resolution pipeline as a branch install.
+ Only items where a matching bay *and* module type are found will be installed;
+ items with no bay or no type are silently skipped (same behaviour as branch).
+ """
+
+ def post(self, request, pk):
+ from dcim.models import Device, Module, ModuleBay, ModuleType
+
+ self.required_object_permissions = {"POST": [("add", Module)]}
+ if error := self.require_all_permissions("POST"):
+ return error
+
+ page_device = get_object_or_404(Device, pk=pk)
+ server_key = request.POST.get("server_key") or self.librenms_api.server_key
+ sync_url = reverse("plugins:netbox_librenms_plugin:device_librenms_sync", kwargs={"pk": pk})
+
+ selected_indices = request.POST.getlist("select")
+ if not selected_indices:
+ messages.warning(request, "No modules selected.")
+ return redirect(f"{sync_url}?tab=modules#librenms-module-table")
+
+ sync_device = _get_sync_device_for_inventory(page_device, server_key)
+ cached_payload = cache.get(self.get_cache_key(sync_device, "inventory", server_key=server_key))
+ cached_data = _extract_inventory_list(cached_payload)
+ if not cached_data:
+ messages.error(request, "No cached inventory data. Please refresh modules first.")
+ return redirect(f"{sync_url}?tab=modules#librenms-module-table")
+
+ try:
+ # Use dict.fromkeys to preserve order while deduplicating
+ selected_list = list(dict.fromkeys(int(i) for i in selected_indices))
+ except ValueError:
+ messages.error(request, "Invalid selection.")
+ return redirect(f"{sync_url}?tab=modules#librenms-module-table")
+
+ index_map = {idx: item for item in cached_data if (idx := item.get("entPhysicalIndex")) is not None}
+ items = [index_map[idx] for idx in selected_list if idx in index_map]
+
+ if not items:
+ messages.warning(request, "None of the selected indices matched cached inventory.")
+ return redirect(f"{sync_url}?tab=modules#librenms-module-table")
+
+ # Load ignore rules once; they're evaluated per-row inside the install
+ # loop using the *resolved* target device serial, since VC rows may
+ # switch to a different member via device_selection_
.
+ from netbox_librenms_plugin.utils import get_enabled_ignore_rules
+ from netbox_librenms_plugin.views.base.modules_view import _check_ignore_rules
+
+ ignore_rules = get_enabled_ignore_rules()
+
+ # Preload all ModuleBayMappings once to avoid N+1 per-item queries.
+ # Manufacturer-scoping happens per-iteration since target_device may
+ # differ across rows (VC members can have different device types).
+ from netbox_librenms_plugin.utils import load_bay_mappings
+ from netbox_librenms_plugin.views.base.modules_view import BaseModuleTableView
+
+ module_types = get_module_types_indexed()
+ all_exact, all_regex = load_bay_mappings()
+
+ installed, skipped, failed = [], [], []
+
+ try:
+ with transaction.atomic():
+ for item in items:
+ ent_index = item.get("entPhysicalIndex")
+ selected_device_id = request.POST.get(f"device_selection_{ent_index}")
+ target_device = _resolve_target_device(page_device, selected_device_id)
+ if ignore_rules:
+ target_serial = (getattr(target_device, "serial", None) or "").strip()
+ rule_action = _check_ignore_rules(
+ item,
+ index_map.get(item.get("entPhysicalContainedIn")),
+ ignore_rules,
+ index_map,
+ target_serial,
+ )
+ if rule_action in {"skip", "transparent"}:
+ skipped.append(f"{item.get('entPhysicalName', '?')}: matched ignore rule")
+ continue
+ mfr_id = getattr(getattr(target_device, "device_type", None), "manufacturer_id", None)
+ exact_mappings = BaseModuleTableView._filter_mappings_by_manufacturer(all_exact, mfr_id)
+ regex_mappings = BaseModuleTableView._filter_mappings_by_manufacturer(all_regex, mfr_id)
+ result = InstallBranchView._install_single(
+ target_device,
+ item,
+ index_map,
+ module_types,
+ ModuleBay,
+ ModuleType,
+ Module,
+ exact_mappings=exact_mappings,
+ regex_mappings=regex_mappings,
+ )
+ if result["status"] == "installed":
+ installed.append(result["name"])
+ elif result["status"] == "skipped":
+ skipped.append(f"{result['name']}: {result['reason']}")
+ else:
+ failed.append(f"{result['name']}: {result['reason']}")
+ except (ValidationError, IntegrityError) as e:
+ messages.error(request, f"Install failed: {e}")
+ return _modules_redirect_response(request, sync_url)
+
+ _report_install_results(request, installed, skipped, failed)
+ return _modules_redirect_response(request, sync_url)
+
+
+class UpdateModuleSerialView(LibreNMSPermissionMixin, NetBoxObjectPermissionMixin, View):
+ """Update the serial number of an already-installed module from LibreNMS inventory data."""
+
+ def post(self, request, pk):
+ from dcim.models import Device, Module
+
+ self.required_object_permissions = {"POST": [("change", Module)]}
+ if error := self.require_all_permissions("POST"):
+ return error
+
+ page_device = get_object_or_404(Device, pk=pk)
+ target_device = _resolve_target_device(page_device, request.POST.get("selected_device_id"))
+ serial = request.POST.get("serial", "").strip()
+ if serial.lower() in _PLACEHOLDER_VALUES:
+ serial = ""
+ sync_url = reverse("plugins:netbox_librenms_plugin:device_librenms_sync", kwargs={"pk": pk})
+
+ try:
+ module_id = int(request.POST.get("module_id"))
+ except (TypeError, ValueError):
+ messages.error(request, "Missing or invalid module ID.")
+ return _modules_redirect_response(request, sync_url)
+
+ try:
+ with transaction.atomic():
+ module = (
+ Module.objects.select_for_update()
+ .select_related("module_type", "module_bay")
+ .filter(pk=module_id, device=target_device)
+ .first()
+ )
+ if not module:
+ messages.error(request, "Module no longer exists.")
+ return _modules_redirect_response(request, sync_url)
+ module.serial = serial
+ module.full_clean()
+ module.save()
+ messages.success(
+ request,
+ f"Updated serial for {module.module_type.model} in {module.module_bay.name} to '{serial}'.",
+ )
+ except (ValidationError, IntegrityError) as e:
+ messages.error(request, f"Failed to update serial: {e}")
+
+ return _modules_redirect_response(request, sync_url)
+
+
+class ModuleMismatchPreviewView(
+ LibreNMSPermissionMixin, NetBoxObjectPermissionMixin, LibreNMSAPIMixin, CacheMixin, View
+):
+ """
+ Return the modal body HTML fragment for the module replace/move dialog.
+
+ Loads the installed module and the corresponding LibreNMS inventory item from
+ cache, detects type/serial mismatch and serial conflicts, then renders the
+ comparison template so the user can choose between Replace, Move, or
+ Update Serial Only.
+ """
+
+ def get(self, request, pk):
+ from dcim.models import Device, Module
+
+ self.required_object_permissions = {"GET": [("view", Device), ("view", Module)]}
+ if error := self.require_object_permissions("GET"):
+ return error
+
+ page_device = get_object_or_404(Device, pk=pk)
+ target_device = _resolve_target_device(page_device, request.GET.get("selected_device_id"))
+ server_key = request.GET.get("server_key") or self.librenms_api.server_key
+
+ try:
+ module_id = int(request.GET.get("module_id"))
+ ent_index_int = int(request.GET.get("ent_index"))
+ except (TypeError, ValueError):
+ return HttpResponse("Missing or invalid module_id/ent_index.", status=400)
+
+ installed_module = get_object_or_404(
+ Module.objects.select_related("module_type", "module_bay", "device"),
+ pk=module_id,
+ device=target_device,
+ )
+
+ sync_device = _get_sync_device_for_inventory(target_device, server_key)
+ cached_payload = cache.get(self.get_cache_key(sync_device, "inventory", server_key=server_key))
+ cached_data = _extract_inventory_list(cached_payload)
+ if not cached_data:
+ return HttpResponse("No cached inventory data. Please refresh modules first.", status=400)
+
+ librenms_item = next(
+ (item for item in cached_data if item.get("entPhysicalIndex") == ent_index_int),
+ None,
+ )
+ if not librenms_item:
+ return HttpResponse("Inventory item not found in cache.", status=400)
+
+ librenms_model = (librenms_item.get("entPhysicalModelName") or "").strip() or "-"
+ librenms_serial = (librenms_item.get("entPhysicalSerialNum") or "").strip()
+ if librenms_serial.lower() in _PLACEHOLDER_VALUES:
+ librenms_serial = ""
+
+ # Detect type mismatch
+ from netbox_librenms_plugin.utils import resolve_module_type
+
+ module_types = get_module_types_indexed()
+ manufacturer = getattr(getattr(target_device, "device_type", None), "manufacturer", None)
+ matched_type = resolve_module_type(
+ librenms_model if librenms_model != "-" else "", module_types, manufacturer=manufacturer
+ )
+
+ type_mismatch = matched_type is not None and installed_module.module_type_id != matched_type.pk
+ installed_serial = (installed_module.serial or "").strip()
+ if installed_serial.lower() in _PLACEHOLDER_VALUES:
+ installed_serial = ""
+ serial_mismatch = bool(
+ not type_mismatch and librenms_serial != installed_serial and (librenms_serial or installed_serial)
+ )
+
+ # Check whether the LibreNMS serial already exists at a different location
+ serial_conflict = None
+ serial_conflict_ambiguous = False
+ if librenms_serial:
+ conflict_qs = (
+ Module.objects.filter(serial=librenms_serial)
+ .exclude(pk=installed_module.pk)
+ .select_related("module_type", "module_bay", "device")
+ )
+ conflict_count = conflict_qs.count()
+ if conflict_count == 1:
+ serial_conflict = conflict_qs.first()
+ elif conflict_count > 1:
+ serial_conflict_ambiguous = True
+
+ return render(
+ request,
+ "netbox_librenms_plugin/htmx/module_mismatch_modal.html",
+ {
+ "device_pk": pk,
+ "installed_module": installed_module,
+ "bay_name": installed_module.module_bay.name,
+ "target_bay_id": installed_module.module_bay_id,
+ "installed_serial": installed_serial,
+ "librenms_model": librenms_model,
+ "librenms_serial": librenms_serial,
+ "type_mismatch": type_mismatch,
+ "serial_mismatch": serial_mismatch,
+ "serial_conflict": serial_conflict,
+ "serial_conflict_ambiguous": serial_conflict_ambiguous,
+ "ent_index": ent_index_int,
+ "server_key": server_key or "",
+ "selected_device_id": target_device.pk,
+ },
+ )
+
+
+class ReplaceModuleView(LibreNMSPermissionMixin, LibreNMSAPIMixin, NetBoxObjectPermissionMixin, CacheMixin, View):
+ """
+ Replace the installed module in a bay with fresh data from LibreNMS inventory.
+
+ Deletes the currently installed module (and optionally removes a conflicting
+ module with the same serial from another location), then installs a new module
+ from cached LibreNMS inventory data.
+ """
+
+ def post(self, request, pk):
+ from dcim.models import Device, Module, ModuleBay, ModuleType # noqa: F401
+
+ self.required_object_permissions = {"POST": [("add", Module), ("change", Module), ("delete", Module)]}
+ if error := self.require_all_permissions("POST"):
+ return error
+
+ page_device = get_object_or_404(Device, pk=pk)
+ target_device = _resolve_target_device(page_device, request.POST.get("selected_device_id"))
+ server_key = request.POST.get("server_key") or self.librenms_api.server_key
+ sync_url = reverse("plugins:netbox_librenms_plugin:device_librenms_sync", kwargs={"pk": pk})
+
+ try:
+ module_id = int(request.POST.get("module_id"))
+ ent_index_int = int(request.POST.get("ent_index"))
+ except (TypeError, ValueError):
+ messages.error(request, "Missing or invalid module_id/ent_index.")
+ return _modules_redirect_response(request, sync_url)
+
+ installed_module = get_object_or_404(
+ Module.objects.select_related("module_type", "module_bay"),
+ pk=module_id,
+ device=target_device,
+ )
+
+ sync_device = _get_sync_device_for_inventory(target_device, server_key)
+ cached_payload = cache.get(self.get_cache_key(sync_device, "inventory", server_key=server_key))
+ cached_data = _extract_inventory_list(cached_payload)
+ if not cached_data:
+ messages.error(request, "No cached inventory data. Please refresh modules first.")
+ return _modules_redirect_response(request, sync_url)
+
+ librenms_item = next(
+ (item for item in cached_data if item.get("entPhysicalIndex") == ent_index_int),
+ None,
+ )
+ if not librenms_item:
+ messages.error(request, "Inventory item not found in cache.")
+ return _modules_redirect_response(request, sync_url)
+
+ model_name = (librenms_item.get("entPhysicalModelName") or "").strip()
+ serial = (librenms_item.get("entPhysicalSerialNum") or "").strip()
+ if serial.lower() in _PLACEHOLDER_VALUES:
+ serial = ""
+
+ module_types = get_module_types_indexed()
+ from netbox_librenms_plugin.utils import resolve_module_type
+
+ manufacturer = getattr(getattr(target_device, "device_type", None), "manufacturer", None)
+ matched_type = resolve_module_type(model_name, module_types, manufacturer=manufacturer)
+
+ if not matched_type:
+ messages.error(request, f"No matching module type found for '{model_name}'.")
+ return _modules_redirect_response(request, sync_url)
+
+ try:
+ conflict_removed_msg = None
+ with transaction.atomic():
+ # Re-fetch with row lock to prevent concurrent modifications
+ installed_module = (
+ Module.objects.select_for_update()
+ .filter(pk=module_id, device=target_device)
+ .select_related("module_type", "module_bay")
+ .first()
+ )
+ if not installed_module:
+ messages.error(request, "Module no longer exists.")
+ return _modules_redirect_response(request, sync_url)
+
+ # Read bay/type from locked row to avoid stale snapshot
+ target_bay = installed_module.module_bay
+ old_type_name = installed_module.module_type.model
+ old_bay_name = target_bay.name
+
+ # Re-derive any serial conflict from the database INSIDE the locked
+ # transaction (and lock those rows too) β checking before the lock
+ # opens a TOCTOU window where a concurrent request could change a
+ # module's serial and we'd then delete a row that no longer
+ # conflicts. Re-querying under select_for_update() guarantees the
+ # set we delete from is the same set we validated.
+ conflict_module = None
+ if serial:
+ conflict_qs = (
+ Module.objects.select_for_update()
+ .filter(serial=serial)
+ .exclude(pk=installed_module.pk)
+ .select_related("module_type", "module_bay", "device")
+ )
+ locked_conflicts = list(conflict_qs)
+ if len(locked_conflicts) > 1:
+ # Roll back and surface a clear error β we don't want to
+ # guess which of N conflicts to remove.
+ raise _SerialConflictAmbiguous(serial)
+ if len(locked_conflicts) == 1:
+ conflict_module = locked_conflicts[0]
+
+ # Remove the serial-conflicting module from its current location.
+ if conflict_module:
+ c_model = conflict_module.module_type.model
+ c_bay = conflict_module.module_bay.name
+ c_device = conflict_module.device.name
+ conflict_module.delete()
+ conflict_removed_msg = f"Removed {c_model} from {c_device}/{c_bay}."
+
+ # Delete the currently installed module in the target bay
+ installed_module.delete()
+
+ # Install fresh module from LibreNMS data
+ new_module = Module(
+ device=target_device,
+ module_bay=target_bay,
+ module_type=matched_type,
+ serial=serial,
+ status="active",
+ )
+ new_module.full_clean()
+ new_module.save()
+
+ if conflict_removed_msg:
+ messages.info(request, conflict_removed_msg)
+ messages.success(
+ request,
+ f"Replaced {old_type_name} with {matched_type.model} in {old_bay_name}"
+ + (f" (serial: {serial})" if serial else "")
+ + ".",
+ )
+ except _SerialConflictAmbiguous as exc:
+ messages.error(
+ request,
+ f"Serial '{exc.serial}' is assigned to multiple modules; cannot determine which to remove. "
+ "Please resolve the conflict manually.",
+ )
+ except (ValidationError, IntegrityError) as e:
+ messages.error(request, f"Replace failed: {e}")
+
+ return _modules_redirect_response(request, sync_url)
+
+
+class MoveModuleView(LibreNMSPermissionMixin, NetBoxObjectPermissionMixin, View):
+ """
+ Move an existing module from its current location to a target bay.
+
+ Handles the case where a module (identified by serial) has been physically
+ moved from one slot to another β possibly on a different device. Updates
+ the module_bay (and device when moving cross-device) rather than deleting
+ and recreating, preserving the module's history.
+ """
+
+ def post(self, request, pk):
+ from dcim.models import Device, Module, ModuleBay
+
+ self.required_object_permissions = {"POST": [("change", Module), ("delete", Module)]}
+ if error := self.require_all_permissions("POST"):
+ return error
+
+ page_device = get_object_or_404(Device, pk=pk)
+ target_device = _resolve_target_device(page_device, request.POST.get("selected_device_id"))
+ sync_url = reverse("plugins:netbox_librenms_plugin:device_librenms_sync", kwargs={"pk": pk})
+
+ try:
+ conflict_module_id = int(request.POST.get("conflict_module_id"))
+ target_bay_id = int(request.POST.get("target_bay_id"))
+ except (TypeError, ValueError):
+ messages.error(request, "Missing or invalid conflict_module_id/target_bay_id.")
+ return _modules_redirect_response(request, sync_url)
+
+ # Optional: current occupant of target bay
+ raw_module_id = request.POST.get("module_id")
+ try:
+ module_id = int(raw_module_id) if raw_module_id else None
+ except (TypeError, ValueError):
+ module_id = None
+
+ get_object_or_404(ModuleBay, pk=target_bay_id, device=target_device)
+
+ try:
+ occupant_removed_msg = None
+ with transaction.atomic():
+ # Lock target bay to prevent concurrent modifications
+ target_bay = ModuleBay.objects.select_for_update().get(pk=target_bay_id, device=target_device)
+
+ # Re-fetch with row lock to prevent concurrent modifications
+ conflict_module = (
+ Module.objects.select_for_update()
+ .filter(pk=conflict_module_id)
+ .select_related("module_type", "module_bay", "device")
+ .first()
+ )
+ if not conflict_module:
+ messages.error(request, "Module no longer exists.")
+ return _modules_redirect_response(request, sync_url)
+
+ # Remove whatever is currently in the target bay (if provided and different)
+ if module_id:
+ occupant = (
+ Module.objects.select_for_update()
+ .filter(pk=module_id, device=target_device, module_bay=target_bay)
+ .first()
+ )
+ if occupant and occupant.pk != conflict_module.pk:
+ occupant_removed_msg = f"Removed {occupant.module_type.model} from {target_bay.name}."
+ occupant.delete()
+
+ # Move the conflict module to the target bay
+ from_bay = conflict_module.module_bay.name
+ from_device = conflict_module.device.name
+ conflict_module.module_bay = target_bay
+ conflict_module.device = target_device
+ conflict_module.full_clean()
+ conflict_module.save()
+
+ if occupant_removed_msg:
+ messages.info(request, occupant_removed_msg)
+ moved_msg = f"Moved {conflict_module.module_type.model}"
+ if from_device != target_device.name:
+ moved_msg += f" from {from_device}"
+ moved_msg += f"/{from_bay} to {target_bay.name}."
+ messages.success(request, moved_msg)
+ except (ValidationError, IntegrityError) as e:
+ messages.error(request, f"Move failed: {e}")
+
+ return _modules_redirect_response(request, sync_url)
+
+
+class AddBayTemplateView(LibreNMSPermissionMixin, NetBoxObjectPermissionMixin, View):
+ """
+ Create a missing ModuleBayTemplate on a Device Type or Module Type so the
+ user can install a sub-component without leaving the modules sync tab.
+
+ GET renders a pre-filled modal fragment that targets ``#htmx-modal-content``;
+ POST creates the bay template and redirects back to the modules tab.
+ """
+
+ TARGET_KINDS = ("device_type", "module_type")
+
+ def _resolve_target(self, target_kind, target_pk):
+ """Load the DeviceType or ModuleType the new bay template will attach to."""
+ from dcim.models import DeviceType, ModuleType
+
+ if target_kind == "device_type":
+ return get_object_or_404(DeviceType, pk=target_pk)
+ if target_kind == "module_type":
+ return get_object_or_404(ModuleType, pk=target_pk)
+ return None
+
+ @staticmethod
+ def _device_manufacturer(device):
+ """Return the device's Manufacturer (or None) for vendor-scoped mapping defaults."""
+ device_type = getattr(device, "device_type", None)
+ return getattr(device_type, "manufacturer", None) if device_type else None
+
+ @staticmethod
+ def _instantiate_template_on_existing(bay_template, target_kind, target):
+ """
+ Materialise the just-saved ``ModuleBayTemplate`` onto every existing
+ device/module of ``target`` so the resolver can match it immediately.
+
+ NetBox auto-creates bays from templates only when a Device/Module is
+ first created β a template added later is invisible to existing
+ instances until manually instantiated. ``target_kind`` selects the
+ scope: device-type templates are instantiated on every Device of that
+ type; module-type templates are instantiated on every installed Module
+ of that type. Pre-existing bays with the resolved name (under the same
+ device/module scope) are skipped so re-adding a template after a
+ partial manual fix is safe.
+ """
+ from dcim.models import Device, Module, ModuleBay
+
+ instantiated = 0
+ if target_kind == "device_type":
+ for device in Device.objects.filter(device_type=target):
+ bay = bay_template.instantiate(device=device)
+ if ModuleBay.objects.filter(device=device, module__isnull=True, name=bay.name).exists():
+ continue
+ bay.full_clean()
+ bay.save()
+ instantiated += 1
+ elif target_kind == "module_type":
+ for module in Module.objects.filter(module_type=target).select_related("device"):
+ bay = bay_template.instantiate(device=module.device, module=module)
+ if ModuleBay.objects.filter(device=module.device, module=module, name=bay.name).exists():
+ continue
+ bay.full_clean()
+ bay.save()
+ instantiated += 1
+ return instantiated
+
+ @staticmethod
+ def _derive_mapping_pattern(librenms_name, netbox_name):
+ """
+ Derive a regex ``ModuleBayMapping`` rule that maps ``librenms_name``
+ to ``netbox_name`` and naturally covers every sibling bay sharing
+ the same LibreNMS-side literal skeleton.
+
+ Both names are tokenised into alternating literal and digit-run
+ segments. Each *distinct* digit value on the LibreNMS side becomes
+ a numbered capture group; subsequent occurrences of the same value
+ emit back-references, so ``"0/FT0"`` produces ``r"^(\\d+)/FT\\1$"``
+ β matching only when both fan-tray digits agree. The NetBox
+ replacement preserves the operator-chosen literals verbatim and
+ back-references the LibreNMS group whose value matches each digit
+ run on the NetBox side.
+
+ Returns ``None`` when:
+ * either name is empty,
+ * the LibreNMS name has no digit run at all,
+ * the NetBox name contains a digit value that does not appear in
+ the LibreNMS name (we'd be inventing a value we can't extract),
+ * the resulting pattern fails to compile or does not round-trip
+ (``compile.fullmatch`` + ``compile.sub`` must reproduce the
+ exact NetBox name).
+
+ Examples::
+
+ ('Sfm 1', 'SFM 1') -> r'^Sfm (\\d+)$' β r'SFM \\1'
+ ('0/FT0', 'Fan Tray 0') -> r'^(\\d+)/FT\\1$' β r'Fan Tray \\1'
+ ('TenGigE0/0/0/0', same) -> r'^TenGigE(\\d+)/\\1/\\1/\\1$'
+ β r'TenGigE\\1/\\1/\\1/\\1'
+ ('0/FT0', 'Fan Tray 1') -> None (libre has no '1')
+ ('Slot A', 'Slot A') -> None (no digit run)
+ """
+ if not librenms_name or not netbox_name:
+ return None
+ token_re = re.compile(r"(\d+|\D+)")
+ libre_tokens = token_re.findall(librenms_name)
+ nb_tokens = token_re.findall(netbox_name)
+
+ digit_groups = {}
+ pattern_parts = ["^"]
+ for tok in libre_tokens:
+ if tok.isdigit():
+ if tok in digit_groups:
+ pattern_parts.append(rf"\{digit_groups[tok]}")
+ else:
+ idx = len(digit_groups) + 1
+ digit_groups[tok] = idx
+ pattern_parts.append(r"(\d+)")
+ else:
+ pattern_parts.append(re.escape(tok))
+ pattern_parts.append("$")
+ if not digit_groups:
+ return None
+
+ replacement_parts = []
+ for tok in nb_tokens:
+ if tok.isdigit():
+ if tok not in digit_groups:
+ return None
+ replacement_parts.append(rf"\{digit_groups[tok]}")
+ else:
+ replacement_parts.append(tok.replace("\\", r"\\"))
+
+ librenms_pattern = "".join(pattern_parts)
+ netbox_replacement = "".join(replacement_parts)
+ try:
+ compiled = re.compile(librenms_pattern)
+ except re.error:
+ return None
+ if not compiled.fullmatch(librenms_name):
+ return None
+ try:
+ if compiled.sub(netbox_replacement, librenms_name) != netbox_name:
+ return None
+ except re.error:
+ return None
+ return {
+ "kind": "regex",
+ "librenms_pattern": librenms_pattern,
+ "netbox_replacement": netbox_replacement,
+ "digit_count": len(digit_groups),
+ }
+
+ @staticmethod
+ def _existing_regex_mapping_covers(librenms_name, librenms_class, manufacturer):
+ """
+ True when an existing regex ModuleBayMapping already matches
+ ``librenms_name`` for the given manufacturer / global scope.
+ Iterates regex rows in Python β the row count is small (one per
+ bay-family) so this is cheap, and Postgres can't compare its
+ re-flavoured patterns server-side anyway.
+ """
+ from netbox_librenms_plugin.models import ModuleBayMapping
+
+ if not librenms_name:
+ return False
+ qs = ModuleBayMapping.objects.filter(
+ librenms_class=librenms_class or "",
+ is_regex=True,
+ )
+ if manufacturer is not None:
+ qs = qs.filter(models.Q(manufacturer=manufacturer) | models.Q(manufacturer__isnull=True))
+ else:
+ qs = qs.filter(manufacturer__isnull=True)
+ for mapping in qs.only("librenms_name"):
+ try:
+ if re.compile(mapping.librenms_name).fullmatch(librenms_name):
+ return True
+ except re.error:
+ continue
+ return False
+
+ @staticmethod
+ def _existing_bay_mapping(librenms_name, librenms_class, manufacturer):
+ """
+ True when a ModuleBayMapping already covers (librenms_name, librenms_class)
+ for the given manufacturer (vendor-scoped) or globally (manufacturer is null).
+
+ Only checks exact mappings β regex mappings are intentionally ignored
+ because the per-row suggestion is for one specific name and we don't
+ want to second-guess broader patterns the operator already wrote.
+ """
+ from netbox_librenms_plugin.models import ModuleBayMapping
+
+ if not librenms_name:
+ return False
+ qs = ModuleBayMapping.objects.filter(
+ librenms_name=librenms_name,
+ librenms_class=librenms_class or "",
+ is_regex=False,
+ )
+ if manufacturer is not None:
+ qs = qs.filter(models.Q(manufacturer=manufacturer) | models.Q(manufacturer__isnull=True))
+ else:
+ qs = qs.filter(manufacturer__isnull=True)
+ return qs.exists()
+
+ def get(self, request, pk):
+ from dcim.models import Device, ModuleBay, ModuleBayTemplate
+
+ # Read-only modal render β only require plugin view permission and
+ # NetBox add-permission on ModuleBayTemplate so users without it never
+ # see a form they cannot submit. POST also instantiates live ModuleBay
+ # rows via _instantiate_template_on_existing(), so require add_modulebay
+ # here too to keep the GET/POST permission contract aligned.
+ self.required_object_permissions = {"GET": [("add", ModuleBayTemplate), ("add", ModuleBay)]}
+ if error := self.require_all_permissions("GET"):
+ return error
+
+ device = get_object_or_404(Device, pk=pk)
+
+ target_kind = request.GET.get("target_kind", "")
+ if target_kind not in self.TARGET_KINDS:
+ return HttpResponse("Invalid target_kind.", status=400)
+ try:
+ target_pk = int(request.GET.get("target_pk", ""))
+ except (TypeError, ValueError):
+ return HttpResponse("Missing or invalid target_pk.", status=400)
+ target = self._resolve_target(target_kind, target_pk)
+
+ librenms_name = request.GET.get("librenms_name", "")
+ librenms_class = request.GET.get("librenms_class", "")
+ suggested_name = request.GET.get("suggested_name", "")
+ manufacturer = self._device_manufacturer(device)
+ mapping_exists = self._existing_bay_mapping(
+ librenms_name, librenms_class, manufacturer
+ ) or self._existing_regex_mapping_covers(librenms_name, librenms_class, manufacturer)
+ # Offer the auto-mapping option only when we have a LibreNMS name to
+ # map *from* and there's no existing mapping covering it. Permission
+ # to add the mapping itself is also required β users without it would
+ # only see the checkbox return a permission error on POST.
+
+ can_add_mapping = request.user.has_perm("netbox_librenms_plugin.add_modulebaymapping")
+ offer_mapping_checkbox = bool(librenms_name) and not mapping_exists and can_add_mapping
+ # Pattern preview against the *current* suggested NetBox name. The
+ # template re-runs this on every keystroke client-side; the
+ # server-side derivation is for the initial render + POST-time check.
+ mapping_pattern = (
+ self._derive_mapping_pattern(librenms_name, suggested_name) if offer_mapping_checkbox else None
+ )
+
+ context = {
+ "device_pk": pk,
+ "target_kind": target_kind,
+ "target_pk": target_pk,
+ "target_label": str(target),
+ "suggested_name": suggested_name,
+ "suggested_position": request.GET.get("suggested_position", ""),
+ "suggested_label": request.GET.get("suggested_label", ""),
+ "librenms_name": librenms_name,
+ "librenms_class": librenms_class,
+ "manufacturer_label": str(manufacturer) if manufacturer else "",
+ "offer_mapping_checkbox": offer_mapping_checkbox,
+ "mapping_exists": mapping_exists,
+ "mapping_pattern": mapping_pattern,
+ "mapping_default_kind": "regex" if mapping_pattern else "exact",
+ }
+ return render(request, "netbox_librenms_plugin/htmx/add_bay_template_modal.html", context)
+
+ def post(self, request, pk):
+ from dcim.models import Device, ModuleBay, ModuleBayTemplate
+
+ from netbox_librenms_plugin.models import ModuleBayMapping
+
+ # POST creates the template AND instantiates live ModuleBay rows on
+ # existing devices/modules via _instantiate_template_on_existing(), so
+ # gate on add_modulebay in addition to add_modulebaytemplate.
+ self.required_object_permissions = {"POST": [("add", ModuleBayTemplate), ("add", ModuleBay)]}
+ if error := self.require_all_permissions("POST"):
+ return error
+
+ device = get_object_or_404(Device, pk=pk)
+ sync_url = reverse("plugins:netbox_librenms_plugin:device_librenms_sync", kwargs={"pk": pk})
+
+ target_kind = request.POST.get("target_kind", "")
+ if target_kind not in self.TARGET_KINDS:
+ messages.error(request, "Invalid target_kind for bay template.")
+ return _modules_redirect_response(request, sync_url)
+ try:
+ target_pk = int(request.POST.get("target_pk", ""))
+ except (TypeError, ValueError):
+ messages.error(request, "Missing or invalid target_pk for bay template.")
+ return _modules_redirect_response(request, sync_url)
+
+ name = (request.POST.get("name") or "").strip()
+ if not name:
+ messages.error(request, "Bay template name is required.")
+ return _modules_redirect_response(request, sync_url)
+ position = (request.POST.get("position") or "").strip()
+ label = (request.POST.get("label") or "").strip()
+ description = (request.POST.get("description") or "").strip()
+
+ # Optional auto-mapping inputs (echoed from the GET-rendered modal).
+ librenms_name = (request.POST.get("librenms_name") or "").strip()
+ librenms_class = (request.POST.get("librenms_class") or "").strip()
+ also_create_mapping = request.POST.get("also_create_mapping") == "1"
+ mapping_kind = request.POST.get("mapping_kind", "exact")
+ if mapping_kind not in ("exact", "regex"):
+ mapping_kind = "exact"
+
+ target = self._resolve_target(target_kind, target_pk)
+ kwargs = {
+ "name": name,
+ "position": position,
+ "label": label,
+ "description": description,
+ }
+ if target_kind == "device_type":
+ kwargs["device_type"] = target
+ else:
+ kwargs["module_type"] = target
+
+ # Only add a mapping when the user actually picked a different
+ # NetBox bay name β if the names match, no mapping is needed.
+ will_add_mapping = (
+ also_create_mapping
+ and librenms_name
+ and librenms_name != name
+ and request.user.has_perm("netbox_librenms_plugin.add_modulebaymapping")
+ )
+ manufacturer = self._device_manufacturer(device) if will_add_mapping else None
+ # Resolve regex pattern + replacement when the user opted in.
+ mapping_libre_value = librenms_name
+ mapping_netbox_value = name
+ mapping_is_regex = False
+ if will_add_mapping and mapping_kind == "regex":
+ pattern = self._derive_mapping_pattern(librenms_name, name)
+ if pattern is None:
+ # Server-side rule didn't fire β fall back to exact rather than
+ # writing an unverified regex from client input.
+ mapping_kind = "exact"
+ else:
+ mapping_libre_value = pattern["librenms_pattern"]
+ mapping_netbox_value = pattern["netbox_replacement"]
+ mapping_is_regex = True
+ if will_add_mapping:
+ if mapping_is_regex:
+ race = self._existing_regex_mapping_covers(librenms_name, librenms_class, manufacturer)
+ else:
+ race = self._existing_bay_mapping(librenms_name, librenms_class, manufacturer)
+ if race:
+ # Race: a mapping was added between modal render and submit.
+ will_add_mapping = False
+
+ try:
+ mapping_created = False
+ instantiated_count = 0
+ with transaction.atomic():
+ bay_template = ModuleBayTemplate(**kwargs)
+ bay_template.full_clean()
+ bay_template.save()
+ instantiated_count = self._instantiate_template_on_existing(bay_template, target_kind, target)
+ if will_add_mapping:
+ mapping = ModuleBayMapping(
+ librenms_name=mapping_libre_value,
+ librenms_class=librenms_class,
+ netbox_bay_name=mapping_netbox_value,
+ is_regex=mapping_is_regex,
+ manufacturer=manufacturer,
+ )
+ mapping.full_clean()
+ mapping.save()
+ mapping_created = True
+ instantiate_note = ""
+ if instantiated_count:
+ noun = "device" if target_kind == "device_type" else "module"
+ plural = "" if instantiated_count == 1 else "s"
+ instantiate_note = f" Bay added to {instantiated_count} existing {noun}{plural}."
+ if mapping_created:
+ vendor_note = f" (scoped to {manufacturer})" if manufacturer else " (global)"
+ kind_note = "regex " if mapping_is_regex else ""
+ messages.success(
+ request,
+ f"Added bay template '{name}' to {target} and {kind_note}ModuleBayMapping "
+ f"'{mapping_libre_value}' β '{mapping_netbox_value}'{vendor_note}.{instantiate_note}",
+ )
+ else:
+ messages.success(request, f"Added bay template '{name}' to {target}.{instantiate_note}")
+ except (ValidationError, IntegrityError) as e:
+ messages.error(request, f"Failed to add bay template: {e}")
+
+ return _modules_redirect_response(request, sync_url)
diff --git a/pyproject.toml b/pyproject.toml
index 5c86227d19..6d4cceb989 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -26,6 +26,10 @@ classifiers=[
requires-python = ">=3.12.0"
+[project.urls]
+"Homepage" = "https://github.com/bonzo81/netbox-librenms-plugin/"
+"Bug Tracker" = "https://github.com/bonzo81/netbox-librenms-plugin/issues/"
+
# New: Restrict package discovery to our plugin only
[tool.setuptools]
include-package-data = true
diff --git a/tests/e2e/__init__.py b/tests/e2e/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py
new file mode 100644
index 0000000000..990870ada7
--- /dev/null
+++ b/tests/e2e/conftest.py
@@ -0,0 +1,16 @@
+"""Conftest for end-to-end Playwright tests.
+
+These tests are excluded from the default pytest discovery via ``testpaths``
+in ``pyproject.toml`` and are intended to be invoked explicitly:
+
+ python -m pytest tests/e2e/test_module_install.py -v -s
+
+Note: ``pyproject.toml`` sets ``DJANGO_SETTINGS_MODULE = "netbox.settings"``
+under ``[tool.pytest.ini_options]``, which pytest-django reads directly from
+the config file (not from the environment). Popping the env var here would
+have no effect on pytest-django's initialisation. The e2e tests do not
+import or use Django models β they drive a running NetBox over HTTP β so
+pytest-django's auto-loading is harmless and we leave it alone. If you ever
+need to skip pytest-django entirely for this suite, invoke pytest with
+``-p no:django``.
+"""
diff --git a/tests/e2e/test_device_type_mapping.py b/tests/e2e/test_device_type_mapping.py
new file mode 100644
index 0000000000..75ed89b3d3
--- /dev/null
+++ b/tests/e2e/test_device_type_mapping.py
@@ -0,0 +1,300 @@
+"""
+End-to-end Playwright test for the Device Type mapping workflow in the import modal.
+
+Tests that after adding a DeviceTypeMapping from the validation modal:
+ 1. The modal stays open and no longer shows "No matching type".
+ 2. The background table row updates to enable the Import button β no page refresh needed.
+
+Configuration (env vars):
+ E2E_TESTS_ENABLED=1 Required to run
+ NETBOX_URL (default http://127.0.0.1:8000)
+ NETBOX_USER / NETBOX_PASS (default admin/admin)
+ E2E_IMPORT_URL Full import page URL to use
+ E2E_HEADLESS=0 Set to 0 to watch browser (default 1)
+
+Run:
+ HTTP_PROXY= HTTPS_PROXY= http_proxy= https_proxy= \\
+ no_proxy=localhost,127.0.0.1 \\
+ E2E_TESTS_ENABLED=1 \\
+ E2E_IMPORT_URL="http://127.0.0.1:8000/plugins/librenms_plugin/librenms-import/?apply_filters=1&librenms_location=1" \\
+ /opt/python-venv/bin/python -m pytest tests/e2e/test_device_type_mapping.py -v -s -p no:django
+"""
+
+import os
+import subprocess
+
+import pytest
+
+NETBOX_URL = os.environ.get("NETBOX_URL", "http://127.0.0.1:8000")
+NETBOX_USER = os.environ.get("NETBOX_USER", "admin")
+NETBOX_PASS = os.environ.get("NETBOX_PASS", "admin")
+IMPORT_URL = os.environ.get(
+ "E2E_IMPORT_URL",
+ f"{NETBOX_URL}/plugins/librenms_plugin/librenms-import/",
+)
+HEADLESS = os.environ.get("E2E_HEADLESS", "1") != "0"
+
+E2E_ENABLED = os.environ.get("E2E_TESTS_ENABLED", "0") == "1"
+
+if not E2E_ENABLED:
+ pytest.skip("E2E tests skipped β set E2E_TESTS_ENABLED=1 to run", allow_module_level=True)
+
+
+def _get_container():
+ result = subprocess.run(["docker", "ps", "--format", "{{.Names}}"], capture_output=True, text=True)
+ matches = [n for n in result.stdout.strip().split("\n") if "devcontainer-devcontainer" in n]
+ if len(matches) == 1:
+ return matches[0]
+ pytest.skip("No devcontainer found")
+
+
+def _netbox_shell(code):
+ import shlex
+
+ container = _get_container()
+ result = subprocess.run(
+ [
+ "docker",
+ "exec",
+ container,
+ "bash",
+ "-c",
+ f"cd /opt/netbox/netbox && python3 manage.py shell -c {shlex.quote(code)}",
+ ],
+ capture_output=True,
+ text=True,
+ env={**os.environ, "PATH": "/usr/bin:/bin", "HOME": "/root"},
+ )
+ lines = [
+ line
+ for line in result.stdout.strip().split("\n")
+ if not line.startswith("π§¬") and "objects imported automatically" not in line
+ ]
+ if result.returncode != 0:
+ raise RuntimeError(f"shell failed: {result.stderr}")
+ return "\n".join(lines).strip()
+
+
+def _delete_mapping(pk: int):
+ _netbox_shell(
+ f"from netbox_librenms_plugin.models import DeviceTypeMapping; "
+ f"from django.core.cache import cache; "
+ f"DeviceTypeMapping.objects.filter(pk={pk}).delete(); "
+ f"cache.clear()"
+ )
+
+
+def _restore_mapping(hardware: str, device_type_id: int):
+ # update_or_create β get_or_create would not revert an existing mapping
+ # whose netbox_device_type was mutated by the test.
+ _netbox_shell(
+ f"from netbox_librenms_plugin.models import DeviceTypeMapping; "
+ f"from dcim.models import DeviceType; "
+ f"dt = DeviceType.objects.get(pk={device_type_id}); "
+ f"DeviceTypeMapping.objects.update_or_create("
+ f" librenms_hardware={hardware!r}, defaults={{'netbox_device_type': dt}})"
+ )
+
+
+def _get_existing_mapping():
+ output = _netbox_shell(
+ "from netbox_librenms_plugin.models import DeviceTypeMapping; "
+ "m = DeviceTypeMapping.objects.first(); "
+ "print(f'{m.pk}|{m.librenms_hardware}|{m.netbox_device_type_id}') if m else print('')"
+ )
+ if not output.strip():
+ return None
+ parts = output.strip().split("|")
+ return {"pk": int(parts[0]), "hardware": parts[1], "device_type_id": int(parts[2])} if len(parts) == 3 else None
+
+
+@pytest.fixture(scope="module")
+def browser():
+ from playwright.sync_api import sync_playwright
+
+ pw = sync_playwright().start()
+ b = pw.chromium.launch(headless=HEADLESS)
+ yield b
+ b.close()
+ pw.stop()
+
+
+@pytest.fixture
+def page(browser):
+ ctx = browser.new_context(ignore_https_errors=True)
+ pg = ctx.new_page()
+ pg.goto(f"{NETBOX_URL}/login/", timeout=15000)
+ pg.fill("#id_username", NETBOX_USER)
+ pg.fill("#id_password", NETBOX_PASS)
+ pg.click("button[type=submit]")
+ pg.wait_for_load_state("networkidle")
+ yield pg
+ ctx.close()
+
+
+def _close_modal(page):
+ """Close the HTMX modal if open (use the X btn-close button)."""
+ close_btn = page.locator("#htmx-modal button.btn-close[data-bs-dismiss='modal']").first
+ if close_btn.count() > 0 and close_btn.is_visible():
+ close_btn.click()
+ # Wait for Bootstrap to actually hide the modal rather than guessing
+ # at the animation duration.
+ page.wait_for_selector("#htmx-modal:not(.show)", timeout=4000)
+
+
+def _find_no_match_modal(page):
+ """
+ Open detail buttons one by one until a modal shows 'No matching type'.
+ Returns (modal_locator, row_locator) or (None, None).
+ """
+ btns = page.locator("button.btn-outline-danger, button.btn-warning").all()
+ for btn in btns[:10]: # Try up to 10 buttons
+ if not btn.is_visible():
+ continue
+ row = btn.locator("xpath=ancestor::tr[1]")
+ btn.click()
+ try:
+ page.wait_for_selector("#htmx-modal-content .modal-header", timeout=8000)
+ except Exception:
+ continue
+ page.wait_for_load_state("networkidle")
+ modal = page.locator("#htmx-modal-content")
+ if "No matching type" in modal.inner_text():
+ return modal, row
+ _close_modal(page)
+ return None, None
+
+
+class TestDeviceTypeMappingModal:
+ """Verify DeviceType mapping from import validation modal updates row in-place."""
+
+ def test_mapping_updates_modal_and_row(self, page):
+ """
+ Setup: temporarily remove a DeviceTypeMapping so a device shows 'No matching type'.
+ Assert: after adding the mapping back via the modal form:
+ - Modal stays open and shows the match (no 'No matching type')
+ - Background row updates to show enabled Import button (no page refresh)
+ """
+ mapping = _get_existing_mapping()
+ if not mapping:
+ pytest.skip("No DeviceTypeMapping in DB β cannot set up test scenario")
+
+ _delete_mapping(mapping["pk"])
+ try:
+ self._run_test(page, mapping)
+ finally:
+ _restore_mapping(mapping["hardware"], mapping["device_type_id"])
+
+ def _run_test(self, page, mapping):
+ page.goto(IMPORT_URL, timeout=20000)
+ page.wait_for_load_state("networkidle")
+
+ modal, row = _find_no_match_modal(page)
+ if modal is None:
+ pytest.skip(
+ f"No 'No matching type' device found at {IMPORT_URL} after removing mapping. "
+ "Set E2E_IMPORT_URL to the filtered import page that shows the affected device."
+ )
+
+ # Ensure a role is selected so the device can become importable
+ role_select = row.locator("select[name^='role_']")
+ if role_select.count() > 0 and not role_select.input_value():
+ role_select.select_option(index=1)
+ page.wait_for_load_state("networkidle")
+ # Re-open modal after role change
+ _close_modal(page)
+ modal, row = _find_no_match_modal(page)
+ if modal is None:
+ pytest.skip("No 'No matching type' device after role selection")
+
+ # Locate the search input
+ search = modal.locator("input[placeholder*='Search device types']")
+ assert search.is_visible(), "DeviceType search input not visible in modal"
+
+ # Diagnose: dump the search input's id and the dropdown div's id
+ search_id = search.get_attribute("id")
+ dropdown_div = modal.locator("[id^='dt-dropdown-']")
+ dropdown_id = dropdown_div.get_attribute("id") if dropdown_div.count() > 0 else "(not found)"
+ print(f"\nSearch input id: {search_id!r}, dropdown div id: {dropdown_id!r}")
+
+ # Check whether the JS event listener is attached by inspecting via JS
+ has_listener = (
+ page.evaluate(f"() => {{ var el = document.getElementById({search_id!r}); return !!el; }}")
+ if search_id
+ else False
+ )
+ print(f"searchEl found via getElementById: {has_listener}")
+
+ # Dispatch input event to trigger dropdown fetch; wait for either
+ # results to appear or a brief HTMX network exchange to settle.
+ page.evaluate(
+ f"() => {{"
+ f" var el = document.getElementById({search_id!r});"
+ f" if (el) {{ el.value = 'a'; el.dispatchEvent(new Event('input', {{bubbles:true}})); }}"
+ f"}}"
+ )
+ try:
+ page.wait_for_selector(
+ "#htmx-modal-content [id^='dt-dropdown-'] a",
+ timeout=2000,
+ state="attached",
+ )
+ except Exception:
+ pass
+
+ # If still no dropdown, type via keyboard and wait again
+ search.click()
+ page.keyboard.type("b", delay=100)
+ try:
+ page.wait_for_selector(
+ "#htmx-modal-content [id^='dt-dropdown-'] a",
+ timeout=2000,
+ state="attached",
+ )
+ except Exception:
+ pass
+
+ # Final check
+ dropdown_items = modal.locator("[id^='dt-dropdown-'] a")
+ print(f"Dropdown item count after attempts: {dropdown_items.count()}")
+
+ page.wait_for_selector(
+ "#htmx-modal-content [id^='dt-dropdown-'] a",
+ timeout=8000,
+ state="attached",
+ )
+
+ dropdown = modal.locator("[id^='dt-dropdown-']")
+ first = dropdown.locator("a").first
+ assert first.count() > 0, "No results in DeviceType dropdown"
+ first.click()
+ # Wait for the hidden input to receive the selected value rather than
+ # polling with a fixed timeout.
+ page.wait_for_function(
+ "() => { var el = document.querySelector(\"#htmx-modal-content input[name='device_type_id']\");"
+ " return el && el.value && el.value.length > 0; }",
+ timeout=4000,
+ )
+
+ hidden_val = modal.locator("input[name='device_type_id']").input_value()
+ assert hidden_val, "Hidden device_type_id not filled after clicking a result"
+
+ # Submit
+ modal.locator("button:has-text('Add Mapping')").click()
+
+ # Wait for the OOB row swap + modal refresh to settle via network idle
+ # rather than a fixed 3s timeout.
+ page.wait_for_load_state("networkidle")
+
+ # --- Assert 1: modal stays open and shows match ---
+ assert modal.is_visible(), "Modal closed after adding mapping β expected to stay open"
+ modal_after = modal.inner_text()
+ assert "No matching type" not in modal_after, f"Modal still shows 'No matching type'.\nModal:\n{modal_after}"
+
+ # --- Assert 2: row updated without page refresh ---
+ import_btn = row.locator("button.btn-success.device-import-btn:not([disabled])")
+ assert import_btn.count() > 0, (
+ "No enabled Import button in row after mapping β "
+ "JS deviceMappingAdded handler may not have triggered the row update.\n"
+ f"Row: {row.inner_text()}"
+ )
diff --git a/tests/e2e/test_module_install.py b/tests/e2e/test_module_install.py
new file mode 100644
index 0000000000..f3b0aa90f6
--- /dev/null
+++ b/tests/e2e/test_module_install.py
@@ -0,0 +1,370 @@
+"""
+End-to-end Playwright tests for LibreNMS plugin module sync workflow.
+
+These tests exercise the full import β modules β install flow against a
+live NetBox + LibreNMS instance inside the devcontainer.
+
+Prerequisites:
+ - NetBox running at NETBOX_URL (default http://172.22.0.4:8000)
+ - LibreNMS server configured in plugin settings
+ - A device linked to LibreNMS that has inventory modules
+ - Playwright installed: pip install playwright && playwright install chromium
+
+Configuration (environment variables):
+ E2E_TESTS_ENABLED=1 Required to run these tests
+ E2E_DEVICE_ID= NetBox device PK to test against (auto-detected if omitted)
+ NETBOX_URL= NetBox base URL (default http://172.22.0.4:8000)
+ NETBOX_USER= Login username (default admin)
+ NETBOX_PASS= Login password (default admin)
+ NETBOX_CONTAINER= Docker container name (auto-detected if omitted)
+
+Run:
+ cd /home/mzieba/workspace/netbox-librenms-plugin
+ HTTP_PROXY= HTTPS_PROXY= http_proxy= https_proxy= \
+ no_proxy=localhost,127.0.0.1,172.22.0.4 \
+ E2E_TESTS_ENABLED=1 python -m pytest tests/e2e/test_module_install.py -v -s
+"""
+
+import os
+import subprocess
+
+import pytest
+
+NETBOX_URL = os.environ.get("NETBOX_URL", "http://172.22.0.4:8000")
+NETBOX_USER = os.environ.get("NETBOX_USER", "admin")
+NETBOX_PASS = os.environ.get("NETBOX_PASS", "admin")
+CONTAINER_NAME = None
+
+E2E_ENABLED = os.environ.get("E2E_TESTS_ENABLED", "0") == "1"
+
+if not E2E_ENABLED:
+ pytest.skip(
+ "E2E tests skipped β set E2E_TESTS_ENABLED=1 to run against a live instance",
+ allow_module_level=True,
+ )
+
+
+def _get_container():
+ """Find the devcontainer name."""
+ global CONTAINER_NAME
+ if CONTAINER_NAME:
+ return CONTAINER_NAME
+ override = os.environ.get("NETBOX_CONTAINER")
+ if override:
+ CONTAINER_NAME = override
+ return override
+ result = subprocess.run(
+ ["docker", "ps", "--format", "{{.Names}}"],
+ capture_output=True,
+ text=True,
+ )
+ if result.returncode != 0:
+ raise RuntimeError(f"docker ps failed (rc={result.returncode}): {result.stderr}")
+ matches = [name for name in result.stdout.strip().split("\n") if "devcontainer-devcontainer" in name]
+ if len(matches) == 1:
+ CONTAINER_NAME = matches[0]
+ return CONTAINER_NAME
+ if len(matches) > 1:
+ raise RuntimeError(f"Multiple candidate devcontainers found: {matches}. Set NETBOX_CONTAINER.")
+ pytest.skip("No devcontainer found")
+
+
+def _netbox_shell(code):
+ """Run Python code in NetBox's Django shell."""
+ import shlex
+
+ container = _get_container()
+ escaped = shlex.quote(code)
+ result = subprocess.run(
+ [
+ "docker",
+ "exec",
+ container,
+ "bash",
+ "-c",
+ f"cd /opt/netbox/netbox && python3 manage.py shell -c {escaped}",
+ ],
+ capture_output=True,
+ text=True,
+ env={**os.environ, "PATH": "/usr/bin:/bin", "HOME": "/root"},
+ )
+ lines = [
+ line
+ for line in result.stdout.strip().split("\n")
+ if not line.startswith("π§¬") and "objects imported automatically" not in line
+ ]
+ if result.returncode != 0:
+ raise RuntimeError(f"netbox shell command failed (rc={result.returncode}): {result.stderr}")
+ return "\n".join(lines).strip()
+
+
+def _detect_device_id():
+ """Find a device linked to LibreNMS that has inventory modules.
+
+ Returns the NetBox device PK, or skips the test session if none found.
+ """
+ output = _netbox_shell(
+ "from dcim.models import Device; "
+ "devs = Device.objects.exclude(custom_field_data__librenms_id=None)"
+ ".exclude(custom_field_data__librenms_id={}).order_by('pk'); "
+ "print(devs.first().pk if devs.exists() else '')"
+ )
+ if not output.strip():
+ pytest.skip("No device with librenms_id found in NetBox")
+ return int(output.strip())
+
+
+def _delete_device_modules(device_id):
+ """Remove all modules from a device."""
+ _netbox_shell(
+ f"from dcim.models import Module; "
+ f"deleted = Module.objects.filter(device_id={device_id}).delete(); "
+ f"print(f'Deleted {{deleted}}')"
+ )
+
+
+def _get_interfaces(device_id):
+ """Get interface names for a device."""
+ output = _netbox_shell(
+ f"from dcim.models import Interface; "
+ f'[print(f\'{{i.name}}|{{i.module.module_type.model if i.module else "-"}}|'
+ f'{{i.module.module_bay.name if i.module else "-"}}\')'
+ f" for i in Interface.objects.filter(device_id={device_id}).order_by('name')]"
+ )
+ results = []
+ for line in output.split("\n"):
+ if "|" in line:
+ name, mod_type, bay = line.split("|")
+ results.append({"name": name, "module_type": mod_type, "bay": bay})
+ return results
+
+
+@pytest.fixture(scope="module")
+def device_id():
+ """Resolve the device PK to test against."""
+ env_id = os.environ.get("E2E_DEVICE_ID")
+ if env_id:
+ return int(env_id)
+ return _detect_device_id()
+
+
+@pytest.fixture(scope="module")
+def browser():
+ """Launch browser for the test module."""
+ from playwright.sync_api import sync_playwright
+
+ pw = sync_playwright().start()
+ b = pw.chromium.launch(headless=True)
+ yield b
+ b.close()
+ pw.stop()
+
+
+@pytest.fixture
+def page(browser):
+ """Create a new page and log in to NetBox."""
+ ctx = browser.new_context(ignore_https_errors=True)
+ pg = ctx.new_page()
+
+ pg.goto(f"{NETBOX_URL}/login/", timeout=10000)
+ pg.fill("#id_username", NETBOX_USER)
+ pg.fill("#id_password", NETBOX_PASS)
+ pg.click("button[type=submit]")
+ pg.wait_for_load_state("networkidle")
+ yield pg
+ ctx.close()
+
+
+class TestModuleInstallWorkflow:
+ """Test the full module sync and install workflow."""
+
+ def _goto_modules_tab(self, page, device_id):
+ """Navigate to the modules sync tab and refresh data."""
+ page.goto(f"{NETBOX_URL}/dcim/devices/{device_id}/librenms-sync/?tab=modules")
+ page.wait_for_load_state("networkidle")
+ page.wait_for_selector('button:has-text("Refresh Modules")', timeout=10000)
+
+ btn = page.query_selector('button:has-text("Refresh Modules")')
+ assert btn is not None, "Refresh Modules button not found"
+ btn.click()
+ page.wait_for_selector("#modules table tr", timeout=30000)
+ page.wait_for_load_state("networkidle")
+
+ def _get_table_rows(self, page):
+ """Parse the module sync table into dicts."""
+ pane = page.query_selector("#modules")
+ assert pane is not None, "Modules pane not found"
+
+ def _text(tr, col):
+ el = tr.query_selector(f'td[data-col="{col}"]')
+ return el.inner_text().strip() if el else ""
+
+ rows = []
+ for tr in pane.query_selector_all("table tr"):
+ if tr.query_selector('td[data-col="name"]') is None:
+ continue
+ rows.append(
+ {
+ "name": _text(tr, "name"),
+ "model": _text(tr, "model"),
+ "serial": _text(tr, "serial"),
+ "bay": _text(tr, "module_bay"),
+ "type": _text(tr, "module_type"),
+ "status": _text(tr, "status"),
+ "tr": tr,
+ }
+ )
+ return rows
+
+ def _find_row_with_button(self, rows, button_text):
+ """Find the first top-level row that has a button matching button_text."""
+ for row in rows:
+ if row["name"].startswith("ββ"):
+ continue
+ btn = row["tr"].query_selector(f'button:has-text("{button_text}")')
+ if btn:
+ return row, btn
+ return None, None
+
+ def test_clean_state_shows_matched_rows(self, page, device_id):
+ """After deleting all modules, table shows rows with Matched status."""
+ _delete_device_modules(device_id)
+ self._goto_modules_tab(page, device_id)
+
+ rows = self._get_table_rows(page)
+ assert len(rows) > 0, "No rows in module sync table"
+
+ matched = [r for r in rows if r["status"] == "Matched"]
+ assert len(matched) > 0, f"Expected at least one Matched row, got statuses: {set(r['status'] for r in rows)}"
+
+ def test_single_install(self, page, device_id):
+ """Installing a single top-level module works."""
+ _delete_device_modules(device_id)
+ self._goto_modules_tab(page, device_id)
+
+ rows = self._get_table_rows(page)
+ row, btn = self._find_row_with_button(rows, "Install")
+ if not btn:
+ pytest.skip("No installable module found in table")
+
+ module_name = row["name"]
+ btn.click()
+ page.wait_for_load_state("networkidle")
+
+ module_count = _netbox_shell(
+ f"from dcim.models import Module; print(Module.objects.filter(device_id={device_id}).count())"
+ )
+ assert int(module_count.strip()) > 0, f"No modules in DB after installing '{module_name}'"
+
+ def test_branch_install(self, page, device_id):
+ """Branch install creates parent module + children."""
+ _delete_device_modules(device_id)
+ self._goto_modules_tab(page, device_id)
+
+ rows = self._get_table_rows(page)
+ row, btn = self._find_row_with_button(rows, "Install Branch")
+ if not btn:
+ pytest.skip("No branch-installable module found in table")
+
+ module_name = row["name"]
+ btn.click()
+ page.wait_for_load_state("networkidle", timeout=60000)
+
+ module_count = _netbox_shell(
+ f"from dcim.models import Module; print(Module.objects.filter(device_id={device_id}).count())"
+ )
+ count = int(module_count.strip())
+ assert count > 1, f"Branch install of '{module_name}' created {count} module(s), expected >1"
+
+ interfaces = _get_interfaces(device_id)
+ module_interfaces = [i for i in interfaces if i["module_type"] != "-"]
+ assert len(module_interfaces) > 0, f"No interfaces linked to modules after branch install of '{module_name}'"
+ for iface in module_interfaces:
+ assert not iface["name"].isdigit(), (
+ f"Interface '{iface['name']}' has bare numeric name β naming rule not applied"
+ )
+
+ def test_branch_install_no_duplicate_errors(self, page, device_id):
+ """Branch install handles already-occupied bays gracefully."""
+ self._goto_modules_tab(page, device_id)
+
+ rows = self._get_table_rows(page)
+ row, btn = self._find_row_with_button(rows, "Install Branch")
+ if not btn:
+ pytest.skip("No branch-installable module found in table")
+
+ module_name = row["name"]
+
+ # First install (may already be installed from prior test)
+ btn.click()
+ page.wait_for_load_state("networkidle", timeout=60000)
+
+ # Navigate back and try again β bays should now be occupied
+ self._goto_modules_tab(page, device_id)
+ rows = self._get_table_rows(page)
+ _, btn2 = self._find_row_with_button(rows, "Install Branch")
+ if not btn2:
+ pytest.skip(f"No Install Branch button after first install of '{module_name}'")
+
+ btn2.click()
+ page.wait_for_load_state("networkidle", timeout=30000)
+
+ body_text = page.query_selector("body").inner_text()
+ assert "Branch install failed" not in body_text, (
+ "Branch install crashed instead of handling occupied bays gracefully"
+ )
+
+ def test_child_bays_hidden_when_parent_not_installed(self, page, device_id):
+ """Children show 'No matching bay' when parent module is not installed."""
+ _delete_device_modules(device_id)
+ self._goto_modules_tab(page, device_id)
+
+ rows = self._get_table_rows(page)
+ children = [r for r in rows if r["name"].startswith("ββ")]
+ if not children:
+ pytest.skip("No child module rows found in table")
+
+ no_bay = [c for c in children if "No matching bay" in c["bay"]]
+ assert len(no_bay) > 0, (
+ "Expected some children to show 'No matching bay' when parent not installed, "
+ f"got bays: {set(c['bay'] for c in children)}"
+ )
+
+ def test_full_workflow(self, page, device_id):
+ """Full workflow: clean β install individuals β branch install β verify."""
+ _delete_device_modules(device_id)
+ self._goto_modules_tab(page, device_id)
+
+ # Step 1: Install a few individual modules
+ for _ in range(3):
+ rows = self._get_table_rows(page)
+ row, btn = self._find_row_with_button(rows, "Install")
+ if not btn:
+ break
+ btn.click()
+ page.wait_for_load_state("networkidle")
+
+ # Step 2: Branch install all available branches
+ while True:
+ rows = self._get_table_rows(page)
+ row, btn = self._find_row_with_button(rows, "Install Branch")
+ if not btn:
+ break
+ btn.click()
+ page.wait_for_load_state("networkidle", timeout=60000)
+ self._goto_modules_tab(page, device_id)
+
+ # Verify: no top-level "Matched" items remain uninstalled
+ self._goto_modules_tab(page, device_id)
+ rows = self._get_table_rows(page)
+ top_level_matched = [r for r in rows if r["status"] == "Matched" and not r["name"].startswith("ββ")]
+ assert len(top_level_matched) == 0, (
+ f"Top-level items still Matched after full workflow: {[r['name'] for r in top_level_matched]}"
+ )
+
+ # Verify interface naming β no bare numeric names
+ interfaces = _get_interfaces(device_id)
+ for iface in interfaces:
+ assert not iface["name"].isdigit(), (
+ f"Interface '{iface['name']}' has bare numeric name β naming rule not applied"
+ )