mirror of
https://github.com/TheNetworkGuy/netbox-zabbix-sync.git
synced 2025-07-16 04:02:56 -06:00
Merge pull request #137 from TheNetworkGuy/hostgroup_fixes2
Fixes bug for hostgroups and removed default values for hostgroups
This commit is contained in:
commit
f15e53185b
1
.github/workflows/quality.yml
vendored
1
.github/workflows/quality.yml
vendored
@ -2,7 +2,6 @@
|
|||||||
name: Pylint Quality control
|
name: Pylint Quality control
|
||||||
|
|
||||||
on:
|
on:
|
||||||
push:
|
|
||||||
pull_request:
|
pull_request:
|
||||||
workflow_call:
|
workflow_call:
|
||||||
|
|
||||||
|
1
.github/workflows/run_tests.yml
vendored
1
.github/workflows/run_tests.yml
vendored
@ -2,7 +2,6 @@
|
|||||||
name: Pytest code testing
|
name: Pytest code testing
|
||||||
|
|
||||||
on:
|
on:
|
||||||
push:
|
|
||||||
pull_request:
|
pull_request:
|
||||||
workflow_call:
|
workflow_call:
|
||||||
|
|
||||||
|
@ -48,6 +48,7 @@ class PhysicalDevice:
|
|||||||
self.zbx_template_names = []
|
self.zbx_template_names = []
|
||||||
self.zbx_templates = []
|
self.zbx_templates = []
|
||||||
self.hostgroups = []
|
self.hostgroups = []
|
||||||
|
self.hostgroup_type = "dev"
|
||||||
self.tenant = nb.tenant
|
self.tenant = nb.tenant
|
||||||
self.config_context = nb.config_context
|
self.config_context = nb.config_context
|
||||||
self.zbxproxy = None
|
self.zbxproxy = None
|
||||||
@ -121,7 +122,7 @@ class PhysicalDevice:
|
|||||||
"""Set the hostgroup for this device"""
|
"""Set the hostgroup for this device"""
|
||||||
# Create new Hostgroup instance
|
# Create new Hostgroup instance
|
||||||
hg = Hostgroup(
|
hg = Hostgroup(
|
||||||
"dev",
|
self.hostgroup_type,
|
||||||
self.nb,
|
self.nb,
|
||||||
self.nb_api_version,
|
self.nb_api_version,
|
||||||
logger=self.logger,
|
logger=self.logger,
|
||||||
|
@ -106,13 +106,8 @@ class Hostgroup:
|
|||||||
"region": {"flag": nested_region_flag, "data": nb_regions},
|
"region": {"flag": nested_region_flag, "data": nb_regions},
|
||||||
}
|
}
|
||||||
|
|
||||||
def generate(self, hg_format=None):
|
def generate(self, hg_format):
|
||||||
"""Generate hostgroup based on a provided format"""
|
"""Generate hostgroup based on a provided format"""
|
||||||
# Set format to default in case its not specified
|
|
||||||
if not hg_format:
|
|
||||||
hg_format = (
|
|
||||||
"site/manufacturer/role" if self.type == "dev" else "cluster/role"
|
|
||||||
)
|
|
||||||
# Split all given names
|
# Split all given names
|
||||||
hg_output = []
|
hg_output = []
|
||||||
hg_items = hg_format.split("/")
|
hg_items = hg_format.split("/")
|
||||||
@ -149,7 +144,6 @@ class Hostgroup:
|
|||||||
)
|
)
|
||||||
self.logger.warning(msg)
|
self.logger.warning(msg)
|
||||||
return None
|
return None
|
||||||
#raise HostgroupError(msg)
|
|
||||||
|
|
||||||
def list_formatoptions(self):
|
def list_formatoptions(self):
|
||||||
"""
|
"""
|
||||||
|
@ -2,7 +2,6 @@
|
|||||||
"""Module that hosts all functions for virtual machine processing"""
|
"""Module that hosts all functions for virtual machine processing"""
|
||||||
from modules.device import PhysicalDevice
|
from modules.device import PhysicalDevice
|
||||||
from modules.exceptions import InterfaceConfigError, SyncInventoryError, TemplateError
|
from modules.exceptions import InterfaceConfigError, SyncInventoryError, TemplateError
|
||||||
from modules.hostgroups import Hostgroup
|
|
||||||
from modules.interface import ZabbixInterface
|
from modules.interface import ZabbixInterface
|
||||||
from modules.config import load_config
|
from modules.config import load_config
|
||||||
# Load config
|
# Load config
|
||||||
@ -16,6 +15,7 @@ class VirtualMachine(PhysicalDevice):
|
|||||||
super().__init__(*args, **kwargs)
|
super().__init__(*args, **kwargs)
|
||||||
self.hostgroup = None
|
self.hostgroup = None
|
||||||
self.zbx_template_names = None
|
self.zbx_template_names = None
|
||||||
|
self.hostgroup_type = "vm"
|
||||||
|
|
||||||
def _inventory_map(self):
|
def _inventory_map(self):
|
||||||
"""use VM inventory maps"""
|
"""use VM inventory maps"""
|
||||||
@ -29,25 +29,6 @@ class VirtualMachine(PhysicalDevice):
|
|||||||
"""use VM tag maps"""
|
"""use VM tag maps"""
|
||||||
return config["vm_tag_map"]
|
return config["vm_tag_map"]
|
||||||
|
|
||||||
def set_hostgroup(self, hg_format, nb_site_groups, nb_regions):
|
|
||||||
"""Set the hostgroup for this device"""
|
|
||||||
# Create new Hostgroup instance
|
|
||||||
hg = Hostgroup(
|
|
||||||
"vm",
|
|
||||||
self.nb,
|
|
||||||
self.nb_api_version,
|
|
||||||
logger=self.logger,
|
|
||||||
nested_sitegroup_flag=config["traverse_site_groups"],
|
|
||||||
nested_region_flag=config["traverse_regions"],
|
|
||||||
nb_groups=nb_site_groups,
|
|
||||||
nb_regions=nb_regions,
|
|
||||||
)
|
|
||||||
# Generate hostgroup based on hostgroup format
|
|
||||||
if isinstance(hg_format, list):
|
|
||||||
self.hostgroups = [hg.generate(f) for f in hg_format]
|
|
||||||
else:
|
|
||||||
self.hostgroups.append(hg.generate(hg_format))
|
|
||||||
|
|
||||||
def set_vm_template(self):
|
def set_vm_template(self):
|
||||||
"""Set Template for VMs. Overwrites default class
|
"""Set Template for VMs. Overwrites default class
|
||||||
to skip a lookup of custom fields."""
|
to skip a lookup of custom fields."""
|
||||||
|
@ -163,10 +163,6 @@ class TestHostgroups(unittest.TestCase):
|
|||||||
"""Test different hostgroup formats for devices."""
|
"""Test different hostgroup formats for devices."""
|
||||||
hostgroup = Hostgroup("dev", self.mock_device, "4.0", self.mock_logger)
|
hostgroup = Hostgroup("dev", self.mock_device, "4.0", self.mock_logger)
|
||||||
|
|
||||||
# Default format: site/manufacturer/role
|
|
||||||
default_result = hostgroup.generate()
|
|
||||||
self.assertEqual(default_result, "TestSite/TestManufacturer/TestRole")
|
|
||||||
|
|
||||||
# Custom format: site/region
|
# Custom format: site/region
|
||||||
custom_result = hostgroup.generate("site/region")
|
custom_result = hostgroup.generate("site/region")
|
||||||
self.assertEqual(custom_result, "TestSite/TestRegion")
|
self.assertEqual(custom_result, "TestSite/TestRegion")
|
||||||
@ -180,7 +176,7 @@ class TestHostgroups(unittest.TestCase):
|
|||||||
hostgroup = Hostgroup("vm", self.mock_vm, "4.0", self.mock_logger)
|
hostgroup = Hostgroup("vm", self.mock_vm, "4.0", self.mock_logger)
|
||||||
|
|
||||||
# Default format: cluster/role
|
# Default format: cluster/role
|
||||||
default_result = hostgroup.generate()
|
default_result = hostgroup.generate("cluster/role")
|
||||||
self.assertEqual(default_result, "TestCluster/TestRole")
|
self.assertEqual(default_result, "TestCluster/TestRole")
|
||||||
|
|
||||||
# Custom format: site/tenant
|
# Custom format: site/tenant
|
||||||
@ -251,7 +247,7 @@ class TestHostgroups(unittest.TestCase):
|
|||||||
hostgroup = Hostgroup("dev", minimal_device, "4.0", self.mock_logger)
|
hostgroup = Hostgroup("dev", minimal_device, "4.0", self.mock_logger)
|
||||||
|
|
||||||
# Generate with default format
|
# Generate with default format
|
||||||
result = hostgroup.generate()
|
result = hostgroup.generate("site/manufacturer/role")
|
||||||
# Site is missing, so only manufacturer and role should be included
|
# Site is missing, so only manufacturer and role should be included
|
||||||
self.assertEqual(result, "MinimalManufacturer/MinimalRole")
|
self.assertEqual(result, "MinimalManufacturer/MinimalRole")
|
||||||
|
|
||||||
@ -259,24 +255,6 @@ class TestHostgroups(unittest.TestCase):
|
|||||||
with self.assertRaises(HostgroupError):
|
with self.assertRaises(HostgroupError):
|
||||||
hostgroup.generate("site/nonexistent/role")
|
hostgroup.generate("site/nonexistent/role")
|
||||||
|
|
||||||
def test_hostgroup_missing_required_attributes(self):
|
|
||||||
"""Test handling when no valid hostgroup can be generated."""
|
|
||||||
# Create a VM with minimal attributes that won't satisfy any format
|
|
||||||
minimal_vm = MagicMock()
|
|
||||||
minimal_vm.name = "minimal-vm"
|
|
||||||
minimal_vm.site = None
|
|
||||||
minimal_vm.tenant = None
|
|
||||||
minimal_vm.platform = None
|
|
||||||
minimal_vm.role = None
|
|
||||||
minimal_vm.cluster = None
|
|
||||||
minimal_vm.custom_fields = {}
|
|
||||||
|
|
||||||
hostgroup = Hostgroup("vm", minimal_vm, "4.0", self.mock_logger)
|
|
||||||
|
|
||||||
# With default format of cluster/role, both are None, so should raise an error
|
|
||||||
with self.assertRaises(HostgroupError):
|
|
||||||
hostgroup.generate()
|
|
||||||
|
|
||||||
def test_nested_region_hostgroups(self):
|
def test_nested_region_hostgroups(self):
|
||||||
"""Test hostgroup generation with nested regions."""
|
"""Test hostgroup generation with nested regions."""
|
||||||
# Mock the build_path function to return a predictable result
|
# Mock the build_path function to return a predictable result
|
||||||
@ -334,6 +312,60 @@ class TestHostgroups(unittest.TestCase):
|
|||||||
calls = [call.write(f"The following options are available for host test-device"),
|
calls = [call.write(f"The following options are available for host test-device"),
|
||||||
call.write('\n')]
|
call.write('\n')]
|
||||||
mock_stdout.assert_has_calls(calls, any_order=True)
|
mock_stdout.assert_has_calls(calls, any_order=True)
|
||||||
|
|
||||||
|
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"
|
||||||
|
]
|
||||||
|
|
||||||
|
# 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")
|
||||||
|
self.assertEqual(results["platform/location"], "TestPlatform/TestLocation")
|
||||||
|
self.assertEqual(results["tenant_group/tenant"], "TestTenantGroup/TestTenant")
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
137
tests/test_list_hostgroup_formats.py
Normal file
137
tests/test_list_hostgroup_formats.py
Normal file
@ -0,0 +1,137 @@
|
|||||||
|
"""Tests for list-based hostgroup formats in configuration."""
|
||||||
|
import unittest
|
||||||
|
from unittest.mock import MagicMock, patch
|
||||||
|
from modules.hostgroups import Hostgroup
|
||||||
|
from modules.exceptions import HostgroupError
|
||||||
|
from modules.tools import verify_hg_format
|
||||||
|
|
||||||
|
|
||||||
|
class TestListHostgroupFormats(unittest.TestCase):
|
||||||
|
"""Test class for list-based hostgroup format functionality."""
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
"""Set up test fixtures."""
|
||||||
|
# Create mock logger
|
||||||
|
self.mock_logger = MagicMock()
|
||||||
|
|
||||||
|
# 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 = 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"
|
||||||
|
cluster_type = MagicMock()
|
||||||
|
cluster_type.name = "TestClusterType"
|
||||||
|
cluster.type = cluster_type
|
||||||
|
self.mock_vm.cluster = cluster
|
||||||
|
|
||||||
|
def test_verify_list_based_hostgroup_format(self):
|
||||||
|
"""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)
|
||||||
|
self.assertIn("TestRole", hostgroups)
|
||||||
|
self.assertIn("TestClusterType/TestCluster", hostgroups)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
Loading…
Reference in New Issue
Block a user