Implemented central configuration and config path that is configurable as path. Updated tests to use self.config instead on re-initializing config.

This commit is contained in:
TheNetworkGuy
2026-02-19 16:17:58 +00:00
parent f7d0989320
commit d5e3199e92
8 changed files with 342 additions and 246 deletions
+121 -3
View File
@@ -5,15 +5,86 @@ from os import environ
from netbox_zabbix_sync.modules.core import Sync from netbox_zabbix_sync.modules.core import Sync
from netbox_zabbix_sync.modules.exceptions import EnvironmentVarError from netbox_zabbix_sync.modules.exceptions import EnvironmentVarError
from netbox_zabbix_sync.modules.logging import get_logger, set_log_levels, setup_logger from netbox_zabbix_sync.modules.logging import get_logger, set_log_levels, setup_logger
from netbox_zabbix_sync.modules.settings import load_config
# Set logging # Set logging
setup_logger() setup_logger()
logger = get_logger() logger = get_logger()
# Boolean settings that can be toggled via --flag / --no-flag
_BOOL_ARGS = [
("clustering", "Enable clustering of devices with virtual chassis setup."),
("create_hostgroups", "Enable hostgroup generation (requires Zabbix permissions)."),
("create_journal", "Create NetBox journal entries on changes."),
("sync_vms", "Enable virtual machine sync."),
(
"full_proxy_sync",
"Enable full proxy sync (removes proxies not in config context).",
),
(
"templates_config_context",
"Use config context as the template source instead of a custom field.",
),
(
"templates_config_context_overrule",
"Give config context templates higher priority than custom field templates.",
),
("traverse_regions", "Use the full parent-region path in hostgroup names."),
("traverse_site_groups", "Use the full parent-site-group path in hostgroup names."),
(
"extended_site_properties",
"Fetch additional site info from NetBox (increases API queries).",
),
("inventory_sync", "Sync NetBox device properties to Zabbix inventory."),
("usermacro_sync", "Sync usermacros from NetBox to Zabbix."),
("tag_sync", "Sync host tags to Zabbix."),
("tag_lower", "Lowercase tag names and values before syncing."),
]
# String settings that can be set via --option VALUE
_STR_ARGS = [
("template_cf", "NetBox custom field name for the Zabbix template.", "FIELD"),
("device_cf", "NetBox custom field name for the Zabbix host ID.", "FIELD"),
(
"hostgroup_format",
"Hostgroup path pattern for physical devices (e.g. site/manufacturer/role).",
"PATTERN",
),
(
"vm_hostgroup_format",
"Hostgroup path pattern for virtual machines (e.g. cluster_type/cluster/role).",
"PATTERN",
),
(
"inventory_mode",
"Zabbix inventory mode: disabled, manual, or automatic.",
"MODE",
),
("tag_name", "Zabbix tag name used when syncing NetBox tags.", "NAME"),
(
"tag_value",
"NetBox tag property to use as the Zabbix tag value (name, slug, or display).",
"PROPERTY",
),
]
def _apply_cli_overrides(config: dict, arguments: argparse.Namespace) -> dict:
"""Override loaded config with any values explicitly provided on the CLI."""
for key, _help in _BOOL_ARGS:
cli_val = getattr(arguments, key, None)
if cli_val is not None:
config[key] = cli_val
for key, _help, _meta in _STR_ARGS:
cli_val = getattr(arguments, key, None)
if cli_val is not None:
config[key] = cli_val
return config
def main(arguments): def main(arguments):
"""Run the sync process.""" """Run the sync process."""
# set environment variables # Set log levels based on verbosity flags
if arguments.verbose: if arguments.verbose:
set_log_levels(logging.WARNING, logging.INFO) set_log_levels(logging.WARNING, logging.INFO)
if arguments.debug: if arguments.debug:
@@ -48,8 +119,12 @@ def main(arguments):
netbox_host = environ.get("NETBOX_HOST") netbox_host = environ.get("NETBOX_HOST")
netbox_token = environ.get("NETBOX_TOKEN") netbox_token = environ.get("NETBOX_TOKEN")
# Load config (defaults → config.py → env vars), then apply CLI overrides
config = load_config(config_file=arguments.config)
config = _apply_cli_overrides(config, arguments)
# Run main sync process # Run main sync process
syncer = Sync() syncer = Sync(config=config)
syncer.connect( syncer.connect(
nb_host=netbox_host, nb_host=netbox_host,
nb_token=netbox_token, nb_token=netbox_token,
@@ -63,8 +138,10 @@ def main(arguments):
def parse_cli(): def parse_cli():
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description="A script to sync Zabbix with NetBox device data." description="Synchronise NetBox device data to Zabbix."
) )
# ── Verbosity ──────────────────────────────────────────────────────────────
parser.add_argument( parser.add_argument(
"-v", "--verbose", help="Turn on verbose logging.", action="store_true" "-v", "--verbose", help="Turn on verbose logging.", action="store_true"
) )
@@ -78,5 +155,46 @@ def parse_cli():
action="store_true", action="store_true",
) )
parser.add_argument("-q", "--quiet", help="Turn off warnings.", action="store_true") parser.add_argument("-q", "--quiet", help="Turn off warnings.", action="store_true")
parser.add_argument(
"-c",
"--config",
help="Path to the config file (default: config.py next to the script or in the current directory).",
metavar="FILE",
default=None,
)
# ── Boolean config overrides ───────────────────────────────────────────────
bool_group = parser.add_argument_group(
"config overrides (boolean)",
"Override boolean settings from config.py. "
"Use --flag to enable or --no-flag to disable. "
"When omitted, the value from config.py (or the built-in default) is used.",
)
for key, help_text in _BOOL_ARGS:
flag = key.replace("_", "-")
bool_group.add_argument(
f"--{flag}",
dest=key,
help=help_text,
action=argparse.BooleanOptionalAction,
default=None,
)
# ── String config overrides ────────────────────────────────────────────────
str_group = parser.add_argument_group(
"config overrides (string)",
"Override string settings from config.py. "
"When omitted, the value from config.py (or the built-in default) is used.",
)
for key, help_text, metavar in _STR_ARGS:
flag = key.replace("_", "-")
str_group.add_argument(
f"--{flag}",
dest=key,
help=help_text,
metavar=metavar,
default=None,
)
args = parser.parse_args() args = parser.parse_args()
main(args) main(args)
+2
View File
@@ -203,6 +203,7 @@ class Sync:
self.nb_version, self.nb_version,
self.config["create_journal"], self.config["create_journal"],
logger, logger,
config=self.config,
) )
logger.debug("Host %s: Started operations on VM.", vm.name) logger.debug("Host %s: Started operations on VM.", vm.name)
vm.set_vm_template() vm.set_vm_template()
@@ -276,6 +277,7 @@ class Sync:
self.nb_version, self.nb_version,
self.config["create_journal"], self.config["create_journal"],
logger, logger,
config=self.config,
) )
logger.debug("Host %s: Started operations on device.", device.name) logger.debug("Host %s: Started operations on device.", device.name)
device.set_template( device.set_template(
+45 -39
View File
@@ -29,8 +29,6 @@ from netbox_zabbix_sync.modules.tools import (
) )
from netbox_zabbix_sync.modules.usermacros import ZabbixUsermacros from netbox_zabbix_sync.modules.usermacros import ZabbixUsermacros
config = load_config()
class PhysicalDevice: class PhysicalDevice:
""" """
@@ -39,8 +37,16 @@ class PhysicalDevice:
""" """
def __init__( def __init__(
self, nb, zabbix, nb_journal_class, nb_version, journal=None, logger=None self,
nb,
zabbix,
nb_journal_class,
nb_version,
journal=None,
logger=None,
config=None,
): ):
self.config = config if config is not None else load_config()
self.nb = nb self.nb = nb
self.id = nb.id self.id = nb.id
self.name = nb.name self.name = nb.name
@@ -76,15 +82,15 @@ class PhysicalDevice:
def _inventory_map(self): def _inventory_map(self):
"""Use device inventory maps""" """Use device inventory maps"""
return config["device_inventory_map"] return self.config["device_inventory_map"]
def _usermacro_map(self): def _usermacro_map(self):
"""Use device inventory maps""" """Use device inventory maps"""
return config["device_usermacro_map"] return self.config["device_usermacro_map"]
def _tag_map(self): def _tag_map(self):
"""Use device host tag maps""" """Use device host tag maps"""
return config["device_tag_map"] return self.config["device_tag_map"]
def _set_basics(self): def _set_basics(self):
""" """
@@ -100,10 +106,10 @@ class PhysicalDevice:
raise SyncInventoryError(e) raise SyncInventoryError(e)
# Check if device has custom field for ZBX ID # Check if device has custom field for ZBX ID
if config["device_cf"] in self.nb.custom_fields: if self.config["device_cf"] in self.nb.custom_fields:
self.zabbix_id = self.nb.custom_fields[config["device_cf"]] self.zabbix_id = self.nb.custom_fields[self.config["device_cf"]]
else: else:
e = f"Host {self.name}: Custom field {config['device_cf']} not present" e = f"Host {self.name}: Custom field {self.config['device_cf']} not present"
self.logger.error(e) self.logger.error(e)
raise SyncInventoryError(e) raise SyncInventoryError(e)
@@ -134,8 +140,8 @@ class PhysicalDevice:
self.nb, self.nb,
self.nb_api_version, self.nb_api_version,
logger=self.logger, logger=self.logger,
nested_sitegroup_flag=config["traverse_site_groups"], nested_sitegroup_flag=self.config["traverse_site_groups"],
nested_region_flag=config["traverse_regions"], nested_region_flag=self.config["traverse_regions"],
nb_groups=nb_site_groups, nb_groups=nb_site_groups,
nb_regions=nb_regions, nb_regions=nb_regions,
) )
@@ -182,12 +188,12 @@ class PhysicalDevice:
# Get Zabbix templates from the device type # Get Zabbix templates from the device type
device_type_cfs = self.nb.device_type.custom_fields device_type_cfs = self.nb.device_type.custom_fields
# Check if the ZBX Template CF is present # Check if the ZBX Template CF is present
if config["template_cf"] in device_type_cfs: if self.config["template_cf"] in device_type_cfs:
# Set value to template # Set value to template
return [device_type_cfs[config["template_cf"]]] return [device_type_cfs[self.config["template_cf"]]]
# Custom field not found, return error # Custom field not found, return error
e = ( e = (
f"Custom field {config['template_cf']} not " f"Custom field {self.config['template_cf']} not "
f"found for {self.nb.device_type.manufacturer.name}" f"found for {self.nb.device_type.manufacturer.name}"
f" - {self.nb.device_type.display}." f" - {self.nb.device_type.display}."
) )
@@ -216,27 +222,27 @@ class PhysicalDevice:
def set_inventory(self, nbdevice): def set_inventory(self, nbdevice):
"""Set host inventory""" """Set host inventory"""
# Set inventory mode. Default is disabled (see class init function). # Set inventory mode. Default is disabled (see class init function).
if config["inventory_mode"] == "disabled": if self.config["inventory_mode"] == "disabled":
if config["inventory_sync"]: if self.config["inventory_sync"]:
self.logger.error( self.logger.error(
"Host %s: Unable to map NetBox inventory to Zabbix." "Host %s: Unable to map NetBox inventory to Zabbix."
"Inventory sync is enabled in config but inventory mode is disabled", "Inventory sync is enabled in config but inventory mode is disabled",
self.name, self.name,
) )
return True return True
if config["inventory_mode"] == "manual": if self.config["inventory_mode"] == "manual":
self.inventory_mode = 0 self.inventory_mode = 0
elif config["inventory_mode"] == "automatic": elif self.config["inventory_mode"] == "automatic":
self.inventory_mode = 1 self.inventory_mode = 1
else: else:
self.logger.error( self.logger.error(
"Host %s: Specified value for inventory mode in config is not valid. Got value %s", "Host %s: Specified value for inventory mode in config is not valid. Got value %s",
self.name, self.name,
config["inventory_mode"], self.config["inventory_mode"],
) )
return False return False
self.inventory = {} self.inventory = {}
if config["inventory_sync"] and self.inventory_mode in [0, 1]: if self.config["inventory_sync"] and self.inventory_mode in [0, 1]:
self.logger.debug("Host %s: Starting inventory mapper.", self.name) self.logger.debug("Host %s: Starting inventory mapper.", self.name)
self.inventory = field_mapper( self.inventory = field_mapper(
self.name, self._inventory_map(), nbdevice, self.logger self.name, self._inventory_map(), nbdevice, self.logger
@@ -382,7 +388,7 @@ class PhysicalDevice:
def _zeroize_cf(self): def _zeroize_cf(self):
"""Sets the hostID custom field in NetBox to zero, """Sets the hostID custom field in NetBox to zero,
effectively destroying the link""" effectively destroying the link"""
self.nb.custom_fields[config["device_cf"]] = None self.nb.custom_fields[self.config["device_cf"]] = None
self.nb.save() self.nb.save()
def _zabbix_hostname_exists(self): def _zabbix_hostname_exists(self):
@@ -427,7 +433,7 @@ class PhysicalDevice:
macros = ZabbixUsermacros( macros = ZabbixUsermacros(
self.nb, self.nb,
self._usermacro_map(), self._usermacro_map(),
config["usermacro_sync"], self.config["usermacro_sync"],
logger=self.logger, logger=self.logger,
host=self.name, host=self.name,
) )
@@ -445,14 +451,14 @@ class PhysicalDevice:
tags = ZabbixTags( tags = ZabbixTags(
self.nb, self.nb,
self._tag_map(), self._tag_map(),
tag_sync=config["tag_sync"], tag_sync=self.config["tag_sync"],
tag_lower=config["tag_lower"], tag_lower=self.config["tag_lower"],
tag_name=config["tag_name"], tag_name=self.config["tag_name"],
tag_value=config["tag_value"], tag_value=self.config["tag_value"],
logger=self.logger, logger=self.logger,
host=self.name, host=self.name,
) )
if config["tag_sync"] is False: if self.config["tag_sync"] is False:
self.tags = [] self.tags = []
return False return False
self.tags = tags.generate() self.tags = tags.generate()
@@ -482,20 +488,20 @@ class PhysicalDevice:
for proxy_type in proxy_types: for proxy_type in proxy_types:
# Check if we should use custom fields for proxy config # Check if we should use custom fields for proxy config
field_config = "proxy_cf" if proxy_type == "proxy" else "proxy_group_cf" field_config = "proxy_cf" if proxy_type == "proxy" else "proxy_group_cf"
if config[field_config]: if self.config[field_config]:
if ( if (
config[field_config] in self.nb.custom_fields self.config[field_config] in self.nb.custom_fields
and self.nb.custom_fields[config[field_config]] and self.nb.custom_fields[self.config[field_config]]
): ):
proxy_name = cf_to_string( proxy_name = cf_to_string(
self.nb.custom_fields[config[field_config]] self.nb.custom_fields[self.config[field_config]]
) )
elif ( elif (
config[field_config] in self.nb.site.custom_fields self.config[field_config] in self.nb.site.custom_fields
and self.nb.site.custom_fields[config[field_config]] and self.nb.site.custom_fields[self.config[field_config]]
): ):
proxy_name = cf_to_string( proxy_name = cf_to_string(
self.nb.site.custom_fields[config[field_config]] self.nb.site.custom_fields[self.config[field_config]]
) )
# Otherwise check if the proxy is configured in NetBox CC # Otherwise check if the proxy is configured in NetBox CC
@@ -586,7 +592,7 @@ class PhysicalDevice:
self.logger.error(msg) self.logger.error(msg)
raise SyncExternalError(msg) from e raise SyncExternalError(msg) from e
# Set NetBox custom field to hostID value. # Set NetBox custom field to hostID value.
self.nb.custom_fields[config["device_cf"]] = int(self.zabbix_id) self.nb.custom_fields[self.config["device_cf"]] = int(self.zabbix_id)
self.nb.save() self.nb.save()
msg = f"Host {self.name}: Created host in Zabbix. (ID:{self.zabbix_id})" msg = f"Host {self.name}: Created host in Zabbix. (ID:{self.zabbix_id})"
self.logger.info(msg) self.logger.info(msg)
@@ -828,7 +834,7 @@ class PhysicalDevice:
else: else:
self.logger.info("Host %s: inventory_mode OUT of sync.", self.name) self.logger.info("Host %s: inventory_mode OUT of sync.", self.name)
self.update_zabbix_host(inventory_mode=str(self.inventory_mode)) self.update_zabbix_host(inventory_mode=str(self.inventory_mode))
if config["inventory_sync"] and self.inventory_mode in [0, 1]: if self.config["inventory_sync"] and self.inventory_mode in [0, 1]:
# Check host inventory mapping # Check host inventory mapping
if host["inventory"] == self.inventory: if host["inventory"] == self.inventory:
self.logger.debug("Host %s: Inventory in-sync.", self.name) self.logger.debug("Host %s: Inventory in-sync.", self.name)
@@ -837,12 +843,12 @@ class PhysicalDevice:
self.update_zabbix_host(inventory=self.inventory) self.update_zabbix_host(inventory=self.inventory)
# Check host usermacros # Check host usermacros
if config["usermacro_sync"]: if self.config["usermacro_sync"]:
# Make a full copy synce we dont want to lose the original value # Make a full copy synce we dont want to lose the original value
# of secret type macros from Netbox # of secret type macros from Netbox
netbox_macros = deepcopy(self.usermacros) netbox_macros = deepcopy(self.usermacros)
# Set the sync bit # Set the sync bit
full_sync_bit = bool(str(config["usermacro_sync"]).lower() == "full") full_sync_bit = bool(str(self.config["usermacro_sync"]).lower() == "full")
for macro in netbox_macros: for macro in netbox_macros:
# If the Macro is a secret and full sync is NOT activated # If the Macro is a secret and full sync is NOT activated
if macro["type"] == str(1) and not full_sync_bit: if macro["type"] == str(1) and not full_sync_bit:
@@ -865,7 +871,7 @@ class PhysicalDevice:
self.update_zabbix_host(macros=self.usermacros) self.update_zabbix_host(macros=self.usermacros)
# Check host tags # Check host tags
if config["tag_sync"]: if self.config["tag_sync"]:
if remove_duplicates( if remove_duplicates(
host["tags"], lambda tag: f"{tag['tag']}{tag['value']}" host["tags"], lambda tag: f"{tag['tag']}{tag['value']}"
) == remove_duplicates( ) == remove_duplicates(
+7 -3
View File
@@ -85,10 +85,14 @@ DEFAULT_CONFIG = {
} }
def load_config(): def load_config(config_file=None):
"""Returns combined config from all sources""" """Returns combined config from all sources"""
# Overwrite default config with config.py # Overwrite default config with config file.
conf = load_config_file(config_default=DEFAULT_CONFIG) # Default config file is config.py but can be overridden by providing a different file path.
conf = load_config_file(
config_default=DEFAULT_CONFIG,
config_file=config_file if config_file else "config.py",
)
# Overwrite default config and config.py with environment variables # Overwrite default config and config.py with environment variables
for key in conf: for key in conf:
value_setting = load_env_variable(key) value_setting = load_env_variable(key)
@@ -7,10 +7,6 @@ from netbox_zabbix_sync.modules.exceptions import (
TemplateError, TemplateError,
) )
from netbox_zabbix_sync.modules.interface import ZabbixInterface from netbox_zabbix_sync.modules.interface import ZabbixInterface
from netbox_zabbix_sync.modules.settings import load_config
# Load config
config = load_config()
class VirtualMachine(PhysicalDevice): class VirtualMachine(PhysicalDevice):
@@ -24,15 +20,15 @@ class VirtualMachine(PhysicalDevice):
def _inventory_map(self): def _inventory_map(self):
"""use VM inventory maps""" """use VM inventory maps"""
return config["vm_inventory_map"] return self.config["vm_inventory_map"]
def _usermacro_map(self): def _usermacro_map(self):
"""use VM usermacro maps""" """use VM usermacro maps"""
return config["vm_usermacro_map"] return self.config["vm_usermacro_map"]
def _tag_map(self): def _tag_map(self):
"""use VM tag maps""" """use VM tag maps"""
return config["vm_tag_map"] return self.config["vm_tag_map"]
def set_vm_template(self): def set_vm_template(self):
"""Set Template for VMs. Overwrites default class """Set Template for VMs. Overwrites default class
+18 -22
View File
@@ -41,17 +41,15 @@ class TestDeviceDeletion(unittest.TestCase):
self.mock_logger = MagicMock() self.mock_logger = MagicMock()
# Create PhysicalDevice instance with mocks # Create PhysicalDevice instance with mocks
with patch( self.device = PhysicalDevice(
"netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"} self.mock_nb_device,
): self.mock_zabbix,
self.device = PhysicalDevice( self.mock_nb_journal,
self.mock_nb_device, "3.0",
self.mock_zabbix, journal=True,
self.mock_nb_journal, logger=self.mock_logger,
"3.0", config={"device_cf": "zabbix_hostid"},
journal=True, )
logger=self.mock_logger,
)
def test_cleanup_successful_deletion(self): def test_cleanup_successful_deletion(self):
"""Test successful device deletion from Zabbix.""" """Test successful device deletion from Zabbix."""
@@ -149,17 +147,15 @@ class TestDeviceDeletion(unittest.TestCase):
def test_create_journal_entry_when_disabled(self): def test_create_journal_entry_when_disabled(self):
"""Test create_journal_entry when journaling is disabled.""" """Test create_journal_entry when journaling is disabled."""
# Setup - create device with journal=False # Setup - create device with journal=False
with patch( device = PhysicalDevice(
"netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"} self.mock_nb_device,
): self.mock_zabbix,
device = PhysicalDevice( self.mock_nb_journal,
self.mock_nb_device, "3.0",
self.mock_zabbix, journal=False, # Disable journaling
self.mock_nb_journal, logger=self.mock_logger,
"3.0", config={"device_cf": "zabbix_hostid"},
journal=False, # Disable journaling )
logger=self.mock_logger,
)
# Execute # Execute
result = device.create_journal_entry("info", "Test message") result = device.create_journal_entry("info", "Test message")
+121 -156
View File
@@ -36,9 +36,14 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_logger = MagicMock() self.mock_logger = MagicMock()
# Create PhysicalDevice instance with mocks # Create PhysicalDevice instance with mocks
with patch( self.device = PhysicalDevice(
"netbox_zabbix_sync.modules.device.config", self.mock_nb_device,
{ self.mock_zabbix,
self.mock_nb_journal,
"3.0",
journal=True,
logger=self.mock_logger,
config={
"device_cf": "zabbix_hostid", "device_cf": "zabbix_hostid",
"template_cf": "zabbix_template", "template_cf": "zabbix_template",
"templates_config_context": False, "templates_config_context": False,
@@ -49,15 +54,7 @@ class TestPhysicalDevice(unittest.TestCase):
"inventory_sync": False, "inventory_sync": False,
"device_inventory_map": {}, "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,
)
def test_init(self): def test_init(self):
"""Test the initialization of the PhysicalDevice class.""" """Test the initialization of the PhysicalDevice class."""
@@ -75,13 +72,7 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_nb_device.name = "test-devïce" self.mock_nb_device.name = "test-devïce"
# We need to patch the search function to simulate finding special characters # We need to patch the search function to simulate finding special characters
with ( with patch("netbox_zabbix_sync.modules.device.search") as mock_search:
patch("netbox_zabbix_sync.modules.device.search") as mock_search,
patch(
"netbox_zabbix_sync.modules.device.config",
{"device_cf": "zabbix_hostid"},
),
):
# Make the search function return True to simulate special characters # Make the search function return True to simulate special characters
mock_search.return_value = True mock_search.return_value = True
@@ -91,6 +82,7 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_nb_journal, self.mock_nb_journal,
"3.0", "3.0",
logger=self.mock_logger, logger=self.mock_logger,
config={"device_cf": "zabbix_hostid"},
) )
# With the mocked search function, the name should be changed to NETBOX_ID format # With the mocked search function, the name should be changed to NETBOX_ID format
@@ -108,16 +100,14 @@ class TestPhysicalDevice(unittest.TestCase):
} }
# Create device with the updated mock # Create device with the updated mock
with patch( device = PhysicalDevice(
"netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"} self.mock_nb_device,
): self.mock_zabbix,
device = PhysicalDevice( self.mock_nb_journal,
self.mock_nb_device, "3.0",
self.mock_zabbix, logger=self.mock_logger,
self.mock_nb_journal, config={"device_cf": "zabbix_hostid"},
"3.0", )
logger=self.mock_logger,
)
# Test that templates are returned correctly # Test that templates are returned correctly
templates = device.get_templates_context() templates = device.get_templates_context()
@@ -129,16 +119,14 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_nb_device.config_context = {"zabbix": {"templates": "Template1"}} self.mock_nb_device.config_context = {"zabbix": {"templates": "Template1"}}
# Create device with the updated mock # Create device with the updated mock
with patch( device = PhysicalDevice(
"netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"} self.mock_nb_device,
): self.mock_zabbix,
device = PhysicalDevice( self.mock_nb_journal,
self.mock_nb_device, "3.0",
self.mock_zabbix, logger=self.mock_logger,
self.mock_nb_journal, config={"device_cf": "zabbix_hostid"},
"3.0", )
logger=self.mock_logger,
)
# Test that template is wrapped in a list # Test that template is wrapped in a list
templates = device.get_templates_context() templates = device.get_templates_context()
@@ -150,16 +138,14 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_nb_device.config_context = {} self.mock_nb_device.config_context = {}
# Create device with the updated mock # Create device with the updated mock
with patch( device = PhysicalDevice(
"netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"} self.mock_nb_device,
): self.mock_zabbix,
device = PhysicalDevice( self.mock_nb_journal,
self.mock_nb_device, "3.0",
self.mock_zabbix, logger=self.mock_logger,
self.mock_nb_journal, config={"device_cf": "zabbix_hostid"},
"3.0", )
logger=self.mock_logger,
)
# Test that TemplateError is raised # Test that TemplateError is raised
with self.assertRaises(TemplateError): with self.assertRaises(TemplateError):
@@ -171,16 +157,14 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_nb_device.config_context = {"zabbix": {}} self.mock_nb_device.config_context = {"zabbix": {}}
# Create device with the updated mock # Create device with the updated mock
with patch( device = PhysicalDevice(
"netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"} self.mock_nb_device,
): self.mock_zabbix,
device = PhysicalDevice( self.mock_nb_journal,
self.mock_nb_device, "3.0",
self.mock_zabbix, logger=self.mock_logger,
self.mock_nb_journal, config={"device_cf": "zabbix_hostid"},
"3.0", )
logger=self.mock_logger,
)
# Test that TemplateError is raised # Test that TemplateError is raised
with self.assertRaises(TemplateError): with self.assertRaises(TemplateError):
@@ -195,17 +179,14 @@ class TestPhysicalDevice(unittest.TestCase):
with patch.object( with patch.object(
PhysicalDevice, "get_templates_context", return_value=["Template1"] PhysicalDevice, "get_templates_context", return_value=["Template1"]
): ):
with patch( device = PhysicalDevice(
"netbox_zabbix_sync.modules.device.config", self.mock_nb_device,
{"device_cf": "zabbix_hostid"}, self.mock_zabbix,
): self.mock_nb_journal,
device = PhysicalDevice( "3.0",
self.mock_nb_device, logger=self.mock_logger,
self.mock_zabbix, config={"device_cf": "zabbix_hostid"},
self.mock_nb_journal, )
"3.0",
logger=self.mock_logger,
)
# Call set_template with prefer_config_context=True # Call set_template with prefer_config_context=True
result = device.set_template( result = device.set_template(
@@ -225,23 +206,20 @@ class TestPhysicalDevice(unittest.TestCase):
"inventory_sync": False, "inventory_sync": False,
} }
with patch("netbox_zabbix_sync.modules.device.config", config_patch): device = PhysicalDevice(
device = PhysicalDevice( self.mock_nb_device,
self.mock_nb_device, self.mock_zabbix,
self.mock_zabbix, self.mock_nb_journal,
self.mock_nb_journal, "3.0",
"3.0", logger=self.mock_logger,
logger=self.mock_logger, config=config_patch,
) )
result = device.set_inventory({})
# Call set_inventory with the config patch still active # Check result
with patch("netbox_zabbix_sync.modules.device.config", config_patch): self.assertTrue(result)
result = device.set_inventory({}) # Default value for disabled inventory
self.assertEqual(device.inventory_mode, -1)
# Check result
self.assertTrue(result)
# Default value for disabled inventory
self.assertEqual(device.inventory_mode, -1)
def test_set_inventory_manual_mode(self): def test_set_inventory_manual_mode(self):
"""Test set_inventory with inventory_mode=manual.""" """Test set_inventory with inventory_mode=manual."""
@@ -252,22 +230,19 @@ class TestPhysicalDevice(unittest.TestCase):
"inventory_sync": False, "inventory_sync": False,
} }
with patch("netbox_zabbix_sync.modules.device.config", config_patch): device = PhysicalDevice(
device = PhysicalDevice( self.mock_nb_device,
self.mock_nb_device, self.mock_zabbix,
self.mock_zabbix, self.mock_nb_journal,
self.mock_nb_journal, "3.0",
"3.0", logger=self.mock_logger,
logger=self.mock_logger, config=config_patch,
) )
result = device.set_inventory({})
# Call set_inventory with the config patch still active # Check result
with patch("netbox_zabbix_sync.modules.device.config", config_patch): self.assertTrue(result)
result = device.set_inventory({}) self.assertEqual(device.inventory_mode, 0) # Manual mode
# Check result
self.assertTrue(result)
self.assertEqual(device.inventory_mode, 0) # Manual mode
def test_set_inventory_automatic_mode(self): def test_set_inventory_automatic_mode(self):
"""Test set_inventory with inventory_mode=automatic.""" """Test set_inventory with inventory_mode=automatic."""
@@ -278,22 +253,19 @@ class TestPhysicalDevice(unittest.TestCase):
"inventory_sync": False, "inventory_sync": False,
} }
with patch("netbox_zabbix_sync.modules.device.config", config_patch): device = PhysicalDevice(
device = PhysicalDevice( self.mock_nb_device,
self.mock_nb_device, self.mock_zabbix,
self.mock_zabbix, self.mock_nb_journal,
self.mock_nb_journal, "3.0",
"3.0", logger=self.mock_logger,
logger=self.mock_logger, config=config_patch,
) )
result = device.set_inventory({})
# Call set_inventory with the config patch still active # Check result
with patch("netbox_zabbix_sync.modules.device.config", config_patch): self.assertTrue(result)
result = device.set_inventory({}) self.assertEqual(device.inventory_mode, 1) # Automatic mode
# Check result
self.assertTrue(result)
self.assertEqual(device.inventory_mode, 1) # Automatic mode
def test_set_inventory_with_inventory_sync(self): def test_set_inventory_with_inventory_sync(self):
"""Test set_inventory with inventory_sync=True.""" """Test set_inventory with inventory_sync=True."""
@@ -305,28 +277,25 @@ class TestPhysicalDevice(unittest.TestCase):
"device_inventory_map": {"name": "name", "serial": "serialno_a"}, "device_inventory_map": {"name": "name", "serial": "serialno_a"},
} }
with patch("netbox_zabbix_sync.modules.device.config", config_patch): device = PhysicalDevice(
device = PhysicalDevice( self.mock_nb_device,
self.mock_nb_device, self.mock_zabbix,
self.mock_zabbix, self.mock_nb_journal,
self.mock_nb_journal, "3.0",
"3.0", logger=self.mock_logger,
logger=self.mock_logger, config=config_patch,
) )
# Create a mock device with the required attributes # Create a mock device with the required attributes
mock_device_data = {"name": "test-device", "serial": "ABC123"} mock_device_data = {"name": "test-device", "serial": "ABC123"}
result = device.set_inventory(mock_device_data)
# Call set_inventory with the config patch still active # Check result
with patch("netbox_zabbix_sync.modules.device.config", config_patch): self.assertTrue(result)
result = device.set_inventory(mock_device_data) self.assertEqual(device.inventory_mode, 0) # Manual mode
self.assertEqual(
# Check result device.inventory, {"name": "test-device", "serialno_a": "ABC123"}
self.assertTrue(result) )
self.assertEqual(device.inventory_mode, 0) # Manual mode
self.assertEqual(
device.inventory, {"name": "test-device", "serialno_a": "ABC123"}
)
def test_iscluster_true(self): def test_iscluster_true(self):
"""Test isCluster when device is part of a cluster.""" """Test isCluster when device is part of a cluster."""
@@ -334,16 +303,14 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_nb_device.virtual_chassis = MagicMock() self.mock_nb_device.virtual_chassis = MagicMock()
# Create device with the updated mock # Create device with the updated mock
with patch( device = PhysicalDevice(
"netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"} self.mock_nb_device,
): self.mock_zabbix,
device = PhysicalDevice( self.mock_nb_journal,
self.mock_nb_device, "3.0",
self.mock_zabbix, logger=self.mock_logger,
self.mock_nb_journal, config={"device_cf": "zabbix_hostid"},
"3.0", )
logger=self.mock_logger,
)
# Check isCluster result # Check isCluster result
self.assertTrue(device.is_cluster()) self.assertTrue(device.is_cluster())
@@ -354,16 +321,14 @@ class TestPhysicalDevice(unittest.TestCase):
self.mock_nb_device.virtual_chassis = None self.mock_nb_device.virtual_chassis = None
# Create device with the updated mock # Create device with the updated mock
with patch( device = PhysicalDevice(
"netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"} self.mock_nb_device,
): self.mock_zabbix,
device = PhysicalDevice( self.mock_nb_journal,
self.mock_nb_device, "3.0",
self.mock_zabbix, logger=self.mock_logger,
self.mock_nb_journal, config={"device_cf": "zabbix_hostid"},
"3.0", )
logger=self.mock_logger,
)
# Check isCluster result # Check isCluster result
self.assertFalse(device.is_cluster()) self.assertFalse(device.is_cluster())
+25 -16
View File
@@ -27,7 +27,7 @@ class TestUsermacroSync(unittest.TestCase):
self.logger = MagicMock() self.logger = MagicMock()
self.usermacro_map = {"serial": "{$HW_SERIAL}"} self.usermacro_map = {"serial": "{$HW_SERIAL}"}
def create_mock_device(self): def create_mock_device(self, config=None):
"""Helper method to create a properly mocked PhysicalDevice""" """Helper method to create a properly mocked PhysicalDevice"""
# Mock the NetBox device with all required attributes # Mock the NetBox device with all required attributes
mock_nb = MagicMock() mock_nb = MagicMock()
@@ -39,6 +39,8 @@ class TestUsermacroSync(unittest.TestCase):
mock_nb.primary_ip.address = "192.168.1.1/24" mock_nb.primary_ip.address = "192.168.1.1/24"
mock_nb.custom_fields = {"zabbix_hostid": None} mock_nb.custom_fields = {"zabbix_hostid": None}
device_config = config if config is not None else {"device_cf": "zabbix_hostid"}
# Create device with proper initialization # Create device with proper initialization
device = PhysicalDevice( device = PhysicalDevice(
nb=mock_nb, nb=mock_nb,
@@ -46,18 +48,21 @@ class TestUsermacroSync(unittest.TestCase):
nb_journal_class=MagicMock(), nb_journal_class=MagicMock(),
nb_version="3.0", nb_version="3.0",
logger=self.logger, logger=self.logger,
config=device_config,
) )
return device return device
@patch(
"netbox_zabbix_sync.modules.device.config",
{"usermacro_sync": False, "device_cf": "zabbix_hostid", "tag_sync": False},
)
@patch.object(PhysicalDevice, "_usermacro_map") @patch.object(PhysicalDevice, "_usermacro_map")
def test_usermacro_sync_false(self, mock_usermacro_map): def test_usermacro_sync_false(self, mock_usermacro_map):
mock_usermacro_map.return_value = self.usermacro_map mock_usermacro_map.return_value = self.usermacro_map
device = self.create_mock_device() device = self.create_mock_device(
config={
"usermacro_sync": False,
"device_cf": "zabbix_hostid",
"tag_sync": False,
}
)
# Call set_usermacros # Call set_usermacros
result = device.set_usermacros() result = device.set_usermacros()
@@ -65,10 +70,6 @@ class TestUsermacroSync(unittest.TestCase):
self.assertEqual(device.usermacros, []) self.assertEqual(device.usermacros, [])
self.assertTrue(result is True or result is None) self.assertTrue(result is True or result is None)
@patch(
"netbox_zabbix_sync.modules.device.config",
{"usermacro_sync": True, "device_cf": "zabbix_hostid", "tag_sync": False},
)
@patch("netbox_zabbix_sync.modules.device.ZabbixUsermacros") @patch("netbox_zabbix_sync.modules.device.ZabbixUsermacros")
@patch.object(PhysicalDevice, "_usermacro_map") @patch.object(PhysicalDevice, "_usermacro_map")
def test_usermacro_sync_true(self, mock_usermacro_map, mock_usermacros_class): def test_usermacro_sync_true(self, mock_usermacro_map, mock_usermacros_class):
@@ -81,7 +82,13 @@ class TestUsermacroSync(unittest.TestCase):
] ]
mock_usermacros_class.return_value = mock_macros_instance mock_usermacros_class.return_value = mock_macros_instance
device = self.create_mock_device() device = self.create_mock_device(
config={
"usermacro_sync": True,
"device_cf": "zabbix_hostid",
"tag_sync": False,
}
)
# Call set_usermacros # Call set_usermacros
device.set_usermacros() device.set_usermacros()
@@ -89,10 +96,6 @@ class TestUsermacroSync(unittest.TestCase):
self.assertIsInstance(device.usermacros, list) self.assertIsInstance(device.usermacros, list)
self.assertGreater(len(device.usermacros), 0) self.assertGreater(len(device.usermacros), 0)
@patch(
"netbox_zabbix_sync.modules.device.config",
{"usermacro_sync": "full", "device_cf": "zabbix_hostid", "tag_sync": False},
)
@patch("netbox_zabbix_sync.modules.device.ZabbixUsermacros") @patch("netbox_zabbix_sync.modules.device.ZabbixUsermacros")
@patch.object(PhysicalDevice, "_usermacro_map") @patch.object(PhysicalDevice, "_usermacro_map")
def test_usermacro_sync_full(self, mock_usermacro_map, mock_usermacros_class): def test_usermacro_sync_full(self, mock_usermacro_map, mock_usermacros_class):
@@ -105,7 +108,13 @@ class TestUsermacroSync(unittest.TestCase):
] ]
mock_usermacros_class.return_value = mock_macros_instance mock_usermacros_class.return_value = mock_macros_instance
device = self.create_mock_device() device = self.create_mock_device(
config={
"usermacro_sync": "full",
"device_cf": "zabbix_hostid",
"tag_sync": False,
}
)
# Call set_usermacros # Call set_usermacros
device.set_usermacros() device.set_usermacros()