NeoAnd/netbox-zabbix-sync
1
0
Fork 0
mirror of https://github.com/TheNetworkGuy/netbox-zabbix-sync.git synced 2026-03-21 20:18:38 -06:00
Code Issues Actions Packages Projects Releases Wiki Activity

Fixed formatting, fixed tests for type checker

Browse Source
This commit is contained in:
TheNetworkGuy
2026-02-11 15:30:53 +00:00
parent d53cc5e7d4
commit 3227bb3165
13 changed files with 351 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/device.py
+13 -12
View File
@@ -167,7 +167,7 @@ class PhysicalDevice:
# Gather templates from the custom field but overrule
# them should there be any device specific templates
if overrule_custom:
try: # noqa: SIM105
try: # noqa: SIM105
self.zbx_template_names = self.get_templates_context()
except TemplateError:
pass
@@ -673,7 +673,7 @@ class PhysicalDevice:
if not self.group_ids:
zbx_groupid_confirmation = self.set_zbx_groupid(groups)
if not zbx_groupid_confirmation and not create_hostgroups:
# Function returns true / false but also sets GroupID
# Function returns true / false but also sets GroupID
e = (
f"Host {self.name}: different hostgroup is required but "
"unable to create hostgroup without generation permission."
@@ -771,10 +771,10 @@ class PhysicalDevice:
# Check if proxy or proxy group is defined.
# Check for proxy_hostid for backwards compatibility with Zabbix <= 6
if (
(self.zbxproxy["idtype"] in host
and host[self.zbxproxy["idtype"]] == self.zbxproxy["id"])
or ("proxy_hostid" in host and
host["proxy_hostid"] == self.zbxproxy["id"])
self.zbxproxy["idtype"] in host
and host[self.zbxproxy["idtype"]] == self.zbxproxy["id"]
) or (
"proxy_hostid" in host and host["proxy_hostid"] == self.zbxproxy["id"]
):
self.logger.debug("Host %s: Proxy in-sync.", self.name)
# Proxy does not match, update Zabbix
@@ -890,10 +890,9 @@ class PhysicalDevice:
if isinstance(item, dict) and key == "details":
for k, i in item.items():
# Check if the key is found in Zabbix and if the value matches
if(
k in host["interfaces"][0][key] and
host["interfaces"][0][key][k] != str(i)
):
if k in host["interfaces"][0][key] and host["interfaces"][
0
][key][k] != str(i):
# If dict has not been created, add it
if key not in updates:
updates[key] = {}
@@ -1009,5 +1008,7 @@ class PhysicalDevice:
# all of the NetBox templates have been confirmed as successful
# and the ZBX template list is empty. This means that
# all of the templates match.
return (len(succesfull_templates) == len(self.zbx_templates)
and len(tmpls_from_zabbix) == 0)
return (
len(succesfull_templates) == len(self.zbx_templates)
and len(tmpls_from_zabbix) == 0
)
modules/hostgroups.py
+4 -4
View File
@@ -117,10 +117,10 @@ class Hostgroup:
if hg_item not in self.format_options:
# If the string is between quotes, use it as a literal in the hostgroup name
minimum_length = 2
if(
len(hg_item) > minimum_length and
hg_item[0] == hg_item[-1] and
hg_item[0] in ("'", '"')
if (
len(hg_item) > minimum_length
and hg_item[0] == hg_item[-1]
and hg_item[0] in ("'", '"')
):
hg_output.append(hg_item[1:-1])
else:
modules/interface.py
+1
View File
@@ -1,6 +1,7 @@
"""
All of the Zabbix interface related configuration
"""
from modules.exceptions import InterfaceConfigError
modules/tags.py
+10 -2
View File
@@ -54,14 +54,22 @@ class ZabbixTags:
Validates tag name
"""
max_tag_name_length = 256
return tag_name and isinstance(tag_name, str) and len(tag_name) <= max_tag_name_length
return (
tag_name
and isinstance(tag_name, str)
and len(tag_name) <= max_tag_name_length
)
def validate_value(self, tag_value):
"""
Validates tag value
"""
max_tag_value_length = 256
return tag_value and isinstance(tag_value, str) and len(tag_value) <= max_tag_value_length
return (
tag_value
and isinstance(tag_value, str)
and len(tag_value) <= max_tag_value_length
)
def render_tag(self, tag_name, tag_value):
"""
modules/virtual_machine.py
+2 -1
View File
@@ -1,5 +1,6 @@
# pylint: disable=duplicate-code
"""Module that hosts all functions for virtual machine processing"""
from modules.config import load_config
from modules.device import PhysicalDevice
from modules.exceptions import InterfaceConfigError, SyncInventoryError, TemplateError
@@ -39,7 +40,7 @@ class VirtualMachine(PhysicalDevice):
except TemplateError as e:
self.logger.warning(e)
return True
def set_interface_details(self):
"""
Overwrites device function to select an agent interface type by default
tests/test_configuration_parsing.py
+24 -12
View File
@@ -1,4 +1,5 @@
"""Tests for configuration parsing in the modules.config module."""
import os
from unittest.mock import MagicMock, patch
@@ -12,8 +13,10 @@ from modules.config import (
def test_load_config_defaults():
"""Test that load_config returns default values when no config file or env vars are present"""
with patch('modules.config.load_config_file', return_value=DEFAULT_CONFIG.copy()), \
patch('modules.config.load_env_variable', return_value=None):
with (
patch("modules.config.load_config_file", return_value=DEFAULT_CONFIG.copy()),
patch("modules.config.load_env_variable", return_value=None),
):
config = load_config()
assert config == DEFAULT_CONFIG
assert config["templates_config_context"] is False
@@ -26,8 +29,10 @@ def test_load_config_file():
mock_config["templates_config_context"] = True
mock_config["sync_vms"] = True
with patch('modules.config.load_config_file', return_value=mock_config), \
patch('modules.config.load_env_variable', return_value=None):
with (
patch("modules.config.load_config_file", return_value=mock_config),
patch("modules.config.load_env_variable", return_value=None),
):
config = load_config()
assert config["templates_config_context"] is True
assert config["sync_vms"] is True
@@ -37,6 +42,7 @@ def test_load_config_file():
def test_load_env_variables():
"""Test that load_config properly loads values from environment variables"""
# Mock env variable loading to return values for specific keys
def mock_load_env(key):
if key == "sync_vms":
@@ -45,8 +51,10 @@ def test_load_env_variables():
return True
return None
with patch('modules.config.load_config_file', return_value=DEFAULT_CONFIG.copy()), \
patch('modules.config.load_env_variable', side_effect=mock_load_env):
with (
patch("modules.config.load_config_file", return_value=DEFAULT_CONFIG.copy()),
patch("modules.config.load_env_variable", side_effect=mock_load_env),
):
config = load_config()
assert config["sync_vms"] is True
assert config["create_journal"] is True
@@ -66,8 +74,10 @@ def test_env_vars_override_config_file():
return True
return None
with patch('modules.config.load_config_file', return_value=mock_config), \
patch('modules.config.load_env_variable', side_effect=mock_load_env):
with (
patch("modules.config.load_config_file", return_value=mock_config),
patch("modules.config.load_env_variable", side_effect=mock_load_env),
):
config = load_config()
# This should be overridden by the env var
assert config["sync_vms"] is True
@@ -78,8 +88,10 @@ def test_env_vars_override_config_file():
def test_load_config_file_function():
"""Test the load_config_file function directly"""
# Test when the file exists
with patch('pathlib.Path.exists', return_value=True), \
patch('importlib.util.spec_from_file_location') as mock_spec:
with (
patch("pathlib.Path.exists", return_value=True),
patch("importlib.util.spec_from_file_location") as mock_spec,
):
# Setup the mock module with attributes
mock_module = MagicMock()
mock_module.templates_config_context = True
@@ -91,7 +103,7 @@ def test_load_config_file_function():
mock_spec_instance.loader.exec_module = lambda x: None
# Patch module_from_spec to return our mock module
with patch('importlib.util.module_from_spec', return_value=mock_module):
with patch("importlib.util.module_from_spec", return_value=mock_module):
config = load_config_file(DEFAULT_CONFIG.copy())
assert config["templates_config_context"] is True
assert config["sync_vms"] is True
@@ -99,7 +111,7 @@ def test_load_config_file_function():
def test_load_config_file_not_found():
"""Test load_config_file when the config file doesn't exist"""
with patch('pathlib.Path.exists', return_value=False):
with patch("pathlib.Path.exists", return_value=False):
result = load_config_file(DEFAULT_CONFIG.copy())
# Should return a dict equal to DEFAULT_CONFIG, not a new object
assert result == DEFAULT_CONFIG
tests/test_device_deletion.py
+26 -15
View File
@@ -1,4 +1,5 @@
"""Tests for device deletion functionality in the PhysicalDevice class."""
import unittest
from unittest.mock import MagicMock, patch
@@ -40,14 +41,14 @@ class TestDeviceDeletion(unittest.TestCase):
self.mock_logger = MagicMock()
# Create PhysicalDevice instance with mocks
with patch('modules.device.config', {"device_cf": "zabbix_hostid"}):
with patch("modules.device.config", {"device_cf": "zabbix_hostid"}):
self.device = PhysicalDevice(
self.mock_nb_device,
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
journal=True,
logger=self.mock_logger
logger=self.mock_logger,
)
def test_cleanup_successful_deletion(self):
@@ -60,12 +61,15 @@ class TestDeviceDeletion(unittest.TestCase):
self.device.cleanup()
# Verify
self.mock_zabbix.host.get.assert_called_once_with(filter={'hostid': '456'}, output=[])
self.mock_zabbix.host.delete.assert_called_once_with('456')
self.mock_zabbix.host.get.assert_called_once_with(
filter={"hostid": "456"}, output=[]
)
self.mock_zabbix.host.delete.assert_called_once_with("456")
self.mock_nb_device.save.assert_called_once()
self.assertIsNone(self.mock_nb_device.custom_fields["zabbix_hostid"])
self.mock_logger.info.assert_called_with(f"Host {self.device.name}: "
"Deleted host from Zabbix.")
self.mock_logger.info.assert_called_with(
f"Host {self.device.name}: Deleted host from Zabbix."
)
def test_cleanup_device_already_deleted(self):
"""Test cleanup when device is already deleted from Zabbix."""
@@ -76,12 +80,15 @@ class TestDeviceDeletion(unittest.TestCase):
self.device.cleanup()
# Verify
self.mock_zabbix.host.get.assert_called_once_with(filter={'hostid': '456'}, output=[])
self.mock_zabbix.host.get.assert_called_once_with(
filter={"hostid": "456"}, output=[]
)
self.mock_zabbix.host.delete.assert_not_called()
self.mock_nb_device.save.assert_called_once()
self.assertIsNone(self.mock_nb_device.custom_fields["zabbix_hostid"])
self.mock_logger.info.assert_called_with(
f"Host {self.device.name}: was already deleted from Zabbix. Removed link in NetBox.")
f"Host {self.device.name}: was already deleted from Zabbix. Removed link in NetBox."
)
def test_cleanup_api_error(self):
"""Test cleanup when Zabbix API returns an error."""
@@ -94,15 +101,17 @@ class TestDeviceDeletion(unittest.TestCase):
self.device.cleanup()
# Verify correct calls were made
self.mock_zabbix.host.get.assert_called_once_with(filter={'hostid': '456'}, output=[])
self.mock_zabbix.host.delete.assert_called_once_with('456')
self.mock_zabbix.host.get.assert_called_once_with(
filter={"hostid": "456"}, output=[]
)
self.mock_zabbix.host.delete.assert_called_once_with("456")
self.mock_nb_device.save.assert_not_called()
self.mock_logger.error.assert_called()
def test_zeroize_cf(self):
"""Test _zeroize_cf method that clears the custom field."""
# Execute
self.device._zeroize_cf() # pylint: disable=protected-access
self.device._zeroize_cf() # pylint: disable=protected-access
# Verify
self.assertIsNone(self.mock_nb_device.custom_fields["zabbix_hostid"])
@@ -138,14 +147,14 @@ class TestDeviceDeletion(unittest.TestCase):
def test_create_journal_entry_when_disabled(self):
"""Test create_journal_entry when journaling is disabled."""
# Setup - create device with journal=False
with patch('modules.device.config', {"device_cf": "zabbix_hostid"}):
with patch("modules.device.config", {"device_cf": "zabbix_hostid"}):
device = PhysicalDevice(
self.mock_nb_device,
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
journal=False, # Disable journaling
logger=self.mock_logger
logger=self.mock_logger,
)
# Execute
@@ -161,8 +170,10 @@ class TestDeviceDeletion(unittest.TestCase):
self.mock_zabbix.host.get.return_value = [{"hostid": "456"}]
# Execute
with patch.object(self.device, 'create_journal_entry') as mock_journal_entry:
with patch.object(self.device, "create_journal_entry") as mock_journal_entry:
self.device.cleanup()
# Verify
mock_journal_entry.assert_called_once_with("warning", "Deleted host from Zabbix")
mock_journal_entry.assert_called_once_with(
"warning", "Deleted host from Zabbix"
)
tests/test_hostgroups.py
+82 -80
View File
@@ -1,4 +1,5 @@
"""Tests for the Hostgroup class in the hostgroups module."""
import unittest
from unittest.mock import MagicMock, patch
@@ -18,27 +19,27 @@ class TestHostgroups(unittest.TestCase):
# Create mock device with all properties
self.mock_device = MagicMock()
self.mock_device.name = "test-device"
# Set up site information
site = MagicMock()
site.name = "TestSite"
# Set up region information
region = MagicMock()
region.name = "TestRegion"
# Ensure region string representation returns the name
region.__str__.return_value = "TestRegion"
site.region = region
# Set up site group information
site_group = MagicMock()
site_group.name = "TestSiteGroup"
# Ensure site group string representation returns the name
site_group.__str__.return_value = "TestSiteGroup"
site.group = site_group
self.mock_device.site = site
# Set up role information (varies based on NetBox version)
self.mock_device_role = MagicMock()
self.mock_device_role.name = "TestRole"
@@ -46,7 +47,7 @@ class TestHostgroups(unittest.TestCase):
self.mock_device_role.__str__.return_value = "TestRole"
self.mock_device.device_role = self.mock_device_role
self.mock_device.role = self.mock_device_role
# Set up tenant information
tenant = MagicMock()
tenant.name = "TestTenant"
@@ -58,45 +59,45 @@ class TestHostgroups(unittest.TestCase):
tenant_group.__str__.return_value = "TestTenantGroup"
tenant.group = tenant_group
self.mock_device.tenant = tenant
# Set up platform information
platform = MagicMock()
platform.name = "TestPlatform"
self.mock_device.platform = platform
# Device-specific properties
device_type = MagicMock()
manufacturer = MagicMock()
manufacturer.name = "TestManufacturer"
device_type.manufacturer = manufacturer
self.mock_device.device_type = device_type
location = MagicMock()
location.name = "TestLocation"
# Ensure location string representation returns the name
location.__str__.return_value = "TestLocation"
self.mock_device.location = location
# Custom fields
self.mock_device.custom_fields = {"test_cf": "TestCF"}
# *** Mock NetBox VM setup ***
# Create mock VM with all properties
self.mock_vm = MagicMock()
self.mock_vm.name = "test-vm"
# Reuse site from device
self.mock_vm.site = site
# Set up role for VM
self.mock_vm.role = self.mock_device_role
# Set up tenant for VM (same as device)
self.mock_vm.tenant = tenant
# Set up platform for VM (same as device)
self.mock_vm.platform = platform
# VM-specific properties
cluster = MagicMock()
cluster.name = "TestCluster"
@@ -104,28 +105,28 @@ class TestHostgroups(unittest.TestCase):
cluster_type.name = "TestClusterType"
cluster.type = cluster_type
self.mock_vm.cluster = cluster
# Custom fields
self.mock_vm.custom_fields = {"test_cf": "TestCF"}
# Mock data for nesting tests
self.mock_regions_data = [
{"name": "ParentRegion", "parent": None, "_depth": 0},
{"name": "TestRegion", "parent": "ParentRegion", "_depth": 1}
{"name": "TestRegion", "parent": "ParentRegion", "_depth": 1},
]
self.mock_groups_data = [
{"name": "ParentSiteGroup", "parent": None, "_depth": 0},
{"name": "TestSiteGroup", "parent": "ParentSiteGroup", "_depth": 1}
{"name": "TestSiteGroup", "parent": "ParentSiteGroup", "_depth": 1},
]
def test_device_hostgroup_creation(self):
"""Test basic device hostgroup creation."""
hostgroup = Hostgroup("dev", self.mock_device, "4.0", self.mock_logger)
# Test the string representation
self.assertEqual(str(hostgroup), "Hostgroup for dev test-device")
# Check format options were set correctly
self.assertEqual(hostgroup.format_options["site"], "TestSite")
self.assertEqual(hostgroup.format_options["region"], "TestRegion")
@@ -136,14 +137,14 @@ class TestHostgroups(unittest.TestCase):
self.assertEqual(hostgroup.format_options["platform"], "TestPlatform")
self.assertEqual(hostgroup.format_options["manufacturer"], "TestManufacturer")
self.assertEqual(hostgroup.format_options["location"], "TestLocation")
def test_vm_hostgroup_creation(self):
"""Test basic VM hostgroup creation."""
hostgroup = Hostgroup("vm", self.mock_vm, "4.0", self.mock_logger)
# Test the string representation
self.assertEqual(str(hostgroup), "Hostgroup for vm test-vm")
# Check format options were set correctly
self.assertEqual(hostgroup.format_options["site"], "TestSite")
self.assertEqual(hostgroup.format_options["region"], "TestRegion")
@@ -154,74 +155,76 @@ class TestHostgroups(unittest.TestCase):
self.assertEqual(hostgroup.format_options["platform"], "TestPlatform")
self.assertEqual(hostgroup.format_options["cluster"], "TestCluster")
self.assertEqual(hostgroup.format_options["cluster_type"], "TestClusterType")
def test_invalid_object_type(self):
"""Test that an invalid object type raises an exception."""
with self.assertRaises(HostgroupError):
Hostgroup("invalid", self.mock_device, "4.0", self.mock_logger)
def test_device_hostgroup_formats(self):
"""Test different hostgroup formats for devices."""
hostgroup = Hostgroup("dev", self.mock_device, "4.0", self.mock_logger)
# Custom format: site/region
custom_result = hostgroup.generate("site/region")
self.assertEqual(custom_result, "TestSite/TestRegion")
# Custom format: site/tenant/platform/location
complex_result = hostgroup.generate("site/tenant/platform/location")
self.assertEqual(complex_result, "TestSite/TestTenant/TestPlatform/TestLocation")
self.assertEqual(
complex_result, "TestSite/TestTenant/TestPlatform/TestLocation"
)
def test_vm_hostgroup_formats(self):
"""Test different hostgroup formats for VMs."""
hostgroup = Hostgroup("vm", self.mock_vm, "4.0", self.mock_logger)
# Default format: cluster/role
default_result = hostgroup.generate("cluster/role")
self.assertEqual(default_result, "TestCluster/TestRole")
# Custom format: site/tenant
custom_result = hostgroup.generate("site/tenant")
self.assertEqual(custom_result, "TestSite/TestTenant")
# Custom format: cluster/cluster_type/platform
complex_result = hostgroup.generate("cluster/cluster_type/platform")
self.assertEqual(complex_result, "TestCluster/TestClusterType/TestPlatform")
def test_device_netbox_version_differences(self):
"""Test hostgroup generation with different NetBox versions."""
# NetBox v2.x
hostgroup_v2 = Hostgroup("dev", self.mock_device, "2.11", self.mock_logger)
self.assertEqual(hostgroup_v2.format_options["role"], "TestRole")
# NetBox v3.x
hostgroup_v3 = Hostgroup("dev", self.mock_device, "3.5", self.mock_logger)
self.assertEqual(hostgroup_v3.format_options["role"], "TestRole")
# NetBox v4.x (already tested in other methods)
def test_custom_field_lookup(self):
"""Test custom field lookup functionality."""
hostgroup = Hostgroup("dev", self.mock_device, "4.0", self.mock_logger)
# Test custom field exists and is populated
cf_result = hostgroup.custom_field_lookup("test_cf")
self.assertTrue(cf_result["result"])
self.assertEqual(cf_result["cf"], "TestCF")
# Test custom field doesn't exist
cf_result = hostgroup.custom_field_lookup("nonexistent_cf")
self.assertFalse(cf_result["result"])
self.assertIsNone(cf_result["cf"])
def test_hostgroup_with_custom_field(self):
"""Test hostgroup generation including a custom field."""
hostgroup = Hostgroup("dev", self.mock_device, "4.0", self.mock_logger)
# Generate with custom field included
result = hostgroup.generate("site/test_cf/role")
self.assertEqual(result, "TestSite/TestCF/TestRole")
def test_missing_hostgroup_format_item(self):
"""Test handling of missing hostgroup format items."""
# Create a device with minimal attributes
@@ -231,124 +234,123 @@ class TestHostgroups(unittest.TestCase):
minimal_device.tenant = None
minimal_device.platform = None
minimal_device.custom_fields = {}
# Create role
role = MagicMock()
role.name = "MinimalRole"
minimal_device.role = role
# Create device_type with manufacturer
device_type = MagicMock()
manufacturer = MagicMock()
manufacturer.name = "MinimalManufacturer"
device_type.manufacturer = manufacturer
minimal_device.device_type = device_type
# Create hostgroup
hostgroup = Hostgroup("dev", minimal_device, "4.0", self.mock_logger)
# Generate with default format
result = hostgroup.generate("site/manufacturer/role")
# Site is missing, so only manufacturer and role should be included
self.assertEqual(result, "MinimalManufacturer/MinimalRole")
# Test with invalid format
with self.assertRaises(HostgroupError):
hostgroup.generate("site/nonexistent/role")
def test_nested_region_hostgroups(self):
"""Test hostgroup generation with nested regions."""
# Mock the build_path function to return a predictable result
with patch('modules.hostgroups.build_path') as mock_build_path:
with patch("modules.hostgroups.build_path") as mock_build_path:
# Configure the mock to return a list of regions in the path
mock_build_path.return_value = ["ParentRegion", "TestRegion"]
# Create hostgroup with nested regions enabled
hostgroup = Hostgroup(
"dev",
self.mock_device,
"4.0",
"dev",
self.mock_device,
"4.0",
self.mock_logger,
nested_region_flag=True,
nb_regions=self.mock_regions_data
nb_regions=self.mock_regions_data,
)
# Generate hostgroup with region
result = hostgroup.generate("site/region/role")
# Should include the parent region
self.assertEqual(result, "TestSite/ParentRegion/TestRegion/TestRole")
def test_nested_sitegroup_hostgroups(self):
"""Test hostgroup generation with nested site groups."""
# Mock the build_path function to return a predictable result
with patch('modules.hostgroups.build_path') as mock_build_path:
with patch("modules.hostgroups.build_path") as mock_build_path:
# Configure the mock to return a list of site groups in the path
mock_build_path.return_value = ["ParentSiteGroup", "TestSiteGroup"]
# Create hostgroup with nested site groups enabled
hostgroup = Hostgroup(
"dev",
self.mock_device,
"4.0",
"dev",
self.mock_device,
"4.0",
self.mock_logger,
nested_sitegroup_flag=True,
nb_groups=self.mock_groups_data
nb_groups=self.mock_groups_data,
)
# Generate hostgroup with site_group
result = hostgroup.generate("site/site_group/role")
# Should include the parent site group
self.assertEqual(result, "TestSite/ParentSiteGroup/TestSiteGroup/TestRole")
def test_vm_list_based_hostgroup_format(self):
"""Test VM hostgroup generation with a list-based format."""
hostgroup = Hostgroup("vm", self.mock_vm, "4.0", self.mock_logger)
# Test with a list of format strings
format_list = ["platform", "role", "cluster_type/cluster"]
# Generate hostgroups for each format in the list
hostgroups = []
for fmt in format_list:
result = hostgroup.generate(fmt)
if result: # Only add non-None results
hostgroups.append(result)
# Verify each expected hostgroup is generated
self.assertEqual(len(hostgroups), 3) # Should have 3 hostgroups
self.assertIn("TestPlatform", hostgroups)
self.assertIn("TestRole", hostgroups)
self.assertIn("TestClusterType/TestCluster", hostgroups)
def test_nested_format_splitting(self):
"""Test that formats with slashes correctly split and resolve each component."""
hostgroup = Hostgroup("vm", self.mock_vm, "4.0", self.mock_logger)
# Test a format with slashes that should be split
complex_format = "cluster_type/cluster"
result = hostgroup.generate(complex_format)
# Verify the format is correctly split and each component resolved
self.assertEqual(result, "TestClusterType/TestCluster")
def test_multiple_hostgroup_formats_device(self):
"""Test device hostgroup generation with multiple formats."""
hostgroup = Hostgroup("dev", self.mock_device, "4.0", self.mock_logger)
# Test with various formats that would be in a list
formats = [
"site",
"manufacturer/role",
"platform/location",
"tenant_group/tenant"
"site",
"manufacturer/role",
"platform/location",
"tenant_group/tenant",
]
# Generate and check each format
results = {}
for fmt in formats:
results[fmt] = hostgroup.generate(fmt)
# Verify results
self.assertEqual(results["site"], "TestSite")
self.assertEqual(results["manufacturer/role"], "TestManufacturer/TestRole")
tests/test_interface.py
+41 -41
View File
@@ -1,5 +1,7 @@
"""Tests for the ZabbixInterface class in the interface module."""
import unittest
from typing import cast
from modules.exceptions import InterfaceConfigError
from modules.interface import ZabbixInterface
@@ -19,11 +21,7 @@ class TestZabbixInterface(unittest.TestCase):
"zabbix": {
"interface_type": 2,
"interface_port": "161",
"snmp": {
"version": 2,
"community": "public",
"bulk": 1
}
"snmp": {"version": 2, "community": "public", "bulk": 1},
}
}
@@ -38,16 +36,13 @@ class TestZabbixInterface(unittest.TestCase):
"authpassphrase": "authpass123",
"privprotocol": "AES",
"privpassphrase": "privpass123",
"contextname": "context1"
}
"contextname": "context1",
},
}
}
self.agent_context = {
"zabbix": {
"interface_type": 1,
"interface_port": "10050"
}
"zabbix": {"interface_type": 1, "interface_port": "10050"}
}
def test_init(self):
@@ -96,27 +91,27 @@ class TestZabbixInterface(unittest.TestCase):
# Test for agent type (1)
interface.interface["type"] = 1
interface._set_default_port() # pylint: disable=protected-access
interface._set_default_port() # pylint: disable=protected-access
self.assertEqual(interface.interface["port"], "10050")
# Test for SNMP type (2)
interface.interface["type"] = 2
interface._set_default_port() # pylint: disable=protected-access
interface._set_default_port() # pylint: disable=protected-access
self.assertEqual(interface.interface["port"], "161")
# Test for IPMI type (3)
interface.interface["type"] = 3
interface._set_default_port() # pylint: disable=protected-access
interface._set_default_port() # pylint: disable=protected-access
self.assertEqual(interface.interface["port"], "623")
# Test for JMX type (4)
interface.interface["type"] = 4
interface._set_default_port() # pylint: disable=protected-access
interface._set_default_port() # pylint: disable=protected-access
self.assertEqual(interface.interface["port"], "12345")
# Test for unsupported type
interface.interface["type"] = 99
result = interface._set_default_port() # pylint: disable=protected-access
result = interface._set_default_port() # pylint: disable=protected-access
self.assertFalse(result)
def test_set_snmp_v2(self):
@@ -128,9 +123,10 @@ class TestZabbixInterface(unittest.TestCase):
interface.set_snmp()
# Check SNMP details
self.assertEqual(interface.interface["details"]["version"], "2")
self.assertEqual(interface.interface["details"]["community"], "public")
self.assertEqual(interface.interface["details"]["bulk"], "1")
details = cast(dict[str, str], interface.interface["details"])
self.assertEqual(details["version"], "2")
self.assertEqual(details["community"], "public")
self.assertEqual(details["bulk"], "1")
def test_set_snmp_v3(self):
"""Test set_snmp with SNMPv3 configuration."""
@@ -141,14 +137,19 @@ class TestZabbixInterface(unittest.TestCase):
interface.set_snmp()
# Check SNMP details
self.assertEqual(interface.interface["details"]["version"], "3")
self.assertEqual(interface.interface["details"]["securityname"], "snmpuser")
self.assertEqual(interface.interface["details"]["securitylevel"], "authPriv")
self.assertEqual(interface.interface["details"]["authprotocol"], "SHA")
self.assertEqual(interface.interface["details"]["authpassphrase"], "authpass123")
self.assertEqual(interface.interface["details"]["privprotocol"], "AES")
self.assertEqual(interface.interface["details"]["privpassphrase"], "privpass123")
self.assertEqual(interface.interface["details"]["contextname"], "context1")
details = cast(dict[str, str], interface.interface["details"])
self.assertEqual(details["version"], "3")
self.assertEqual(details["securityname"], "snmpuser")
self.assertEqual(details["securitylevel"], "authPriv")
self.assertEqual(details["authprotocol"], "SHA")
self.assertEqual(
details["authpassphrase"], "authpass123"
)
self.assertEqual(details["privprotocol"], "AES")
self.assertEqual(
details["privpassphrase"], "privpass123"
)
self.assertEqual(details["contextname"], "context1")
def test_set_snmp_no_snmp_config(self):
"""Test set_snmp with missing SNMP configuration."""
@@ -169,7 +170,7 @@ class TestZabbixInterface(unittest.TestCase):
"interface_type": 2,
"snmp": {
"version": 4 # Invalid version
}
},
}
}
interface = ZabbixInterface(context, self.test_ip)
@@ -187,7 +188,7 @@ class TestZabbixInterface(unittest.TestCase):
"interface_type": 2,
"snmp": {
"community": "public" # No version specified
}
},
}
}
interface = ZabbixInterface(context, self.test_ip)
@@ -214,9 +215,12 @@ class TestZabbixInterface(unittest.TestCase):
# Check interface properties
self.assertEqual(interface.interface["type"], "2")
self.assertEqual(interface.interface["port"], "161")
self.assertEqual(interface.interface["details"]["version"], "2")
self.assertEqual(interface.interface["details"]["community"], "{$SNMP_COMMUNITY}")
self.assertEqual(interface.interface["details"]["bulk"], "1")
details = cast(dict[str, str], interface.interface["details"])
self.assertEqual(details["version"], "2")
self.assertEqual(
details["community"], "{$SNMP_COMMUNITY}"
)
self.assertEqual(details["bulk"], "1")
def test_set_default_agent(self):
"""Test set_default_agent method."""
@@ -230,14 +234,7 @@ class TestZabbixInterface(unittest.TestCase):
def test_snmpv2_no_community(self):
"""Test SNMPv2 with no community string specified."""
# Create context with SNMPv2 but no community
context = {
"zabbix": {
"interface_type": 2,
"snmp": {
"version": 2
}
}
}
context = {"zabbix": {"interface_type": 2, "snmp": {"version": 2}}}
interface = ZabbixInterface(context, self.test_ip)
interface.get_context() # Set the interface type
@@ -245,4 +242,7 @@ class TestZabbixInterface(unittest.TestCase):
interface.set_snmp()
# Should use default community string
self.assertEqual(interface.interface["details"]["community"], "{$SNMP_COMMUNITY}")
details = cast(dict[str, str], interface.interface["details"])
self.assertEqual(
details["community"], "{$SNMP_COMMUNITY}"
)
tests/test_list_hostgroup_formats.py
+24 -23
View File
@@ -1,4 +1,5 @@
"""Tests for list-based hostgroup formats in configuration."""
import unittest
from unittest.mock import MagicMock
@@ -18,56 +19,56 @@ class TestListHostgroupFormats(unittest.TestCase):
# Create mock device
self.mock_device = MagicMock()
self.mock_device.name = "test-device"
# Set up site information
site = MagicMock()
site.name = "TestSite"
# Set up region information
region = MagicMock()
region.name = "TestRegion"
region.__str__.return_value = "TestRegion"
site.region = region
# Set device site
self.mock_device.site = site
# Set up role information
self.mock_device_role = MagicMock()
self.mock_device_role.name = "TestRole"
self.mock_device_role.__str__.return_value = "TestRole"
self.mock_device_role.__str__.return_value = "TestRole"
self.mock_device.role = self.mock_device_role
# Set up rack information
rack = MagicMock()
rack.name = "TestRack"
self.mock_device.rack = rack
# Set up platform information
platform = MagicMock()
platform.name = "TestPlatform"
self.mock_device.platform = platform
# Device-specific properties
device_type = MagicMock()
manufacturer = MagicMock()
manufacturer.name = "TestManufacturer"
device_type.manufacturer = manufacturer
self.mock_device.device_type = device_type
# Create mock VM
self.mock_vm = MagicMock()
self.mock_vm.name = "test-vm"
# Reuse site from device
self.mock_vm.site = site
# Set up role for VM
self.mock_vm.role = self.mock_device_role
# Set up platform for VM
self.mock_vm.platform = platform
# VM-specific properties
cluster = MagicMock()
cluster.name = "TestCluster"
@@ -80,53 +81,53 @@ class TestListHostgroupFormats(unittest.TestCase):
"""Test verification of list-based hostgroup formats."""
# List format with valid items
valid_format = ["region", "site", "rack"]
# List format with nested path
valid_nested_format = ["region", "site/rack"]
# List format with invalid item
invalid_format = ["region", "invalid_item", "rack"]
# Should not raise exception for valid formats
verify_hg_format(valid_format, hg_type="dev", logger=self.mock_logger)
verify_hg_format(valid_nested_format, hg_type="dev", logger=self.mock_logger)
# Should raise exception for invalid format
with self.assertRaises(HostgroupError):
verify_hg_format(invalid_format, hg_type="dev", logger=self.mock_logger)
def test_simulate_hostgroup_generation_from_config(self):
"""Simulate how the main script would generate hostgroups from list-based config."""
# Mock configuration with list-based hostgroup format
config_format = ["region", "site", "rack"]
hostgroup = Hostgroup("dev", self.mock_device, "4.0", self.mock_logger)
# Simulate the main script's hostgroup generation process
hostgroups = []
for fmt in config_format:
result = hostgroup.generate(fmt)
if result:
hostgroups.append(result)
# Check results
self.assertEqual(len(hostgroups), 3)
self.assertIn("TestRegion", hostgroups)
self.assertIn("TestSite", hostgroups)
self.assertIn("TestRack", hostgroups)
def test_vm_hostgroup_format_from_config(self):
"""Test VM hostgroup generation with list-based format."""
# Mock VM configuration with mixed format
config_format = ["platform", "role", "cluster_type/cluster"]
hostgroup = Hostgroup("vm", self.mock_vm, "4.0", self.mock_logger)
# Simulate the main script's hostgroup generation process
hostgroups = []
for fmt in config_format:
result = hostgroup.generate(fmt)
if result:
hostgroups.append(result)
# Check results
self.assertEqual(len(hostgroups), 3)
self.assertIn("TestPlatform", hostgroups)
tests/test_physical_device.py
+72 -77
View File
@@ -1,4 +1,5 @@
"""Tests for the PhysicalDevice class in the device module."""
import unittest
from unittest.mock import MagicMock, patch
@@ -35,24 +36,27 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_logger = MagicMock()
# Create PhysicalDevice instance with mocks
with patch('modules.device.config',
{"device_cf": "zabbix_hostid",
"template_cf": "zabbix_template",
"templates_config_context": False,
"templates_config_context_overrule": False,
"traverse_regions": False,
"traverse_site_groups": False,
"inventory_mode": "disabled",
"inventory_sync": False,
"device_inventory_map": {}
}):
with patch(
"modules.device.config",
{
"device_cf": "zabbix_hostid",
"template_cf": "zabbix_template",
"templates_config_context": False,
"templates_config_context_overrule": False,
"traverse_regions": False,
"traverse_site_groups": False,
"inventory_mode": "disabled",
"inventory_sync": False,
"device_inventory_map": {},
},
):
self.device = PhysicalDevice(
self.mock_nb_device,
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
journal=True,
logger=self.mock_logger
logger=self.mock_logger,
)
def test_init(self):
@@ -71,8 +75,10 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_nb_device.name = "test-devïce"
# We need to patch the search function to simulate finding special characters
with patch('modules.device.search') as mock_search, \
patch('modules.device.config', {"device_cf": "zabbix_hostid"}):
with (
patch("modules.device.search") as mock_search,
patch("modules.device.config", {"device_cf": "zabbix_hostid"}),
):
# Make the search function return True to simulate special characters
mock_search.return_value = True
@@ -81,7 +87,7 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
logger=self.mock_logger
logger=self.mock_logger,
)
# With the mocked search function, the name should be changed to NETBOX_ID format
@@ -95,19 +101,17 @@ class TestPhysicalDevice(unittest.TestCase):
"""Test get_templates_context with valid config."""
# Set up config_context with valid template data
self.mock_nb_device.config_context = {
"zabbix": {
"templates": ["Template1", "Template2"]
}
"zabbix": {"templates": ["Template1", "Template2"]}
}
# Create device with the updated mock
with patch('modules.device.config', {"device_cf": "zabbix_hostid"}):
with patch("modules.device.config", {"device_cf": "zabbix_hostid"}):
device = PhysicalDevice(
self.mock_nb_device,
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
logger=self.mock_logger
logger=self.mock_logger,
)
# Test that templates are returned correctly
@@ -117,20 +121,16 @@ class TestPhysicalDevice(unittest.TestCase):
def test_get_templates_context_with_string(self):
"""Test get_templates_context with a string instead of list."""
# Set up config_context with a string template
self.mock_nb_device.config_context = {
"zabbix": {
"templates": "Template1"
}
}
self.mock_nb_device.config_context = {"zabbix": {"templates": "Template1"}}
# Create device with the updated mock
with patch('modules.device.config', {"device_cf": "zabbix_hostid"}):
with patch("modules.device.config", {"device_cf": "zabbix_hostid"}):
device = PhysicalDevice(
self.mock_nb_device,
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
logger=self.mock_logger
logger=self.mock_logger,
)
# Test that template is wrapped in a list
@@ -143,13 +143,13 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_nb_device.config_context = {}
# Create device with the updated mock
with patch('modules.device.config', {"device_cf": "zabbix_hostid"}):
with patch("modules.device.config", {"device_cf": "zabbix_hostid"}):
device = PhysicalDevice(
self.mock_nb_device,
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
logger=self.mock_logger
logger=self.mock_logger,
)
# Test that TemplateError is raised
@@ -162,13 +162,13 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_nb_device.config_context = {"zabbix": {}}
# Create device with the updated mock
with patch('modules.device.config', {"device_cf": "zabbix_hostid"}):
with patch("modules.device.config", {"device_cf": "zabbix_hostid"}):
device = PhysicalDevice(
self.mock_nb_device,
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
logger=self.mock_logger
logger=self.mock_logger,
)
# Test that TemplateError is raised
@@ -178,25 +178,25 @@ class TestPhysicalDevice(unittest.TestCase):
def test_set_template_with_config_context(self):
"""Test set_template with templates_config_context=True."""
# Set up config_context with templates
self.mock_nb_device.config_context = {
"zabbix": {
"templates": ["Template1"]
}
}
self.mock_nb_device.config_context = {"zabbix": {"templates": ["Template1"]}}
# Mock get_templates_context to return expected templates
with patch.object(PhysicalDevice, 'get_templates_context', return_value=["Template1"]):
with patch('modules.device.config', {"device_cf": "zabbix_hostid"}):
with patch.object(
PhysicalDevice, "get_templates_context", return_value=["Template1"]
):
with patch("modules.device.config", {"device_cf": "zabbix_hostid"}):
device = PhysicalDevice(
self.mock_nb_device,
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
logger=self.mock_logger
logger=self.mock_logger,
)
# Call set_template with prefer_config_context=True
result = device.set_template(prefer_config_context=True, overrule_custom=False)
result = device.set_template(
prefer_config_context=True, overrule_custom=False
)
# Check result and template names
self.assertTrue(result)
@@ -208,20 +208,20 @@ class TestPhysicalDevice(unittest.TestCase):
config_patch = {
"device_cf": "zabbix_hostid",
"inventory_mode": "disabled",
"inventory_sync": False
"inventory_sync": False,
}
with patch('modules.device.config', config_patch):
with patch("modules.device.config", config_patch):
device = PhysicalDevice(
self.mock_nb_device,
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
logger=self.mock_logger
logger=self.mock_logger,
)
# Call set_inventory with the config patch still active
with patch('modules.device.config', config_patch):
with patch("modules.device.config", config_patch):
result = device.set_inventory({})
# Check result
@@ -235,20 +235,20 @@ class TestPhysicalDevice(unittest.TestCase):
config_patch = {
"device_cf": "zabbix_hostid",
"inventory_mode": "manual",
"inventory_sync": False
"inventory_sync": False,
}
with patch('modules.device.config', config_patch):
with patch("modules.device.config", config_patch):
device = PhysicalDevice(
self.mock_nb_device,
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
logger=self.mock_logger
logger=self.mock_logger,
)
# Call set_inventory with the config patch still active
with patch('modules.device.config', config_patch):
with patch("modules.device.config", config_patch):
result = device.set_inventory({})
# Check result
@@ -261,20 +261,20 @@ class TestPhysicalDevice(unittest.TestCase):
config_patch = {
"device_cf": "zabbix_hostid",
"inventory_mode": "automatic",
"inventory_sync": False
"inventory_sync": False,
}
with patch('modules.device.config', config_patch):
with patch("modules.device.config", config_patch):
device = PhysicalDevice(
self.mock_nb_device,
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
logger=self.mock_logger
logger=self.mock_logger,
)
# Call set_inventory with the config patch still active
with patch('modules.device.config', config_patch):
with patch("modules.device.config", config_patch):
result = device.set_inventory({})
# Check result
@@ -288,38 +288,31 @@ class TestPhysicalDevice(unittest.TestCase):
"device_cf": "zabbix_hostid",
"inventory_mode": "manual",
"inventory_sync": True,
"device_inventory_map": {
"name": "name",
"serial": "serialno_a"
}
"device_inventory_map": {"name": "name", "serial": "serialno_a"},
}
with patch('modules.device.config', config_patch):
with patch("modules.device.config", config_patch):
device = PhysicalDevice(
self.mock_nb_device,
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
logger=self.mock_logger
logger=self.mock_logger,
)
# Create a mock device with the required attributes
mock_device_data = {
"name": "test-device",
"serial": "ABC123"
}
mock_device_data = {"name": "test-device", "serial": "ABC123"}
# Call set_inventory with the config patch still active
with patch('modules.device.config', config_patch):
with patch("modules.device.config", config_patch):
result = device.set_inventory(mock_device_data)
# Check result
self.assertTrue(result)
self.assertEqual(device.inventory_mode, 0) # Manual mode
self.assertEqual(device.inventory, {
"name": "test-device",
"serialno_a": "ABC123"
})
self.assertEqual(
device.inventory, {"name": "test-device", "serialno_a": "ABC123"}
)
def test_iscluster_true(self):
"""Test isCluster when device is part of a cluster."""
@@ -327,13 +320,13 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_nb_device.virtual_chassis = MagicMock()
# Create device with the updated mock
with patch('modules.device.config', {"device_cf": "zabbix_hostid"}):
with patch("modules.device.config", {"device_cf": "zabbix_hostid"}):
device = PhysicalDevice(
self.mock_nb_device,
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
logger=self.mock_logger
logger=self.mock_logger,
)
# Check isCluster result
@@ -345,26 +338,27 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_nb_device.virtual_chassis = None
# Create device with the updated mock
with patch('modules.device.config', {"device_cf": "zabbix_hostid"}):
with patch("modules.device.config", {"device_cf": "zabbix_hostid"}):
device = PhysicalDevice(
self.mock_nb_device,
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
logger=self.mock_logger
logger=self.mock_logger,
)
# Check isCluster result
self.assertFalse(device.is_cluster())
def test_promote_master_device_primary(self):
"""Test promoteMasterDevice when device is primary in cluster."""
# Set up virtual chassis with master device
mock_vc = MagicMock()
mock_vc.name = "virtual-chassis-1"
mock_master = MagicMock()
mock_master.id = self.mock_nb_device.id # Set master ID to match the current device
mock_master.id = (
self.mock_nb_device.id
) # Set master ID to match the current device
mock_vc.master = mock_master
self.mock_nb_device.virtual_chassis = mock_vc
@@ -374,7 +368,7 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
logger=self.mock_logger
logger=self.mock_logger,
)
# Call promoteMasterDevice and check the result
@@ -385,14 +379,15 @@ class TestPhysicalDevice(unittest.TestCase):
# Device name should be updated to virtual chassis name
self.assertEqual(device.name, "virtual-chassis-1")
def test_promote_master_device_secondary(self):
"""Test promoteMasterDevice when device is secondary in cluster."""
# Set up virtual chassis with a different master device
mock_vc = MagicMock()
mock_vc.name = "virtual-chassis-1"
mock_master = MagicMock()
mock_master.id = self.mock_nb_device.id + 1 # Different ID than the current device
mock_master.id = (
self.mock_nb_device.id + 1
) # Different ID than the current device
mock_vc.master = mock_master
self.mock_nb_device.virtual_chassis = mock_vc
@@ -402,7 +397,7 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_zabbix,
self.mock_nb_journal,
"3.0",
logger=self.mock_logger
logger=self.mock_logger,
)
# Call promoteMasterDevice and check the result
tests/test_tools.py
+7 -3
View File
@@ -12,6 +12,7 @@ def test_sanatize_log_output_secrets():
assert sanitized["macros"][0]["value"] == "********"
assert sanitized["macros"][1]["value"] == "notsecret"
def test_sanatize_log_output_interface_secrets():
data = {
"interfaceid": 123,
@@ -20,8 +21,8 @@ def test_sanatize_log_output_interface_secrets():
"privpassphrase": "anothersecret",
"securityname": "sensitiveuser",
"community": "public",
"other": "normalvalue"
}
"other": "normalvalue",
},
}
sanitized = sanatize_log_output(data)
# Sensitive fields should be sanitized
@@ -34,6 +35,7 @@ def test_sanatize_log_output_interface_secrets():
# interfaceid should be removed
assert "interfaceid" not in sanitized
def test_sanatize_log_output_interface_macros():
data = {
"interfaceid": 123,
@@ -42,7 +44,7 @@ def test_sanatize_log_output_interface_macros():
"privpassphrase": "{$SECRET_MACRO}",
"securityname": "{$USER_MACRO}",
"community": "{$SNNMP_COMMUNITY}",
}
},
}
sanitized = sanatize_log_output(data)
# Macro values should not be sanitized
@@ -52,11 +54,13 @@ def test_sanatize_log_output_interface_macros():
assert sanitized["details"]["community"] == "{$SNNMP_COMMUNITY}"
assert "interfaceid" not in sanitized
def test_sanatize_log_output_plain_data():
data = {"foo": "bar", "baz": 123}
sanitized = sanatize_log_output(data)
assert sanitized == data
def test_sanatize_log_output_non_dict():
data = [1, 2, 3]
sanitized = sanatize_log_output(data)
tests/test_usermacros.py
+45 -42
View File
@@ -20,6 +20,7 @@ class DummyNB:
return self.config_context[key]
raise KeyError(key)
class TestUsermacroSync(unittest.TestCase):
def setUp(self):
self.nb = DummyNB(serial="1234")
@@ -37,77 +38,82 @@ class TestUsermacroSync(unittest.TestCase):
mock_nb.config_context = {}
mock_nb.primary_ip.address = "192.168.1.1/24"
mock_nb.custom_fields = {"zabbix_hostid": None}
# Create device with proper initialization
device = PhysicalDevice(
nb=mock_nb,
zabbix=MagicMock(),
nb_journal_class=MagicMock(),
nb_version="3.0",
logger=self.logger
logger=self.logger,
)
# Override the usermacro map method for testing
device._usermacro_map = MagicMock(return_value=self.usermacro_map)
return device
@patch("modules.device.config", {
"usermacro_sync": False,
"device_cf": "zabbix_hostid",
"tag_sync": False
})
def test_usermacro_sync_false(self):
@patch(
"modules.device.config",
{"usermacro_sync": False, "device_cf": "zabbix_hostid", "tag_sync": False},
)
@patch.object(PhysicalDevice, '_usermacro_map')
def test_usermacro_sync_false(self, mock_usermacro_map):
mock_usermacro_map.return_value = self.usermacro_map
device = self.create_mock_device()
# Call set_usermacros
result = device.set_usermacros()
self.assertEqual(device.usermacros, [])
self.assertTrue(result is True or result is None)
@patch("modules.device.config", {
"usermacro_sync": True,
"device_cf": "zabbix_hostid",
"tag_sync": False
})
@patch(
"modules.device.config",
{"usermacro_sync": True, "device_cf": "zabbix_hostid", "tag_sync": False},
)
@patch("modules.device.ZabbixUsermacros")
def test_usermacro_sync_true(self, mock_usermacros_class):
@patch.object(PhysicalDevice, '_usermacro_map')
def test_usermacro_sync_true(self, mock_usermacro_map, mock_usermacros_class):
mock_usermacro_map.return_value = self.usermacro_map
# Mock the ZabbixUsermacros class to return some test data
mock_macros_instance = MagicMock()
mock_macros_instance.sync = True # This is important - sync must be True
mock_macros_instance.generate.return_value = [{"macro": "{$HW_SERIAL}", "value": "1234"}]
mock_macros_instance.generate.return_value = [
{"macro": "{$HW_SERIAL}", "value": "1234"}
]
mock_usermacros_class.return_value = mock_macros_instance
device = self.create_mock_device()
# Call set_usermacros
device.set_usermacros()
self.assertIsInstance(device.usermacros, list)
self.assertGreater(len(device.usermacros), 0)
@patch("modules.device.config", {
"usermacro_sync": "full",
"device_cf": "zabbix_hostid",
"tag_sync": False
})
@patch(
"modules.device.config",
{"usermacro_sync": "full", "device_cf": "zabbix_hostid", "tag_sync": False},
)
@patch("modules.device.ZabbixUsermacros")
def test_usermacro_sync_full(self, mock_usermacros_class):
@patch.object(PhysicalDevice, '_usermacro_map')
def test_usermacro_sync_full(self, mock_usermacro_map, mock_usermacros_class):
mock_usermacro_map.return_value = self.usermacro_map
# Mock the ZabbixUsermacros class to return some test data
mock_macros_instance = MagicMock()
mock_macros_instance.sync = True # This is important - sync must be True
mock_macros_instance.generate.return_value = [{"macro": "{$HW_SERIAL}", "value": "1234"}]
mock_macros_instance.generate.return_value = [
{"macro": "{$HW_SERIAL}", "value": "1234"}
]
mock_usermacros_class.return_value = mock_macros_instance
device = self.create_mock_device()
# Call set_usermacros
device.set_usermacros()
self.assertIsInstance(device.usermacros, list)
self.assertGreater(len(device.usermacros), 0)
class TestZabbixUsermacros(unittest.TestCase):
def setUp(self):
self.nb = DummyNB()
@@ -128,7 +134,9 @@ class TestZabbixUsermacros(unittest.TestCase):
def test_render_macro_dict(self):
macros = ZabbixUsermacros(self.nb, {}, False, logger=self.logger)
macro = macros.render_macro("{$FOO}", {"value": "bar", "type": "secret", "description": "desc"})
macro = macros.render_macro(
"{$FOO}", {"value": "bar", "type": "secret", "description": "desc"}
)
self.assertEqual(macro["macro"], "{$FOO}")
self.assertEqual(macro["value"], "bar")
self.assertEqual(macro["type"], "1")
@@ -164,13 +172,7 @@ class TestZabbixUsermacros(unittest.TestCase):
self.assertEqual(result[1]["macro"], "{$BAR}")
def test_generate_from_config_context(self):
config_context = {
"zabbix": {
"usermacros": {
"{$TEST_MACRO}": "test_value"
}
}
}
config_context = {"zabbix": {"usermacros": {"{$TEST_MACRO}": "test_value"}}}
nb = DummyNB(config_context=config_context)
macros = ZabbixUsermacros(nb, {}, True, logger=self.logger)
result = macros.generate()
@@ -178,5 +180,6 @@ class TestZabbixUsermacros(unittest.TestCase):
self.assertEqual(result[0]["macro"], "{$TEST_MACRO}")
self.assertEqual(result[0]["value"], "test_value")
if __name__ == "__main__":
unittest.main()