diff --git a/netbox_librenms_plugin/import_utils/device_operations.py b/netbox_librenms_plugin/import_utils/device_operations.py index 6fb760529..6b87935b0 100644 --- a/netbox_librenms_plugin/import_utils/device_operations.py +++ b/netbox_librenms_plugin/import_utils/device_operations.py @@ -749,7 +749,7 @@ def import_single_device( 'interfaces': int, 'cables': int, 'ip_addresses': int - } + }, } """ try: diff --git a/netbox_librenms_plugin/import_utils/vm_operations.py b/netbox_librenms_plugin/import_utils/vm_operations.py index a630b5563..99ae10a8a 100644 --- a/netbox_librenms_plugin/import_utils/vm_operations.py +++ b/netbox_librenms_plugin/import_utils/vm_operations.py @@ -151,6 +151,10 @@ def bulk_import_vms( # Use job logger if available, otherwise standard logger log = job.logger if job else logger + # Resolve options once before the loop — they do not change per-VM + use_sysname_opt = sync_options.get("use_sysname", True) if sync_options else True + strip_domain_opt = sync_options.get("strip_domain", False) if sync_options else False + for idx, vm_id in enumerate(vm_ids, start=1): # Check for job cancellation before first VM and every 5 thereafter if job and (idx == 1 or idx % 5 == 0) and _is_job_cancelled(job): @@ -173,8 +177,6 @@ def bulk_import_vms( continue # Validate as VM - use_sysname_opt = sync_options.get("use_sysname", True) if sync_options else True - strip_domain_opt = sync_options.get("strip_domain", False) if sync_options else False validation = validate_device_for_import( libre_device, import_as_vm=True, diff --git a/netbox_librenms_plugin/static/netbox_librenms_plugin/js/librenms_import.js b/netbox_librenms_plugin/static/netbox_librenms_plugin/js/librenms_import.js index a2facc706..edfda206a 100644 --- a/netbox_librenms_plugin/static/netbox_librenms_plugin/js/librenms_import.js +++ b/netbox_librenms_plugin/static/netbox_librenms_plugin/js/librenms_import.js @@ -349,8 +349,8 @@ * Persists toggle state to user preferences on change. */ function initializeTogglePrefs() { - const sysname = document.getElementById('use-sysname-toggle'); - const strip = document.getElementById('strip-domain-toggle'); + const sysname = document.getElementById('use-sysname-toggle-cb'); + const strip = document.getElementById('strip-domain-toggle-cb'); if (sysname) sysname.addEventListener('change', function () { savePref('use_sysname', this.checked); }); if (strip) strip.addEventListener('change', function () { savePref('strip_domain', this.checked); }); } @@ -1199,10 +1199,14 @@ const dismissTrigger = event.target.closest('[data-bs-dismiss="modal"]'); if (dismissTrigger) { - event.preventDefault(); - - // Check if it's in the HTMX modal - if (modalElement.contains(dismissTrigger)) { + // Only handle dismiss triggers whose nearest .modal ancestor IS + // the outer HTMX modal. Buttons inside nested modals (e.g. the + // Promote-to-host modal rendered inside #htmx-modal-content) + // must be left for Bootstrap's own dismiss handler so they + // close the inner modal, not the outer one. + const nearestModal = dismissTrigger.closest('.modal'); + if (nearestModal === modalElement) { + event.preventDefault(); hideModal(modalElement, fallbackBackdropRef); } } diff --git a/netbox_librenms_plugin/tables/ipaddresses.py b/netbox_librenms_plugin/tables/ipaddresses.py index 8c9e1bfe2..0324c6de2 100644 --- a/netbox_librenms_plugin/tables/ipaddresses.py +++ b/netbox_librenms_plugin/tables/ipaddresses.py @@ -33,6 +33,7 @@ class Meta: row_attrs = { "data-interface": lambda record: record["ip_address"], "data-name": lambda record: record["ip_address"], + "data-mgmt-ip": lambda record: "true" if record.get("is_mgmt_ip") else "", } selection = ToggleColumn( diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/_ipaddress_sync_content.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/_ipaddress_sync_content.html index e73bb8396..535ecaa55 100644 --- a/netbox_librenms_plugin/templates/netbox_librenms_plugin/_ipaddress_sync_content.html +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/_ipaddress_sync_content.html @@ -10,11 +10,21 @@ {% if ip_sync.server_key %}{% endif %}
-
+
+
+ + + +
{% if ip_sync.cache_expiry %}
@@ -74,6 +84,45 @@
+ {% else %}
diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/_dt_mapping_form.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/_dt_mapping_form.html new file mode 100644 index 000000000..ee846ed00 --- /dev/null +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/_dt_mapping_form.html @@ -0,0 +1,115 @@ +{% comment %} +Reusable "Add Device Type Mapping" form for the import validation modal. + +Required context: + - libre_device (dict-like with device_id, hardware) +Optional context: + - preselect_device_type: a DeviceType instance to pre-fill the typeahead + (used in the existing-device + mismatch branch so a single click maps + `libre_device.hardware` → `existing_device.device_type`). + - submit_label (defaults to "Add Mapping") +{% endcomment %} +
+ {% csrf_token %} + {% if server_key %}{% endif %} +
+ + + + +
+ {# Disabled until a typeahead item is chosen (or a preselect value is present). #} + +
+ diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/_platform_manage_icon.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/_platform_manage_icon.html new file mode 100644 index 000000000..65fb5b07e --- /dev/null +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/_platform_manage_icon.html @@ -0,0 +1,15 @@ +{% comment %} +Compact pencil icon that opens the combined "manage platform" modal (map +LibreNMS OS to an existing platform, or create a new one). Keeps the platform +cell uncluttered. Requires: libre_device, server_key. +{% endcomment %} +{% if libre_device.os and libre_device.os != "-" %} + +{% endif %} diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/_platform_mapping_form.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/_platform_mapping_form.html new file mode 100644 index 000000000..9de8f381b --- /dev/null +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/_platform_mapping_form.html @@ -0,0 +1,113 @@ +{% comment %} +Reusable "Add Platform Mapping" form for the import validation modal. + +Required context: + - libre_device (dict-like with device_id, os) +Optional: + - preselect_platform: a Platform instance to pre-fill the typeahead. + - submit_label +{% endcomment %} +
+ {% csrf_token %} + {% if server_key %}{% endif %} +
+ + + + +
+ {# Disabled until a typeahead item is chosen (or a preselect value is present). #} + +
+ diff --git a/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/create_platform_modal.html b/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/create_platform_modal.html index f2b557e0d..27db062c8 100644 --- a/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/create_platform_modal.html +++ b/netbox_librenms_plugin/templates/netbox_librenms_plugin/htmx/create_platform_modal.html @@ -13,10 +13,19 @@ {# server_key - (optional) LibreNMS server key, forwarded as hidden field #}
+{% if libre_device %} +{# Map the LibreNMS OS to an EXISTING platform (separate form -> add_platform_mapping). #} +
+
Map {{ librenms_os }} to an existing platform
+ {% include "netbox_librenms_plugin/htmx/_platform_mapping_form.html" with preselect_platform=current_platform submit_label="Map to this platform" %} +
— or create a new platform below —
+
+
+{% endif %} {% csrf_token %} @@ -27,9 +36,9 @@
diff --git a/netbox_librenms_plugin/tests/test_coverage_base_views.py b/netbox_librenms_plugin/tests/test_coverage_base_views.py index f2456d7be..5a691570a 100644 --- a/netbox_librenms_plugin/tests/test_coverage_base_views.py +++ b/netbox_librenms_plugin/tests/test_coverage_base_views.py @@ -2259,3 +2259,57 @@ def test_netbox_only_interfaces_non_vc_device_name_from_obj(self): gi01 = next((i for i in netbox_only if i["name"] == "Gi0/1"), None) assert gi01 is not None assert gi01["device_name"] == "router-1" + + +# ============================================================================= +# BaseIPAddressTableView._flag_management_ip +# ============================================================================= + + +class TestBaseIPAddressTableViewFlagManagementIp: + """Tests for marking the LibreNMS management-IP row (Set Primary IP support).""" + + def _make_view(self, librenms_id=42): + from netbox_librenms_plugin.views.base.ip_addresses_view import BaseIPAddressTableView + + view = object.__new__(BaseIPAddressTableView) + view.librenms_id = librenms_id + view._librenms_api = MagicMock() + view._librenms_api.server_key = "default" + return view + + def test_flags_matching_entry(self): + view = self._make_view() + view._librenms_api.get_device_info.return_value = (True, {"ip": "10.0.0.1"}) + data = [{"ip_address": "10.0.0.5"}, {"ip_address": "10.0.0.1"}] + view._flag_management_ip(data) + assert data[0].get("is_mgmt_ip") is None + assert data[1].get("is_mgmt_ip") is True + + def test_no_flag_when_no_match(self): + view = self._make_view() + view._librenms_api.get_device_info.return_value = (True, {"ip": "192.0.2.9"}) + data = [{"ip_address": "10.0.0.1"}] + view._flag_management_ip(data) + assert data[0].get("is_mgmt_ip") is None + + def test_no_flag_when_no_librenms_id(self): + view = self._make_view(librenms_id=None) + data = [{"ip_address": "10.0.0.1"}] + view._flag_management_ip(data) + view._librenms_api.get_device_info.assert_not_called() + assert data[0].get("is_mgmt_ip") is None + + def test_no_flag_when_device_info_fails(self): + view = self._make_view() + view._librenms_api.get_device_info.return_value = (False, None) + data = [{"ip_address": "10.0.0.1"}] + view._flag_management_ip(data) + assert data[0].get("is_mgmt_ip") is None + + def test_no_flag_when_mgmt_ip_blank(self): + view = self._make_view() + view._librenms_api.get_device_info.return_value = (True, {"ip": ""}) + data = [{"ip_address": "10.0.0.1"}] + view._flag_management_ip(data) + assert data[0].get("is_mgmt_ip") is None diff --git a/netbox_librenms_plugin/tests/test_coverage_sync_views.py b/netbox_librenms_plugin/tests/test_coverage_sync_views.py index 552b67e25..131b8abc0 100644 --- a/netbox_librenms_plugin/tests/test_coverage_sync_views.py +++ b/netbox_librenms_plugin/tests/test_coverage_sync_views.py @@ -2583,3 +2583,92 @@ def test_no_vlans_created_shows_warning(self): with patch.object(view, "_redirect", return_value=MagicMock()): view._handle_create_vlans(req, mock_obj, "device", 1) mock_msg.warning.assert_called_once() + + +class TestSyncIPAddressesViewSetPrimaryIp: + """Phase 1: auto-match the LibreNMS management IP and set it as Primary IP.""" + + def _setup_view(self): + from netbox_librenms_plugin.views.sync.ip_addresses import SyncIPAddressesView + + view = _make_view(SyncIPAddressesView) + view._post_server_key = "default" + return view + + def test_same_host(self): + from netbox_librenms_plugin.views.sync.ip_addresses import SyncIPAddressesView as V + + assert V._same_host("10.0.0.1", "10.0.0.1") is True + assert V._same_host("10.0.0.1", "10.0.0.2") is False + assert V._same_host("not-an-ip", "10.0.0.1") is False + # IPv6 equality across differing textual forms + assert V._same_host("2001:db8::1", "2001:0db8:0000:0000:0000:0000:0000:0001") is True + + def test_set_primary_ip_sets_ipv4_and_is_idempotent(self): + from netbox_librenms_plugin.views.sync.ip_addresses import SyncIPAddressesView as V + + ip_obj = MagicMock(family=4, pk=42) + obj = MagicMock() + obj.primary_ip4_id = None + assert V._set_primary_ip(obj, ip_obj) is True + assert obj.primary_ip4 is ip_obj + obj.save.assert_called_once() + + # Already pointing at this IP -> no change, no extra save + obj.save.reset_mock() + obj.primary_ip4_id = 42 + assert V._set_primary_ip(obj, ip_obj) is False + obj.save.assert_not_called() + + def test_set_primary_ip_uses_v6_field(self): + from netbox_librenms_plugin.views.sync.ip_addresses import SyncIPAddressesView as V + + ip_obj = MagicMock(family=6, pk=7) + obj = MagicMock() + obj.primary_ip6_id = None + assert V._set_primary_ip(obj, ip_obj) is True + assert obj.primary_ip6 is ip_obj + + def _run_process(self, view, cached, *, mgmt_ip, set_primary=True, interface=True): + selected = ["10.0.0.1"] + created_ip = MagicMock(family=4, pk=42) + obj = MagicMock() + obj.primary_ip4_id = None + with patch("netbox_librenms_plugin.views.sync.ip_addresses.resolve_set_primary_ip", return_value=set_primary): + with patch.object(view, "get_management_ip", return_value=mgmt_ip) as mock_mgmt: + with patch("netbox_librenms_plugin.views.sync.ip_addresses.transaction", _atomic_txn()): + with patch("netbox_librenms_plugin.views.sync.ip_addresses.IPAddress") as mock_ip_cls: + mock_ip_cls.objects.filter.return_value.first.return_value = None + mock_ip_cls.objects.create.return_value = created_ip + with patch("netbox_librenms_plugin.views.sync.ip_addresses.Interface") as mock_iface_cls: + mock_iface_cls.objects.get.return_value = MagicMock() + with patch.object(view, "get_vrf_selection", return_value=None): + results = view.process_ip_sync(view.request, selected, cached, obj, "device") + return results, obj, created_ip, mock_mgmt + + def test_primary_set_when_matched_and_interface_assigned(self): + view = self._setup_view() + cached = [{"ip_address": "10.0.0.1", "ip_with_mask": "10.0.0.1/24", "interface_url": "/api/dcim/interfaces/5/"}] + results, obj, created_ip, _ = self._run_process(view, cached, mgmt_ip="10.0.0.1") + assert results["primary_set"] == ["10.0.0.1"] + assert obj.primary_ip4 is created_ip + + def test_primary_skipped_when_no_interface(self): + view = self._setup_view() + cached = [{"ip_address": "10.0.0.1", "ip_with_mask": "10.0.0.1/24", "interface_url": None}] + results, obj, _, _ = self._run_process(view, cached, mgmt_ip="10.0.0.1") + assert results["primary_set"] == [] + obj.save.assert_not_called() + + def test_primary_skipped_when_ip_does_not_match_mgmt(self): + view = self._setup_view() + cached = [{"ip_address": "10.0.0.1", "ip_with_mask": "10.0.0.1/24", "interface_url": "/api/dcim/interfaces/5/"}] + results, obj, _, _ = self._run_process(view, cached, mgmt_ip="10.9.9.9") + assert results["primary_set"] == [] + + def test_toggle_off_skips_mgmt_lookup_and_primary(self): + view = self._setup_view() + cached = [{"ip_address": "10.0.0.1", "ip_with_mask": "10.0.0.1/24", "interface_url": "/api/dcim/interfaces/5/"}] + results, obj, _, mock_mgmt = self._run_process(view, cached, mgmt_ip="10.0.0.1", set_primary=False) + assert results["primary_set"] == [] + mock_mgmt.assert_not_called() diff --git a/netbox_librenms_plugin/tests/test_import_utils.py b/netbox_librenms_plugin/tests/test_import_utils.py index 27551436f..f7d8620c0 100644 --- a/netbox_librenms_plugin/tests/test_import_utils.py +++ b/netbox_librenms_plugin/tests/test_import_utils.py @@ -5895,3 +5895,70 @@ def _capture_create(**kwargs): create_virtual_chassis_with_members(master_device, members_info, {"device_id": 1}) assert mock_device_cls.objects.create.call_count == 1 + + +class TestResolveSetPrimaryIp: + """Phase 1: resolve_set_primary_ip cascade (POST/GET toggle -> user pref -> False).""" + + def _make_request(self, post=None, get=None): + from unittest.mock import MagicMock + + request = MagicMock() + request.POST = post or {} + request.GET = get or {} + request.user = MagicMock() + return request + + def test_defaults_false_when_nothing_set(self): + from unittest.mock import patch + + from netbox_librenms_plugin.utils import resolve_set_primary_ip + + request = self._make_request() + with patch("netbox_librenms_plugin.utils.get_user_pref", return_value=None): + assert resolve_set_primary_ip(request) is False + + def test_post_toggle_on(self): + from unittest.mock import patch + + from netbox_librenms_plugin.utils import resolve_set_primary_ip + + request = self._make_request(post={"set-primary-ip-toggle": "on"}) + with patch("netbox_librenms_plugin.utils.get_user_pref", return_value=None): + assert resolve_set_primary_ip(request) is True + + def test_post_toggle_off_overrides_pref(self): + from unittest.mock import patch + + from netbox_librenms_plugin.utils import resolve_set_primary_ip + + request = self._make_request(post={"set-primary-ip-toggle": "off"}) + with patch("netbox_librenms_plugin.utils.get_user_pref", return_value=True): + assert resolve_set_primary_ip(request) is False + + def test_underscore_key_variant(self): + from unittest.mock import patch + + from netbox_librenms_plugin.utils import resolve_set_primary_ip + + request = self._make_request(post={"set_primary_ip": "1"}) + with patch("netbox_librenms_plugin.utils.get_user_pref", return_value=None): + assert resolve_set_primary_ip(request) is True + + def test_get_used_when_not_in_post(self): + from unittest.mock import patch + + from netbox_librenms_plugin.utils import resolve_set_primary_ip + + request = self._make_request(get={"set-primary-ip-toggle": "true"}) + with patch("netbox_librenms_plugin.utils.get_user_pref", return_value=None): + assert resolve_set_primary_ip(request) is True + + def test_user_pref_bool_used_when_no_toggle(self): + from unittest.mock import patch + + from netbox_librenms_plugin.utils import resolve_set_primary_ip + + request = self._make_request() + with patch("netbox_librenms_plugin.utils.get_user_pref", return_value=True): + assert resolve_set_primary_ip(request) is True diff --git a/netbox_librenms_plugin/tests/test_template_comments.py b/netbox_librenms_plugin/tests/test_template_comments.py new file mode 100644 index 000000000..6f49df4f7 --- /dev/null +++ b/netbox_librenms_plugin/tests/test_template_comments.py @@ -0,0 +1,36 @@ +"""Guard against multi-line Django ``{# #}`` comments. + +Django's ``{# #}`` comment syntax must stay on a single line. A multi-line +``{# ... #}`` is NOT recognised as a comment and renders as literal text in the +page (e.g. the platform-cell pencil-icon partial once leaked its header comment +into the import modal). ``get_template()`` only *parses* templates, so this slips +past template-compile checks — hence this static scan. Use ``{% comment %}`` for +multi-line comments instead. +""" + +import pathlib + +import netbox_librenms_plugin + +TEMPLATES_DIR = pathlib.Path(netbox_librenms_plugin.__file__).parent / "templates" + + +def _html_templates(): + return sorted(TEMPLATES_DIR.rglob("*.html")) + + +def test_templates_found(): + """Sanity check that the scan actually has templates to look at.""" + assert _html_templates(), f"No .html templates found under {TEMPLATES_DIR}" + + +def test_no_multiline_single_hash_comments(): + """No line may contain an unbalanced ``{#`` / ``#}`` (a multi-line ``{# #}``).""" + offenders = [] + for path in _html_templates(): + for lineno, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1): + if line.count("{#") != line.count("#}"): + offenders.append(f"{path.relative_to(TEMPLATES_DIR)}:{lineno}: {line.strip()}") + assert not offenders, "Multi-line `{# #}` comments render as literal text — use `{% comment %}`:\n" + "\n".join( + offenders + ) diff --git a/netbox_librenms_plugin/urls.py b/netbox_librenms_plugin/urls.py index 4d7881c17..73b954c54 100644 --- a/netbox_librenms_plugin/urls.py +++ b/netbox_librenms_plugin/urls.py @@ -13,6 +13,7 @@ from .views import ( AddDeviceToLibreNMSView, AddDeviceTypeMappingView, + AddPlatformMappingView, AssignVCSerialView, BulkImportConfirmView, BulkImportDevicesView, @@ -429,6 +430,11 @@ AddDeviceTypeMappingView.as_view(), name="add_device_type_mapping", ), + path( + "device-import/add-platform-mapping//", + AddPlatformMappingView.as_view(), + name="add_platform_mapping", + ), path( "device-import/create-platform//", CreatePlatformFromImportView.as_view(), diff --git a/netbox_librenms_plugin/utils.py b/netbox_librenms_plugin/utils.py index 8da32ed15..95864b00d 100644 --- a/netbox_librenms_plugin/utils.py +++ b/netbox_librenms_plugin/utils.py @@ -363,6 +363,53 @@ def _is_truthy(val): return use_sysname, strip_domain +def same_host(a, b) -> bool: + """True if two address strings refer to the same host IP (version-aware).""" + from ipaddress import ip_address + + try: + return ip_address(a) == ip_address(b) + except ValueError: + return False + + +def resolve_set_primary_ip(request) -> bool: + """Resolve the "set Primary IP from the LibreNMS management IP" flag. + + Cascade (mirrors :func:`resolve_naming_preferences`, minus a settings + default since this is a per-sync action choice): + + 1. POST/GET ``set-primary-ip-toggle`` (or ``set_primary_ip``) wins + -- set by the IP-sync tab toggle. + 2. Otherwise the user's saved preference + ``plugins.netbox_librenms_plugin.set_primary_ip``. + 3. Otherwise ``False`` (opt-in). + + When enabled, :class:`SyncIPAddressesView` sets ``primary_ip4``/``primary_ip6`` + on the device/VM for the synced IP that matches the LibreNMS management IP, + provided that IP ends up assigned to one of the object's interfaces. + """ + _TRUTHY = frozenset({"on", "true", "1"}) + _KEYS = ("set-primary-ip-toggle", "set_primary_ip-toggle", "set_primary_ip") + + def _is_truthy(val): + return val.lower() in _TRUTHY if val is not None else False + + post_val = next((request.POST.get(k) for k in _KEYS if k in request.POST), None) + get_val = next((request.GET.get(k) for k in _KEYS if k in request.GET), None) + + if post_val is not None: + return _is_truthy(post_val) + if get_val is not None: + return _is_truthy(get_val) + + pref = get_user_pref(request, "plugins.netbox_librenms_plugin.set_primary_ip") + if pref is not None: + return _is_truthy(pref) if isinstance(pref, str) else bool(pref) + + return False + + def get_interface_name_field(request: Optional[HttpRequest] = None) -> str: """ Get interface name field with request override support. diff --git a/netbox_librenms_plugin/views/__init__.py b/netbox_librenms_plugin/views/__init__.py index 4ce30a5c5..ece34847a 100644 --- a/netbox_librenms_plugin/views/__init__.py +++ b/netbox_librenms_plugin/views/__init__.py @@ -110,6 +110,7 @@ PlatformMappingView, ) from .imports.actions import AddDeviceTypeMappingView # noqa: F401 +from .imports.actions import AddPlatformMappingView # noqa: F401 from .object_sync import ( # noqa: F401 DeviceCableTableView, DeviceInterfaceTableView, diff --git a/netbox_librenms_plugin/views/base/ip_addresses_view.py b/netbox_librenms_plugin/views/base/ip_addresses_view.py index a11edc82f..a8f289deb 100644 --- a/netbox_librenms_plugin/views/base/ip_addresses_view.py +++ b/netbox_librenms_plugin/views/base/ip_addresses_view.py @@ -12,7 +12,12 @@ from virtualization.models import VirtualMachine from netbox_librenms_plugin.tables.ipaddresses import IPAddressTable -from netbox_librenms_plugin.utils import get_interface_name_field, get_librenms_device_id +from netbox_librenms_plugin.utils import ( + get_interface_name_field, + get_librenms_device_id, + resolve_set_primary_ip, + same_host, +) from netbox_librenms_plugin.views.mixins import CacheMixin, LibreNMSAPIMixin, LibreNMSPermissionMixin logger = logging.getLogger(__name__) @@ -99,8 +104,33 @@ def enrich_ip_data(self, ip_data, obj, interface_name_field): enriched_data.append(enriched_ip) + self._flag_management_ip(enriched_data) return enriched_data + def _flag_management_ip(self, enriched_data): + """Mark the entry whose IP equals the device's LibreNMS management IP. + + Used by the "Set Primary IP" toggle on the IP-sync tab to auto-select + the right row. Best-effort: a lookup failure simply leaves nothing + flagged so the sync table still renders. + """ + librenms_id = getattr(self, "librenms_id", None) + if not librenms_id: + return + try: + success, info = self.librenms_api.get_device_info(librenms_id) + except Exception: # pragma: no cover - defensive + return + if not success or not isinstance(info, dict): + return + mgmt_ip = (info.get("ip") or "").strip() + if not mgmt_ip: + return + for entry in enriched_data: + if same_host(entry.get("ip_address", ""), mgmt_ip): + entry["is_mgmt_ip"] = True + break + def _prefetch_netbox_data(self, obj): """Prefetch all necessary NetBox data to minimize database queries""" # Get all interfaces for the device @@ -275,6 +305,7 @@ def _prepare_context(self, request, obj, interface_name_field, fetch_fresh=False "object": obj, "cache_expiry": cache_expiry, "server_key": server_key, + "set_primary_ip": resolve_set_primary_ip(request), } def get_context_data(self, request, obj): diff --git a/netbox_librenms_plugin/views/imports/actions.py b/netbox_librenms_plugin/views/imports/actions.py index be442d93d..f0ccd016c 100644 --- a/netbox_librenms_plugin/views/imports/actions.py +++ b/netbox_librenms_plugin/views/imports/actions.py @@ -35,7 +35,11 @@ fetch_model_by_id, ) from netbox_librenms_plugin.tables.device_status import DeviceImportTable -from netbox_librenms_plugin.utils import resolve_naming_preferences, save_user_pref, set_librenms_device_id +from netbox_librenms_plugin.utils import ( + resolve_naming_preferences, + save_user_pref, + set_librenms_device_id, +) from netbox_librenms_plugin.views.mixins import LibreNMSAPIMixin, LibreNMSPermissionMixin, NetBoxObjectPermissionMixin logger = logging.getLogger(__name__) @@ -655,8 +659,6 @@ def post(self, request): # noqa: PLR0912 - branching keeps responses explicit ) # Show notification and redirect - matching NetBox's native pattern - from django.utils.safestring import mark_safe - messages.info( request, mark_safe( @@ -744,21 +746,24 @@ def post(self, request): # noqa: PLR0912 - branching keeps responses explicit failed_count = len(device_result.get("failed", [])) + len(vm_result.get("failed", [])) skipped_count = len(device_result.get("skipped", [])) + len(vm_result.get("skipped", [])) + # Build summary messages for both paths. For HTMX responses, messages.* + # is consumed/cleared by device_import_row.html before reaching the browser; + # htmx_toasts carries the same text as an explicit OOB swap instead. + htmx_toasts = [] # [(bg_class, mdi_class, label, text), ...] + if success_count: - messages.success( - request, - f"Successfully imported {success_count} LibreNMS device{'s' if success_count != 1 else ''}", - ) + _msg = f"Successfully imported {success_count} LibreNMS device{'s' if success_count != 1 else ''}" + messages.success(request, _msg) + htmx_toasts.append(("text-bg-success", "mdi-check-circle", "Success", _msg)) + if failed_count: - messages.error( - request, - f"Failed to import {failed_count} device{'s' if failed_count != 1 else ''}", - ) + _msg = f"Failed to import {failed_count} device{'s' if failed_count != 1 else ''}" + messages.error(request, _msg) + htmx_toasts.append(("text-bg-danger", "mdi-alert-circle", "Error", _msg)) if skipped_count: - messages.warning( - request, - f"Skipped {skipped_count} existing device{'s' if skipped_count != 1 else ''}", - ) + _msg = f"Skipped {skipped_count} existing device{'s' if skipped_count != 1 else ''}" + messages.warning(request, _msg) + htmx_toasts.append(("text-bg-warning", "mdi-alert", "Warning", _msg)) if request.headers.get("HX-Request"): # Return updated rows for all imported devices using HTMX OOB swaps @@ -819,7 +824,38 @@ def post(self, request): # noqa: PLR0912 - branching keeps responses explicit ).content.decode("utf-8") updated_rows_html.append(row_html) - # Return concatenated row HTML with closeModal trigger + # Append all summary toasts as a single OOB swap. The messages.* + # queue is consumed/cleared by device_import_row.html, so they never + # reach the browser in the HTMX path; this OOB fragment takes over. + if htmx_toasts: + toast_items = mark_safe( + "".join( + format_html( + '", + bg_cls, + icon_cls, + label, + text, + ) + for bg_cls, icon_cls, label, text in htmx_toasts + ) + ) + updated_rows_html.append( + format_html( + '
{}
', + toast_items, + ) + ) return HttpResponse( "\n".join(updated_rows_html), headers={"HX-Trigger": '{"closeModal": null}'}, @@ -1578,10 +1614,12 @@ def get(self, request, device_id): _, validation, _ = self.get_validated_device_with_selections(device_id, request) device_pk = None selected_manufacturer_pk = None + current_platform = None if validation: existing = validation.get("existing_device") if existing: device_pk = existing.pk + current_platform = getattr(existing, "platform", None) device_type = getattr(existing, "device_type", None) if device_type: selected_manufacturer_pk = device_type.manufacturer_id @@ -1604,6 +1642,9 @@ def get(self, request, device_id): "server_key": self.librenms_api.server_key, "use_htmx": True, "htmx_include": htmx_include, + # Enable the "map to existing platform" section of the combined modal. + "libre_device": libre_device, + "current_platform": current_platform, }, ) @@ -1771,6 +1812,7 @@ class SaveUserPrefView(LibreNMSPermissionMixin, View): ALLOWED_PREFS = { "use_sysname": "plugins.netbox_librenms_plugin.use_sysname", "strip_domain": "plugins.netbox_librenms_plugin.strip_domain", + "set_primary_ip": "plugins.netbox_librenms_plugin.set_primary_ip", "interface_name_field": "plugins.netbox_librenms_plugin.interface_name_field", } @@ -1789,3 +1831,121 @@ def post(self, request): save_user_pref(request, self.ALLOWED_PREFS[key], value) return JsonResponse({"status": "ok"}) + + +class AddPlatformMappingView( + LibreNMSPermissionMixin, NetBoxObjectPermissionMixin, LibreNMSAPIMixin, DeviceImportHelperMixin, View +): + """HTMX view to create a PlatformMapping from the import validation modal.""" + + def post(self, request, device_id): + """Create a PlatformMapping linking the LibreNMS OS string to a NetBox Platform.""" + if error := self.require_write_permission(): + return error + + from dcim.models import Platform + from netbox_librenms_plugin.librenms_api import LibreNMSAPI + from netbox_librenms_plugin.models import PlatformMapping + + post_server_key = (request.POST.get("server_key") or "").strip() + if post_server_key: + 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 _htmx_error_response("Device not found in LibreNMS.") + + librenms_os = (libre_device.get("os") or "").strip() + if not librenms_os or librenms_os == "-": + return _htmx_error_response("Device has no OS string — cannot create mapping.") + + platform_id = request.POST.get("platform_id", "").strip() + if not platform_id: + return _htmx_error_response("Please select a platform before submitting.") + + try: + platform_id = int(platform_id) + except (ValueError, TypeError): + return _htmx_error_response("Invalid platform selection.") + + try: + platform = Platform.objects.get(pk=platform_id) + except Platform.DoesNotExist: + return _htmx_error_response("Selected platform not found.") + + if PlatformMapping.objects.filter(librenms_os__iexact=librenms_os).count() > 1: + return _htmx_error_response( + "Multiple mappings exist for this OS string. Remove duplicates before updating." + ) + existing_mapping = PlatformMapping.objects.filter(librenms_os__iexact=librenms_os).first() + self.required_object_permissions = { + "POST": [("change", PlatformMapping) if existing_mapping else ("add", PlatformMapping)] + } + if error := self.require_object_permissions("POST"): + return error + + try: + with transaction.atomic(): + # Lock the row to close the TOCTOU window between the upfront + # permission check and the actual write. select_for_update cannot + # lock absent rows, so the create branch handles IntegrityError. + # Materialise the locked rows in one query — count() would drop + # the FOR UPDATE clause, leaving the rows unlocked. + locked_rows = list( + PlatformMapping.objects.select_for_update().filter(librenms_os__iexact=librenms_os)[:2] + ) + if len(locked_rows) > 1: + return _htmx_error_response( + "Multiple mappings exist for this OS string. Remove duplicates before updating." + ) + locked = locked_rows[0] if locked_rows else None + if locked and not existing_mapping: + # Concurrent request created the mapping after our upfront read. + # Only escalate to change permission if we would actually mutate. + if locked.netbox_platform_id != platform_id: + self.required_object_permissions = {"POST": [("change", PlatformMapping)]} + if error := self.require_object_permissions("POST"): + return error + if existing_mapping and not locked: + # Mapping was deleted between our upfront read and the lock. + # We are about to CREATE a new row, so require add permission. + self.required_object_permissions = {"POST": [("add", PlatformMapping)]} + if error := self.require_object_permissions("POST"): + return error + if locked: + if locked.netbox_platform_id != platform_id: + locked.netbox_platform = platform + locked.full_clean() + locked.save() + else: + try: + PlatformMapping.objects.create( + librenms_os=librenms_os.lower(), + netbox_platform=platform, + ) + except IntegrityError: + return _htmx_error_response("Mapping was created concurrently. Please try again.") + except Exception as exc: + logger.exception("AddPlatformMappingView: failed to save mapping: %s", exc) + return _htmx_error_response("Error saving mapping. Please try again.") + + cache_key = get_import_device_cache_key(device_id, self.librenms_api.server_key) + cache.delete(cache_key) + + 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), + ) + + 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") diff --git a/netbox_librenms_plugin/views/sync/ip_addresses.py b/netbox_librenms_plugin/views/sync/ip_addresses.py index 3fb43820c..e1b602af3 100644 --- a/netbox_librenms_plugin/views/sync/ip_addresses.py +++ b/netbox_librenms_plugin/views/sync/ip_addresses.py @@ -1,3 +1,4 @@ +import logging from urllib.parse import quote_plus from dcim.models import Device, Interface @@ -11,6 +12,7 @@ from ipam.models import VRF, IPAddress from virtualization.models import VirtualMachine, VMInterface +from netbox_librenms_plugin.utils import resolve_set_primary_ip, same_host from netbox_librenms_plugin.views.mixins import ( CacheMixin, LibreNMSAPIMixin, @@ -18,6 +20,8 @@ NetBoxObjectPermissionMixin, ) +logger = logging.getLogger(__name__) + class SyncIPAddressesView(LibreNMSPermissionMixin, NetBoxObjectPermissionMixin, LibreNMSAPIMixin, CacheMixin, View): """Synchronize IP addresses from LibreNMS cache into NetBox.""" @@ -100,13 +104,68 @@ def post(self, request, object_type, pk): return redirect(self.get_ip_tab_url(obj)) - def process_ip_sync(self, request, selected_ips, cached_ips, obj, object_type): - """Create or update IP addresses in NetBox from cached LibreNMS data.""" - results = {"created": [], "updated": [], "unchanged": [], "failed": []} + def get_management_ip(self, obj): + """Return the LibreNMS management/polling IP for *obj*, or None. + + Used to decide which synced IP (if any) should become the object's + Primary IP. Best-effort: any lookup failure yields None so the sync + itself is never blocked. + """ + try: + librenms_id = self.librenms_api.get_librenms_id(obj) + if not librenms_id: + return None + success, info = self.librenms_api.get_device_info(librenms_id) + if not success or not isinstance(info, dict): + return None + return (info.get("ip") or "").strip() or None + except Exception: # pragma: no cover - defensive + return None - with transaction.atomic(): - for ip_address in selected_ips: - try: + @staticmethod + def _same_host(a, b): + """True if two address strings refer to the same host IP.""" + return same_host(a, b) + + @staticmethod + def _set_primary_ip(obj, ip_obj): + """Point obj.primary_ip4/6 (by family) at *ip_obj*. Returns True if changed. + + The caller guarantees ``ip_obj`` is assigned to one of the object's + interfaces, so this satisfies NetBox's ``primary_ip`` constraint. + """ + field = "primary_ip6" if ip_obj.family == 6 else "primary_ip4" + if getattr(obj, f"{field}_id") == ip_obj.pk: + return False + setattr(obj, field, ip_obj) + obj.save() + return True + + def process_ip_sync(self, request, selected_ips, cached_ips, obj, object_type): + """Create or update IP addresses in NetBox from cached LibreNMS data. + + When the "set Primary IP" toggle is on, the synced IP that matches the + LibreNMS management IP — and ends up assigned to one of the object's + interfaces — is also set as the object's ``primary_ip4``/``primary_ip6``. + """ + results = { + "created": [], + "updated": [], + "unchanged": [], + "failed": [], + "primary_set": [], + "primary_no_interface": [], + "errors": {}, + } + + set_primary = resolve_set_primary_ip(request) + mgmt_ip = self.get_management_ip(obj) if set_primary else None + + for ip_address in selected_ips: + try: + # Per-IP savepoint so one bad address rolls back only itself and + # surfaces a real error, instead of poisoning the whole batch. + with transaction.atomic(): ip_data = next(ip for ip in cached_ips if ip["ip_address"] == ip_address) vrf = self.get_vrf_selection(request, ip_address) @@ -121,7 +180,10 @@ def process_ip_sync(self, request, selected_ips, cached_ips, obj, object_type): ip_with_mask = ip_data["ip_with_mask"] - existing_ip = IPAddress.objects.filter(address=ip_with_mask).first() + # Scope the lookup to the target VRF: the same address can + # legitimately exist in multiple VRFs, and matching on address + # alone would hijack an unrelated IP and rewrite its VRF. + existing_ip = IPAddress.objects.filter(address=ip_with_mask, vrf=vrf).first() if existing_ip: if existing_ip.assigned_object != interface or existing_ip.vrf != vrf: @@ -131,8 +193,9 @@ def process_ip_sync(self, request, selected_ips, cached_ips, obj, object_type): results["updated"].append(ip_address) else: results["unchanged"].append(ip_address) + ip_obj = existing_ip else: - IPAddress.objects.create( + ip_obj = IPAddress.objects.create( address=ip_with_mask, assigned_object=interface, status="active", @@ -140,10 +203,21 @@ def process_ip_sync(self, request, selected_ips, cached_ips, obj, object_type): ) results["created"].append(ip_address) - except Exception: # pragma: no cover - defensive - results["failed"].append(ip_address) + # Primary-IP auto-match for the management IP. NetBox requires + # the IP be interface-assigned to be a primary, so when the + # interface is missing we flag it rather than silently skip. + if mgmt_ip and self._same_host(ip_data["ip_address"], mgmt_ip): + if interface is None: + results["primary_no_interface"].append(ip_address) + elif self._set_primary_ip(obj, ip_obj): + results["primary_set"].append(ip_address) + + except Exception as exc: + logger.warning("IP sync failed for %s: %s", ip_address, exc, exc_info=True) + results["failed"].append(ip_address) + results["errors"][ip_address] = str(exc) or exc.__class__.__name__ - return results + return results def display_sync_results(self, request, results): """Display flash messages summarizing the IP sync results.""" @@ -151,10 +225,21 @@ def display_sync_results(self, request, results): messages.success(request, f"Created IP addresses: {', '.join(results['created'])}") if results["updated"]: messages.success(request, f"Updated IP addresses: {', '.join(results['updated'])}") + if results.get("primary_set"): + messages.success(request, f"Set as Primary IP: {', '.join(results['primary_set'])}") + if results.get("primary_no_interface"): + messages.warning( + request, + "Primary IP not set for " + f"{', '.join(results['primary_no_interface'])} — no NetBox interface for this IP. " + "Sync interfaces first, then re-run.", + ) if results["unchanged"]: messages.warning( request, f"IP addresses already exist: {', '.join(results['unchanged'])}", ) if results["failed"]: - messages.error(request, f"Failed to sync IP addresses: {', '.join(results['failed'])}") + errors = results.get("errors", {}) + detail = ", ".join(f"{ip} ({errors[ip]})" if errors.get(ip) else ip for ip in results["failed"]) + messages.error(request, f"Failed to sync IP addresses: {detail}")