Garoshi/netbox-librenms-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
netbox-librenms-plugin/netbox_librenms_plugin/tests/test_import_utils.py
Vlastislav Svatek 673e67106e
Some checks failed
ci / deploy (push) Has been cancelled
Details
CodeQL Advanced / Analyze (actions) (push) Has been cancelled
Details
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Details
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Details
first commit
2026-06-05 10:39:05 +02:00

5895 lines
246 KiB
Python
Raw Permalink Blame History

"""
Tests for netbox_librenms_plugin.import_utils module.
Phase 2 tests covering cache key generation, device name determination,
device retrieval, and device validation functions.
"""
from unittest.mock import MagicMock, patch
# =============================================================================
# TestCacheKeyGeneration - 4 tests
# =============================================================================
class TestCacheKeyGeneration:
"""Test cache key generation functions."""
def test_get_cache_metadata_key_basic(self):
"""Generate cache metadata key with minimal filters."""
from netbox_librenms_plugin.import_utils import get_cache_metadata_key
key = get_cache_metadata_key(server_key="default", filters={}, vc_enabled=False)
assert "default" in key
assert "librenms_filter_cache_metadata" in key
assert isinstance(key, str)
def test_get_cache_metadata_key_all_params(self):
"""Generate cache metadata key with all filter parameters."""
from netbox_librenms_plugin.import_utils import get_cache_metadata_key
key = get_cache_metadata_key(
server_key="production",
filters={"location": "DC1", "type": "network", "hostname": "switch*"},
vc_enabled=True,
)
assert "production" in key
# Filter values are hashed, not embedded directly in the key
assert "librenms_filter_cache_metadata" in key
assert "True" in key or "true" in key.lower()
def test_get_validated_device_cache_key(self):
"""Generate validated device cache key."""
from netbox_librenms_plugin.import_utils import get_validated_device_cache_key
key = get_validated_device_cache_key(
server_key="default",
filters={"location": "NYC"},
device_id=123,
vc_enabled=True,
)
assert "validated_device" in key
assert "default" in key
assert "123" in key
assert "vc" in key
def test_get_import_device_cache_key(self):
"""Generate raw device data cache key."""
from netbox_librenms_plugin.import_utils import get_import_device_cache_key
key = get_import_device_cache_key(device_id=456, server_key="secondary")
assert "import_device_data" in key
assert "secondary" in key
assert "456" in key
def test_validated_device_cache_key_unique_per_naming_mode(self):
"""Different naming preferences produce different cache keys."""
from netbox_librenms_plugin.import_utils import get_validated_device_cache_key
base_args = dict(server_key="default", filters={}, device_id=123, vc_enabled=False)
key_default = get_validated_device_cache_key(**base_args)
key_no_sysname = get_validated_device_cache_key(**base_args, use_sysname=False)
key_strip = get_validated_device_cache_key(**base_args, strip_domain=True)
assert key_default != key_no_sysname
assert key_default != key_strip
assert key_no_sysname != key_strip
def test_cache_metadata_key_unique_per_naming_mode(self):
"""Different naming preferences produce different metadata cache keys."""
from netbox_librenms_plugin.import_utils import get_cache_metadata_key
base_args = dict(server_key="default", filters={}, vc_enabled=False)
key_default = get_cache_metadata_key(**base_args)
key_no_sysname = get_cache_metadata_key(**base_args, use_sysname=False)
key_strip = get_cache_metadata_key(**base_args, strip_domain=True)
assert key_default != key_no_sysname
assert key_default != key_strip
# =============================================================================
# TestDeviceNameDetermination - 6 tests
# =============================================================================
class TestDeviceNameDetermination:
"""Test device name determination logic."""
def test_determine_device_name_prefers_sysname(self):
"""sysName should be preferred over hostname when use_sysname=True."""
from netbox_librenms_plugin.import_utils import _determine_device_name
device_data = {"sysName": "switch-01", "hostname": "switch-01.example.com"}
name = _determine_device_name(device_data, use_sysname=True)
assert name == "switch-01"
def test_determine_device_name_falls_back_to_hostname(self):
"""hostname used when sysName missing."""
from netbox_librenms_plugin.import_utils import _determine_device_name
device_data = {"hostname": "router-01.example.com"}
name = _determine_device_name(device_data, use_sysname=True)
assert name == "router-01.example.com"
def test_determine_device_name_strips_domain(self):
"""FQDN domain suffix should be stripped when strip_domain=True."""
from netbox_librenms_plugin.import_utils import _determine_device_name
device_data = {
"sysName": "router-core.datacenter.example.com",
"hostname": "10.0.0.1",
}
name = _determine_device_name(device_data, use_sysname=True, strip_domain=True)
assert name == "router-core"
def test_determine_device_name_handles_empty_sysname(self):
"""Empty sysName should fall back to hostname."""
from netbox_librenms_plugin.import_utils import _determine_device_name
device_data = {"sysName": "", "hostname": "fallback-host"}
name = _determine_device_name(device_data, use_sysname=True)
assert name == "fallback-host"
def test_determine_device_name_preserves_short_names(self):
"""Names without dots should remain unchanged."""
from netbox_librenms_plugin.import_utils import _determine_device_name
device_data = {"sysName": "shortname", "hostname": "192.168.1.1"}
name = _determine_device_name(device_data, use_sysname=True, strip_domain=True)
assert name == "shortname"
def test_determine_device_name_handles_ip_address(self):
"""IP addresses should not be stripped even with strip_domain=True."""
from netbox_librenms_plugin.import_utils import _determine_device_name
device_data = {"sysName": "192.168.1.1", "hostname": "192.168.1.1"}
name = _determine_device_name(device_data, use_sysname=True, strip_domain=True)
# IP addresses should not have domain stripped
assert name == "192.168.1.1"
def test_determine_device_name_fallback_to_device_id(self):
"""Fallback to device_id when no name available."""
from netbox_librenms_plugin.import_utils import _determine_device_name
device_data = {}
name = _determine_device_name(device_data, device_id=999)
assert name == "device-999"
# =============================================================================
# TestDeviceRetrieval - 10 tests
# =============================================================================
class TestDeviceRetrieval:
"""Test device retrieval and filtering functions."""
@patch("netbox_librenms_plugin.import_utils.filters.cache")
@patch("netbox_librenms_plugin.import_utils.filters.LibreNMSAPI")
def test_get_librenms_devices_for_import_success(self, mock_api_class, mock_cache):
"""Retrieve devices from LibreNMS API."""
mock_cache.get.return_value = None # Cache miss
mock_api = MagicMock()
mock_api.list_devices.return_value = (
True,
[
{"device_id": 1, "hostname": "switch-01"},
{"device_id": 2, "hostname": "switch-02"},
],
)
mock_api.cache_timeout = 300
from netbox_librenms_plugin.import_utils import get_librenms_devices_for_import
devices = get_librenms_devices_for_import(api=mock_api, filters={})
assert len(devices) == 2
assert devices[0]["hostname"] == "switch-01"
@patch("netbox_librenms_plugin.import_utils.filters.cache")
def test_get_librenms_devices_for_import_uses_cache(self, mock_cache):
"""Cached results returned on repeat call."""
cached_devices = [
{"device_id": 1, "hostname": "cached-device"},
]
mock_cache.get.return_value = cached_devices
mock_api = MagicMock()
from netbox_librenms_plugin.import_utils import get_librenms_devices_for_import
devices = get_librenms_devices_for_import(api=mock_api, filters={})
assert len(devices) == 1
assert devices[0]["hostname"] == "cached-device"
mock_api.list_devices.assert_not_called()
@patch("netbox_librenms_plugin.import_utils.filters.cache")
def test_get_librenms_devices_for_import_cache_miss(self, mock_cache):
"""API called when cache empty."""
mock_cache.get.return_value = None
mock_api = MagicMock()
mock_api.list_devices.return_value = (
True,
[
{"device_id": 3, "hostname": "fresh-device"},
],
)
mock_api.cache_timeout = 300
from netbox_librenms_plugin.import_utils import get_librenms_devices_for_import
devices = get_librenms_devices_for_import(api=mock_api, filters={}, force_refresh=True)
mock_api.list_devices.assert_called_once()
assert len(devices) == 1
@patch("netbox_librenms_plugin.import_utils.filters.cache")
def test_get_device_count_for_filters_success(self, mock_cache):
"""Returns correct count from API."""
mock_cache.get.return_value = [
{"device_id": 1, "hostname": "switch-01", "status": 1},
{"device_id": 2, "hostname": "switch-02", "status": 1},
{"device_id": 3, "hostname": "switch-03", "status": 0},
]
mock_api = MagicMock()
from netbox_librenms_plugin.import_utils import get_device_count_for_filters
count = get_device_count_for_filters(api=mock_api, filters={})
assert count == 3
@patch("netbox_librenms_plugin.import_utils.filters.cache")
def test_get_device_count_excludes_disabled(self, mock_cache):
"""Count respects show_disabled filter parameter: disabled==1 devices excluded."""
mock_cache.get.return_value = [
{"device_id": 1, "hostname": "switch-01", "disabled": 0, "status": 1},
{"device_id": 2, "hostname": "switch-02", "disabled": 0, "status": 0},
{"device_id": 3, "hostname": "switch-03", "disabled": 1, "status": 1}, # disabled in LibreNMS
]
mock_api = MagicMock()
from netbox_librenms_plugin.import_utils import get_device_count_for_filters
count = get_device_count_for_filters(api=mock_api, filters={}, show_disabled=False)
assert count == 2
def test_get_import_device_cache_key_default_server(self):
"""Generate cache key with explicit default server key."""
from netbox_librenms_plugin.import_utils import get_import_device_cache_key
key = get_import_device_cache_key(device_id=123, server_key="default")
assert "default" in key
assert "123" in key
def test_get_validated_device_cache_key_no_vc(self):
"""Generate cache key without VC enabled."""
from netbox_librenms_plugin.import_utils import get_validated_device_cache_key
key = get_validated_device_cache_key(server_key="default", filters={}, device_id=100, vc_enabled=False)
assert "novc" in key
def test_get_validated_device_cache_key_with_vc(self):
"""Generate cache key with VC enabled."""
from netbox_librenms_plugin.import_utils import get_validated_device_cache_key
key_vc = get_validated_device_cache_key(server_key="default", filters={}, device_id=100, vc_enabled=True)
key_novc = get_validated_device_cache_key(server_key="default", filters={}, device_id=100, vc_enabled=False)
# Keys should be different based on VC setting
assert key_vc != key_novc
def test_empty_virtual_chassis_data(self):
"""Empty VC data helper returns correct structure."""
from netbox_librenms_plugin.import_utils import empty_virtual_chassis_data
data = empty_virtual_chassis_data()
assert data["is_stack"] is False
assert data["member_count"] == 0
assert data["members"] == []
assert data["detection_error"] is None
@patch("netbox_librenms_plugin.import_utils.virtual_chassis.cache")
def test_get_virtual_chassis_data_returns_empty_without_api(self, mock_cache):
"""Get VC data returns empty structure without API."""
from netbox_librenms_plugin.import_utils import get_virtual_chassis_data
result = get_virtual_chassis_data(api=None, device_id=123)
assert result["is_stack"] is False
assert result["member_count"] == 0
# =============================================================================
# TestDeviceValidation - 15 tests
# =============================================================================
class TestDeviceValidation:
"""Test device validation for import."""
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_validate_device_site_match_found(
self,
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
):
"""Site matched successfully."""
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = None
mock_site = MagicMock(id=1, name="DC1")
mock_find_site.return_value = {
"found": True,
"site": mock_site,
"match_type": "exact",
"confidence": 1.0,
}
mock_find_platform.return_value = {
"found": False,
"platform": None,
"match_type": None,
}
mock_match_type.return_value = {
"matched": False,
"device_type": None,
"match_type": None,
}
mock_role.objects.all.return_value = []
mock_cluster.objects.all.return_value = []
mock_site_model.objects.all.return_value = [mock_site]
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
"location": "DC1",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["site"]["found"] is True
assert result["site"]["site"] == mock_site
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_validate_device_site_not_found(
self,
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
):
"""Site not found adds validation issue."""
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = None
mock_find_site.return_value = {
"found": False,
"site": None,
"match_type": None,
"confidence": 0.0,
}
mock_find_platform.return_value = {
"found": False,
"platform": None,
"match_type": None,
}
mock_match_type.return_value = {
"matched": False,
"device_type": None,
"match_type": None,
}
mock_role.objects.all.return_value = []
mock_cluster.objects.all.return_value = []
mock_site_model.objects.all.return_value = []
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
"location": "Unknown Location",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["site"]["found"] is False
assert any("site" in issue.lower() for issue in result["issues"])
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceType")
def test_validate_device_platform_match_found(
self,
mock_device_type,
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
):
"""Platform matched successfully."""
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = None
mock_device_type.objects.all.return_value = []
mock_find_site.return_value = {
"found": False,
"site": None,
"match_type": None,
"confidence": 0.0,
}
mock_platform = MagicMock(id=1, name="ios")
mock_find_platform.return_value = {
"found": True,
"platform": mock_platform,
"match_type": "exact",
}
mock_match_type.return_value = {
"matched": False,
"device_type": None,
"match_type": None,
}
mock_role.objects.all.return_value = []
mock_cluster.objects.all.return_value = []
mock_site_model.objects.all.return_value = []
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
"os": "ios",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["platform"]["found"] is True
assert result["platform"]["platform"] == mock_platform
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_validate_device_platform_not_found(
self,
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
):
"""Platform not found adds warning (not blocking)."""
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = None
mock_find_site.return_value = {
"found": False,
"site": None,
"match_type": None,
"confidence": 0.0,
}
mock_find_platform.return_value = {
"found": False,
"platform": None,
"match_type": None,
}
mock_match_type.return_value = {
"matched": False,
"device_type": None,
"match_type": None,
}
mock_role.objects.all.return_value = []
mock_cluster.objects.all.return_value = []
mock_site_model.objects.all.return_value = []
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
"os": "unknown_os",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["platform"]["found"] is False
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_validate_device_type_match_found(
self,
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
):
"""Device type matched successfully."""
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = None
mock_find_site.return_value = {
"found": False,
"site": None,
"match_type": None,
"confidence": 0.0,
}
mock_find_platform.return_value = {
"found": False,
"platform": None,
"match_type": None,
}
mock_dt = MagicMock(id=1, model="C9300-48P")
mock_match_type.return_value = {
"matched": True,
"device_type": mock_dt,
"match_type": "exact",
}
mock_role.objects.all.return_value = []
mock_cluster.objects.all.return_value = []
mock_site_model.objects.all.return_value = []
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
"hardware": "C9300-48P",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["device_type"]["found"] is True
assert result["device_type"]["device_type"] == mock_dt
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceType")
def test_validate_device_type_not_found(
self,
mock_device_type,
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
):
"""Device type not found adds validation issue."""
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = None
mock_device_type.objects.all.return_value = []
mock_find_site.return_value = {
"found": False,
"site": None,
"match_type": None,
"confidence": 0.0,
}
mock_find_platform.return_value = {
"found": False,
"platform": None,
"match_type": None,
}
mock_match_type.return_value = {
"matched": False,
"device_type": None,
"match_type": None,
}
mock_role.objects.all.return_value = []
mock_cluster.objects.all.return_value = []
mock_site_model.objects.all.return_value = []
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
"hardware": "Unknown Hardware",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["device_type"]["found"] is False
assert any("device type" in issue.lower() for issue in result["issues"])
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_validate_device_role_required(
self,
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
):
"""Missing role flagged as required."""
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = None
mock_site = MagicMock(id=1, name="DC1")
mock_find_site.return_value = {
"found": True,
"site": mock_site,
"match_type": "exact",
"confidence": 1.0,
}
mock_find_platform.return_value = {
"found": False,
"platform": None,
"match_type": None,
}
mock_dt = MagicMock(id=1, model="C9300-48P")
mock_match_type.return_value = {
"matched": True,
"device_type": mock_dt,
"match_type": "exact",
}
mock_role.objects.all.return_value = [MagicMock(id=1, name="Access Switch")]
mock_cluster.objects.all.return_value = []
mock_site_model.objects.all.return_value = [mock_site]
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
"location": "DC1",
"hardware": "C9300-48P",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["device_role"]["found"] is False
assert any("role" in issue.lower() for issue in result["issues"])
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_validate_device_handles_empty_location(
self,
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
):
"""Empty location handled gracefully."""
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = None
mock_find_site.return_value = {
"found": False,
"site": None,
"match_type": None,
"confidence": 0.0,
}
mock_find_platform.return_value = {
"found": False,
"platform": None,
"match_type": None,
}
mock_match_type.return_value = {
"matched": False,
"device_type": None,
"match_type": None,
}
mock_role.objects.all.return_value = []
mock_cluster.objects.all.return_value = []
mock_site_model.objects.all.return_value = []
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
"location": "",
}
# Should not raise exception
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result is not None
assert result["site"]["found"] is False
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_validate_device_handles_empty_os(
self,
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
):
"""Empty OS handled gracefully."""
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = None
mock_find_site.return_value = {
"found": False,
"site": None,
"match_type": None,
"confidence": 0.0,
}
mock_find_platform.return_value = {
"found": False,
"platform": None,
"match_type": None,
}
mock_match_type.return_value = {
"matched": False,
"device_type": None,
"match_type": None,
}
mock_role.objects.all.return_value = []
mock_cluster.objects.all.return_value = []
mock_site_model.objects.all.return_value = []
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
"os": "",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result is not None
assert result["platform"]["found"] is False
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceType")
def test_validate_device_handles_empty_hardware(
self,
mock_device_type,
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
):
"""Empty hardware handled gracefully."""
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = None
mock_device_type.objects.all.return_value = []
mock_find_site.return_value = {
"found": False,
"site": None,
"match_type": None,
"confidence": 0.0,
}
mock_find_platform.return_value = {
"found": False,
"platform": None,
"match_type": None,
}
mock_match_type.return_value = {
"matched": False,
"device_type": None,
"match_type": None,
}
mock_role.objects.all.return_value = []
mock_cluster.objects.all.return_value = []
mock_site_model.objects.all.return_value = []
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
"hardware": "",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result is not None
assert result["device_type"]["found"] is False
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_validate_device_duplicate_detection(
self,
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
):
"""Existing device detected."""
existing_device = MagicMock()
existing_device.name = "switch-01"
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = existing_device
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["existing_device"] == existing_device
assert result["can_import"] is False
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_validate_device_returns_complete_state(
self,
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
):
"""All expected fields in result."""
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = None
mock_find_site.return_value = {
"found": False,
"site": None,
"match_type": None,
"confidence": 0.0,
}
mock_find_platform.return_value = {
"found": False,
"platform": None,
"match_type": None,
}
mock_match_type.return_value = {
"matched": False,
"device_type": None,
"match_type": None,
}
mock_role.objects.all.return_value = []
mock_cluster.objects.all.return_value = []
mock_site_model.objects.all.return_value = []
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
# Check all expected keys exist
assert "is_ready" in result
assert "can_import" in result
assert "import_as_vm" in result
assert "existing_device" in result
assert "issues" in result
assert "warnings" in result
assert "site" in result
assert "device_type" in result
assert "device_role" in result
assert "cluster" in result
assert "platform" in result
@patch("netbox_librenms_plugin.import_utils.device_operations.cache")
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_validate_device_import_as_vm(
self,
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
mock_cache,
):
"""Import as VM mode uses cluster instead of site/device_type."""
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = None
mock_find_site.return_value = {
"found": False,
"site": None,
"match_type": None,
"confidence": 0.0,
}
mock_find_platform.return_value = {
"found": False,
"platform": None,
"match_type": None,
}
mock_match_type.return_value = {
"matched": False,
"device_type": None,
"match_type": None,
}
mock_role.objects.all.return_value = []
mock_clusters = [MagicMock(id=1, name="VMware Cluster")]
mock_cluster.objects.all.return_value = mock_clusters
mock_cache.get.return_value = None # Force cache miss to trigger Cluster.objects.all()
mock_site_model.objects.all.return_value = []
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "vm-01",
}
result = validate_device_for_import(device_data, import_as_vm=True, include_vc_detection=False)
assert result["import_as_vm"] is True
assert result["cluster"]["available_clusters"] == mock_clusters
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_validate_device_existing_vm_blocks_import(
self,
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
):
"""Existing VM detection blocks import."""
existing_vm = MagicMock()
existing_vm.name = "vm-01"
mock_vm.objects.filter.return_value.first.return_value = existing_vm
mock_device.objects.filter.return_value.first.return_value = None
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "vm-01",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["existing_device"] == existing_vm
assert result["can_import"] is False
assert result["import_as_vm"] is True
class TestDeviceNamingPreferencesLegacy:
"""Test that validation honours use_sysname and strip_domain user preferences."""
COMMON_PATCHES = [
"netbox_librenms_plugin.import_utils.device_operations.Site",
"netbox_librenms_plugin.import_utils.device_operations.Rack",
"netbox_librenms_plugin.import_utils.device_operations.Cluster",
"netbox_librenms_plugin.import_utils.device_operations.DeviceRole",
"netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type",
"netbox_librenms_plugin.import_utils.device_operations.find_matching_platform",
"netbox_librenms_plugin.import_utils.device_operations.find_matching_site",
"netbox_librenms_plugin.import_utils.device_operations.Device",
"virtualization.models.VirtualMachine",
]
def _setup_no_existing(self, mocks):
"""Configure mocks so no existing device is found."""
mock_vm = mocks[-1] # VirtualMachine
mock_device = mocks[-2] # Device
mock_find_site = mocks[-3]
mock_find_platform = mocks[-4]
mock_match_type = mocks[-5]
mock_role = mocks[-6]
mock_rack = mocks[-8]
mock_site_model = mocks[-9]
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = None
mock_find_site.return_value = {
"found": False,
"site": None,
"match_type": None,
"confidence": 0.0,
}
mock_find_platform.return_value = {
"found": False,
"platform": None,
"match_type": None,
}
mock_match_type.return_value = {
"matched": False,
"device_type": None,
"match_type": None,
}
mock_role.objects.all.return_value = []
mock_rack.objects.filter.return_value = []
mock_site_model.objects.all.return_value = []
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_resolved_name_uses_sysname_by_default(self, *mocks):
"""Default use_sysname=True uses sysName for resolved_name."""
self._setup_no_existing(mocks)
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "10.0.0.1",
"sysName": "core-switch",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["resolved_name"] == "core-switch"
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_resolved_name_uses_hostname_when_sysname_disabled(self, *mocks):
"""use_sysname=False uses hostname for resolved_name."""
self._setup_no_existing(mocks)
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "10.0.0.1",
"sysName": "core-switch",
}
result = validate_device_for_import(
device_data,
include_vc_detection=False,
use_sysname=False,
)
assert result["resolved_name"] == "10.0.0.1"
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_resolved_name_strips_domain(self, *mocks):
"""strip_domain=True strips the domain suffix."""
self._setup_no_existing(mocks)
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01.example.com",
"sysName": "switch-01.example.com",
}
result = validate_device_for_import(
device_data,
include_vc_detection=False,
strip_domain=True,
)
assert result["resolved_name"] == "switch-01"
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_duplicate_detection_uses_resolved_name(self, *mocks):
"""Duplicate detection should match against the resolved name, not raw hostname."""
self._setup_no_existing(mocks)
mock_device = mocks[-2] # Device
# The first filter call (librenms_id) returns None,
# the second filter call (name__iexact) returns the existing device.
existing = MagicMock()
existing.name = "core-switch"
existing.serial = ""
mock_device.objects.filter.return_value.first.side_effect = [None, existing]
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 999,
"hostname": "10.0.0.1",
"sysName": "core-switch",
}
# use_sysname=True (default): resolved name is "core-switch"
# so duplicate detection should find existing device "core-switch"
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["existing_device"] == existing
assert result["existing_match_type"] == "hostname"
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_backward_compatible_defaults(self, *mocks):
"""Calling without naming params produces resolved_name in result."""
self._setup_no_existing(mocks)
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
# resolved_name should be present and match sysName fallback to hostname
assert "resolved_name" in result
assert result["resolved_name"] == "switch-01"
class TestNameMatchesWithNamingPreferencesLegacy:
"""
Test that name_matches/name_sync_available respect naming preferences and VC patterns.
The name comparison should use the resolved name (result of _determine_device_name())
which accounts for use_sysname and strip_domain, not the raw LibreNMS sysName.
For VC members, it should also account for the VC naming pattern.
"""
COMMON_PATCHES = [
"netbox_librenms_plugin.import_utils.device_operations.Site",
"netbox_librenms_plugin.import_utils.device_operations.Rack",
"netbox_librenms_plugin.import_utils.device_operations.Cluster",
"netbox_librenms_plugin.import_utils.device_operations.DeviceRole",
"netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type",
"netbox_librenms_plugin.import_utils.device_operations.find_matching_platform",
"netbox_librenms_plugin.import_utils.device_operations.find_matching_site",
"netbox_librenms_plugin.import_utils.device_operations.Device",
"virtualization.models.VirtualMachine",
]
def _start_patches(self):
"""Start all common patches and return mocks in standard order."""
self._patchers = [patch(p) for p in self.COMMON_PATCHES]
mocks = [p.start() for p in self._patchers]
(
self.mock_site_model,
self.mock_rack,
self.mock_cluster,
self.mock_role,
self.mock_match_type,
self.mock_find_platform,
self.mock_find_site,
self.mock_device,
self.mock_vm,
) = mocks
def _stop_patches(self):
"""Stop all patches."""
for p in self._patchers:
p.stop()
def _configure_standard_mocks(self):
"""Configure standard mock returns for site/platform/type/role."""
self.mock_find_site.return_value = {
"found": True,
"site": MagicMock(),
"match_type": "exact",
"confidence": 1.0,
}
self.mock_find_platform.return_value = {"found": False, "platform": None, "match_type": None}
self.mock_match_type.return_value = {"matched": False, "device_type": None, "match_type": None}
self.mock_role.objects.all.return_value = []
self.mock_cluster.objects.all.return_value = []
self.mock_rack.objects.filter.return_value = []
self.mock_site_model.objects.all.return_value = []
def _setup_librenms_id_match(self, existing_device, as_vm=False):
"""Configure mocks so that a device is found by librenms_id.
Uses a Q-aware side_effect so only filter() calls targeting a
``librenms_id`` field return the existing device; other filter() calls
(e.g. name lookups, serial lookups) return an empty queryset.
"""
from unittest.mock import MagicMock
def _librenms_id_filter_side_effect(hit):
def side_effect(*args, **kwargs):
mock_qs = MagicMock()
# Match when the first positional arg is a Q that references librenms_id
if args:
q = args[0]
if hasattr(q, "children") and any(
isinstance(child, tuple) and "librenms_id" in child[0] for child in q.children
):
mock_qs.first.return_value = hit
return mock_qs
mock_qs.first.return_value = None
return mock_qs
return side_effect
if as_vm:
self.mock_vm.objects.filter.side_effect = _librenms_id_filter_side_effect(existing_device)
self.mock_device.objects.filter.side_effect = _librenms_id_filter_side_effect(None)
else:
self.mock_device.objects.filter.side_effect = _librenms_id_filter_side_effect(existing_device)
self.mock_vm.objects.filter.side_effect = _librenms_id_filter_side_effect(None)
def setup_method(self):
"""Set up common patches."""
self._start_patches()
def teardown_method(self):
"""Tear down patches."""
self._stop_patches()
def test_name_matches_with_strip_domain(self):
"""strip_domain=True: FQDN in LibreNMS matches short name in NetBox."""
existing = MagicMock()
existing.name = "router"
existing.serial = ""
existing.virtual_chassis = None
existing.vc_position = None
self._setup_librenms_id_match(existing)
self._configure_standard_mocks()
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "router.example.com",
"sysName": "router.example.com",
}
result = validate_device_for_import(
device_data,
include_vc_detection=False,
strip_domain=True,
)
assert result["existing_match_type"] == "librenms_id"
assert result["name_matches"] is True
assert result["name_sync_available"] is False
def test_name_matches_uses_hostname_when_sysname_disabled(self):
"""use_sysname=False: matches against hostname instead of sysName."""
existing = MagicMock()
existing.name = "10.0.0.1"
existing.serial = ""
existing.virtual_chassis = None
existing.vc_position = None
self._setup_librenms_id_match(existing)
self._configure_standard_mocks()
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "10.0.0.1",
"sysName": "core-switch",
}
result = validate_device_for_import(
device_data,
include_vc_detection=False,
use_sysname=False,
)
assert result["name_matches"] is True
assert result["name_sync_available"] is False
def test_name_mismatch_offers_sync_with_resolved_name(self):
"""When names don't match, suggested_name is the resolved name, not raw sysName."""
existing = MagicMock()
existing.name = "old-device"
existing.serial = ""
existing.virtual_chassis = None
existing.vc_position = None
self._setup_librenms_id_match(existing)
self._configure_standard_mocks()
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "new-switch.example.com",
"sysName": "new-switch.example.com",
}
result = validate_device_for_import(
device_data,
include_vc_detection=False,
strip_domain=True,
)
assert result["name_sync_available"] is True
# suggested_name should be the resolved (stripped) name, not raw sysName
assert result["suggested_name"] == "new-switch"
@patch("netbox_librenms_plugin.import_utils.device_operations._generate_vc_member_name")
def test_name_matches_vc_member(self, mock_vc_name):
"""VC member: name matches when existing device name matches generated VC name."""
mock_vc_name.return_value = "switch-M2"
existing = MagicMock()
existing.name = "switch-M2"
existing.serial = "SN123"
existing.virtual_chassis = MagicMock() # Not None → device is a VC member
existing.vc_position = 2
self._setup_librenms_id_match(existing)
self._configure_standard_mocks()
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch",
"sysName": "switch",
}
result = validate_device_for_import(
device_data,
include_vc_detection=False,
)
assert result["name_matches"] is True
assert result["name_sync_available"] is False
# _generate_vc_member_name should be called with resolved name, position, serial
mock_vc_name.assert_called_with("switch", 2, serial="SN123")
@patch("netbox_librenms_plugin.import_utils.device_operations._generate_vc_member_name")
def test_name_matches_vc_member_with_strip_domain(self, mock_vc_name):
"""VC member + strip_domain: FQDN resolved to short name matches VC pattern."""
mock_vc_name.return_value = "siteA-9300-1 (2)"
existing = MagicMock()
existing.name = "siteA-9300-1 (2)"
existing.serial = "SN456"
existing.virtual_chassis = MagicMock()
existing.vc_position = 2
self._setup_librenms_id_match(existing)
self._configure_standard_mocks()
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 555,
"hostname": "siteA-9300-1.example.net.com",
"sysName": "siteA-9300-1.example.net.com",
}
result = validate_device_for_import(
device_data,
include_vc_detection=False,
strip_domain=True,
)
assert result["name_matches"] is True
assert result["name_sync_available"] is False
# Resolved name should be "siteA-9300-1" (stripped), then VC name generated
mock_vc_name.assert_called_with("siteA-9300-1", 2, serial="SN456")
@patch("netbox_librenms_plugin.import_utils.device_operations._generate_vc_member_name")
def test_vc_member_name_mismatch_suggests_vc_name(self, mock_vc_name):
"""VC member name mismatch: suggested_name should be the expected VC name."""
mock_vc_name.return_value = "new-switch-M2"
existing = MagicMock()
existing.name = "old-switch-M2"
existing.serial = "SN789"
existing.virtual_chassis = MagicMock()
existing.vc_position = 2
self._setup_librenms_id_match(existing)
self._configure_standard_mocks()
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "new-switch",
"sysName": "new-switch",
}
result = validate_device_for_import(
device_data,
include_vc_detection=False,
)
assert result["name_matches"] is False
assert result["name_sync_available"] is True
assert result["suggested_name"] == "new-switch-M2"
def test_vm_name_matches_with_strip_domain(self):
"""VM name comparison also uses resolved name, not raw sysName."""
existing_vm = MagicMock()
existing_vm.name = "vm-server"
self._setup_librenms_id_match(existing_vm, as_vm=True)
self._configure_standard_mocks()
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "vm-server.example.com",
"sysName": "vm-server.example.com",
}
result = validate_device_for_import(
device_data,
include_vc_detection=False,
strip_domain=True,
)
assert result["import_as_vm"] is True
assert result["name_matches"] is True
def test_name_matches_exact_without_vc(self):
"""Standalone device: exact name match works without VC check."""
existing = MagicMock()
existing.name = "core-router"
existing.serial = ""
existing.virtual_chassis = None
existing.vc_position = None
self._setup_librenms_id_match(existing)
self._configure_standard_mocks()
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "core-router",
"sysName": "core-router",
}
result = validate_device_for_import(
device_data,
include_vc_detection=False,
)
assert result["name_matches"] is True
assert result["name_sync_available"] is False
class TestSerialNumberMatching:
"""Test serial number matching in device validation."""
SERIAL_PATCHES = [
"netbox_librenms_plugin.import_utils.device_operations.Site",
"netbox_librenms_plugin.import_utils.device_operations.Rack",
"netbox_librenms_plugin.import_utils.device_operations.Cluster",
"netbox_librenms_plugin.import_utils.device_operations.DeviceRole",
"netbox_librenms_plugin.import_utils.device_operations.DeviceType",
"netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type",
"netbox_librenms_plugin.import_utils.device_operations.find_matching_platform",
"netbox_librenms_plugin.import_utils.device_operations.find_matching_site",
"netbox_librenms_plugin.import_utils.device_operations.Device",
"virtualization.models.VirtualMachine",
]
def _start_patches(self):
"""Start all common patches and return mocks in standard order."""
self._patchers = [patch(p) for p in self.SERIAL_PATCHES]
mocks = [p.start() for p in self._patchers]
(
self.mock_site_model,
self.mock_rack,
self.mock_cluster,
self.mock_role,
self.mock_device_type,
self.mock_match_type,
self.mock_find_platform,
self.mock_find_site,
self.mock_device,
self.mock_vm,
) = mocks
self.mock_device_type.objects.all.return_value = []
def _stop_patches(self):
"""Stop all patches."""
for p in self._patchers:
p.stop()
def setup_method(self):
"""Set up common patches for serial number tests."""
self._start_patches()
def teardown_method(self):
"""Tear down patches."""
self._stop_patches()
def test_serial_match_blocks_import(self):
"""Device with matching serial blocks import."""
existing = MagicMock()
existing.name = "existing-device"
existing.serial = "ABC123"
self.mock_vm.objects.filter.return_value.first.return_value = None
def device_filter(*args, **kwargs):
result = MagicMock()
if "serial" in kwargs:
result.first.return_value = existing
else:
result.first.return_value = None
return result
self.mock_device.objects.filter.side_effect = device_filter
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {"device_id": 1, "hostname": "new-hostname", "serial": "ABC123"}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["can_import"] is False
assert result["existing_match_type"] == "serial"
assert result["existing_device"] == existing
def test_serial_match_same_hostname_offers_link(self):
"""Serial + hostname match offers link action."""
existing = MagicMock()
existing.name = "switch-01"
existing.serial = "ABC123"
self.mock_vm.objects.filter.return_value.first.return_value = None
def device_filter(*args, **kwargs):
result = MagicMock()
if "serial" in kwargs:
result.first.return_value = existing
else:
result.first.return_value = None
return result
self.mock_device.objects.filter.side_effect = device_filter
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {"device_id": 1, "hostname": "switch-01", "serial": "ABC123"}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["serial_action"] == "link"
assert result["existing_match_type"] == "serial"
assert "not linked to LibreNMS" in result["warnings"][0]
def test_serial_match_diff_hostname_offers_hostname_differs(self):
"""Serial matches but hostname differs offers hostname_differs action."""
existing = MagicMock()
existing.name = "old-hostname"
existing.serial = "ABC123"
self.mock_vm.objects.filter.return_value.first.return_value = None
def device_filter(*args, **kwargs):
result = MagicMock()
if "serial" in kwargs:
result.first.return_value = existing
else:
result.first.return_value = None
return result
self.mock_device.objects.filter.side_effect = device_filter
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {"device_id": 1, "hostname": "new-hostname", "serial": "ABC123"}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["serial_action"] == "hostname_differs"
assert result["existing_match_type"] == "serial"
assert "hostname differs" in result["warnings"][0]
def test_hostname_match_diff_serial_offers_update(self):
"""Hostname matches but serial differs offers update_serial action."""
existing = MagicMock()
existing.name = "switch-01"
existing.serial = "OLD_SERIAL"
self.mock_vm.objects.filter.return_value.first.return_value = None
def device_filter(*args, **kwargs):
result = MagicMock()
if "name__iexact" in kwargs:
result.first.return_value = existing
elif "serial" in kwargs:
result.first.return_value = None
result.exclude.return_value.first.return_value = None
else:
result.first.return_value = None
return result
self.mock_device.objects.filter.side_effect = device_filter
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {"device_id": 1, "hostname": "switch-01", "serial": "NEW_SERIAL"}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["serial_action"] == "update_serial"
assert result["existing_match_type"] == "hostname"
assert "Hardware may have been replaced" in result["warnings"][0]
def _setup_no_match_mocks(self):
"""Configure mocks for tests where no device match is expected."""
self.mock_vm.objects.filter.return_value.first.return_value = None
self.mock_device.objects.filter.return_value.first.return_value = None
self.mock_find_site.return_value = {"found": False, "site": None, "match_type": None, "confidence": 0}
self.mock_find_platform.return_value = {"found": False, "platform": None, "match_type": None}
self.mock_match_type.return_value = {"matched": False, "device_type": None, "match_type": None}
self.mock_role.objects.all.return_value = []
self.mock_cluster.objects.all.return_value = []
self.mock_site_model.objects.all.return_value = []
def test_serial_dash_ignored(self):
"""Serial '-' is not treated as a match."""
self._setup_no_match_mocks()
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {"device_id": 1, "hostname": "switch-01", "serial": "-"}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["existing_match_type"] is None
assert result["serial_action"] is None
def test_serial_empty_ignored(self):
"""Empty serial skips serial matching."""
self._setup_no_match_mocks()
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {"device_id": 1, "hostname": "switch-01", "serial": ""}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["existing_match_type"] is None
assert result["serial_action"] is None
def test_serial_none_ignored(self):
"""None serial skips serial matching."""
self._setup_no_match_mocks()
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {"device_id": 1, "hostname": "switch-01", "serial": None}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["existing_match_type"] is None
assert result["serial_action"] is None
def test_hostname_match_serial_conflict_warns(self):
"""Hostname matches, incoming serial already on another device warns about conflict."""
hostname_device = MagicMock()
hostname_device.name = "switch-01"
hostname_device.serial = "OLD_SERIAL"
serial_conflict_device = MagicMock()
serial_conflict_device.name = "other-device"
self.mock_vm.objects.filter.return_value.first.return_value = None
def device_filter(*args, **kwargs):
result = MagicMock()
if "name__iexact" in kwargs:
result.first.return_value = hostname_device
elif "serial" in kwargs:
result.exclude.return_value.first.return_value = serial_conflict_device
else:
result.first.return_value = None
return result
self.mock_device.objects.filter.side_effect = device_filter
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {"device_id": 1, "hostname": "switch-01", "serial": "CONFLICTING_SERIAL"}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["serial_action"] == "conflict"
assert result["existing_match_type"] == "hostname"
assert "Serial conflict" in result["warnings"][0]
def test_librenms_id_match_shows_serial_confirmed(self):
"""librenms_id match with matching serial shows confirmation."""
existing = MagicMock()
existing.name = "switch-01"
existing.serial = "ABC123"
existing.virtual_chassis = None # Not a VC member → use plain hostname comparison
existing.vc_position = None
self.mock_vm.objects.filter.return_value.first.return_value = None
def device_filter(*args, **kwargs):
result = MagicMock()
q_has_librenms = any("librenms_id" in str(arg) for arg in args) or any(
k.startswith("custom_field_data__librenms_id") for k in kwargs
)
if q_has_librenms:
result.first.return_value = existing
else:
result.first.return_value = None
return result
self.mock_device.objects.filter.side_effect = device_filter
self.mock_find_site.return_value = {
"found": True,
"site": MagicMock(),
"match_type": "exact",
"confidence": 1.0,
}
self.mock_find_platform.return_value = {"found": False, "platform": None, "match_type": None}
self.mock_match_type.return_value = {"matched": False, "device_type": None, "match_type": None}
self.mock_role.objects.all.return_value = []
self.mock_cluster.objects.all.return_value = []
self.mock_site_model.objects.all.return_value = []
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {"device_id": 1, "hostname": "switch-01", "sysName": "switch-01", "serial": "ABC123"}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["existing_match_type"] == "librenms_id"
assert result["can_import"] is False
assert result["serial_confirmed"] is True
assert result["name_matches"] is True
def test_librenms_id_match_detects_serial_drift(self):
"""librenms_id match with different serial warns about drift."""
existing = MagicMock()
existing.name = "switch-01"
existing.serial = "OLD_SERIAL"
existing.virtual_chassis = None
existing.vc_position = None
self.mock_vm.objects.filter.return_value.first.return_value = None
def device_filter(*args, **kwargs):
result = MagicMock()
q_has_librenms = any("librenms_id" in str(arg) for arg in args) or any(
k.startswith("custom_field_data__librenms_id") for k in kwargs
)
if q_has_librenms:
result.first.return_value = existing
elif "serial" in kwargs:
result.first.return_value = None
result.exclude.return_value.first.return_value = None
else:
result.first.return_value = None
return result
self.mock_device.objects.filter.side_effect = device_filter
self.mock_find_site.return_value = {
"found": True,
"site": MagicMock(),
"match_type": "exact",
"confidence": 1.0,
}
self.mock_find_platform.return_value = {"found": False, "platform": None, "match_type": None}
self.mock_match_type.return_value = {"matched": False, "device_type": None, "match_type": None}
self.mock_role.objects.all.return_value = []
self.mock_cluster.objects.all.return_value = []
self.mock_site_model.objects.all.return_value = []
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {"device_id": 1, "hostname": "switch-01", "serial": "NEW_SERIAL"}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["existing_match_type"] == "librenms_id"
assert result["serial_action"] == "update_serial"
assert any("Hardware may have been replaced" in w for w in result["warnings"])
def test_librenms_id_match_still_validates_site(self):
"""librenms_id match continues to populate site/type validation."""
existing = MagicMock()
existing.name = "switch-01"
existing.serial = ""
existing.virtual_chassis = None
existing.vc_position = None
self.mock_vm.objects.filter.return_value.first.return_value = None
def device_filter(*args, **kwargs):
result = MagicMock()
q_has_librenms = any("librenms_id" in str(arg) for arg in args) or any(
k.startswith("custom_field_data__librenms_id") for k in kwargs
)
if q_has_librenms:
result.first.return_value = existing
else:
result.first.return_value = None
return result
self.mock_device.objects.filter.side_effect = device_filter
mock_site = MagicMock(id=1, name="DC1")
self.mock_find_site.return_value = {"found": True, "site": mock_site, "match_type": "exact", "confidence": 1.0}
self.mock_find_platform.return_value = {"found": False, "platform": None, "match_type": None}
mock_dt = MagicMock()
self.mock_match_type.return_value = {"matched": True, "device_type": mock_dt, "match_type": "exact"}
self.mock_role.objects.all.return_value = []
self.mock_cluster.objects.all.return_value = []
self.mock_site_model.objects.all.return_value = []
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {"device_id": 1, "hostname": "switch-01", "location": "DC1", "hardware": "WS-C4900M"}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["existing_match_type"] == "librenms_id"
assert result["can_import"] is False
assert result["is_ready"] is False
# Site and device_type should still be populated
assert result["site"]["found"] is True
assert result["site"]["site"] == mock_site
assert result["device_type"]["found"] is True
def test_existing_device_role_populated(self):
"""Existing device's role should be shown in validation details."""
existing = MagicMock()
existing.name = "switch-01"
existing.serial = "ABC123"
existing.virtual_chassis = None
existing.vc_position = None
mock_existing_role = MagicMock()
mock_existing_role.name = "Access Switch"
existing.role = mock_existing_role
self.mock_vm.objects.filter.return_value.first.return_value = None
def device_filter(*args, **kwargs):
result = MagicMock()
if "serial" in kwargs:
result.first.return_value = existing
else:
result.first.return_value = None
return result
self.mock_device.objects.filter.side_effect = device_filter
self.mock_find_site.return_value = {"found": False, "site": None, "match_type": None, "confidence": 0}
self.mock_find_platform.return_value = {"found": False, "platform": None, "match_type": None}
self.mock_match_type.return_value = {"matched": True, "device_type": MagicMock(), "match_type": "exact"}
self.mock_role.objects.all.return_value = [mock_existing_role]
self.mock_cluster.objects.all.return_value = []
self.mock_site_model.objects.all.return_value = []
with patch("netbox_librenms_plugin.import_utils.device_operations.cache") as mock_cache:
mock_cache.get.return_value = None
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
"serial": "ABC123",
"location": "",
"hardware": "",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["existing_device"] == existing
assert result["device_role"]["found"] is True
assert result["device_role"]["role"] == mock_existing_role
def test_device_type_mismatch_flagged(self):
"""Device type mismatch between existing device and LibreNMS should be flagged."""
existing = MagicMock()
existing.name = "switch-01"
existing.serial = "ABC123"
existing.virtual_chassis = None
existing.vc_position = None
existing_device_type = MagicMock()
existing_device_type.pk = 1
existing_device_type.__str__ = lambda self: "Old Type"
existing.device_type = existing_device_type
existing.role = MagicMock()
librenms_device_type = MagicMock()
librenms_device_type.pk = 2
librenms_device_type.__str__ = lambda self: "New Type"
self.mock_vm.objects.filter.return_value.first.return_value = None
def device_filter(*args, **kwargs):
result = MagicMock()
if "serial" in kwargs:
result.first.return_value = existing
else:
result.first.return_value = None
return result
self.mock_device.objects.filter.side_effect = device_filter
self.mock_find_site.return_value = {"found": False, "site": None, "match_type": None, "confidence": 0}
self.mock_find_platform.return_value = {"found": False, "platform": None, "match_type": None}
self.mock_match_type.return_value = {
"matched": True,
"device_type": librenms_device_type,
"match_type": "exact",
}
self.mock_role.objects.all.return_value = []
self.mock_cluster.objects.all.return_value = []
self.mock_site_model.objects.all.return_value = []
with patch("netbox_librenms_plugin.import_utils.device_operations.cache") as mock_cache:
mock_cache.get.return_value = None
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
"serial": "ABC123",
"location": "",
"hardware": "New Type",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["device_type_mismatch"] is True
assert any("Device type mismatch" in w for w in result["warnings"])
def test_no_device_type_mismatch_when_types_match(self):
"""No mismatch flag when existing device type matches LibreNMS."""
existing = MagicMock()
existing.name = "switch-01"
existing.serial = "ABC123"
existing.virtual_chassis = None
existing.vc_position = None
same_device_type = MagicMock()
same_device_type.pk = 1
existing.device_type = same_device_type
existing.role = MagicMock()
self.mock_vm.objects.filter.return_value.first.return_value = None
def device_filter(*args, **kwargs):
result = MagicMock()
if "serial" in kwargs:
result.first.return_value = existing
else:
result.first.return_value = None
return result
self.mock_device.objects.filter.side_effect = device_filter
self.mock_find_site.return_value = {"found": False, "site": None, "match_type": None, "confidence": 0}
self.mock_find_platform.return_value = {"found": False, "platform": None, "match_type": None}
self.mock_match_type.return_value = {"matched": True, "device_type": same_device_type, "match_type": "exact"}
self.mock_role.objects.all.return_value = []
self.mock_cluster.objects.all.return_value = []
self.mock_site_model.objects.all.return_value = []
with patch("netbox_librenms_plugin.import_utils.device_operations.cache") as mock_cache:
mock_cache.get.return_value = None
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
"serial": "ABC123",
"location": "",
"hardware": "Same Type",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["device_type_mismatch"] is False
class TestNameMatchesWithNamingPreferences:
"""Test VC-aware name matching with use_sysname/strip_domain preferences."""
PATCHES = [
"netbox_librenms_plugin.import_utils.device_operations.Site",
"netbox_librenms_plugin.import_utils.device_operations.Rack",
"netbox_librenms_plugin.import_utils.device_operations.Cluster",
"netbox_librenms_plugin.import_utils.device_operations.DeviceRole",
"netbox_librenms_plugin.import_utils.device_operations.DeviceType",
"netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type",
"netbox_librenms_plugin.import_utils.device_operations.find_matching_platform",
"netbox_librenms_plugin.import_utils.device_operations.find_matching_site",
"netbox_librenms_plugin.import_utils.device_operations.Device",
"virtualization.models.VirtualMachine",
]
def setup_method(self):
self._patchers = [patch(p) for p in self.PATCHES]
mocks = [p.start() for p in self._patchers]
(
self.mock_site,
self.mock_rack,
self.mock_cluster,
self.mock_role,
self.mock_device_type,
self.mock_match_type,
self.mock_find_platform,
self.mock_find_site,
self.mock_device,
self.mock_vm,
) = mocks
self.mock_device_type.objects.all.return_value = []
self.mock_vm.objects.filter.return_value.first.return_value = None
self.mock_find_site.return_value = {
"found": True,
"site": MagicMock(),
"match_type": "exact",
"confidence": 1.0,
}
self.mock_find_platform.return_value = {"found": False, "platform": None, "match_type": None}
self.mock_match_type.return_value = {"matched": False, "device_type": None, "match_type": None}
self.mock_role.objects.all.return_value = []
self.mock_cluster.objects.all.return_value = []
self.mock_site.objects.all.return_value = []
def teardown_method(self):
for p in self._patchers:
p.stop()
def _make_existing(self, name, serial="SN001", virtual_chassis=None, vc_position=None):
existing = MagicMock()
existing.name = name
existing.serial = serial
existing.virtual_chassis = virtual_chassis
existing.vc_position = vc_position
existing.custom_field_data = {"librenms_id": {"default": 42}}
return existing
def _setup_librenms_id_filter(self, existing):
def device_filter(*args, **kwargs):
result = MagicMock()
q_has_librenms = any("librenms_id" in str(arg) for arg in args) or any(
k.startswith("custom_field_data__librenms_id") for k in kwargs
)
result.first.return_value = existing if q_has_librenms else None
result.exclude.return_value.first.return_value = None
return result
self.mock_device.objects.filter.side_effect = device_filter
def test_strip_domain_name_matches(self):
"""strip_domain=True resolves 'switch-01.example.com' to 'switch-01', matching existing device."""
from netbox_librenms_plugin.import_utils import validate_device_for_import
existing = self._make_existing("switch-01")
self._setup_librenms_id_filter(existing)
device_data = {
"device_id": 42,
"hostname": "switch-01.example.com",
"sysName": "switch-01.example.com",
"serial": "SN001",
}
result = validate_device_for_import(device_data, include_vc_detection=False, strip_domain=True)
assert result["name_matches"] is True
assert result["name_sync_available"] is False
def test_sysname_disabled_uses_hostname(self):
"""use_sysname=False falls back to hostname for name comparison."""
from netbox_librenms_plugin.import_utils import validate_device_for_import
existing = self._make_existing("switch-hostname")
self._setup_librenms_id_filter(existing)
device_data = {
"device_id": 42,
"hostname": "switch-hostname",
"sysName": "switch-sysname",
"serial": "SN001",
}
result = validate_device_for_import(device_data, include_vc_detection=False, use_sysname=False)
assert result["name_matches"] is True
assert result["resolved_name"] == "switch-hostname"
def test_name_mismatch_offers_sync(self):
"""When resolved name differs from existing device name, name_sync_available is set."""
from netbox_librenms_plugin.import_utils import validate_device_for_import
existing = self._make_existing("old-name")
self._setup_librenms_id_filter(existing)
device_data = {
"device_id": 42,
"hostname": "new-name",
"sysName": "new-name",
"serial": "SN001",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["name_matches"] is False
assert result["name_sync_available"] is True
assert result["suggested_name"] == "new-name"
def test_vc_member_name_matches(self):
"""Existing VC member name is compared against vc_member_name(hostname, vc_position)."""
from netbox_librenms_plugin.import_utils import validate_device_for_import
from netbox_librenms_plugin.import_utils.virtual_chassis import _generate_vc_member_name
mock_vc = MagicMock()
expected_name = _generate_vc_member_name("stack-master", 2, serial="SN001")
existing = self._make_existing(expected_name, serial="SN001", virtual_chassis=mock_vc, vc_position=2)
self._setup_librenms_id_filter(existing)
device_data = {
"device_id": 42,
"hostname": "stack-master",
"sysName": "stack-master",
"serial": "SN001",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["name_matches"] is True
def test_vc_member_name_mismatch_suggests_vc_name(self):
"""When VC member name differs, suggested_name is the expected VC member name."""
from netbox_librenms_plugin.import_utils import validate_device_for_import
from netbox_librenms_plugin.import_utils.virtual_chassis import _generate_vc_member_name
mock_vc = MagicMock()
existing = self._make_existing("wrong-name", serial="SN001", virtual_chassis=mock_vc, vc_position=2)
self._setup_librenms_id_filter(existing)
device_data = {
"device_id": 42,
"hostname": "stack-master",
"sysName": "stack-master",
"serial": "SN001",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
expected_name = _generate_vc_member_name("stack-master", 2, serial="SN001")
assert result["name_matches"] is False
assert result["name_sync_available"] is True
assert result["suggested_name"] == expected_name
def test_vc_member_with_strip_domain(self):
"""strip_domain applies before VC member name comparison."""
from netbox_librenms_plugin.import_utils import validate_device_for_import
from netbox_librenms_plugin.import_utils.virtual_chassis import _generate_vc_member_name
mock_vc = MagicMock()
expected_name = _generate_vc_member_name("stack", 1, serial="SN001")
existing = self._make_existing(expected_name, serial="SN001", virtual_chassis=mock_vc, vc_position=1)
self._setup_librenms_id_filter(existing)
device_data = {
"device_id": 42,
"hostname": "stack.example.com",
"sysName": "stack.example.com",
"serial": "SN001",
}
result = validate_device_for_import(device_data, include_vc_detection=False, strip_domain=True)
assert result["name_matches"] is True
def test_naming_criteria_populated(self):
"""naming_criteria dict is set in result with use_sysname/strip_domain/source."""
from netbox_librenms_plugin.import_utils import validate_device_for_import
self.mock_device.objects.filter.return_value.first.return_value = None
device_data = {
"device_id": 99,
"hostname": "router-01",
"sysName": "router-sysname",
}
result = validate_device_for_import(
device_data, include_vc_detection=False, use_sysname=True, strip_domain=False
)
criteria = result["naming_criteria"]
assert criteria is not None
assert criteria["use_sysname"] is True
assert criteria["strip_domain"] is False
assert criteria["raw_sysname"] == "router-sysname"
assert criteria["raw_hostname"] == "router-01"
assert criteria["source"] == "sysname"
def test_naming_criteria_source_hostname_when_sysname_disabled(self):
"""naming_criteria source is 'hostname' when use_sysname=False."""
from netbox_librenms_plugin.import_utils import validate_device_for_import
self.mock_device.objects.filter.return_value.first.return_value = None
device_data = {
"device_id": 99,
"hostname": "router-01",
"sysName": "router-sysname",
}
result = validate_device_for_import(
device_data, include_vc_detection=False, use_sysname=False, strip_domain=False
)
assert result["naming_criteria"]["source"] == "hostname"
def test_naming_criteria_source_sysname_when_sysname_disabled_but_hostname_empty(self):
"""
When use_sysname=False and hostname is empty, source falls back to 'sysname'.
Before the fix, source was incorrectly reported as 'hostname' even
though the resolved name actually came from sysName.
"""
from netbox_librenms_plugin.import_utils import validate_device_for_import
self.mock_device.objects.filter.return_value.first.return_value = None
device_data = {
"device_id": 99,
"hostname": "",
"sysName": "router-sysname",
}
result = validate_device_for_import(
device_data, include_vc_detection=False, use_sysname=False, strip_domain=False
)
assert result["naming_criteria"]["source"] == "sysname", (
"When hostname is empty, source must be 'sysname', not 'hostname'"
)
def test_naming_criteria_source_hostname_fallback_when_both_empty(self):
"""When both hostname and sysName are empty, source is 'device-{id}' (no-name guard)."""
from netbox_librenms_plugin.import_utils import validate_device_for_import
self.mock_device.objects.filter.return_value.first.return_value = None
device_data = {"device_id": 99, "hostname": "", "sysName": ""}
result = validate_device_for_import(
device_data, include_vc_detection=False, use_sysname=False, strip_domain=False
)
# Both empty → no-name guard returns 'device-{id}' as source
assert result["naming_criteria"]["source"] == "device-99"
class TestLegacyLibreNMSIdMigration:
"""Test detection of legacy bare-integer librenms_id format during device validation."""
PATCHES = [
"netbox_librenms_plugin.import_utils.device_operations.Site",
"netbox_librenms_plugin.import_utils.device_operations.Rack",
"netbox_librenms_plugin.import_utils.device_operations.Cluster",
"netbox_librenms_plugin.import_utils.device_operations.DeviceRole",
"netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type",
"netbox_librenms_plugin.import_utils.device_operations.find_matching_platform",
"netbox_librenms_plugin.import_utils.device_operations.find_matching_site",
"netbox_librenms_plugin.import_utils.device_operations.Device",
"virtualization.models.VirtualMachine",
]
def setup_method(self):
self._patchers = [patch(p) for p in self.PATCHES]
mocks = [p.start() for p in self._patchers]
(
self.mock_site_model,
self.mock_rack,
self.mock_cluster,
self.mock_role,
self.mock_match_type,
self.mock_find_platform,
self.mock_find_site,
self.mock_device,
self.mock_vm,
) = mocks
self.mock_find_site.return_value = {
"found": True,
"site": MagicMock(),
"match_type": "exact",
"confidence": 1.0,
}
self.mock_find_platform.return_value = {"found": False, "platform": None, "match_type": None}
self.mock_match_type.return_value = {"matched": False, "device_type": None, "match_type": None}
self.mock_role.objects.all.return_value = []
self.mock_cluster.objects.all.return_value = []
self.mock_site_model.objects.all.return_value = []
self.mock_vm.objects.filter.return_value.first.return_value = None
def teardown_method(self):
for p in self._patchers:
p.stop()
def _make_existing(self, librenms_id_value, serial="SN001"):
existing = MagicMock()
existing.name = "switch-01"
existing.serial = serial
existing.virtual_chassis = None
existing.vc_position = None
existing.custom_field_data = {"librenms_id": librenms_id_value}
return existing
def _setup_device_filter(self, existing):
def device_filter(*args, **kwargs):
result = MagicMock()
q_has_librenms = any("librenms_id" in str(arg) for arg in args) or any(
k.startswith("custom_field_data__librenms_id") for k in kwargs
)
result.first.return_value = existing if q_has_librenms else None
return result
self.mock_device.objects.filter.side_effect = device_filter
def test_legacy_int_sets_needs_migration_flag(self):
"""Device with bare-integer librenms_id sets librenms_id_needs_migration=True."""
existing = self._make_existing(librenms_id_value=42, serial="SN001")
self._setup_device_filter(existing)
from netbox_librenms_plugin.import_utils import validate_device_for_import
result = validate_device_for_import(
{"device_id": 42, "hostname": "switch-01", "serial": "SN001"},
include_vc_detection=False,
)
assert result["existing_match_type"] == "librenms_id"
assert result["librenms_id_needs_migration"] is True
assert result["serial_confirmed"] is True
def test_legacy_int_no_serial_still_sets_flag(self):
"""Legacy int format sets the migration flag even when serial is absent."""
existing = self._make_existing(librenms_id_value=42, serial="")
self._setup_device_filter(existing)
from netbox_librenms_plugin.import_utils import validate_device_for_import
result = validate_device_for_import(
{"device_id": 42, "hostname": "switch-01"},
include_vc_detection=False,
)
assert result["librenms_id_needs_migration"] is True
assert result["serial_confirmed"] is False
def test_json_format_does_not_set_flag(self):
"""Device with JSON librenms_id does NOT set librenms_id_needs_migration."""
existing = self._make_existing(librenms_id_value={"default": 42}, serial="SN001")
self._setup_device_filter(existing)
from netbox_librenms_plugin.import_utils import validate_device_for_import
result = validate_device_for_import(
{"device_id": 42, "hostname": "switch-01", "serial": "SN001"},
include_vc_detection=False,
)
assert result["existing_match_type"] == "librenms_id"
assert result["librenms_id_needs_migration"] is False
def test_migrate_legacy_librenms_id_helper(self):
"""migrate_legacy_librenms_id converts int to {server_key: int}."""
from netbox_librenms_plugin.utils import migrate_legacy_librenms_id
obj = MagicMock()
obj.custom_field_data = {"librenms_id": 42}
result = migrate_legacy_librenms_id(obj, "primary")
assert result is True
assert obj.custom_field_data["librenms_id"] == {"primary": 42}
def test_migrate_legacy_librenms_id_noop_for_json(self):
"""migrate_legacy_librenms_id is a no-op when value is already a dict."""
from netbox_librenms_plugin.utils import migrate_legacy_librenms_id
obj = MagicMock()
obj.custom_field_data = {"librenms_id": {"primary": 42}}
result = migrate_legacy_librenms_id(obj, "primary")
assert result is False
assert obj.custom_field_data["librenms_id"] == {"primary": 42}
def test_migrate_legacy_librenms_id_noop_for_none(self):
"""migrate_legacy_librenms_id is a no-op when librenms_id is absent."""
from netbox_librenms_plugin.utils import migrate_legacy_librenms_id
obj = MagicMock()
obj.custom_field_data = {}
result = migrate_legacy_librenms_id(obj, "primary")
assert result is False
class TestDeviceConflictActionView:
"""Test DeviceConflictActionView conflict resolution actions."""
def _create_view(self):
"""Create a DeviceConflictActionView instance with mocked dependencies."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = DeviceConflictActionView()
view._librenms_api = MagicMock()
view._librenms_api.server_key = "default"
return view
def _create_request(self, action, existing_device_id, use_sysname=False, strip_domain=False):
"""
Create a mock request with POST data and permission stubs.
The returned request should be bound to the view (view.request = request)
before calling view.post() so permission checks and business logic
operate on the same request object, matching real Django CBV behavior.
"""
request = MagicMock()
request.user.has_perm.return_value = True
# Always include both toggles so resolve_naming_preferences never falls through
# to the user-pref/settings DB path, which would hit the real database.
post_data = {
"action": action,
"existing_device_id": str(existing_device_id),
"use-sysname-toggle": "on" if use_sysname else "off",
"strip-domain-toggle": "on" if strip_domain else "off",
}
request.POST = post_data
return request
@patch("netbox_librenms_plugin.views.imports.actions.cache")
@patch("netbox_librenms_plugin.views.imports.actions.get_import_device_cache_key")
def test_link_action_sets_librenms_id_and_name(self, mock_cache_key, mock_cache):
"""Link action should set librenms_id and update name from sysName."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
existing_device = MagicMock()
existing_device.pk = 42
existing_device.custom_field_data = {}
existing_device.name = "84.116.251.35"
libre_device = {
"device_id": 10,
"hostname": "84.116.251.35",
"sysName": "switch-01.example.com",
"serial": "ABC123",
}
validation = {"can_import": False, "existing_device": existing_device}
selections = {}
request = self._create_request("link", 42, use_sysname=True)
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch.object(DeviceConflictActionView, "render_device_row") as mock_render,
patch("dcim.models.Device") as mock_device_cls,
patch("netbox_librenms_plugin.views.imports.actions.transaction") as mock_tx,
):
mock_tx.atomic.return_value = MagicMock()
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = False
mock_validate.return_value = (libre_device, validation, selections)
mock_render.return_value = MagicMock()
view.request = request
view.post(request, device_id=10)
assert existing_device.custom_field_data["librenms_id"] == {"default": 10}
assert existing_device.name == "switch-01.example.com"
existing_device.save.assert_called_once()
@patch("netbox_librenms_plugin.views.imports.actions.cache")
@patch("netbox_librenms_plugin.views.imports.actions.get_import_device_cache_key")
def test_link_action_uses_non_default_server_key(self, mock_cache_key, mock_cache):
"""Link action should store librenms_id under the active server_key, not always 'default'."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
view._librenms_api.server_key = "production"
existing_device = MagicMock()
existing_device.pk = 42
existing_device.custom_field_data = {}
existing_device.name = "84.116.251.35"
libre_device = {
"device_id": 10,
"hostname": "84.116.251.35",
"sysName": "switch-01.example.com",
"serial": "ABC123",
}
validation = {"can_import": False, "existing_device": existing_device}
selections = {}
request = self._create_request("link", 42, use_sysname=True)
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch.object(DeviceConflictActionView, "render_device_row") as mock_render,
patch("dcim.models.Device") as mock_device_cls,
patch("netbox_librenms_plugin.views.imports.actions.transaction") as mock_tx,
):
mock_tx.atomic.return_value = MagicMock()
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.filter.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = False
mock_validate.return_value = (libre_device, validation, selections)
mock_render.return_value = MagicMock()
view.request = request
view.post(request, device_id=10)
assert existing_device.custom_field_data["librenms_id"] == {"production": 10}
@patch("netbox_librenms_plugin.views.imports.actions.cache")
@patch("netbox_librenms_plugin.views.imports.actions.get_import_device_cache_key")
def test_update_action_sets_hostname_serial_and_librenms_id(self, mock_cache_key, mock_cache):
"""Update action should set hostname, serial, and librenms_id."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
existing_device = MagicMock()
existing_device.pk = 42
existing_device.custom_field_data = {}
existing_device.name = "old-name"
existing_device.serial = "OLD-SERIAL"
libre_device = {
"device_id": 10,
"hostname": "84.116.251.35",
"sysName": "new-name.example.com",
"serial": "NEW-SERIAL",
}
validation = {"can_import": False, "existing_device": existing_device}
selections = {}
request = self._create_request("update", 42, use_sysname=True)
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch.object(DeviceConflictActionView, "render_device_row") as mock_render,
patch("dcim.models.Device") as mock_device_cls,
patch("netbox_librenms_plugin.views.imports.actions.transaction") as mock_tx,
):
mock_tx.atomic.return_value = MagicMock()
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = False
mock_validate.return_value = (libre_device, validation, selections)
mock_render.return_value = MagicMock()
view.request = request
view.post(request, device_id=10)
assert existing_device.custom_field_data["librenms_id"] == {"default": 10}
assert existing_device.serial == "NEW-SERIAL"
assert existing_device.name == "new-name.example.com"
existing_device.save.assert_called_once()
@patch("netbox_librenms_plugin.views.imports.actions.cache")
@patch("netbox_librenms_plugin.views.imports.actions.get_import_device_cache_key")
def test_update_action_uses_non_default_server_key(self, mock_cache_key, mock_cache):
"""Update action should store librenms_id under the active server key."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
view._librenms_api = MagicMock()
view._librenms_api.server_key = "production"
existing_device = MagicMock()
existing_device.pk = 42
existing_device.custom_field_data = {}
existing_device.name = "old-name"
existing_device.serial = "OLD-SERIAL"
libre_device = {
"device_id": 10,
"hostname": "switch-01",
"sysName": "switch-01",
"serial": "NEW-SERIAL",
"resolved_name": "switch-01",
}
validation = {"can_import": False, "existing_device": existing_device, "resolved_name": "switch-01"}
selections = {}
request = self._create_request("update", 42)
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch.object(DeviceConflictActionView, "render_device_row") as mock_render,
patch("dcim.models.Device") as mock_device_cls,
patch("netbox_librenms_plugin.views.imports.actions.transaction") as mock_tx,
):
mock_tx.atomic.return_value = MagicMock()
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = False
mock_validate.return_value = (libre_device, validation, selections)
mock_render.return_value = MagicMock()
view.request = request
view.post(request, device_id=10)
assert existing_device.custom_field_data["librenms_id"] == {"production": 10}
@patch("netbox_librenms_plugin.views.imports.actions.cache")
@patch("netbox_librenms_plugin.views.imports.actions.get_import_device_cache_key")
def test_update_serial_action_updates_serial_only(self, mock_cache_key, mock_cache):
"""Update serial action should update serial and librenms_id but not hostname."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
existing_device = MagicMock()
existing_device.pk = 42
existing_device.custom_field_data = {}
existing_device.name = "switch-01"
existing_device.serial = "OLD-SERIAL"
libre_device = {"device_id": 10, "hostname": "switch-01", "serial": "NEW-SERIAL"}
validation = {"can_import": False, "existing_device": existing_device}
selections = {}
request = self._create_request("update_serial", 42)
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch.object(DeviceConflictActionView, "render_device_row") as mock_render,
patch("dcim.models.Device") as mock_device_cls,
patch("netbox_librenms_plugin.views.imports.actions.transaction") as mock_tx,
):
mock_tx.atomic.return_value = MagicMock()
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = False
mock_validate.return_value = (libre_device, validation, selections)
mock_render.return_value = MagicMock()
view.request = request
view.post(request, device_id=10)
assert existing_device.custom_field_data["librenms_id"] == {"default": 10}
assert existing_device.serial == "NEW-SERIAL"
# Name should NOT be changed by update_serial
assert existing_device.name == "switch-01"
existing_device.save.assert_called_once()
@patch("netbox_librenms_plugin.views.imports.actions.cache")
@patch("netbox_librenms_plugin.views.imports.actions.get_import_device_cache_key")
def test_update_skips_dash_serial(self, mock_cache_key, mock_cache):
"""Update should not set serial to '-' (LibreNMS placeholder)."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
existing_device = MagicMock()
existing_device.pk = 42
existing_device.custom_field_data = {}
existing_device.name = "switch-01"
existing_device.serial = "EXISTING"
libre_device = {"device_id": 10, "hostname": "switch-01", "serial": "-"}
validation = {"can_import": False, "existing_device": existing_device}
selections = {}
request = self._create_request("update_serial", 42)
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch.object(DeviceConflictActionView, "render_device_row") as mock_render,
patch("dcim.models.Device") as mock_device_cls,
patch("netbox_librenms_plugin.views.imports.actions.transaction") as mock_tx,
):
mock_tx.atomic.return_value = MagicMock()
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = False
mock_validate.return_value = (libre_device, validation, selections)
mock_render.return_value = MagicMock()
view.request = request
view.post(request, device_id=10)
# Serial should NOT be updated to '-'
assert existing_device.serial == "EXISTING"
def test_missing_action_returns_400(self):
"""Missing action or existing_device_id should return 400."""
view = self._create_view()
request = MagicMock()
request.user.has_perm.return_value = True
request.POST = {}
view.request = request
response = view.post(request, device_id=10)
assert response.status_code == 400
def test_unknown_action_returns_400(self):
"""Unknown action should return 400."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
request = self._create_request("invalid_action", 42)
existing_device = MagicMock()
existing_device.pk = 42
libre_device = {"device_id": 10, "hostname": "switch-01", "serial": "ABC"}
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch.object(DeviceConflictActionView, "require_object_permissions", return_value=None),
patch("dcim.models.Device") as mock_device_cls,
):
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.filter.return_value.exclude.return_value.first.return_value = None
# Include existing_device so the validated-conflict-target guard passes;
# we want to exercise the unknown-action branch, not the missing-device guard.
mock_validate.return_value = (libre_device, {"existing_device": existing_device}, {})
view.request = request
response = view.post(request, device_id=10)
assert response.status_code == 400
@patch("netbox_librenms_plugin.views.imports.actions.cache")
@patch("netbox_librenms_plugin.views.imports.actions.get_import_device_cache_key")
def test_sync_name_action_updates_name(self, mock_cache_key, mock_cache):
"""Sync name action should update device name using sysName."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
existing_device = MagicMock()
existing_device.pk = 42
existing_device.custom_field_data = {"librenms_id": 10}
existing_device.name = "84.116.251.35"
libre_device = {
"device_id": 10,
"hostname": "84.116.251.35",
"sysName": "switch-01.example.com",
"serial": "ABC123",
}
validation = {"can_import": False, "existing_device": existing_device}
selections = {}
request = self._create_request("sync_name", 42, use_sysname=True)
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch.object(DeviceConflictActionView, "render_device_row") as mock_render,
patch("dcim.models.Device") as mock_device_cls,
):
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.filter.return_value.exclude.return_value.first.return_value = None
mock_validate.return_value = (libre_device, validation, selections)
mock_render.return_value = MagicMock()
view.request = request
view.post(request, device_id=10)
assert existing_device.name == "switch-01.example.com"
existing_device.save.assert_called_once()
assert existing_device.custom_field_data["librenms_id"] == 10
@patch("netbox_librenms_plugin.views.imports.actions.cache")
@patch("netbox_librenms_plugin.views.imports.actions.get_import_device_cache_key")
def test_device_type_mismatch_blocked_without_force(self, mock_cache_key, mock_cache):
"""Action should be blocked when device_type_mismatch is True and force is not set."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
existing_device = MagicMock()
existing_device.pk = 42
existing_device.custom_field_data = {}
libre_device = {"device_id": 10, "hostname": "switch-01", "serial": "ABC123"}
validation = {"can_import": False, "device_type_mismatch": True, "existing_device": existing_device}
selections = {}
request = self._create_request("link", 42, use_sysname=True)
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch("dcim.models.Device") as mock_device_cls,
):
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.filter.return_value.exclude.return_value.first.return_value = None
mock_validate.return_value = (libre_device, validation, selections)
view.request = request
response = view.post(request, device_id=10)
assert response.status_code == 400
existing_device.save.assert_not_called()
@patch("netbox_librenms_plugin.views.imports.actions.cache")
@patch("netbox_librenms_plugin.views.imports.actions.get_import_device_cache_key")
def test_device_type_mismatch_allowed_with_force(self, mock_cache_key, mock_cache):
"""Action should proceed when device_type_mismatch is True and force is set."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
existing_device = MagicMock()
existing_device.pk = 42
existing_device.custom_field_data = {}
existing_device.name = "old-name"
libre_device = {
"device_id": 10,
"hostname": "switch-01",
"sysName": "switch-01.example.com",
"serial": "ABC123",
}
validation = {"can_import": False, "device_type_mismatch": True, "existing_device": existing_device}
selections = {}
request = self._create_request("link", 42, use_sysname=True)
request.POST["force"] = "on"
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch.object(DeviceConflictActionView, "render_device_row") as mock_render,
patch("dcim.models.Device") as mock_device_cls,
patch("netbox_librenms_plugin.views.imports.actions.transaction") as mock_tx,
):
mock_tx.atomic.return_value = MagicMock()
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = False
mock_validate.return_value = (libre_device, validation, selections)
mock_render.return_value = MagicMock()
view.request = request
view.post(request, device_id=10)
assert existing_device.custom_field_data["librenms_id"] == {"default": 10}
existing_device.save.assert_called_once()
@patch("netbox_librenms_plugin.views.imports.actions.cache")
@patch("netbox_librenms_plugin.views.imports.actions.get_import_device_cache_key")
def test_force_with_mismatch_updates_device_type(self, mock_cache_key, mock_cache):
"""Force with device_type_mismatch should update existing device's device_type."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
existing_device = MagicMock()
existing_device.pk = 42
existing_device.custom_field_data = {}
existing_device.name = "old-name"
librenms_device_type = MagicMock()
librenms_device_type.pk = 99
libre_device = {
"device_id": 10,
"hostname": "switch-01",
"sysName": "switch-01.example.com",
"serial": "ABC123",
}
validation = {
"can_import": False,
"device_type_mismatch": True,
"device_type": {"device_type": librenms_device_type},
"existing_device": existing_device,
}
selections = {}
request = self._create_request("link", 42, use_sysname=True)
request.POST["force"] = "on"
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch.object(DeviceConflictActionView, "render_device_row") as mock_render,
patch("dcim.models.Device") as mock_device_cls,
patch("netbox_librenms_plugin.views.imports.actions.transaction") as mock_tx,
):
mock_tx.atomic.return_value = MagicMock()
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = False
mock_validate.return_value = (libre_device, validation, selections)
mock_render.return_value = MagicMock()
view.request = request
view.post(request, device_id=10)
assert existing_device.device_type == librenms_device_type
assert existing_device.custom_field_data["librenms_id"] == {"default": 10}
existing_device.save.assert_called_once()
@patch("netbox_librenms_plugin.views.imports.actions.cache")
@patch("netbox_librenms_plugin.views.imports.actions.get_import_device_cache_key")
def test_update_type_action_changes_device_type(self, mock_cache_key, mock_cache):
"""update_type action should change device type on existing device."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
existing_device = MagicMock()
existing_device.pk = 42
existing_device.custom_field_data = {"librenms_id": 10}
existing_device.name = "switch-01"
old_device_type = MagicMock()
existing_device.device_type = old_device_type
new_device_type = MagicMock()
new_device_type.pk = 99
libre_device = {
"device_id": 10,
"hostname": "switch-01",
"serial": "ABC123",
}
validation = {
"can_import": False,
"device_type_mismatch": True,
"device_type": {"device_type": new_device_type},
"existing_device": existing_device,
}
selections = {}
request = self._create_request("update_type", 42)
request.POST["force"] = "on"
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch.object(DeviceConflictActionView, "render_device_row") as mock_render,
patch("dcim.models.Device") as mock_device_cls,
):
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.filter.return_value.exclude.return_value.first.return_value = None
mock_validate.return_value = (libre_device, validation, selections)
mock_render.return_value = MagicMock()
view.request = request
view.post(request, device_id=10)
assert existing_device.device_type == new_device_type
existing_device.save.assert_called_once()
assert existing_device.custom_field_data["librenms_id"] == 10
@patch("netbox_librenms_plugin.views.imports.actions.cache")
@patch("netbox_librenms_plugin.views.imports.actions.get_import_device_cache_key")
def test_sync_serial_action(self, mock_cache_key, mock_cache):
"""sync_serial action should update serial from LibreNMS."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
existing_device = MagicMock()
existing_device.serial = "OLD123"
libre_device = {"device_id": 10, "serial": "NEW456", "sysName": "test"}
validation = {"existing_device": existing_device, "device_type_mismatch": False}
selections = {}
request = self._create_request("sync_serial", 42)
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch.object(DeviceConflictActionView, "render_device_row") as mock_render,
patch("dcim.models.Device") as mock_device_cls,
patch("netbox_librenms_plugin.views.imports.actions.transaction") as mock_tx,
):
mock_tx.atomic.return_value = MagicMock()
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.first.return_value = None
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = False
mock_validate.return_value = (libre_device, validation, selections)
mock_render.return_value = MagicMock()
view.request = request
view.post(request, device_id=10)
assert existing_device.serial == "NEW456"
existing_device.save.assert_called_once()
@patch("netbox_librenms_plugin.views.imports.actions.cache")
@patch("netbox_librenms_plugin.views.imports.actions.get_import_device_cache_key")
def test_sync_platform_action(self, mock_cache_key, mock_cache):
"""sync_platform action should update platform from LibreNMS OS."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
existing_device = MagicMock()
existing_device.platform = None
libre_device = {"device_id": 10, "os": "ios", "sysName": "test"}
validation = {"existing_device": existing_device, "device_type_mismatch": False}
selections = {}
mock_platform = MagicMock()
request = self._create_request("sync_platform", 42)
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch.object(DeviceConflictActionView, "render_device_row") as mock_render,
patch("dcim.models.Device") as mock_device_cls,
# Patch find_matching_platform at the utility module level — the action imports
# it from netbox_librenms_plugin.utils, so that is the correct seam to mock.
patch("netbox_librenms_plugin.utils.find_matching_platform") as mock_find_platform,
):
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.filter.return_value.exclude.return_value.first.return_value = None
mock_find_platform.return_value = {"found": True, "platform": mock_platform, "match_type": "exact"}
mock_validate.return_value = (libre_device, validation, selections)
mock_render.return_value = MagicMock()
view.request = request
view.post(request, device_id=10)
assert existing_device.platform == mock_platform
existing_device.save.assert_called_once()
@patch("netbox_librenms_plugin.views.imports.actions.cache")
@patch("netbox_librenms_plugin.views.imports.actions.get_import_device_cache_key")
def test_sync_device_type_action(self, mock_cache_key, mock_cache):
"""sync_device_type action should update device type from LibreNMS hardware match."""
from netbox_librenms_plugin.views.imports.actions import DeviceConflictActionView
view = self._create_view()
existing_device = MagicMock()
new_device_type = MagicMock()
libre_device = {"device_id": 10, "hardware": "Catalyst C4900M", "sysName": "test"}
validation = {"existing_device": existing_device, "device_type_mismatch": False}
selections = {}
request = self._create_request("sync_device_type", 42)
with (
patch.object(DeviceConflictActionView, "get_validated_device_with_selections") as mock_validate,
patch.object(DeviceConflictActionView, "render_device_row") as mock_render,
patch("dcim.models.Device") as mock_device_cls,
patch("netbox_librenms_plugin.utils.match_librenms_hardware_to_device_type") as mock_hw_match,
):
mock_device_cls.objects.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.get.return_value = existing_device
mock_device_cls.objects.select_for_update.return_value.filter.return_value.exclude.return_value.first.return_value = None
mock_hw_match.return_value = {"matched": True, "device_type": new_device_type}
mock_validate.return_value = (libre_device, validation, selections)
mock_render.return_value = MagicMock()
view.request = request
view.post(request, device_id=10)
assert existing_device.device_type == new_device_type
existing_device.save.assert_called_once()
class TestBuildSyncInfo:
"""Test DeviceValidationDetailsView._build_sync_info method."""
def test_all_synced(self):
"""When serial, platform, device type all match, all_synced is True."""
from netbox_librenms_plugin.views.imports.actions import DeviceValidationDetailsView
existing = MagicMock()
existing.serial = "ABC123"
platform = MagicMock()
platform.pk = 1
existing.platform = platform
device_type = MagicMock()
device_type.pk = 5
existing.device_type = device_type
libre_device = {"serial": "ABC123", "os": "ios", "hardware": "Catalyst C4900M"}
with (
patch("netbox_librenms_plugin.utils.find_matching_platform") as mock_platform_match,
patch("netbox_librenms_plugin.utils.match_librenms_hardware_to_device_type") as mock_hw_match,
):
mock_platform_match.return_value = {"found": True, "platform": platform}
mock_hw_match.return_value = {"matched": True, "device_type": device_type}
result = DeviceValidationDetailsView._build_sync_info(libre_device, existing)
assert result["all_synced"] is True
assert result["serial_synced"] is True
assert result["platform_synced"] is True
assert result["device_type_synced"] is True
def test_serial_out_of_sync(self):
"""When serial differs, serial_synced is False."""
from netbox_librenms_plugin.views.imports.actions import DeviceValidationDetailsView
existing = MagicMock()
existing.serial = "OLD123"
existing.platform = None
device_type = MagicMock()
device_type.pk = 5
existing.device_type = device_type
libre_device = {"serial": "NEW456", "os": "-", "hardware": "-"}
result = DeviceValidationDetailsView._build_sync_info(libre_device, existing)
assert result["serial_synced"] is False
assert result["all_synced"] is False
def test_platform_out_of_sync(self):
"""When platform differs, platform_synced is False."""
from netbox_librenms_plugin.views.imports.actions import DeviceValidationDetailsView
existing = MagicMock()
existing.serial = "ABC123"
old_platform = MagicMock()
old_platform.pk = 1
existing.platform = old_platform
device_type = MagicMock()
device_type.pk = 5
existing.device_type = device_type
new_platform = MagicMock()
new_platform.pk = 2
libre_device = {"serial": "ABC123", "os": "junos", "hardware": "-"}
with patch("netbox_librenms_plugin.utils.find_matching_platform") as mock_platform_match:
mock_platform_match.return_value = {"found": True, "platform": new_platform}
result = DeviceValidationDetailsView._build_sync_info(libre_device, existing)
assert result["platform_synced"] is False
assert result["all_synced"] is False
def test_platform_no_match_found_returns_bool(self):
"""When find_matching_platform returns no match, platform_synced must be False (not None)."""
from netbox_librenms_plugin.views.imports.actions import DeviceValidationDetailsView
existing = MagicMock()
existing.serial = "ABC123"
existing.platform = MagicMock() # device has a platform set
device_type = MagicMock()
device_type.pk = 5
existing.device_type = device_type
libre_device = {"serial": "ABC123", "os": "ios", "hardware": "-"}
with patch("netbox_librenms_plugin.utils.find_matching_platform") as mock_platform_match:
mock_platform_match.return_value = {"found": False, "platform": None}
result = DeviceValidationDetailsView._build_sync_info(libre_device, existing)
# Without bool() cast this would be None; verify it's exactly False (type-stable)
assert result["platform_synced"] is False
assert isinstance(result["platform_synced"], bool)
def test_platform_synced_no_netbox_platform_returns_bool(self):
"""When device has no platform in NetBox and os is non-dash, platform_synced must be bool."""
from netbox_librenms_plugin.views.imports.actions import DeviceValidationDetailsView
existing = MagicMock()
existing.serial = "ABC123"
existing.platform = None # no platform on device
device_type = MagicMock()
device_type.pk = 5
existing.device_type = device_type
libre_device = {"serial": "ABC123", "os": "eos", "hardware": "-"}
with patch("netbox_librenms_plugin.utils.find_matching_platform") as mock_platform_match:
mock_platform_match.return_value = {"found": True, "platform": MagicMock()}
result = DeviceValidationDetailsView._build_sync_info(libre_device, existing)
# None and ... returns None; bool() cast ensures False
assert result["platform_synced"] is False
assert isinstance(result["platform_synced"], bool)
def test_hardware_no_match_device_type_out_of_sync(self):
"""When hardware is present but no device type match found, device_type_synced is False."""
from netbox_librenms_plugin.views.imports.actions import DeviceValidationDetailsView
existing = MagicMock()
existing.serial = "ABC123"
existing.platform = None
device_type = MagicMock()
device_type.pk = 5
existing.device_type = device_type
libre_device = {"serial": "ABC123", "os": "-", "hardware": "UnknownHardwareXYZ"}
with patch("netbox_librenms_plugin.utils.match_librenms_hardware_to_device_type") as mock_hw_match:
mock_hw_match.return_value = {"matched": False, "device_type": None}
result = DeviceValidationDetailsView._build_sync_info(libre_device, existing)
assert result["device_type_synced"] is False
assert result["all_synced"] is False
class TestImportSingleDeviceLazyValidation:
"""import_single_device must pass api=api to validate_device_for_import when validation is None."""
def test_api_passed_to_validate(self):
from netbox_librenms_plugin.import_utils.device_operations import import_single_device
mock_api = MagicMock()
mock_api.server_key = "prod"
mock_validation = {
"existing_device": MagicMock(name="existing"),
"can_import": False,
}
with (
patch(
"netbox_librenms_plugin.import_utils.device_operations.LibreNMSAPI",
return_value=mock_api,
),
patch(
"netbox_librenms_plugin.import_utils.device_operations.validate_device_for_import",
return_value=mock_validation,
) as mock_validate,
):
import_single_device(
42,
server_key="prod",
sync_options={"use_sysname": True, "strip_domain": False},
validation=None,
libre_device={"device_id": 42, "hostname": "test"},
)
mock_validate.assert_called_once()
assert mock_validate.call_args[1].get("api") is mock_api
class TestDeviceNamingPreferences:
"""Test that validation honours use_sysname and strip_domain user preferences."""
def _setup_no_existing(self, mocks):
"""Configure mocks so no existing device is found."""
(
mock_site_model,
mock_rack,
mock_cluster,
mock_role,
mock_match_type,
mock_find_platform,
mock_find_site,
mock_device,
mock_vm,
) = mocks
mock_vm.objects.filter.return_value.first.return_value = None
mock_device.objects.filter.return_value.first.return_value = None
mock_find_site.return_value = {
"found": False,
"site": None,
"match_type": None,
"confidence": 0.0,
}
mock_find_platform.return_value = {
"found": False,
"platform": None,
"match_type": None,
}
mock_match_type.return_value = {
"matched": False,
"device_type": None,
"match_type": None,
}
mock_role.objects.all.return_value = []
mock_site_model.objects.all.return_value = []
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_resolved_name_uses_sysname_by_default(self, *mocks):
"""Default use_sysname=True uses sysName for resolved_name."""
self._setup_no_existing(mocks)
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "10.0.0.1",
"sysName": "core-switch",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["resolved_name"] == "core-switch"
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_resolved_name_uses_hostname_when_sysname_disabled(self, *mocks):
"""use_sysname=False uses hostname for resolved_name."""
self._setup_no_existing(mocks)
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "10.0.0.1",
"sysName": "core-switch",
}
result = validate_device_for_import(
device_data,
include_vc_detection=False,
use_sysname=False,
)
assert result["resolved_name"] == "10.0.0.1"
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_resolved_name_strips_domain(self, *mocks):
"""strip_domain=True strips the domain suffix."""
self._setup_no_existing(mocks)
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01.example.com",
"sysName": "switch-01.example.com",
}
result = validate_device_for_import(
device_data,
include_vc_detection=False,
strip_domain=True,
)
assert result["resolved_name"] == "switch-01"
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_duplicate_detection_uses_resolved_name(self, *mocks):
"""Duplicate detection should match against the resolved name, not raw hostname."""
self._setup_no_existing(mocks)
# Unpack using same order as _setup_no_existing / @patch decorators (bottom-up)
(
_mock_site,
_mock_rack,
_mock_cluster,
_mock_role,
_mock_hw,
_mock_platform,
_mock_find_site,
mock_device,
_mock_vm,
) = mocks
existing = MagicMock()
existing.name = "core-switch"
existing.serial = ""
existing.virtual_chassis = None
existing.vc_position = None
mock_device.objects.filter.return_value.first.side_effect = [None, existing]
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 999,
"hostname": "10.0.0.1",
"sysName": "core-switch",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert result["existing_device"] == existing
assert result["existing_match_type"] == "hostname"
@patch("virtualization.models.VirtualMachine")
@patch("netbox_librenms_plugin.import_utils.device_operations.Device")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_site")
@patch("netbox_librenms_plugin.import_utils.device_operations.find_matching_platform")
@patch("netbox_librenms_plugin.import_utils.device_operations.match_librenms_hardware_to_device_type")
@patch("netbox_librenms_plugin.import_utils.device_operations.DeviceRole")
@patch("netbox_librenms_plugin.import_utils.device_operations.Cluster")
@patch("netbox_librenms_plugin.import_utils.device_operations.Rack")
@patch("netbox_librenms_plugin.import_utils.device_operations.Site")
def test_backward_compatible_defaults(self, *mocks):
"""Calling without naming params produces resolved_name in result."""
self._setup_no_existing(mocks)
from netbox_librenms_plugin.import_utils import validate_device_for_import
device_data = {
"device_id": 1,
"hostname": "switch-01",
}
result = validate_device_for_import(device_data, include_vc_detection=False)
assert "resolved_name" in result
assert result["resolved_name"] == "switch-01"
class TestProcessDeviceFilters:
"""Tests for process_device_filters and related bulk_import utilities."""
def test_show_disabled_filters_integer_disabled_1(self):
"""show_disabled=False should exclude devices with disabled==1 (int)."""
from unittest.mock import MagicMock, patch
devices = [
{"device_id": 1, "hostname": "a", "disabled": 0, "status": 1},
{"device_id": 2, "hostname": "b", "disabled": 1, "status": 1}, # disabled in LibreNMS
]
with (
patch(
"netbox_librenms_plugin.import_utils.bulk_import.get_librenms_devices_for_import",
return_value=(devices, False),
),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.validate_device_for_import",
side_effect=lambda d, **kw: {
"resolved_name": d["hostname"],
"is_ready": True,
"can_import": True,
"status": "active",
"existing_device": None,
"import_as_vm": False,
"existing_match_type": None,
},
),
patch("netbox_librenms_plugin.import_utils.bulk_import.prefetch_vc_data_for_devices"),
patch("netbox_librenms_plugin.import_utils.bulk_import.cache"),
patch("netbox_librenms_plugin.import_utils.bulk_import.get_cache_metadata_key", return_value="key"),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.get_validated_device_cache_key", return_value="vkey"
),
):
from netbox_librenms_plugin.import_utils.bulk_import import process_device_filters
api = MagicMock()
api.server_key = "default"
result = process_device_filters(
api, filters={}, vc_detection_enabled=False, clear_cache=False, show_disabled=False
)
# Only enabled device (disabled==0) should be processed
assert len(result) == 1
assert result[0]["hostname"] == "a"
def test_show_disabled_keeps_unreachable_enabled_device(self):
"""show_disabled=False should keep devices that are enabled (disabled==0) even if status==0."""
from unittest.mock import MagicMock, patch
devices = [
{"device_id": 1, "hostname": "a", "disabled": 0, "status": 0}, # down but enabled
{"device_id": 2, "hostname": "b", "disabled": 1, "status": 0}, # down and disabled
]
with (
patch(
"netbox_librenms_plugin.import_utils.bulk_import.get_librenms_devices_for_import",
return_value=(devices, False),
),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.validate_device_for_import",
side_effect=lambda d, **kw: {
"resolved_name": d["hostname"],
"is_ready": True,
"can_import": True,
"status": "active",
"existing_device": None,
"import_as_vm": False,
"existing_match_type": None,
},
),
patch("netbox_librenms_plugin.import_utils.bulk_import.prefetch_vc_data_for_devices"),
patch("netbox_librenms_plugin.import_utils.bulk_import.cache"),
patch("netbox_librenms_plugin.import_utils.bulk_import.get_cache_metadata_key", return_value="key"),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.get_validated_device_cache_key", return_value="vkey"
),
):
from netbox_librenms_plugin.import_utils.bulk_import import process_device_filters
api = MagicMock()
api.server_key = "default"
result = process_device_filters(
api, filters={}, vc_detection_enabled=False, clear_cache=False, show_disabled=False
)
# Device a is enabled (disabled==0) and should be kept even though status==0
assert len(result) == 1
assert result[0]["hostname"] == "a"
def test_show_disabled_true_includes_all(self):
"""show_disabled=True should include both active and inactive devices."""
from unittest.mock import MagicMock, patch
devices = [
{"device_id": 1, "hostname": "a", "status": 1},
{"device_id": 2, "hostname": "b", "status": 0},
]
with (
patch(
"netbox_librenms_plugin.import_utils.bulk_import.get_librenms_devices_for_import",
return_value=(devices, False),
),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.validate_device_for_import",
side_effect=lambda d, **kw: {
"resolved_name": d["hostname"],
"is_ready": True,
"can_import": True,
"status": "active",
"existing_device": None,
"import_as_vm": False,
"existing_match_type": None,
},
),
patch("netbox_librenms_plugin.import_utils.bulk_import.prefetch_vc_data_for_devices"),
patch("netbox_librenms_plugin.import_utils.bulk_import.cache"),
patch("netbox_librenms_plugin.import_utils.bulk_import.get_cache_metadata_key", return_value="key"),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.get_validated_device_cache_key", return_value="vkey"
),
):
from netbox_librenms_plugin.import_utils.bulk_import import process_device_filters
api = MagicMock()
api.server_key = "default"
result = process_device_filters(
api, filters={}, vc_detection_enabled=False, clear_cache=False, show_disabled=True
)
assert len(result) == 2
def test_empty_return_helper(self):
"""_empty_return should return ([], False) when return_cache_status=True, else []."""
from netbox_librenms_plugin.import_utils.bulk_import import _empty_return
assert _empty_return(True) == ([], False)
assert _empty_return(False) == []
def test_bulk_import_devices_uses_resolved_server_key(self):
"""bulk_import_devices_shared should pass api.server_key to import_single_device."""
from unittest.mock import MagicMock, patch
with (
patch("netbox_librenms_plugin.import_utils.bulk_import.LibreNMSAPI") as mock_api_cls,
patch("netbox_librenms_plugin.import_utils.bulk_import.import_single_device") as mock_import,
patch("netbox_librenms_plugin.import_utils.bulk_import.validate_device_for_import") as mock_validate,
patch("netbox_librenms_plugin.import_utils.bulk_import.require_permissions"),
):
mock_api = MagicMock()
mock_api.server_key = "resolved-key"
mock_api.get_device_info.return_value = (True, {"device_id": 1, "hostname": "sw"})
mock_api_cls.return_value = mock_api
mock_import.return_value = {"success": True, "device": MagicMock(), "is_vm": False}
user = MagicMock()
from netbox_librenms_plugin.import_utils.bulk_import import bulk_import_devices_shared
bulk_import_devices_shared([1], user=user, server_key=None)
# The resolved api.server_key ("resolved-key") must be passed, not None
assert mock_import.call_args is not None
assert mock_import.call_args.kwargs.get("server_key") == "resolved-key"
assert mock_validate.call_args is not None
# Import-time VC detection is always enabled (restored pre-regression behavior).
assert mock_validate.call_args.kwargs.get("include_vc_detection") is True
class TestVCPositionHandling:
"""Test VC position normalization and suggested name generation."""
def test_clone_vc_data_position_fallback_is_one_based(self):
"""_clone_virtual_chassis_data fallback must be 1-based (idx+1, not idx)."""
from netbox_librenms_plugin.import_utils.virtual_chassis import _clone_virtual_chassis_data
data = {"is_stack": True, "member_count": 2, "members": [{"serial": "S1"}, {"serial": "S2"}]}
result = _clone_virtual_chassis_data(data)
positions = [m["position"] for m in result["members"]]
# First member: idx=0 → position should be 1, not 0
assert positions[0] == 1
assert positions[1] == 2
def test_clone_vc_data_preserves_explicit_positions(self):
"""_clone_virtual_chassis_data must preserve explicitly set positions."""
from netbox_librenms_plugin.import_utils.virtual_chassis import _clone_virtual_chassis_data
data = {
"is_stack": True,
"member_count": 2,
"members": [{"serial": "S1", "position": 3}, {"serial": "S2", "position": 5}],
}
result = _clone_virtual_chassis_data(data)
assert result["members"][0]["position"] == 3
assert result["members"][1]["position"] == 5
def test_clone_vc_data_bad_position_falls_back_to_one_based(self):
"""_clone_virtual_chassis_data falls back to idx+1 for non-int position."""
from netbox_librenms_plugin.import_utils.virtual_chassis import _clone_virtual_chassis_data
data = {
"is_stack": True,
"member_count": 2,
"members": [{"serial": "S1", "position": "bad"}, {"serial": "S2", "position": None}],
}
result = _clone_virtual_chassis_data(data)
# idx=0 → fallback 1, idx=1 → fallback 2
assert result["members"][0]["position"] == 1
assert result["members"][1]["position"] == 2
def test_suggested_name_uses_position_directly(self):
"""
Suggested name generation must use position directly (not position+1).
This test verifies that _generate_vc_member_name is called with the
already-1-based position value, not position+1.
"""
from netbox_librenms_plugin.import_utils.virtual_chassis import _generate_vc_member_name
# position=1 should produce name with "1", not "2"
name = _generate_vc_member_name("switch-1", 1, pattern="-M{position}")
assert name == "switch-1-M1", f"Expected 'switch-1-M1', got '{name}'"
# position=2 should produce "2", not "3"
name = _generate_vc_member_name("switch-1", 2, pattern="-M{position}")
assert name == "switch-1-M2", f"Expected 'switch-1-M2', got '{name}'"
def test_update_vc_member_suggested_names_no_off_by_one(self):
"""
update_vc_member_suggested_names must use stored 1-based positions directly.
Previously bays_by_depth applied an extra +1 to positions that were
already 1-based, producing suggested names like "switch-M2" for position 1.
"""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import (
update_vc_member_suggested_names,
)
vc_data = {
"is_stack": True,
"member_count": 2,
"members": [
{"serial": "S1", "position": 1},
{"serial": "S2", "position": 2},
],
}
with patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
):
result = update_vc_member_suggested_names(vc_data, "switch-01")
names = [m["suggested_name"] for m in result["members"]]
# Position 1 → "switch-01-M1", NOT "switch-01-M2"
assert names[0] == "switch-01-M1", f"Expected 'switch-01-M1' but got {names[0]!r} — off-by-one regression"
assert names[1] == "switch-01-M2", f"Expected 'switch-01-M2' but got {names[1]!r} — off-by-one regression"
def test_update_vc_member_suggested_names_preserves_position(self):
"""update_vc_member_suggested_names must write final position back to member dict."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import (
update_vc_member_suggested_names,
)
vc_data = {
"is_stack": True,
"member_count": 1,
"members": [{"serial": "S1", "position": 3}],
}
with patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
):
result = update_vc_member_suggested_names(vc_data, "router")
member = result["members"][0]
assert member["position"] == 3
assert member["suggested_name"] == "router-M3"
def test_update_vc_member_suggested_names_fallback_for_zero_position(self):
"""Position 0 must be replaced with 1-based fallback (idx+1)."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import (
update_vc_member_suggested_names,
)
vc_data = {
"is_stack": True,
"member_count": 2,
"members": [{"serial": "S1", "position": 0}, {"serial": "S2", "position": -1}],
}
with patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
):
result = update_vc_member_suggested_names(vc_data, "sw")
positions = [m["position"] for m in result["members"]]
assert positions[0] == 1, f"Zero position must fall back to 1, got {positions[0]}"
assert positions[1] == 2, f"Negative position must fall back to 2 (idx+1), got {positions[1]}"
# ---------------------------------------------------------------------------
# Additional virtual_chassis.py coverage
# ---------------------------------------------------------------------------
class TestEmptyVirtualChassisData:
"""Tests for empty_virtual_chassis_data helper."""
def test_returns_expected_structure(self):
from netbox_librenms_plugin.import_utils.virtual_chassis import empty_virtual_chassis_data
result = empty_virtual_chassis_data()
assert result["is_stack"] is False
assert result["member_count"] == 0
assert result["members"] == []
assert result["detection_error"] is None
def test_returns_new_dict_each_call(self):
"""Each call returns an independent dict (not a shared reference)."""
from netbox_librenms_plugin.import_utils.virtual_chassis import empty_virtual_chassis_data
a = empty_virtual_chassis_data()
b = empty_virtual_chassis_data()
a["members"].append("x")
assert b["members"] == []
class TestCloneVirtualChassisDataAdditional:
"""Additional _clone_virtual_chassis_data edge cases."""
def test_none_input_returns_empty(self):
from netbox_librenms_plugin.import_utils.virtual_chassis import _clone_virtual_chassis_data
result = _clone_virtual_chassis_data(None)
assert result["is_stack"] is False
assert result["members"] == []
def test_empty_dict_returns_empty(self):
from netbox_librenms_plugin.import_utils.virtual_chassis import _clone_virtual_chassis_data
result = _clone_virtual_chassis_data({})
assert result["is_stack"] is False
assert result["members"] == []
def test_full_data_defensive_copy(self):
"""Members list is a new list; mutating it does not affect the source."""
from netbox_librenms_plugin.import_utils.virtual_chassis import _clone_virtual_chassis_data
data = {
"is_stack": True,
"member_count": 1,
"members": [{"serial": "SN1", "position": 1}],
"detection_error": None,
}
result = _clone_virtual_chassis_data(data)
result["members"].append({"serial": "SN-NEW", "position": 2})
assert len(data["members"]) == 1 # original untouched
def test_detection_error_preserved(self):
"""detection_error field from source data is preserved."""
from netbox_librenms_plugin.import_utils.virtual_chassis import _clone_virtual_chassis_data
data = {
"is_stack": True,
"member_count": 1,
"members": [],
"detection_error": "Some error",
}
result = _clone_virtual_chassis_data(data)
assert result["detection_error"] == "Some error"
def test_member_with_zero_position_replaced_by_one_based(self):
"""A member with position=0 is replaced by idx+1 (1-based)."""
from netbox_librenms_plugin.import_utils.virtual_chassis import _clone_virtual_chassis_data
data = {
"is_stack": True,
"member_count": 2,
"members": [{"serial": "S0", "position": 0}, {"serial": "S2", "position": 2}],
}
result = _clone_virtual_chassis_data(data)
assert result["members"][0]["position"] == 1 # 0 → idx+1 = 1
assert result["members"][1]["position"] == 2 # kept as-is
def test_member_count_falls_back_to_len_when_zero(self):
"""member_count=0 in source is replaced by len(members)."""
from netbox_librenms_plugin.import_utils.virtual_chassis import _clone_virtual_chassis_data
data = {
"is_stack": True,
"member_count": 0,
"members": [{"serial": "S1", "position": 1}, {"serial": "S2", "position": 2}],
}
result = _clone_virtual_chassis_data(data)
assert result["member_count"] == 2
class TestVCCacheKey:
"""Tests for _vc_cache_key."""
def test_cache_key_format(self):
from netbox_librenms_plugin.import_utils.virtual_chassis import _vc_cache_key
mock_api = MagicMock()
mock_api.server_key = "default"
key = _vc_cache_key(mock_api, 42)
assert "librenms_vc_detection" in key
assert "default" in key
assert "42" in key
def test_cache_key_includes_server_key(self):
from netbox_librenms_plugin.import_utils.virtual_chassis import _vc_cache_key
api_a = MagicMock()
api_a.server_key = "server-a"
api_b = MagicMock()
api_b.server_key = "server-b"
assert _vc_cache_key(api_a, 1) != _vc_cache_key(api_b, 1)
def test_cache_key_differs_for_different_device_ids(self):
from netbox_librenms_plugin.import_utils.virtual_chassis import _vc_cache_key
mock_api = MagicMock()
mock_api.server_key = "default"
assert _vc_cache_key(mock_api, 1) != _vc_cache_key(mock_api, 2)
def test_missing_server_key_falls_back_to_default(self):
"""api without server_key attribute uses 'default' as fallback."""
from netbox_librenms_plugin.import_utils.virtual_chassis import _vc_cache_key
mock_api = MagicMock(spec=[]) # no attributes
key = _vc_cache_key(mock_api, 10)
assert "default" in key
class TestGetVirtualChassisData:
"""Tests for get_virtual_chassis_data."""
def test_none_api_returns_empty(self):
from netbox_librenms_plugin.import_utils.virtual_chassis import get_virtual_chassis_data
result = get_virtual_chassis_data(None, 1)
assert result["is_stack"] is False
assert result["members"] == []
def test_none_device_id_returns_empty(self):
from netbox_librenms_plugin.import_utils.virtual_chassis import get_virtual_chassis_data
mock_api = MagicMock()
result = get_virtual_chassis_data(mock_api, None)
assert result["is_stack"] is False
def test_cache_hit_returns_cloned_data(self):
"""Cached data is returned without calling detect_virtual_chassis_from_inventory."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import get_virtual_chassis_data
mock_api = MagicMock()
mock_api.server_key = "default"
cached = {
"is_stack": True,
"member_count": 2,
"members": [{"serial": "S1", "position": 1}, {"serial": "S2", "position": 2}],
"detection_error": None,
}
with (
patch("netbox_librenms_plugin.import_utils.virtual_chassis.cache") as mock_cache,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.detect_virtual_chassis_from_inventory"
) as mock_detect,
):
mock_cache.get.return_value = cached
result = get_virtual_chassis_data(mock_api, 42)
assert result["is_stack"] is True
assert result["member_count"] == 2
mock_detect.assert_not_called()
def test_cache_miss_calls_detect_and_stores_result(self):
"""On cache miss, detect_virtual_chassis_from_inventory is called and result cached."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import get_virtual_chassis_data
mock_api = MagicMock()
mock_api.server_key = "default"
mock_api.cache_timeout = 300
detection_result = {"is_stack": False, "member_count": 0, "members": []}
with (
patch("netbox_librenms_plugin.import_utils.virtual_chassis.cache") as mock_cache,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.detect_virtual_chassis_from_inventory",
return_value=detection_result,
) as mock_detect,
):
mock_cache.get.return_value = None # cache miss
result = get_virtual_chassis_data(mock_api, 42)
mock_detect.assert_called_once_with(mock_api, 42)
mock_cache.set.assert_called_once()
assert result["is_stack"] is False
def test_cache_miss_detect_returns_none_stores_empty(self):
"""When detect returns None (non-stack or API failure), empty result is cached to suppress repeated hits."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import get_virtual_chassis_data
mock_api = MagicMock()
mock_api.server_key = "default"
mock_api.cache_timeout = 300
with (
patch("netbox_librenms_plugin.import_utils.virtual_chassis.cache") as mock_cache,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.detect_virtual_chassis_from_inventory",
return_value=None,
),
):
mock_cache.get.return_value = None
result = get_virtual_chassis_data(mock_api, 99)
mock_cache.set.assert_called_once()
set_args = mock_cache.set.call_args
cached_val = set_args[0][1]
assert cached_val["is_stack"] is False
assert cached_val["member_count"] == 0
assert result["is_stack"] is False
def test_force_refresh_bypasses_cache(self):
"""force_refresh=True skips the cache.get check."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import get_virtual_chassis_data
mock_api = MagicMock()
mock_api.server_key = "default"
mock_api.cache_timeout = 300
with (
patch("netbox_librenms_plugin.import_utils.virtual_chassis.cache") as mock_cache,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.detect_virtual_chassis_from_inventory",
return_value=None,
),
):
mock_cache.get.return_value = {"is_stack": True, "member_count": 1, "members": [], "detection_error": None}
get_virtual_chassis_data(mock_api, 1, force_refresh=True)
# cache.get should NOT have been consulted
mock_cache.get.assert_not_called()
class TestPrefetchVCData:
"""Tests for prefetch_vc_data_for_devices."""
def test_none_api_returns_immediately(self):
"""None api causes early return without touching get_virtual_chassis_data."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import prefetch_vc_data_for_devices
with patch("netbox_librenms_plugin.import_utils.virtual_chassis.get_virtual_chassis_data") as mock_get:
prefetch_vc_data_for_devices(None, [1, 2, 3])
mock_get.assert_not_called()
def test_empty_device_ids_returns_immediately(self):
"""Empty device_ids list causes early return."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import prefetch_vc_data_for_devices
mock_api = MagicMock()
with patch("netbox_librenms_plugin.import_utils.virtual_chassis.get_virtual_chassis_data") as mock_get:
prefetch_vc_data_for_devices(mock_api, [])
mock_get.assert_not_called()
def test_connection_error_stops_processing(self):
"""BrokenPipeError / ConnectionError stops the loop (return, not continue)."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import prefetch_vc_data_for_devices
mock_api = MagicMock()
with patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.get_virtual_chassis_data",
side_effect=ConnectionError("Connection reset"),
) as mock_get:
prefetch_vc_data_for_devices(mock_api, [1, 2, 3])
# Only the first call fires before the connection error stops processing
assert mock_get.call_count == 1
def test_broken_pipe_error_stops_processing(self):
"""BrokenPipeError is treated the same as ConnectionError."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import prefetch_vc_data_for_devices
mock_api = MagicMock()
with patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.get_virtual_chassis_data",
side_effect=BrokenPipeError("Pipe broken"),
) as mock_get:
prefetch_vc_data_for_devices(mock_api, [10, 20])
assert mock_get.call_count == 1
def test_generic_exception_continues_to_next_device(self):
"""Non-connection exceptions are logged but processing continues."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import prefetch_vc_data_for_devices
mock_api = MagicMock()
with patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.get_virtual_chassis_data",
side_effect=ValueError("Unexpected"),
) as mock_get:
prefetch_vc_data_for_devices(mock_api, [1, 2, 3])
# All devices attempted despite the error
assert mock_get.call_count == 3
def test_success_calls_get_for_each_device(self):
"""All device IDs are prefetched when no errors occur."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import prefetch_vc_data_for_devices
mock_api = MagicMock()
with patch("netbox_librenms_plugin.import_utils.virtual_chassis.get_virtual_chassis_data") as mock_get:
prefetch_vc_data_for_devices(mock_api, [10, 20, 30])
assert mock_get.call_count == 3
class TestDetectVirtualChassisFromInventory:
"""Tests for detect_virtual_chassis_from_inventory."""
def test_no_root_items_returns_none(self):
"""Returns None when get_inventory_filtered returns no root items."""
from netbox_librenms_plugin.import_utils.virtual_chassis import detect_virtual_chassis_from_inventory
mock_api = MagicMock()
mock_api.get_device_info.return_value = (True, {"sysName": "sw1"})
mock_api.get_inventory_filtered.return_value = (False, None)
result = detect_virtual_chassis_from_inventory(mock_api, 1)
assert result is None
def test_empty_root_items_returns_none(self):
"""Returns None when root items list is empty."""
from netbox_librenms_plugin.import_utils.virtual_chassis import detect_virtual_chassis_from_inventory
mock_api = MagicMock()
mock_api.get_device_info.return_value = (True, {"sysName": "sw1"})
mock_api.get_inventory_filtered.return_value = (True, [])
result = detect_virtual_chassis_from_inventory(mock_api, 1)
assert result is None
def test_no_stack_or_chassis_parent_returns_none(self):
"""Returns None when no root item has class 'stack' or 'chassis'."""
from netbox_librenms_plugin.import_utils.virtual_chassis import detect_virtual_chassis_from_inventory
mock_api = MagicMock()
mock_api.get_device_info.return_value = (True, {"sysName": "sw1"})
mock_api.get_inventory_filtered.return_value = (
True,
[{"entPhysicalClass": "other", "entPhysicalIndex": 1}],
)
result = detect_virtual_chassis_from_inventory(mock_api, 1)
assert result is None
def test_single_child_chassis_returns_none(self):
"""Returns None when only one child chassis is found (not a stack)."""
from netbox_librenms_plugin.import_utils.virtual_chassis import detect_virtual_chassis_from_inventory
mock_api = MagicMock()
mock_api.get_device_info.return_value = (True, {"sysName": "sw1"})
mock_api.get_inventory_filtered.side_effect = [
(True, [{"entPhysicalClass": "stack", "entPhysicalIndex": 100}]),
(True, [{"entPhysicalClass": "chassis", "entPhysicalIndex": 200}]),
]
result = detect_virtual_chassis_from_inventory(mock_api, 1)
assert result is None
def test_stack_detected_with_two_chassis(self):
"""Returns stack dict when two or more chassis are found under the parent."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import detect_virtual_chassis_from_inventory
mock_api = MagicMock()
mock_api.get_device_info.return_value = (True, {"sysName": "sw1"})
mock_api.get_inventory_filtered.side_effect = [
(True, [{"entPhysicalClass": "stack", "entPhysicalIndex": 100}]),
(
True,
[
{
"entPhysicalClass": "chassis",
"entPhysicalIndex": 201,
"entPhysicalParentRelPos": 1,
"entPhysicalSerialNum": "SN1",
"entPhysicalModelName": "C9300-48P",
"entPhysicalName": "Switch 1",
"entPhysicalDescr": "Cisco Catalyst 9300",
},
{
"entPhysicalClass": "chassis",
"entPhysicalIndex": 202,
"entPhysicalParentRelPos": 2,
"entPhysicalSerialNum": "SN2",
"entPhysicalModelName": "C9300-48P",
"entPhysicalName": "Switch 2",
"entPhysicalDescr": "Cisco Catalyst 9300",
},
],
),
]
with patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
):
result = detect_virtual_chassis_from_inventory(mock_api, 1)
assert result is not None
assert result["is_stack"] is True
assert result["member_count"] == 2
assert len(result["members"]) == 2
assert result["members"][0]["serial"] == "SN1"
assert result["members"][1]["serial"] == "SN2"
def test_stack_members_sorted_by_position(self):
"""Members are sorted by position ascending."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import detect_virtual_chassis_from_inventory
mock_api = MagicMock()
mock_api.get_device_info.return_value = (True, {"sysName": "sw1"})
mock_api.get_inventory_filtered.side_effect = [
(True, [{"entPhysicalClass": "stack", "entPhysicalIndex": 100}]),
(
True,
[
{"entPhysicalClass": "chassis", "entPhysicalParentRelPos": 3, "entPhysicalIndex": 203},
{"entPhysicalClass": "chassis", "entPhysicalParentRelPos": 1, "entPhysicalIndex": 201},
{"entPhysicalClass": "chassis", "entPhysicalParentRelPos": 2, "entPhysicalIndex": 202},
],
),
]
with patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
):
result = detect_virtual_chassis_from_inventory(mock_api, 1)
positions = [m["position"] for m in result["members"]]
assert positions == [1, 2, 3]
def test_zero_position_replaced_by_one_based_index(self):
"""entPhysicalParentRelPos=0 is replaced by idx+1."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import detect_virtual_chassis_from_inventory
mock_api = MagicMock()
mock_api.get_device_info.return_value = (True, {"sysName": "sw1"})
mock_api.get_inventory_filtered.side_effect = [
(True, [{"entPhysicalClass": "stack", "entPhysicalIndex": 100}]),
(
True,
[
{"entPhysicalClass": "chassis", "entPhysicalParentRelPos": 0, "entPhysicalIndex": 201},
{"entPhysicalClass": "chassis", "entPhysicalParentRelPos": 2, "entPhysicalIndex": 202},
],
),
]
with patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
):
result = detect_virtual_chassis_from_inventory(mock_api, 1)
positions = [m["position"] for m in result["members"]]
assert 0 not in positions
assert 1 in positions
def test_no_master_name_uses_member_prefix(self):
"""When device_info has no sysName/hostname, suggested_name uses 'Member-N'."""
from netbox_librenms_plugin.import_utils.virtual_chassis import detect_virtual_chassis_from_inventory
mock_api = MagicMock()
mock_api.get_device_info.return_value = (False, None) # no master name
mock_api.get_inventory_filtered.side_effect = [
(True, [{"entPhysicalClass": "stack", "entPhysicalIndex": 100}]),
(
True,
[
{"entPhysicalClass": "chassis", "entPhysicalParentRelPos": 1, "entPhysicalIndex": 201},
{"entPhysicalClass": "chassis", "entPhysicalParentRelPos": 2, "entPhysicalIndex": 202},
],
),
]
result = detect_virtual_chassis_from_inventory(mock_api, 1)
assert result is not None
assert result["members"][0]["suggested_name"].startswith("Member-")
def test_child_items_fetch_fails_returns_none(self):
"""Returns None when the second get_inventory_filtered call fails."""
from netbox_librenms_plugin.import_utils.virtual_chassis import detect_virtual_chassis_from_inventory
mock_api = MagicMock()
mock_api.get_device_info.return_value = (True, {"sysName": "sw1"})
mock_api.get_inventory_filtered.side_effect = [
(True, [{"entPhysicalClass": "stack", "entPhysicalIndex": 100}]),
(False, None), # child fetch fails
]
result = detect_virtual_chassis_from_inventory(mock_api, 1)
assert result is None
def test_exception_returns_none(self):
"""Unhandled exception inside the function returns None."""
from netbox_librenms_plugin.import_utils.virtual_chassis import detect_virtual_chassis_from_inventory
mock_api = MagicMock()
mock_api.get_device_info.side_effect = RuntimeError("Unexpected")
result = detect_virtual_chassis_from_inventory(mock_api, 1)
assert result is None
class TestLoadVCMemberNamePattern:
"""Tests for _load_vc_member_name_pattern."""
def test_returns_pattern_from_settings(self):
"""Returns vc_member_name_pattern from LibreNMSSettings when found."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import _load_vc_member_name_pattern
mock_settings = MagicMock()
mock_settings.vc_member_name_pattern = "-SW{position}"
with patch("netbox_librenms_plugin.models.LibreNMSSettings") as mock_cls:
mock_cls.objects.order_by.return_value.first.return_value = mock_settings
result = _load_vc_member_name_pattern()
assert result == "-SW{position}"
def test_no_settings_returns_default(self):
"""Returns '-M{position}' when LibreNMSSettings.objects.order_by().first() returns None."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import _load_vc_member_name_pattern
with patch("netbox_librenms_plugin.models.LibreNMSSettings") as mock_cls:
mock_cls.objects.order_by.return_value.first.return_value = None
result = _load_vc_member_name_pattern()
assert result == "-M{position}"
def test_exception_returns_default(self):
"""Returns '-M{position}' when the DB query raises an exception."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import _load_vc_member_name_pattern
with patch("netbox_librenms_plugin.models.LibreNMSSettings") as mock_cls:
mock_cls.objects.order_by.side_effect = Exception("DB offline")
result = _load_vc_member_name_pattern()
assert result == "-M{position}"
class TestGenerateVCMemberNameAdditional:
"""Additional tests for _generate_vc_member_name."""
def test_with_serial_in_pattern(self):
"""Pattern using {serial} placeholder substitutes the serial number."""
from netbox_librenms_plugin.import_utils.virtual_chassis import _generate_vc_member_name
name = _generate_vc_member_name("switch-1", 2, serial="ABC123", pattern=" [{serial}]")
assert name == "switch-1 [ABC123]"
def test_empty_serial_produces_empty_brackets(self):
"""Empty serial with {serial} pattern results in empty brackets."""
from netbox_librenms_plugin.import_utils.virtual_chassis import _generate_vc_member_name
name = _generate_vc_member_name("switch-1", 1, serial="", pattern=" [{serial}]")
assert name == "switch-1 []"
def test_invalid_placeholder_falls_back_to_default(self):
"""A KeyError from an unknown placeholder triggers the '-M{position}' fallback."""
from netbox_librenms_plugin.import_utils.virtual_chassis import _generate_vc_member_name
name = _generate_vc_member_name("switch-1", 3, pattern="-{nonexistent_key}")
assert name == "switch-1-M3"
def test_none_pattern_loads_from_settings(self):
"""When pattern=None, _load_vc_member_name_pattern is called to fetch the pattern."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import _generate_vc_member_name
with patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
) as mock_load:
name = _generate_vc_member_name("router", 5, pattern=None)
mock_load.assert_called_once()
assert name == "router-M5"
def test_master_name_placeholder(self):
"""Pattern can also reference {master_name}."""
from netbox_librenms_plugin.import_utils.virtual_chassis import _generate_vc_member_name
name = _generate_vc_member_name("sw", 2, pattern="-{master_name}-pos{position}")
assert name == "sw-sw-pos2"
class TestUpdateVCMemberSuggestedNamesAdditional:
"""Additional tests for update_vc_member_suggested_names."""
def test_not_stack_returns_vc_data_unchanged(self):
"""When is_stack=False, the function returns immediately without modifying members."""
from netbox_librenms_plugin.import_utils.virtual_chassis import update_vc_member_suggested_names
vc_data = {
"is_stack": False,
"members": [{"serial": "S1", "position": 1, "suggested_name": "old-name"}],
}
result = update_vc_member_suggested_names(vc_data, "sw")
# suggested_name must not be regenerated
assert result["members"][0]["suggested_name"] == "old-name"
def test_none_vc_data_returns_none(self):
"""None input is returned as-is (falsy guard)."""
from netbox_librenms_plugin.import_utils.virtual_chassis import update_vc_member_suggested_names
result = update_vc_member_suggested_names(None, "sw")
assert result is None
def test_no_members_returns_empty_members(self):
"""is_stack=True with empty members list processes without error."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import update_vc_member_suggested_names
vc_data = {"is_stack": True, "members": []}
with patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
):
result = update_vc_member_suggested_names(vc_data, "sw")
assert result["members"] == []
class TestCreateVirtualChassisWithMembers:
"""Tests for create_virtual_chassis_with_members."""
def test_raises_when_vc_create_fails(self):
"""Exception from VirtualChassis.objects.create is re-raised to the caller."""
from contextlib import contextmanager
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import create_virtual_chassis_with_members
master_device = MagicMock()
master_device.name = "sw1"
master_device.pk = 1
master_device.serial = ""
@contextmanager
def mock_atomic():
yield
with (
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.transaction.atomic",
mock_atomic,
),
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._generate_vc_member_name",
return_value="sw1-M1",
),
patch("netbox_librenms_plugin.import_utils.virtual_chassis.Device") as mock_device_cls,
patch("netbox_librenms_plugin.import_utils.virtual_chassis.VirtualChassis") as mock_vc_cls,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
),
):
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = False
mock_vc_cls.objects.create.side_effect = Exception("DB error")
import pytest
with pytest.raises(Exception, match="DB error"):
create_virtual_chassis_with_members(master_device, [], {"device_id": 1})
def test_success_with_no_members(self):
"""Happy path with empty members_info creates VC and returns it."""
from contextlib import contextmanager
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import create_virtual_chassis_with_members
master_device = MagicMock()
master_device.name = "sw1"
master_device.pk = 1
master_device.serial = ""
mock_vc = MagicMock()
mock_vc.members.count.return_value = 1
@contextmanager
def mock_atomic():
yield
with (
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.transaction.atomic",
mock_atomic,
),
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._generate_vc_member_name",
return_value="sw1-M1",
),
patch("netbox_librenms_plugin.import_utils.virtual_chassis.Device") as mock_device_cls,
patch("netbox_librenms_plugin.import_utils.virtual_chassis.VirtualChassis") as mock_vc_cls,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
),
):
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = False
mock_vc_cls.objects.create.return_value = mock_vc
result = create_virtual_chassis_with_members(master_device, [], {"device_id": 1})
assert result == mock_vc
mock_vc_cls.objects.create.assert_called_once()
def test_calls_module_bay_counter_sync(self):
"""VC creation calls counter sync helper after assigning master."""
from contextlib import contextmanager
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import create_virtual_chassis_with_members
master_device = MagicMock()
master_device.name = "sw1"
master_device.pk = 1
master_device.serial = ""
mock_vc = MagicMock()
mock_vc.members.count.return_value = 1
@contextmanager
def mock_atomic():
yield
with (
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.transaction.atomic",
mock_atomic,
),
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._generate_vc_member_name",
return_value="sw1-M1",
),
patch("netbox_librenms_plugin.import_utils.virtual_chassis.Device") as mock_device_cls,
patch("netbox_librenms_plugin.import_utils.virtual_chassis.VirtualChassis") as mock_vc_cls,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
),
patch("netbox_librenms_plugin.import_utils.virtual_chassis._sync_module_bay_counter") as mock_sync,
):
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = False
mock_vc_cls.objects.create.return_value = mock_vc
create_virtual_chassis_with_members(master_device, [], {"device_id": 1})
mock_sync.assert_called_once_with(master_device)
def test_master_save_uses_update_fields(self):
"""Master save should update only VC/name fields, not stale counter fields."""
from contextlib import contextmanager
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import create_virtual_chassis_with_members
master_device = MagicMock()
master_device.name = "sw1"
master_device.pk = 1
master_device.serial = ""
mock_vc = MagicMock()
mock_vc.members.count.return_value = 1
@contextmanager
def mock_atomic():
yield
with (
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.transaction.atomic",
mock_atomic,
),
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._generate_vc_member_name",
return_value="sw1-M1",
),
patch("netbox_librenms_plugin.import_utils.virtual_chassis.Device") as mock_device_cls,
patch("netbox_librenms_plugin.import_utils.virtual_chassis.VirtualChassis") as mock_vc_cls,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
),
patch("netbox_librenms_plugin.import_utils.virtual_chassis._sync_module_bay_counter"),
):
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = False
mock_vc_cls.objects.create.return_value = mock_vc
create_virtual_chassis_with_members(master_device, [], {"device_id": 1})
master_device.save.assert_called_once_with(update_fields=["virtual_chassis", "vc_position", "name"])
class TestSyncModuleBayCounter:
"""Tests for module_bay_count synchronization helper."""
def test_syncs_counter_when_actual_differs(self):
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import _sync_module_bay_counter
device = MagicMock()
device.pk = 42
device.name = "sw1"
device.module_bay_count = 0
device.modulebays.count.return_value = 2
with patch("netbox_librenms_plugin.import_utils.virtual_chassis.Device") as mock_device_cls:
_sync_module_bay_counter(device)
mock_device_cls.objects.filter.assert_called_once_with(pk=42)
mock_device_cls.objects.filter.return_value.update.assert_called_once_with(module_bay_count=2)
assert device.module_bay_count == 2
def test_no_op_when_counter_matches(self):
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import _sync_module_bay_counter
device = MagicMock()
device.pk = 42
device.name = "sw1"
device.module_bay_count = 3
device.modulebays.count.return_value = 3
with patch("netbox_librenms_plugin.import_utils.virtual_chassis.Device") as mock_device_cls:
_sync_module_bay_counter(device)
mock_device_cls.objects.filter.assert_not_called()
class TestBulkImportCancellation:
"""Test that bulk_import_devices_shared respects RQ and DB cancellation."""
def _run_bulk_import(self, mock_rq_job=None, db_status="running", device_ids=None):
"""Helper: run bulk_import with provided mocks, return import call count."""
from unittest.mock import MagicMock, patch
if device_ids is None:
device_ids = [1, 2, 3, 4, 5, 6]
job = MagicMock()
job.job.job_id = "test-uuid"
job_status = MagicMock()
job_status.value = db_status
job.job.status = job_status
job.logger = MagicMock()
with (
patch("netbox_librenms_plugin.import_utils.bulk_import.LibreNMSAPI") as mock_api_cls,
patch("netbox_librenms_plugin.import_utils.bulk_import.import_single_device") as mock_import,
patch("netbox_librenms_plugin.import_utils.bulk_import.validate_device_for_import"),
patch("netbox_librenms_plugin.import_utils.bulk_import.require_permissions"),
# Inline imports in the loop use django_rq.get_queue / rq.job.Job directly
patch("django_rq.get_queue") as mock_get_queue,
patch("rq.job.Job") as mock_rqjob_cls,
):
mock_api = MagicMock()
mock_api.server_key = "default"
mock_api.get_device_info.return_value = (True, {"device_id": 1, "hostname": "sw"})
mock_api_cls.return_value = mock_api
mock_import.return_value = {"success": True, "device": MagicMock(), "is_vm": False}
if mock_rq_job is not None:
mock_conn = MagicMock()
mock_queue = MagicMock()
mock_queue.connection = mock_conn
mock_get_queue.return_value = mock_queue
mock_rqjob_cls.fetch.return_value = mock_rq_job
else:
# Simulate RQ unavailable — get_queue raises, triggers DB fallback
mock_get_queue.side_effect = Exception("RQ unavailable")
from netbox_librenms_plugin.import_utils.bulk_import import bulk_import_devices_shared
result = bulk_import_devices_shared(device_ids, user=MagicMock(), server_key=None, job=job)
return mock_import.call_count, result
def test_rq_stopped_cancels_import_loop(self):
"""When RQ job is_stopped, import loop should break early."""
rq_job = MagicMock()
rq_job.is_stopped = True
rq_job.is_failed = False
rq_job.get_status.return_value = "stopped"
# With 6 devices and RQ stopped on first check (idx=1), at most 1 device processed
count, result = self._run_bulk_import(mock_rq_job=rq_job, device_ids=[1, 2, 3, 4, 5, 6])
assert count == 0 # break before first import
assert result.get("cancelled") is True
def test_rq_failed_cancels_import_loop(self):
"""When RQ job is_failed, import loop should break early."""
rq_job = MagicMock()
rq_job.is_stopped = False
rq_job.is_failed = True
rq_job.get_status.return_value = "failed"
count, result = self._run_bulk_import(mock_rq_job=rq_job, device_ids=[1, 2, 3, 4, 5, 6])
assert count == 0
assert result.get("cancelled") is True
def test_rq_unavailable_treats_as_not_cancelled(self):
"""When RQ is unavailable, _is_job_cancelled returns False so import continues."""
# mock_rq_job=None triggers the side_effect=Exception path
count, result = self._run_bulk_import(mock_rq_job=None, db_status="failed", device_ids=[1, 2, 3])
# Redis unavailable → not cancelled → all devices processed
assert count == 3
assert result.get("cancelled") is False
def test_healthy_job_runs_all_devices(self):
"""When job is healthy, all devices should be imported."""
rq_job = MagicMock()
rq_job.is_stopped = False
rq_job.is_failed = False
rq_job.get_status.return_value = "started"
count, result = self._run_bulk_import(mock_rq_job=rq_job, device_ids=[1, 2, 3])
assert count == 3
assert result.get("cancelled") is False
# ---------------------------------------------------------------------------
# Tests for VC permission guard in bulk_import_devices_shared (closes #31)
# ---------------------------------------------------------------------------
class TestBulkImportVCPermission:
"""Test VC creation behavior during bulk import."""
def _make_stack_validation(self):
from unittest.mock import MagicMock
v = MagicMock()
v.get.side_effect = lambda k, d=None: {
"is_ready": True,
"import_as_vm": False,
"existing_device": None,
"virtual_chassis": {
"is_stack": True,
"members": [
{"serial": "SN-A", "position": 1},
{"serial": "SN-B", "position": 2},
],
},
}.get(k, d)
return v
def _make_non_stack_validation(self):
from unittest.mock import MagicMock
v = MagicMock()
v.get.side_effect = lambda k, d=None: {
"is_ready": True,
"import_as_vm": False,
"existing_device": None,
"virtual_chassis": {
"is_stack": False,
"members": [],
},
}.get(k, d)
return v
def test_stack_device_not_imported_without_vc_permission(self):
"""Missing dcim.add_virtualchassis permission should block stack device import."""
from unittest.mock import MagicMock, patch
mock_device = MagicMock()
user = MagicMock()
user.has_perm.side_effect = lambda p: p != "dcim.add_virtualchassis"
with (
patch("netbox_librenms_plugin.import_utils.bulk_import.require_permissions"),
patch("netbox_librenms_plugin.import_utils.bulk_import.LibreNMSAPI"),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.validate_device_for_import",
return_value=self._make_stack_validation(),
),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.import_single_device",
return_value={"success": True, "device": mock_device, "message": "ok", "is_vm": False},
) as mock_import_single,
patch(
"netbox_librenms_plugin.import_utils.bulk_import.create_virtual_chassis_with_members",
return_value=MagicMock(name="vc"),
) as mock_create_vc,
):
from netbox_librenms_plugin.import_utils.bulk_import import bulk_import_devices_shared
result = bulk_import_devices_shared(
device_ids=[1],
user=user,
libre_devices_cache={1: {"device_id": 1, "hostname": "sw"}},
)
mock_import_single.assert_not_called()
mock_create_vc.assert_not_called()
user.has_perm.assert_called_with("dcim.add_virtualchassis")
assert len(result["success"]) == 0
assert len(result["failed"]) == 1
assert "missing permission dcim.add_virtualchassis" in result["failed"][0]["error"]
assert result["virtual_chassis_created"] == 0
def test_vc_creation_proceeds_with_vc_permission(self):
"""User has dcim.add_virtualchassis → VC creation proceeds normally."""
from unittest.mock import MagicMock, patch
mock_device = MagicMock()
mock_vc = MagicMock()
mock_vc.name = "VC-Stack"
user = MagicMock()
user.has_perm.return_value = True
with (
patch("netbox_librenms_plugin.import_utils.bulk_import.require_permissions"),
patch("netbox_librenms_plugin.import_utils.bulk_import.LibreNMSAPI"),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.validate_device_for_import",
return_value=self._make_stack_validation(),
),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.import_single_device",
return_value={"success": True, "device": mock_device, "message": "ok", "is_vm": False},
),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.create_virtual_chassis_with_members",
return_value=mock_vc,
) as mock_create_vc,
):
from netbox_librenms_plugin.import_utils.bulk_import import bulk_import_devices_shared
result = bulk_import_devices_shared(
device_ids=[1],
user=user,
libre_devices_cache={1: {"device_id": 1, "hostname": "sw"}},
)
mock_create_vc.assert_called_once()
user.has_perm.assert_called_with("dcim.add_virtualchassis")
assert result["virtual_chassis_created"] == 1
def test_non_stack_import_works_without_vc_permission(self):
"""Non-stack devices should still import when add_device permissions are present."""
from unittest.mock import MagicMock, patch
mock_device = MagicMock()
user = MagicMock()
user.has_perm.side_effect = lambda p: p != "dcim.add_virtualchassis"
with (
patch("netbox_librenms_plugin.import_utils.bulk_import.require_permissions"),
patch("netbox_librenms_plugin.import_utils.bulk_import.LibreNMSAPI"),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.validate_device_for_import",
return_value=self._make_non_stack_validation(),
),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.import_single_device",
return_value={"success": True, "device": mock_device, "message": "ok", "is_vm": False},
) as mock_import_single,
patch(
"netbox_librenms_plugin.import_utils.bulk_import.create_virtual_chassis_with_members",
return_value=MagicMock(name="vc"),
) as mock_create_vc,
):
from netbox_librenms_plugin.import_utils.bulk_import import bulk_import_devices_shared
result = bulk_import_devices_shared(
device_ids=[1],
user=user,
libre_devices_cache={1: {"device_id": 1, "hostname": "sw"}},
)
mock_import_single.assert_called_once()
mock_create_vc.assert_not_called()
assert len(result["success"]) == 1
assert len(result["failed"]) == 0
def test_mixed_stack_and_non_stack_without_vc_permission(self):
"""Stack device fails, but non-stack device still imports when VC permission is missing."""
from unittest.mock import MagicMock, patch
mock_device = MagicMock()
user = MagicMock()
user.has_perm.side_effect = lambda p: p != "dcim.add_virtualchassis"
validations = [self._make_stack_validation(), self._make_non_stack_validation()]
with (
patch("netbox_librenms_plugin.import_utils.bulk_import.require_permissions"),
patch("netbox_librenms_plugin.import_utils.bulk_import.LibreNMSAPI"),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.validate_device_for_import",
side_effect=validations,
),
patch(
"netbox_librenms_plugin.import_utils.bulk_import.import_single_device",
return_value={"success": True, "device": mock_device, "message": "ok", "is_vm": False},
) as mock_import_single,
patch(
"netbox_librenms_plugin.import_utils.bulk_import.create_virtual_chassis_with_members",
return_value=MagicMock(name="vc"),
) as mock_create_vc,
):
from netbox_librenms_plugin.import_utils.bulk_import import bulk_import_devices_shared
result = bulk_import_devices_shared(
device_ids=[1, 2],
user=user,
libre_devices_cache={
1: {"device_id": 1, "hostname": "stack-sw"},
2: {"device_id": 2, "hostname": "edge-sw"},
},
)
mock_import_single.assert_called_once()
mock_create_vc.assert_not_called()
assert len(result["success"]) == 1
assert len(result["failed"]) == 1
assert result["failed"][0]["device_id"] == 1
# ---------------------------------------------------------------------------
# Tests for DeviceValidationDetailsView._build_id_server_info
# ---------------------------------------------------------------------------
class TestBuildIdServerInfo:
"""Test DeviceValidationDetailsView._build_id_server_info method."""
def _make_device(self, librenms_id_value):
from unittest.mock import MagicMock
device = MagicMock()
device.custom_field_data = {"librenms_id": librenms_id_value}
return device
def test_returns_none_for_legacy_int(self):
from netbox_librenms_plugin.views.imports.actions import DeviceValidationDetailsView
device = self._make_device(42)
result = DeviceValidationDetailsView._build_id_server_info(device)
assert result is None
def test_returns_none_for_missing_cf(self):
from netbox_librenms_plugin.views.imports.actions import DeviceValidationDetailsView
device = self._make_device(None)
result = DeviceValidationDetailsView._build_id_server_info(device)
assert result is None
def test_single_server_resolves_display_name(self):
from unittest.mock import patch
from netbox_librenms_plugin.views.imports.actions import DeviceValidationDetailsView
device = self._make_device({"production": 42})
plugins_cfg = {
"netbox_librenms_plugin": {
"servers": {
"production": {"display_name": "Production LibreNMS", "librenms_url": "https://prod.example.com"},
}
}
}
with patch("django.conf.settings") as mock_settings:
mock_settings.PLUGINS_CONFIG = plugins_cfg
result = DeviceValidationDetailsView._build_id_server_info(device)
assert result is not None
assert len(result) == 1
assert result[0]["server_key"] == "production"
assert result[0]["display_name"] == "Production LibreNMS"
assert result[0]["device_id"] == 42
def test_unconfigured_server_uses_key_as_display_name(self):
from unittest.mock import patch
from netbox_librenms_plugin.views.imports.actions import DeviceValidationDetailsView
device = self._make_device({"deleted-server": 77})
plugins_cfg = {"netbox_librenms_plugin": {"servers": {}}}
with patch("django.conf.settings") as mock_settings:
mock_settings.PLUGINS_CONFIG = plugins_cfg
result = DeviceValidationDetailsView._build_id_server_info(device)
assert result is not None
assert result[0]["display_name"] == "deleted-server"
def test_empty_dict_returns_none(self):
from netbox_librenms_plugin.views.imports.actions import DeviceValidationDetailsView
device = self._make_device({})
result = DeviceValidationDetailsView._build_id_server_info(device)
assert result is None
# ---------------------------------------------------------------------------
# Tests for _refresh_existing_device sys_name fallback fix
# ---------------------------------------------------------------------------
class TestRefreshExistingDeviceSysNameFallback:
"""Test that _refresh_existing_device tries sys_name even when hostname is empty."""
def test_sysname_used_when_hostname_empty(self):
"""When hostname is empty but sys_name matches, the device is found in validation."""
from unittest.mock import MagicMock, patch
from netbox_librenms_plugin.import_utils.bulk_import import _refresh_existing_device
mock_device = MagicMock()
mock_device.pk = 99
mock_device.name = "router-01"
mock_device.custom_field_data = {"librenms_id": None}
libre_device = {
"device_id": 55,
"hostname": "", # empty hostname
"sysName": "router-01",
"serial": "SN-MATCH",
}
validation = {
"existing_device": None,
"existing_vm": None,
"import_as_vm": False,
"is_ready": False,
"can_import": False,
}
# sys_name lookup: filter(name__iexact="router-01") returns mock_device
# hostname lookup: filter(name__iexact="") returns None
def make_qs(return_val):
qs = MagicMock()
qs.first.return_value = return_val
return qs
with patch("netbox_librenms_plugin.import_utils.bulk_import.find_by_librenms_id", return_value=None):
import dcim.models as dcim_models
import virtualization.models as virt_models
with (
patch.object(
dcim_models.Device.objects,
"filter",
side_effect=lambda **kw: make_qs(mock_device if kw.get("name__iexact") == "router-01" else None),
),
patch.object(virt_models.VirtualMachine.objects, "filter", return_value=make_qs(None)),
):
_refresh_existing_device(validation, libre_device=libre_device, server_key="default")
assert validation["existing_device"] is mock_device
def test_hostname_lookup_succeeds_without_sysname(self):
"""When hostname is non-empty and matches, validation is updated correctly."""
from unittest.mock import MagicMock, patch
from netbox_librenms_plugin.import_utils.bulk_import import _refresh_existing_device
mock_device = MagicMock()
mock_device.pk = 10
mock_device.name = "sw-01"
mock_device.custom_field_data = {"librenms_id": None}
libre_device = {
"device_id": 10,
"hostname": "sw-01",
"sysName": "sw-01-sysname",
"serial": "",
}
validation = {
"existing_device": None,
"existing_vm": None,
"import_as_vm": False,
"is_ready": False,
"can_import": False,
}
def make_qs(return_val):
qs = MagicMock()
qs.first.return_value = return_val
return qs
with patch("netbox_librenms_plugin.import_utils.bulk_import.find_by_librenms_id", return_value=None):
import dcim.models as dcim_models
import virtualization.models as virt_models
with (
patch.object(
dcim_models.Device.objects,
"filter",
side_effect=lambda **kw: make_qs(mock_device if kw.get("name__iexact") == "sw-01" else None),
),
patch.object(virt_models.VirtualMachine.objects, "filter", return_value=make_qs(None)),
):
_refresh_existing_device(validation, libre_device=libre_device, server_key="default")
assert validation["existing_device"] is mock_device
# ---------------------------------------------------------------------------
# Tests for _get_hostname_for_action helper
# ---------------------------------------------------------------------------
class TestGetHostnameForAction:
"""Test _get_hostname_for_action helper in actions.py."""
def test_returns_resolved_name_when_set(self):
from unittest.mock import MagicMock
from netbox_librenms_plugin.views.imports.actions import _get_hostname_for_action
request = MagicMock()
validation = {"resolved_name": "cached-name"}
libre_device = {"hostname": "raw-hostname", "sysName": "raw-sysname"}
result = _get_hostname_for_action(request, validation, libre_device)
assert result == "cached-name"
def test_falls_back_to_determine_device_name(self):
from unittest.mock import MagicMock, patch
from netbox_librenms_plugin.views.imports.actions import _get_hostname_for_action
request = MagicMock()
validation = {} # no resolved_name
libre_device = {"hostname": "host.example.com", "sysName": "host"}
with patch("netbox_librenms_plugin.views.imports.actions.resolve_naming_preferences") as mock_prefs:
mock_prefs.return_value = (False, False) # use_sysname=False, strip_domain=False
with patch("netbox_librenms_plugin.views.imports.actions._determine_device_name") as mock_name:
mock_name.return_value = "host.example.com"
result = _get_hostname_for_action(request, validation, libre_device)
assert result == "host.example.com"
mock_prefs.assert_called_once_with(request)
mock_name.assert_called_once()
# ---------------------------------------------------------------------------
# Tests for resolve_naming_preferences underscore-variant key support
# ---------------------------------------------------------------------------
class TestResolveNamingPreferencesKeys:
"""Test that resolve_naming_preferences handles both hyphenated and underscored keys."""
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_hyphenated_post_key_use_sysname(self):
from unittest.mock import patch
from netbox_librenms_plugin.utils import resolve_naming_preferences
request = self._make_request(post={"use-sysname-toggle": "on", "strip-domain-toggle": "off"})
with patch("netbox_librenms_plugin.utils.get_user_pref", return_value=None):
use_sysname, strip_domain = resolve_naming_preferences(request)
assert use_sysname is True
assert strip_domain is False
def test_underscored_post_key_use_sysname(self):
"""Underscore variant 'use_sysname-toggle' should also be recognised."""
from unittest.mock import patch
from netbox_librenms_plugin.utils import resolve_naming_preferences
request = self._make_request(post={"use_sysname-toggle": "on", "strip_domain-toggle": "on"})
with patch("netbox_librenms_plugin.utils.get_user_pref", return_value=None):
use_sysname, strip_domain = resolve_naming_preferences(request)
assert use_sysname is True
assert strip_domain is True
def test_get_key_used_when_not_in_post(self):
from unittest.mock import patch
from netbox_librenms_plugin.utils import resolve_naming_preferences
request = self._make_request(get={"use-sysname-toggle": "off", "strip-domain-toggle": "on"})
with patch("netbox_librenms_plugin.utils.get_user_pref", return_value=None):
use_sysname, strip_domain = resolve_naming_preferences(request)
assert use_sysname is False
assert strip_domain is True
def test_user_pref_used_when_no_toggle_in_request(self):
from unittest.mock import patch
from netbox_librenms_plugin.utils import resolve_naming_preferences
request = self._make_request()
with patch("netbox_librenms_plugin.utils.get_user_pref") as mock_pref:
mock_pref.side_effect = lambda req, key: False if "use_sysname" in key else True
use_sysname, strip_domain = resolve_naming_preferences(request)
assert use_sysname is False
assert strip_domain is True
def test_post_takes_precedence_over_user_pref(self):
from unittest.mock import patch
from netbox_librenms_plugin.utils import resolve_naming_preferences
request = self._make_request(post={"use-sysname-toggle": "off"})
# user_pref would say True — POST should win
with patch("netbox_librenms_plugin.utils.get_user_pref", return_value=True):
use_sysname, _ = resolve_naming_preferences(request)
assert use_sysname is False
def test_truthy_string_true_value(self):
"""'true' and '1' (in addition to 'on') should be treated as True."""
from unittest.mock import patch
from netbox_librenms_plugin.utils import resolve_naming_preferences
for truthy_val in ("true", "True", "TRUE", "1"):
request = self._make_request(post={"use-sysname-toggle": truthy_val, "strip-domain-toggle": "off"})
with patch("netbox_librenms_plugin.utils.get_user_pref", return_value=None):
use_sysname, _ = resolve_naming_preferences(request)
assert use_sysname is True, f"Expected True for value {truthy_val!r}"
def test_falsy_string_false_value(self):
"""Unrecognised strings should be treated as False."""
from unittest.mock import patch
from netbox_librenms_plugin.utils import resolve_naming_preferences
for falsy_val in ("off", "false", "0", "", "no"):
request = self._make_request(post={"use-sysname-toggle": falsy_val, "strip-domain-toggle": "off"})
with patch("netbox_librenms_plugin.utils.get_user_pref", return_value=None):
use_sysname, _ = resolve_naming_preferences(request)
assert use_sysname is False, f"Expected False for value {falsy_val!r}"
# ---------------------------------------------------------------------------
# Tests for vc_domain stack dedup key fix
# ---------------------------------------------------------------------------
class TestVCDomainStackDedup:
"""Test that bulk_import_devices_shared deduplicates VC creation by member serials."""
def test_vc_domain_uses_member_serials(self):
"""vc_domain for two stack members with the same serials should be identical."""
# The logic lives inline; test the produced key directly from vc_data
members = [
{"serial": "SN100", "position": 1},
{"serial": "SN200", "position": 2},
]
member_serials = sorted(m.get("serial") for m in members if m.get("serial"))
vc_domain = f"librenms-stack-{','.join(member_serials)}"
# Same members from a different device's perspective should produce the same key
assert vc_domain == "librenms-stack-SN100,SN200"
def test_vc_domain_fallback_to_device_id_when_no_serials(self):
"""When no member serials are available, device_id is used as fallback."""
members = [
{"position": 1},
{"position": 2},
]
member_serials = sorted(m.get("serial") for m in members if m.get("serial"))
device_id = 42
vc_domain = f"librenms-stack-{','.join(member_serials)}" if member_serials else f"librenms-{device_id}"
assert vc_domain == "librenms-42"
def test_different_stacks_produce_different_keys(self):
"""Two stacks with different serials produce distinct dedup keys."""
members_a = [{"serial": "SN-A1"}, {"serial": "SN-A2"}]
members_b = [{"serial": "SN-B1"}, {"serial": "SN-B2"}]
key_a = f"librenms-stack-{','.join(sorted(m['serial'] for m in members_a))}"
key_b = f"librenms-stack-{','.join(sorted(m['serial'] for m in members_b))}"
assert key_a != key_b
class TestVirtualChassisEdgeBranches:
"""Targeted tests for exception branches not covered by main tests."""
def test_detect_vc_invalid_position_string_falls_back(self):
"""When entPhysicalParentRelPos is a non-numeric string, position falls back to idx+1."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import detect_virtual_chassis_from_inventory
mock_api = MagicMock()
mock_api.get_device_info.return_value = (True, {"sysName": "sw1"})
mock_api.get_inventory_filtered.side_effect = [
(True, [{"entPhysicalClass": "stack", "entPhysicalIndex": 100}]),
(
True,
[
{"entPhysicalClass": "chassis", "entPhysicalParentRelPos": "bad", "entPhysicalIndex": 201},
{"entPhysicalClass": "chassis", "entPhysicalParentRelPos": "invalid", "entPhysicalIndex": 202},
],
),
]
with patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
):
result = detect_virtual_chassis_from_inventory(mock_api, 1)
# invalid string → idx+1 fallback (1-based: idx=0→1, idx=1→2)
positions = sorted(m["position"] for m in result["members"])
assert positions == [1, 2]
def test_update_vc_suggested_names_invalid_position_string_falls_back(self):
"""Non-numeric position string in member triggers except branch → idx+1 fallback."""
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import update_vc_member_suggested_names
vc_data = {
"is_stack": True,
"member_count": 2,
"members": [{"serial": "S1", "position": "bad"}, {"serial": "S2", "position": None}],
}
with patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
):
result = update_vc_member_suggested_names(vc_data, "sw")
positions = [m["position"] for m in result["members"]]
assert positions[0] == 1 # idx=0 → 1
assert positions[1] == 2 # idx=1 → 2
def _make_atomic(self):
from contextlib import contextmanager
@contextmanager
def _atomic():
yield
return _atomic
def _base_patches(self):
from unittest.mock import patch
return [
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.transaction.atomic",
self._make_atomic(),
),
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
),
]
def test_create_vc_master_name_conflict_keeps_original(self):
"""When the renamed master clashes, master_base_name stays as original."""
from contextlib import contextmanager
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import create_virtual_chassis_with_members
master_device = MagicMock()
master_device.name = "sw1"
master_device.pk = 1
master_device.serial = ""
master_device.rack = None
master_device.location = None
mock_vc = MagicMock()
mock_vc.members.count.return_value = 1
@contextmanager
def mock_atomic():
yield
with (
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.transaction.atomic",
mock_atomic,
),
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._generate_vc_member_name",
return_value="sw1-M1",
),
patch("netbox_librenms_plugin.import_utils.virtual_chassis.Device") as mock_device_cls,
patch("netbox_librenms_plugin.import_utils.virtual_chassis.VirtualChassis") as mock_vc_cls,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
),
):
# Name conflict: renamed master already exists
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = True
mock_vc_cls.objects.create.return_value = mock_vc
result = create_virtual_chassis_with_members(master_device, [], {"device_id": 1})
# VC still created; master.name was NOT changed (conflict)
assert result == mock_vc
assert master_device.name == "sw1"
def test_create_vc_member_serial_matches_master_skipped(self):
"""Member whose serial equals master serial is skipped."""
from contextlib import contextmanager
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import create_virtual_chassis_with_members
master_device = MagicMock()
master_device.name = "sw1"
master_device.pk = 1
master_device.serial = "SERIAL-MASTER"
master_device.rack = None
master_device.location = None
mock_vc = MagicMock()
mock_vc.members.count.return_value = 1
@contextmanager
def mock_atomic():
yield
with (
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.transaction.atomic",
mock_atomic,
),
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._generate_vc_member_name",
return_value="sw1-M1",
),
patch("netbox_librenms_plugin.import_utils.virtual_chassis.Device") as mock_device_cls,
patch("netbox_librenms_plugin.import_utils.virtual_chassis.VirtualChassis") as mock_vc_cls,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
),
):
mock_device_cls.objects.filter.return_value.exclude.return_value.exists.return_value = False
mock_device_cls.objects.filter.return_value.exists.return_value = False
mock_vc_cls.objects.create.return_value = mock_vc
# One member with same serial as master → should be skipped
members_info = [{"serial": "SERIAL-MASTER", "position": 2, "name": "sw1-2"}]
create_virtual_chassis_with_members(master_device, members_info, {"device_id": 1})
# Device.objects.create should NOT be called (member skipped)
mock_device_cls.objects.create.assert_not_called()
def test_create_vc_member_duplicate_serial_skipped(self):
"""Member with a serial that already exists in DB is skipped."""
from contextlib import contextmanager
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import create_virtual_chassis_with_members
master_device = MagicMock()
master_device.name = "sw1"
master_device.pk = 1
master_device.serial = ""
master_device.rack = None
master_device.location = None
mock_vc = MagicMock()
mock_vc.members.count.return_value = 1
@contextmanager
def mock_atomic():
yield
def _filter_exists(*args, **kwargs):
# First call: check renamed master name conflict (exclude().exists()) → False
# Subsequent calls: check duplicate serial → True (for serial)
mock = MagicMock()
mock.exclude.return_value.exists.return_value = False
mock.exists.return_value = True # serial already exists
return mock
with (
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.transaction.atomic",
mock_atomic,
),
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._generate_vc_member_name",
return_value="sw1-M2",
),
patch("netbox_librenms_plugin.import_utils.virtual_chassis.Device") as mock_device_cls,
patch("netbox_librenms_plugin.import_utils.virtual_chassis.VirtualChassis") as mock_vc_cls,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
),
):
mock_device_cls.objects.filter.side_effect = _filter_exists
mock_vc_cls.objects.create.return_value = mock_vc
members_info = [{"serial": "DUP-SERIAL", "position": 2, "name": "sw1-2"}]
create_virtual_chassis_with_members(master_device, members_info, {"device_id": 1})
mock_device_cls.objects.create.assert_not_called()
def test_create_vc_member_created_successfully(self):
"""Normal member (no duplicate serial/name) is created via Device.objects.create."""
from contextlib import contextmanager
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import create_virtual_chassis_with_members
master_device = MagicMock()
master_device.name = "sw1"
master_device.pk = 1
master_device.serial = ""
master_device.rack = None
master_device.location = None
master_device.platform = None
master_device.role = MagicMock()
master_device.device_type = MagicMock()
master_device.site = MagicMock()
mock_vc = MagicMock()
mock_vc.members.count.return_value = 2
@contextmanager
def mock_atomic():
yield
def _filter_side_effect(*args, **kwargs):
mock = MagicMock()
mock.exclude.return_value.exists.return_value = False # no name conflict
mock.exists.return_value = False # no duplicate serial or name
return mock
with (
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.transaction.atomic",
mock_atomic,
),
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._generate_vc_member_name",
return_value="sw1-M2",
),
patch("netbox_librenms_plugin.import_utils.virtual_chassis.Device") as mock_device_cls,
patch("netbox_librenms_plugin.import_utils.virtual_chassis.VirtualChassis") as mock_vc_cls,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
),
):
mock_device_cls.objects.filter.side_effect = _filter_side_effect
mock_vc_cls.objects.create.return_value = mock_vc
members_info = [{"serial": "NEW-SERIAL", "position": 2, "name": "sw1-2"}]
result = create_virtual_chassis_with_members(master_device, members_info, {"device_id": 1})
mock_device_cls.objects.create.assert_called_once()
assert result == mock_vc
def test_create_vc_member_count_warning_when_fewer_created(self):
"""Warning is logged when members_created < expected_members."""
from contextlib import contextmanager
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import create_virtual_chassis_with_members
master_device = MagicMock()
master_device.name = "sw1"
master_device.pk = 1
master_device.serial = ""
master_device.rack = None
master_device.location = None
mock_vc = MagicMock()
mock_vc.members.count.return_value = 1
@contextmanager
def mock_atomic():
yield
def _filter_side_effect(*args, **kwargs):
mock = MagicMock()
mock.exclude.return_value.exists.return_value = False
# serial check: True → member skipped
mock.exists.return_value = True
return mock
with (
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.transaction.atomic",
mock_atomic,
),
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._generate_vc_member_name",
return_value="sw1-M2",
),
patch("netbox_librenms_plugin.import_utils.virtual_chassis.Device") as mock_device_cls,
patch("netbox_librenms_plugin.import_utils.virtual_chassis.VirtualChassis") as mock_vc_cls,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
),
patch("netbox_librenms_plugin.import_utils.virtual_chassis.logger") as mock_logger,
):
mock_device_cls.objects.filter.side_effect = _filter_side_effect
mock_vc_cls.objects.create.return_value = mock_vc
# 2 members expected, both skipped → warning
members_info = [
{"serial": "S1", "position": 2, "name": "sw1-2"},
{"serial": "S2", "position": 3, "name": "sw1-3"},
]
create_virtual_chassis_with_members(master_device, members_info, {"device_id": 1})
# Warning should be called for count mismatch
mock_logger.warning.assert_called()
def test_create_vc_member_zero_position_and_name_conflict(self):
"""Member position=0 → discovered_pos=None, and name conflict → skip."""
from contextlib import contextmanager
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import create_virtual_chassis_with_members
master_device = MagicMock()
master_device.name = "sw1"
master_device.pk = 1
master_device.serial = ""
master_device.rack = None
master_device.location = None
master_device.platform = None
master_device.role = MagicMock()
master_device.device_type = MagicMock()
master_device.site = MagicMock()
mock_vc = MagicMock()
mock_vc.members.count.return_value = 1
@contextmanager
def mock_atomic():
yield
filter_call_count = [0]
def _filter_side_effect(*args, **kwargs):
mock = MagicMock()
mock.exclude.return_value.exists.return_value = False # no renamed-master conflict
filter_call_count[0] += 1
# call 1: renamed-master name conflict check (.exclude().exists()) → handled above
# call 2: serial duplicate check (.exists()) → False (serial doesn't exist)
# call 3: member name conflict check (.exists()) → True (name already taken)
mock.exists.return_value = filter_call_count[0] == 3
return mock
with (
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.transaction.atomic",
mock_atomic,
),
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._generate_vc_member_name",
return_value="sw1-M2",
),
patch("netbox_librenms_plugin.import_utils.virtual_chassis.Device") as mock_device_cls,
patch("netbox_librenms_plugin.import_utils.virtual_chassis.VirtualChassis") as mock_vc_cls,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
),
):
mock_device_cls.objects.filter.side_effect = _filter_side_effect
mock_vc_cls.objects.create.return_value = mock_vc
# position=0 → discovered_pos normalized to None; serial present but name conflicts
members_info = [{"serial": "S-UNIQUE", "position": 0, "name": "sw1-2"}]
create_virtual_chassis_with_members(master_device, members_info, {"device_id": 1})
# Member skipped due to name conflict (not created)
mock_device_cls.objects.create.assert_not_called()
def test_create_vc_member_invalid_position_string_uses_sequential(self):
"""Member with position='abc' (non-int) triggers except branch → uses sequential counter."""
from contextlib import contextmanager
from unittest.mock import patch
from netbox_librenms_plugin.import_utils.virtual_chassis import create_virtual_chassis_with_members
master_device = MagicMock()
master_device.name = "sw1"
master_device.pk = 1
master_device.serial = ""
master_device.rack = None
master_device.location = None
master_device.platform = None
master_device.role = MagicMock()
master_device.device_type = MagicMock()
master_device.site = MagicMock()
mock_vc = MagicMock()
mock_vc.members.count.return_value = 2
created_positions = []
@contextmanager
def mock_atomic():
yield
def _filter_side_effect(*args, **kwargs):
mock = MagicMock()
mock.exclude.return_value.exists.return_value = False
mock.exists.return_value = False
return mock
def _capture_create(**kwargs):
created_positions.append(kwargs.get("vc_position"))
return MagicMock()
with (
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis.transaction.atomic",
mock_atomic,
),
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._generate_vc_member_name",
return_value="sw1-M2",
),
patch("netbox_librenms_plugin.import_utils.virtual_chassis.Device") as mock_device_cls,
patch("netbox_librenms_plugin.import_utils.virtual_chassis.VirtualChassis") as mock_vc_cls,
patch(
"netbox_librenms_plugin.import_utils.virtual_chassis._load_vc_member_name_pattern",
return_value="-M{position}",
),
):
mock_device_cls.objects.filter.side_effect = _filter_side_effect
mock_device_cls.objects.create.side_effect = _capture_create
mock_vc_cls.objects.create.return_value = mock_vc
# "abc" position → except branch → sequential fallback (position=2, then +=1)
members_info = [{"serial": "S1", "position": "abc", "name": "m1"}]
create_virtual_chassis_with_members(master_device, members_info, {"device_id": 1})
assert mock_device_cls.objects.create.call_count == 1
Reference in New Issue View Git Blame Copy Permalink