From d5e3199e92b73470e4c97be3893443808d72bbab Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Thu, 19 Feb 2026 16:17:58 +0000 Subject: [PATCH] Implemented central configuration and config path that is configurable as path. Updated tests to use self.config instead on re-initializing config. --- netbox_zabbix_sync/modules/cli.py | 124 +++++++- netbox_zabbix_sync/modules/core.py | 2 + netbox_zabbix_sync/modules/device.py | 84 +++--- netbox_zabbix_sync/modules/settings.py | 10 +- netbox_zabbix_sync/modules/virtual_machine.py | 10 +- tests/test_device_deletion.py | 40 ++- tests/test_physical_device.py | 277 ++++++++---------- tests/test_usermacros.py | 41 ++- 8 files changed, 342 insertions(+), 246 deletions(-) diff --git a/netbox_zabbix_sync/modules/cli.py b/netbox_zabbix_sync/modules/cli.py index 016e28a..4b8110d 100644 --- a/netbox_zabbix_sync/modules/cli.py +++ b/netbox_zabbix_sync/modules/cli.py @@ -5,15 +5,86 @@ from os import environ from netbox_zabbix_sync.modules.core import Sync 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.settings import load_config # Set logging setup_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): """Run the sync process.""" - # set environment variables + # Set log levels based on verbosity flags if arguments.verbose: set_log_levels(logging.WARNING, logging.INFO) if arguments.debug: @@ -48,8 +119,12 @@ def main(arguments): netbox_host = environ.get("NETBOX_HOST") 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 - syncer = Sync() + syncer = Sync(config=config) syncer.connect( nb_host=netbox_host, nb_token=netbox_token, @@ -63,8 +138,10 @@ def main(arguments): def parse_cli(): parser = argparse.ArgumentParser( - description="A script to sync Zabbix with NetBox device data." + description="Synchronise NetBox device data to Zabbix." ) + + # ── Verbosity ────────────────────────────────────────────────────────────── parser.add_argument( "-v", "--verbose", help="Turn on verbose logging.", action="store_true" ) @@ -78,5 +155,46 @@ def parse_cli(): 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() main(args) diff --git a/netbox_zabbix_sync/modules/core.py b/netbox_zabbix_sync/modules/core.py index 05ff624..bd95f99 100644 --- a/netbox_zabbix_sync/modules/core.py +++ b/netbox_zabbix_sync/modules/core.py @@ -203,6 +203,7 @@ class Sync: self.nb_version, self.config["create_journal"], logger, + config=self.config, ) logger.debug("Host %s: Started operations on VM.", vm.name) vm.set_vm_template() @@ -276,6 +277,7 @@ class Sync: self.nb_version, self.config["create_journal"], logger, + config=self.config, ) logger.debug("Host %s: Started operations on device.", device.name) device.set_template( diff --git a/netbox_zabbix_sync/modules/device.py b/netbox_zabbix_sync/modules/device.py index c3e7dd8..7d95172 100644 --- a/netbox_zabbix_sync/modules/device.py +++ b/netbox_zabbix_sync/modules/device.py @@ -29,8 +29,6 @@ from netbox_zabbix_sync.modules.tools import ( ) from netbox_zabbix_sync.modules.usermacros import ZabbixUsermacros -config = load_config() - class PhysicalDevice: """ @@ -39,8 +37,16 @@ class PhysicalDevice: """ 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.id = nb.id self.name = nb.name @@ -76,15 +82,15 @@ class PhysicalDevice: def _inventory_map(self): """Use device inventory maps""" - return config["device_inventory_map"] + return self.config["device_inventory_map"] def _usermacro_map(self): """Use device inventory maps""" - return config["device_usermacro_map"] + return self.config["device_usermacro_map"] def _tag_map(self): """Use device host tag maps""" - return config["device_tag_map"] + return self.config["device_tag_map"] def _set_basics(self): """ @@ -100,10 +106,10 @@ class PhysicalDevice: raise SyncInventoryError(e) # Check if device has custom field for ZBX ID - if config["device_cf"] in self.nb.custom_fields: - self.zabbix_id = self.nb.custom_fields[config["device_cf"]] + if self.config["device_cf"] in self.nb.custom_fields: + self.zabbix_id = self.nb.custom_fields[self.config["device_cf"]] 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) raise SyncInventoryError(e) @@ -134,8 +140,8 @@ class PhysicalDevice: self.nb, self.nb_api_version, logger=self.logger, - nested_sitegroup_flag=config["traverse_site_groups"], - nested_region_flag=config["traverse_regions"], + nested_sitegroup_flag=self.config["traverse_site_groups"], + nested_region_flag=self.config["traverse_regions"], nb_groups=nb_site_groups, nb_regions=nb_regions, ) @@ -182,12 +188,12 @@ class PhysicalDevice: # Get Zabbix templates from the device type device_type_cfs = self.nb.device_type.custom_fields # 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 - return [device_type_cfs[config["template_cf"]]] + return [device_type_cfs[self.config["template_cf"]]] # Custom field not found, return error 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" - {self.nb.device_type.display}." ) @@ -216,27 +222,27 @@ class PhysicalDevice: def set_inventory(self, nbdevice): """Set host inventory""" # Set inventory mode. Default is disabled (see class init function). - if config["inventory_mode"] == "disabled": - if config["inventory_sync"]: + if self.config["inventory_mode"] == "disabled": + if self.config["inventory_sync"]: self.logger.error( "Host %s: Unable to map NetBox inventory to Zabbix." "Inventory sync is enabled in config but inventory mode is disabled", self.name, ) return True - if config["inventory_mode"] == "manual": + if self.config["inventory_mode"] == "manual": self.inventory_mode = 0 - elif config["inventory_mode"] == "automatic": + elif self.config["inventory_mode"] == "automatic": self.inventory_mode = 1 else: self.logger.error( "Host %s: Specified value for inventory mode in config is not valid. Got value %s", self.name, - config["inventory_mode"], + self.config["inventory_mode"], ) return False 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.inventory = field_mapper( self.name, self._inventory_map(), nbdevice, self.logger @@ -382,7 +388,7 @@ class PhysicalDevice: def _zeroize_cf(self): """Sets the hostID custom field in NetBox to zero, effectively destroying the link""" - self.nb.custom_fields[config["device_cf"]] = None + self.nb.custom_fields[self.config["device_cf"]] = None self.nb.save() def _zabbix_hostname_exists(self): @@ -427,7 +433,7 @@ class PhysicalDevice: macros = ZabbixUsermacros( self.nb, self._usermacro_map(), - config["usermacro_sync"], + self.config["usermacro_sync"], logger=self.logger, host=self.name, ) @@ -445,14 +451,14 @@ class PhysicalDevice: tags = ZabbixTags( self.nb, self._tag_map(), - tag_sync=config["tag_sync"], - tag_lower=config["tag_lower"], - tag_name=config["tag_name"], - tag_value=config["tag_value"], + tag_sync=self.config["tag_sync"], + tag_lower=self.config["tag_lower"], + tag_name=self.config["tag_name"], + tag_value=self.config["tag_value"], logger=self.logger, host=self.name, ) - if config["tag_sync"] is False: + if self.config["tag_sync"] is False: self.tags = [] return False self.tags = tags.generate() @@ -482,20 +488,20 @@ class PhysicalDevice: for proxy_type in proxy_types: # Check if we should use custom fields for proxy config field_config = "proxy_cf" if proxy_type == "proxy" else "proxy_group_cf" - if config[field_config]: + if self.config[field_config]: if ( - config[field_config] in self.nb.custom_fields - and self.nb.custom_fields[config[field_config]] + self.config[field_config] in self.nb.custom_fields + and self.nb.custom_fields[self.config[field_config]] ): proxy_name = cf_to_string( - self.nb.custom_fields[config[field_config]] + self.nb.custom_fields[self.config[field_config]] ) elif ( - config[field_config] in self.nb.site.custom_fields - and self.nb.site.custom_fields[config[field_config]] + self.config[field_config] in self.nb.site.custom_fields + and self.nb.site.custom_fields[self.config[field_config]] ): 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 @@ -586,7 +592,7 @@ class PhysicalDevice: self.logger.error(msg) raise SyncExternalError(msg) from e # 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() msg = f"Host {self.name}: Created host in Zabbix. (ID:{self.zabbix_id})" self.logger.info(msg) @@ -828,7 +834,7 @@ class PhysicalDevice: else: self.logger.info("Host %s: inventory_mode OUT of sync.", self.name) 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 if host["inventory"] == self.inventory: self.logger.debug("Host %s: Inventory in-sync.", self.name) @@ -837,12 +843,12 @@ class PhysicalDevice: self.update_zabbix_host(inventory=self.inventory) # 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 # of secret type macros from Netbox netbox_macros = deepcopy(self.usermacros) # 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: # If the Macro is a secret and full sync is NOT activated if macro["type"] == str(1) and not full_sync_bit: @@ -865,7 +871,7 @@ class PhysicalDevice: self.update_zabbix_host(macros=self.usermacros) # Check host tags - if config["tag_sync"]: + if self.config["tag_sync"]: if remove_duplicates( host["tags"], lambda tag: f"{tag['tag']}{tag['value']}" ) == remove_duplicates( diff --git a/netbox_zabbix_sync/modules/settings.py b/netbox_zabbix_sync/modules/settings.py index e5509c6..1d07380 100644 --- a/netbox_zabbix_sync/modules/settings.py +++ b/netbox_zabbix_sync/modules/settings.py @@ -85,10 +85,14 @@ DEFAULT_CONFIG = { } -def load_config(): +def load_config(config_file=None): """Returns combined config from all sources""" - # Overwrite default config with config.py - conf = load_config_file(config_default=DEFAULT_CONFIG) + # Overwrite default config with config file. + # 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 for key in conf: value_setting = load_env_variable(key) diff --git a/netbox_zabbix_sync/modules/virtual_machine.py b/netbox_zabbix_sync/modules/virtual_machine.py index a88865c..a9f57bf 100644 --- a/netbox_zabbix_sync/modules/virtual_machine.py +++ b/netbox_zabbix_sync/modules/virtual_machine.py @@ -7,10 +7,6 @@ from netbox_zabbix_sync.modules.exceptions import ( TemplateError, ) 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): @@ -24,15 +20,15 @@ class VirtualMachine(PhysicalDevice): def _inventory_map(self): """use VM inventory maps""" - return config["vm_inventory_map"] + return self.config["vm_inventory_map"] def _usermacro_map(self): """use VM usermacro maps""" - return config["vm_usermacro_map"] + return self.config["vm_usermacro_map"] def _tag_map(self): """use VM tag maps""" - return config["vm_tag_map"] + return self.config["vm_tag_map"] def set_vm_template(self): """Set Template for VMs. Overwrites default class diff --git a/tests/test_device_deletion.py b/tests/test_device_deletion.py index 0d78f36..2ee154f 100644 --- a/tests/test_device_deletion.py +++ b/tests/test_device_deletion.py @@ -41,17 +41,15 @@ class TestDeviceDeletion(unittest.TestCase): self.mock_logger = MagicMock() # Create PhysicalDevice instance with mocks - with patch( - "netbox_zabbix_sync.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, - ) + self.device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + journal=True, + logger=self.mock_logger, + config={"device_cf": "zabbix_hostid"}, + ) def test_cleanup_successful_deletion(self): """Test successful device deletion from Zabbix.""" @@ -149,17 +147,15 @@ 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( - "netbox_zabbix_sync.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, - ) + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + journal=False, # Disable journaling + logger=self.mock_logger, + config={"device_cf": "zabbix_hostid"}, + ) # Execute result = device.create_journal_entry("info", "Test message") diff --git a/tests/test_physical_device.py b/tests/test_physical_device.py index 0ee6ffc..133de1d 100644 --- a/tests/test_physical_device.py +++ b/tests/test_physical_device.py @@ -36,9 +36,14 @@ class TestPhysicalDevice(unittest.TestCase): self.mock_logger = MagicMock() # Create PhysicalDevice instance with mocks - with patch( - "netbox_zabbix_sync.modules.device.config", - { + self.device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + journal=True, + logger=self.mock_logger, + config={ "device_cf": "zabbix_hostid", "template_cf": "zabbix_template", "templates_config_context": False, @@ -49,15 +54,7 @@ class TestPhysicalDevice(unittest.TestCase): "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, - ) + ) def test_init(self): """Test the initialization of the PhysicalDevice class.""" @@ -75,13 +72,7 @@ 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("netbox_zabbix_sync.modules.device.search") as mock_search, - patch( - "netbox_zabbix_sync.modules.device.config", - {"device_cf": "zabbix_hostid"}, - ), - ): + with patch("netbox_zabbix_sync.modules.device.search") as mock_search: # Make the search function return True to simulate special characters mock_search.return_value = True @@ -91,6 +82,7 @@ class TestPhysicalDevice(unittest.TestCase): self.mock_nb_journal, "3.0", logger=self.mock_logger, + config={"device_cf": "zabbix_hostid"}, ) # 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 - with patch( - "netbox_zabbix_sync.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, - ) + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger, + config={"device_cf": "zabbix_hostid"}, + ) # Test that templates are returned correctly templates = device.get_templates_context() @@ -129,16 +119,14 @@ class TestPhysicalDevice(unittest.TestCase): self.mock_nb_device.config_context = {"zabbix": {"templates": "Template1"}} # Create device with the updated mock - with patch( - "netbox_zabbix_sync.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, - ) + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger, + config={"device_cf": "zabbix_hostid"}, + ) # Test that template is wrapped in a list templates = device.get_templates_context() @@ -150,16 +138,14 @@ class TestPhysicalDevice(unittest.TestCase): self.mock_nb_device.config_context = {} # Create device with the updated mock - with patch( - "netbox_zabbix_sync.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, - ) + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger, + config={"device_cf": "zabbix_hostid"}, + ) # Test that TemplateError is raised with self.assertRaises(TemplateError): @@ -171,16 +157,14 @@ class TestPhysicalDevice(unittest.TestCase): self.mock_nb_device.config_context = {"zabbix": {}} # Create device with the updated mock - with patch( - "netbox_zabbix_sync.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, - ) + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger, + config={"device_cf": "zabbix_hostid"}, + ) # Test that TemplateError is raised with self.assertRaises(TemplateError): @@ -195,17 +179,14 @@ class TestPhysicalDevice(unittest.TestCase): with patch.object( PhysicalDevice, "get_templates_context", return_value=["Template1"] ): - with patch( - "netbox_zabbix_sync.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, - ) + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger, + config={"device_cf": "zabbix_hostid"}, + ) # Call set_template with prefer_config_context=True result = device.set_template( @@ -225,23 +206,20 @@ class TestPhysicalDevice(unittest.TestCase): "inventory_sync": False, } - with patch("netbox_zabbix_sync.modules.device.config", config_patch): - device = PhysicalDevice( - self.mock_nb_device, - self.mock_zabbix, - self.mock_nb_journal, - "3.0", - logger=self.mock_logger, - ) + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger, + config=config_patch, + ) + result = device.set_inventory({}) - # Call set_inventory with the config patch still active - with patch("netbox_zabbix_sync.modules.device.config", config_patch): - result = device.set_inventory({}) - - # Check result - self.assertTrue(result) - # 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): """Test set_inventory with inventory_mode=manual.""" @@ -252,22 +230,19 @@ class TestPhysicalDevice(unittest.TestCase): "inventory_sync": False, } - with patch("netbox_zabbix_sync.modules.device.config", config_patch): - device = PhysicalDevice( - self.mock_nb_device, - self.mock_zabbix, - self.mock_nb_journal, - "3.0", - logger=self.mock_logger, - ) + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger, + config=config_patch, + ) + result = device.set_inventory({}) - # Call set_inventory with the config patch still active - with patch("netbox_zabbix_sync.modules.device.config", config_patch): - result = device.set_inventory({}) - - # Check result - self.assertTrue(result) - 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): """Test set_inventory with inventory_mode=automatic.""" @@ -278,22 +253,19 @@ class TestPhysicalDevice(unittest.TestCase): "inventory_sync": False, } - with patch("netbox_zabbix_sync.modules.device.config", config_patch): - device = PhysicalDevice( - self.mock_nb_device, - self.mock_zabbix, - self.mock_nb_journal, - "3.0", - logger=self.mock_logger, - ) + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger, + config=config_patch, + ) + result = device.set_inventory({}) - # Call set_inventory with the config patch still active - with patch("netbox_zabbix_sync.modules.device.config", config_patch): - result = device.set_inventory({}) - - # Check result - self.assertTrue(result) - 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): """Test set_inventory with inventory_sync=True.""" @@ -305,28 +277,25 @@ class TestPhysicalDevice(unittest.TestCase): "device_inventory_map": {"name": "name", "serial": "serialno_a"}, } - with patch("netbox_zabbix_sync.modules.device.config", config_patch): - device = PhysicalDevice( - self.mock_nb_device, - self.mock_zabbix, - self.mock_nb_journal, - "3.0", - logger=self.mock_logger, - ) + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger, + config=config_patch, + ) - # Create a mock device with the required attributes - mock_device_data = {"name": "test-device", "serial": "ABC123"} + # Create a mock device with the required attributes + mock_device_data = {"name": "test-device", "serial": "ABC123"} + result = device.set_inventory(mock_device_data) - # Call set_inventory with the config patch still active - with patch("netbox_zabbix_sync.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"} - ) + # Check result + 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): """Test isCluster when device is part of a cluster.""" @@ -334,16 +303,14 @@ class TestPhysicalDevice(unittest.TestCase): self.mock_nb_device.virtual_chassis = MagicMock() # Create device with the updated mock - with patch( - "netbox_zabbix_sync.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, - ) + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger, + config={"device_cf": "zabbix_hostid"}, + ) # Check isCluster result self.assertTrue(device.is_cluster()) @@ -354,16 +321,14 @@ class TestPhysicalDevice(unittest.TestCase): self.mock_nb_device.virtual_chassis = None # Create device with the updated mock - with patch( - "netbox_zabbix_sync.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, - ) + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger, + config={"device_cf": "zabbix_hostid"}, + ) # Check isCluster result self.assertFalse(device.is_cluster()) diff --git a/tests/test_usermacros.py b/tests/test_usermacros.py index 45576fc..1b13080 100644 --- a/tests/test_usermacros.py +++ b/tests/test_usermacros.py @@ -27,7 +27,7 @@ class TestUsermacroSync(unittest.TestCase): self.logger = MagicMock() 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""" # Mock the NetBox device with all required attributes mock_nb = MagicMock() @@ -39,6 +39,8 @@ class TestUsermacroSync(unittest.TestCase): mock_nb.primary_ip.address = "192.168.1.1/24" 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 device = PhysicalDevice( nb=mock_nb, @@ -46,18 +48,21 @@ class TestUsermacroSync(unittest.TestCase): nb_journal_class=MagicMock(), nb_version="3.0", logger=self.logger, + config=device_config, ) return device - @patch( - "netbox_zabbix_sync.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() + device = self.create_mock_device( + config={ + "usermacro_sync": False, + "device_cf": "zabbix_hostid", + "tag_sync": False, + } + ) # Call set_usermacros result = device.set_usermacros() @@ -65,10 +70,6 @@ class TestUsermacroSync(unittest.TestCase): self.assertEqual(device.usermacros, []) 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.object(PhysicalDevice, "_usermacro_map") 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 - device = self.create_mock_device() + device = self.create_mock_device( + config={ + "usermacro_sync": True, + "device_cf": "zabbix_hostid", + "tag_sync": False, + } + ) # Call set_usermacros device.set_usermacros() @@ -89,10 +96,6 @@ class TestUsermacroSync(unittest.TestCase): self.assertIsInstance(device.usermacros, list) 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.object(PhysicalDevice, "_usermacro_map") 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 - device = self.create_mock_device() + device = self.create_mock_device( + config={ + "usermacro_sync": "full", + "device_cf": "zabbix_hostid", + "tag_sync": False, + } + ) # Call set_usermacros device.set_usermacros()