diff --git a/.github/workflows/publish-image.yml b/.github/workflows/publish-image.yml index 615b784..4531303 100644 --- a/.github/workflows/publish-image.yml +++ b/.github/workflows/publish-image.yml @@ -1,5 +1,9 @@ name: Build and Push Docker Image +on: + push: + branches: + - main permissions: contents: read packages: write diff --git a/.github/workflows/quality.yml b/.github/workflows/quality.yml index 7b01f6f..4421765 100644 --- a/.github/workflows/quality.yml +++ b/.github/workflows/quality.yml @@ -1,11 +1,12 @@ --- name: Pylint Quality control -on: - workflow_call +on: + push: + pull_request: jobs: - build: + python_quality_testing: runs-on: ubuntu-latest strategy: matrix: @@ -23,4 +24,4 @@ jobs: pip install -r requirements.txt - name: Analysing the code with pylint run: | - pylint --module-naming-style=any $(git ls-files '*.py') + pylint --module-naming-style=any modules/* netbox_zabbix_sync.py diff --git a/.github/workflows/run_tests.yml b/.github/workflows/run_tests.yml new file mode 100644 index 0000000..9093c96 --- /dev/null +++ b/.github/workflows/run_tests.yml @@ -0,0 +1,25 @@ +--- +name: Pytest code testing + +on: + push: + pull_request: + +jobs: + test_code: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: 3.12 + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install pytest pytest-mock + pip install -r requirements.txt + - name: Testing the code with PyTest + run: | + cp config.py.example config.py + pytest tests diff --git a/.gitignore b/.gitignore index 2a3448b..0693f26 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,5 @@ Pipfile.lock # Byte-compiled / optimized / DLL files __pycache__/ *.py[cod] +.vscode +.flake \ No newline at end of file diff --git a/README.md b/README.md index 0bd6bfc..62a6673 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,7 @@ # NetBox to Zabbix synchronization A script to create, update and delete Zabbix hosts using NetBox device objects. +Currently compatible with Zabbix 7.0. Zabbix 7.2 is unfortunately not supported and will result in the script failing. ## Installation via Docker @@ -48,9 +49,15 @@ Make sure that you have a python environment with the following packages installed. You can also use the `requirements.txt` file for installation with pip. -``` +```sh +# Packages: pynetbox pyzabbix + +# Install them through requirements.txt from a venv: +virtualenv .venv +source .venv/bin/activate +.venv/bin/pip --require-virtualenv install -r requirements.txt ``` ### Config file @@ -58,7 +65,7 @@ pyzabbix First time user? Copy the `config.py.example` file to `config.py`. This file is used for modifying filters and setting variables such as custom field names. -``` +```sh cp config.py.example config.py ``` @@ -84,8 +91,8 @@ ZABBIX_TOKEN=othersecrettoken If you are using custom SSL certificates for NetBox and/or Zabbix, you can set the following environment variable to the path of your CA bundle file: -```bash -REQUEST_CA_BUNDLE=/path/to/your/ca-bundle.crt +```sh +export REQUESTS_CA_BUNDLE=/path/to/your/ca-bundle.crt ``` ### NetBox custom fields diff --git a/modules/config.py b/modules/config.py new file mode 100644 index 0000000..586227d --- /dev/null +++ b/modules/config.py @@ -0,0 +1,124 @@ +""" +Module for parsing configuration from the top level config.py file +""" +from pathlib import Path +from importlib import util +from os import environ +from logging import getLogger + +logger = getLogger(__name__) + +# PLEASE NOTE: This is a sample config file. Please do NOT make any edits in this file! +# You should create your own config.py and it will overwrite the default config. + +DEFAULT_CONFIG = { + "templates_config_context": False, + "templates_config_context_overrule": False, + "template_cf": "zabbix_template", + "device_cf": "zabbix_hostid", + "clustering": False, + "create_hostgroups": True, + "create_journal": False, + "sync_vms": False, + "vm_hostgroup_format": "cluster_type/cluster/role", + "full_proxy_sync": False, + "zabbix_device_removal": ["Decommissioning", "Inventory"], + "zabbix_device_disable": ["Offline", "Planned", "Staged", "Failed"], + "hostgroup_format": "site/manufacturer/role", + "traverse_regions": False, + "traverse_site_groups": False, + "nb_device_filter": {"name__n": "null"}, + "nb_vm_filter": {"name__n": "null"}, + "inventory_mode": "disabled", + "inventory_sync": False, + "device_inventory_map": { + "asset_tag": "asset_tag", + "virtual_chassis/name": "chassis", + "status/label": "deployment_status", + "location/name": "location", + "latitude": "location_lat", + "longitude": "location_lon", + "comments": "notes", + "name": "name", + "rack/name": "site_rack", + "serial": "serialno_a", + "device_type/model": "type", + "device_type/manufacturer/name": "vendor", + "oob_ip/address": "oob_ip" + }, + "vm_inventory_map": { + "status/label": "deployment_status", + "comments": "notes", + "name": "name" + }, + "usermacro_sync": False, + "device_usermacro_map": { + "serial": "{$HW_SERIAL}", + "role/name": "{$DEV_ROLE}", + "url": "{$NB_URL}", + "id": "{$NB_ID}" + }, + "vm_usermacro_map": { + "memory": "{$TOTAL_MEMORY}", + "role/name": "{$DEV_ROLE}", + "url": "{$NB_URL}", + "id": "{$NB_ID}" + }, + "tag_sync": False, + "tag_lower": True, + "tag_name": 'NetBox', + "tag_value": "name", + "device_tag_map": { + "site/name": "site", + "rack/name": "rack", + "platform/name": "target" + }, + "vm_tag_map": { + "site/name": "site", + "cluster/name": "cluster", + "platform/name": "target" + } +} + + +def load_config(): + """Returns combined config from all sources""" + # Overwrite default config with config.py + conf = load_config_file(config_default=DEFAULT_CONFIG) + # Overwrite default config and config.py with environment variables + for key in conf: + value_setting = load_env_variable(key) + if value_setting is not None: + conf[key] = value_setting + return conf + + +def load_env_variable(config_environvar): + """Returns config from environment variable""" + prefix = "NBZX_" + config_environvar = prefix + config_environvar.upper() + if config_environvar in environ: + return environ[config_environvar] + return None + + +def load_config_file(config_default, config_file="config.py"): + """Returns config from config.py file""" + # Check if config.py exists and load it + # If it does not exist, return the default config + config_path = Path(config_file) + if config_path.exists(): + dconf = config_default.copy() + # Dynamically import the config module + spec = util.spec_from_file_location("config", config_path) + config_module = util.module_from_spec(spec) + spec.loader.exec_module(config_module) + # Update DEFAULT_CONFIG with variables from the config module + for key in dconf: + if hasattr(config_module, key): + dconf[key] = getattr(config_module, key) + return dconf + logger.warning( + "Config file %s not found. Using default config " + "and environment variables.", config_file) + return None diff --git a/modules/device.py b/modules/device.py index b8d038d..971a5b3 100644 --- a/modules/device.py +++ b/modules/device.py @@ -2,9 +2,9 @@ """ Device specific handeling for NetBox to Zabbix """ + from copy import deepcopy from logging import getLogger -from os import sys from re import search from zabbix_utils import APIRequestError @@ -21,31 +21,9 @@ from modules.interface import ZabbixInterface from modules.tags import ZabbixTags from modules.tools import field_mapper, remove_duplicates from modules.usermacros import ZabbixUsermacros +from modules.config import load_config -try: - from config import ( - device_cf, - device_inventory_map, - device_tag_map, - device_usermacro_map, - inventory_mode, - inventory_sync, - tag_lower, - tag_name, - tag_sync, - tag_value, - template_cf, - traverse_regions, - traverse_site_groups, - usermacro_sync, - ) -except ModuleNotFoundError: - print( - "Configuration file config.py not found in main directory." - "Please create the file or rename the config.py.example file to config.py." - ) - sys.exit(0) - +config = load_config() class PhysicalDevice: # pylint: disable=too-many-instance-attributes, too-many-arguments, too-many-positional-arguments @@ -90,15 +68,15 @@ class PhysicalDevice: def _inventory_map(self): """Use device inventory maps""" - return device_inventory_map + return config["device_inventory_map"] def _usermacro_map(self): """Use device inventory maps""" - return device_usermacro_map + return config["device_usermacro_map"] def _tag_map(self): """Use device host tag maps""" - return device_tag_map + return config["device_tag_map"] def _setBasics(self): """ @@ -114,10 +92,10 @@ class PhysicalDevice: raise SyncInventoryError(e) # Check if device has custom field for ZBX ID - if device_cf in self.nb.custom_fields: - self.zabbix_id = self.nb.custom_fields[device_cf] + if config["device_cf"] in self.nb.custom_fields: + self.zabbix_id = self.nb.custom_fields[config["device_cf"]] else: - e = f"Host {self.name}: Custom field {device_cf} not present" + e = f'Host {self.name}: Custom field {config["device_cf"]} not present' self.logger.warning(e) raise SyncInventoryError(e) @@ -146,8 +124,8 @@ class PhysicalDevice: self.nb, self.nb_api_version, logger=self.logger, - nested_sitegroup_flag=traverse_site_groups, - nested_region_flag=traverse_regions, + nested_sitegroup_flag=config['traverse_site_groups'], + nested_region_flag=config['traverse_regions'], nb_groups=nb_site_groups, nb_regions=nb_regions, ) @@ -183,18 +161,20 @@ 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 template_cf in device_type_cfs: + if config["template_cf"] in device_type_cfs: # Set value to template - return [device_type_cfs[template_cf]] + return [device_type_cfs[config["template_cf"]]] # Custom field not found, return error e = ( - f"Custom field {template_cf} not " + f"Custom field {config['template_cf']} not " f"found for {self.nb.device_type.manufacturer.name}" f" - {self.nb.device_type.display}." ) self.logger.warning(e) raise TemplateError(e) + + def get_templates_context(self): """Get Zabbix templates from the device context""" if "zabbix" not in self.config_context: @@ -217,25 +197,24 @@ class PhysicalDevice: def set_inventory(self, nbdevice): """Set host inventory""" # Set inventory mode. Default is disabled (see class init function). - if inventory_mode == "disabled": - if inventory_sync: - self.logger.error( - f"Host {self.name}: Unable to map NetBox inventory to Zabbix. " - "Inventory sync is enabled in config but inventory mode is disabled." - ) + if config["inventory_mode"] == "disabled": + if config["inventory_sync"]: + self.logger.error(f"Host {self.name}: Unable to map NetBox inventory to Zabbix. " + "Inventory sync is enabled in " + "config but inventory mode is disabled.") return True - if inventory_mode == "manual": + if config["inventory_mode"] == "manual": self.inventory_mode = 0 - elif inventory_mode == "automatic": + elif config["inventory_mode"] == "automatic": self.inventory_mode = 1 else: self.logger.error( f"Host {self.name}: Specified value for inventory mode in" - f" config is not valid. Got value {inventory_mode}" + f" config is not valid. Got value {config['inventory_mode']}" ) return False self.inventory = {} - if inventory_sync and self.inventory_mode in [0, 1]: + if config["inventory_sync"] and self.inventory_mode in [0, 1]: self.logger.debug(f"Host {self.name}: Starting inventory mapper") self.inventory = field_mapper( self.name, self._inventory_map(), nbdevice, self.logger @@ -371,7 +350,7 @@ class PhysicalDevice: def _zeroize_cf(self): """Sets the hostID custom field in NetBox to zero, effectively destroying the link""" - self.nb.custom_fields[device_cf] = None + self.nb.custom_fields[config["device_cf"]] = None self.nb.save() def _zabbixHostnameExists(self): @@ -415,7 +394,7 @@ class PhysicalDevice: macros = ZabbixUsermacros( self.nb, self._usermacro_map(), - usermacro_sync, + config['usermacro_sync'], logger=self.logger, host=self.name, ) @@ -432,10 +411,10 @@ class PhysicalDevice: tags = ZabbixTags( self.nb, self._tag_map(), - tag_sync, - tag_lower, - tag_name=tag_name, - tag_value=tag_value, + config['tag_sync'], + config['tag_lower'], + tag_name=config['tag_name'], + tag_value=config['tag_value'], logger=self.logger, host=self.name, ) @@ -453,7 +432,7 @@ class PhysicalDevice: input: List of all proxies and proxy groups in standardized format """ # check if the key Zabbix is defined in the config context - if not "zabbix" in self.nb.config_context: + if "zabbix" not in self.nb.config_context: return False if ( "proxy" in self.nb.config_context["zabbix"] @@ -550,7 +529,7 @@ class PhysicalDevice: self.logger.error(msg) raise SyncExternalError(msg) from e # Set NetBox custom field to hostID value. - self.nb.custom_fields[device_cf] = int(self.zabbix_id) + self.nb.custom_fields[config["device_cf"]] = int(self.zabbix_id) self.nb.save() msg = f"Host {self.name}: Created host in Zabbix." self.logger.info(msg) @@ -724,10 +703,8 @@ class PhysicalDevice: # Check if a proxy has been defined if self.zbxproxy: # Check if proxy or proxy group is defined - if ( - self.zbxproxy["idtype"] in host - and host[self.zbxproxy["idtype"]] == self.zbxproxy["id"] - ): + if (self.zbxproxy["idtype"] in host and + host[self.zbxproxy["idtype"]] == self.zbxproxy["id"]): self.logger.debug(f"Host {self.name}: proxy in-sync.") # Backwards compatibility for Zabbix <= 6 elif "proxy_hostid" in host and host["proxy_hostid"] == self.zbxproxy["id"]: @@ -785,7 +762,7 @@ class PhysicalDevice: else: self.logger.warning(f"Host {self.name}: inventory_mode OUT of sync.") self.updateZabbixHost(inventory_mode=str(self.inventory_mode)) - if inventory_sync and self.inventory_mode in [0, 1]: + if config["inventory_sync"] and self.inventory_mode in [0, 1]: # Check host inventory mapping if host["inventory"] == self.inventory: self.logger.debug(f"Host {self.name}: inventory in-sync.") @@ -794,10 +771,10 @@ class PhysicalDevice: self.updateZabbixHost(inventory=self.inventory) # Check host usermacros - if usermacro_sync: + if config['usermacro_sync']: macros_filtered = [] # Do not re-sync secret usermacros unless sync is set to 'full' - if str(usermacro_sync).lower() != "full": + if str(config['usermacro_sync']).lower() != "full": for m in deepcopy(self.usermacros): if m["type"] == str(1): # Remove the value as the api doesn't return it @@ -811,7 +788,7 @@ class PhysicalDevice: self.updateZabbixHost(macros=self.usermacros) # Check host usermacros - if tag_sync: + if config['tag_sync']: if remove_duplicates(host["tags"], sortkey="tag") == self.tags: self.logger.debug(f"Host {self.name}: tags in-sync.") else: diff --git a/modules/virtual_machine.py b/modules/virtual_machine.py index 34e3394..7ee6659 100644 --- a/modules/virtual_machine.py +++ b/modules/virtual_machine.py @@ -1,27 +1,12 @@ # pylint: disable=duplicate-code """Module that hosts all functions for virtual machine processing""" - -from os import sys - from modules.device import PhysicalDevice from modules.exceptions import InterfaceConfigError, SyncInventoryError, TemplateError from modules.hostgroups import Hostgroup from modules.interface import ZabbixInterface - -try: - from config import ( - traverse_regions, - traverse_site_groups, - vm_inventory_map, - vm_tag_map, - vm_usermacro_map, - ) -except ModuleNotFoundError: - print( - "Configuration file config.py not found in main directory." - "Please create the file or rename the config.py.example file to config.py." - ) - sys.exit(0) +from modules.config import load_config +# Load config +config = load_config() class VirtualMachine(PhysicalDevice): @@ -34,15 +19,15 @@ class VirtualMachine(PhysicalDevice): def _inventory_map(self): """use VM inventory maps""" - return vm_inventory_map + return config["vm_inventory_map"] def _usermacro_map(self): """use VM usermacro maps""" - return vm_usermacro_map + return config["vm_usermacro_map"] def _tag_map(self): """use VM tag maps""" - return 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""" @@ -52,8 +37,8 @@ class VirtualMachine(PhysicalDevice): self.nb, self.nb_api_version, logger=self.logger, - nested_sitegroup_flag=traverse_site_groups, - nested_region_flag=traverse_regions, + nested_sitegroup_flag=config["traverse_site_groups"], + nested_region_flag=config["traverse_regions"], nb_groups=nb_site_groups, nb_regions=nb_regions, ) diff --git a/netbox_zabbix_sync.py b/netbox_zabbix_sync.py index 79d8a20..ec14b4e 100755 --- a/netbox_zabbix_sync.py +++ b/netbox_zabbix_sync.py @@ -11,35 +11,14 @@ from pynetbox import api from pynetbox.core.query import RequestError as NBRequestError from requests.exceptions import ConnectionError as RequestsConnectionError from zabbix_utils import APIRequestError, ProcessingError, ZabbixAPI - +from modules.config import load_config from modules.device import PhysicalDevice from modules.exceptions import EnvironmentVarError, HostgroupError, SyncError from modules.logging import get_logger, set_log_levels, setup_logger from modules.tools import convert_recordset, proxy_prepper from modules.virtual_machine import VirtualMachine -try: - from config import ( - clustering, - create_hostgroups, - create_journal, - full_proxy_sync, - hostgroup_format, - nb_device_filter, - nb_vm_filter, - sync_vms, - templates_config_context, - templates_config_context_overrule, - vm_hostgroup_format, - zabbix_device_disable, - zabbix_device_removal, - ) -except ModuleNotFoundError: - print( - "Configuration file config.py not found in main directory." - "Please create the file or rename the config.py.example file to config.py." - ) - sys.exit(1) +config = load_config() setup_logger() @@ -85,7 +64,7 @@ def main(arguments): # Set NetBox API netbox = api(netbox_host, token=netbox_token, threading=True) # Check if the provided Hostgroup layout is valid - hg_objects = hostgroup_format.split("/") + hg_objects = config["hostgroup_format"].split("/") allowed_objects = [ "location", "role", @@ -145,10 +124,11 @@ def main(arguments): else: proxy_name = "name" # Get all Zabbix and NetBox data - netbox_devices = list(netbox.dcim.devices.filter(**nb_device_filter)) + netbox_devices = list(netbox.dcim.devices.filter(**config["nb_device_filter"])) netbox_vms = [] - if sync_vms: - netbox_vms = list(netbox.virtualization.virtual_machines.filter(**nb_vm_filter)) + if config["sync_vms"]: + netbox_vms = list( + netbox.virtualization.virtual_machines.filter(**config["nb_vm_filter"])) netbox_site_groups = convert_recordset((netbox.dcim.site_groups.all())) netbox_regions = convert_recordset(netbox.dcim.regions.all()) netbox_journals = netbox.extras.journal_entries @@ -172,15 +152,15 @@ def main(arguments): # Go through all NetBox devices for nb_vm in netbox_vms: try: - vm = VirtualMachine( - nb_vm, zabbix, netbox_journals, nb_version, create_journal, logger - ) - logger.debug(f"Host {vm.name}: Started operations on VM.") + vm = VirtualMachine(nb_vm, zabbix, netbox_journals, nb_version, + config["create_journal"], logger) + logger.debug(f"Host {vm.name}: started operations on VM.") vm.set_vm_template() # Check if a valid template has been found for this VM. if not vm.zbx_template_names: continue - vm.set_hostgroup(vm_hostgroup_format, netbox_site_groups, netbox_regions) + vm.set_hostgroup(config["vm_hostgroup_format"], + netbox_site_groups, netbox_regions) # Check if a valid hostgroup has been found for this VM. if not vm.hostgroup: continue @@ -188,7 +168,7 @@ def main(arguments): vm.set_usermacros() vm.set_tags() # Checks if device is in cleanup state - if vm.status in zabbix_device_removal: + if vm.status in config["zabbix_device_removal"]: if vm.zabbix_id: # Delete device from Zabbix # and remove hostID from NetBox. @@ -203,7 +183,7 @@ def main(arguments): ) continue # Check if the VM is in the disabled state - if vm.status in zabbix_device_disable: + if vm.status in config["zabbix_device_disable"]: vm.zabbix_state = 1 # Check if VM is already in Zabbix if vm.zabbix_id: @@ -211,12 +191,12 @@ def main(arguments): zabbix_groups, zabbix_templates, zabbix_proxy_list, - full_proxy_sync, - create_hostgroups, + config["full_proxy_sync"], + config["create_hostgroups"], ) continue # Add hostgroup is config is set - if create_hostgroups: + if config["create_hostgroups"]: # Create new hostgroup. Potentially multiple groups if nested hostgroups = vm.createZabbixHostgroup(zabbix_groups) # go through all newly created hostgroups @@ -231,17 +211,16 @@ def main(arguments): for nb_device in netbox_devices: try: # Set device instance set data such as hostgroup and template information. - device = PhysicalDevice( - nb_device, zabbix, netbox_journals, nb_version, create_journal, logger - ) + device = PhysicalDevice(nb_device, zabbix, netbox_journals, nb_version, + config["create_journal"], logger) logger.debug(f"Host {device.name}: started operations on device.") - device.set_template( - templates_config_context, templates_config_context_overrule - ) + device.set_template(config["templates_config_context"], + config["templates_config_context_overrule"]) # Check if a valid template has been found for this VM. if not device.zbx_template_names: continue - device.set_hostgroup(hostgroup_format, netbox_site_groups, netbox_regions) + device.set_hostgroup( + config["hostgroup_format"], netbox_site_groups, netbox_regions) # Check if a valid hostgroup has been found for this VM. if not device.hostgroup: continue @@ -250,7 +229,7 @@ def main(arguments): device.set_tags() # Checks if device is part of cluster. # Requires clustering variable - if device.isCluster() and clustering: + if device.isCluster() and config["clustering"]: # Check if device is primary or secondary if device.promoteMasterDevice(): e = f"Device {device.name}: is " f"part of cluster and primary." @@ -265,7 +244,7 @@ def main(arguments): logger.info(e) continue # Checks if device is in cleanup state - if device.status in zabbix_device_removal: + if device.status in config["zabbix_device_removal"]: if device.zabbix_id: # Delete device from Zabbix # and remove hostID from NetBox. @@ -280,7 +259,7 @@ def main(arguments): ) continue # Check if the device is in the disabled state - if device.status in zabbix_device_disable: + if device.status in config["zabbix_device_disable"]: device.zabbix_state = 1 # Check if device is already in Zabbix if device.zabbix_id: @@ -288,12 +267,12 @@ def main(arguments): zabbix_groups, zabbix_templates, zabbix_proxy_list, - full_proxy_sync, - create_hostgroups, + config["full_proxy_sync"], + config["create_hostgroups"], ) continue # Add hostgroup is config is set - if create_hostgroups: + if config["create_hostgroups"]: # Create new hostgroup. Potentially multiple groups if nested hostgroups = device.createZabbixHostgroup(zabbix_groups) # go through all newly created hostgroups diff --git a/requirements.txt b/requirements.txt index 33f4b90..8da5ce5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,2 @@ pynetbox -zabbix-utils==2.0.1 +zabbix-utils==2.0.1 \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_configuration_parsing.py b/tests/test_configuration_parsing.py new file mode 100644 index 0000000..23438b4 --- /dev/null +++ b/tests/test_configuration_parsing.py @@ -0,0 +1,140 @@ +"""Tests for configuration parsing in the modules.config module.""" +from unittest.mock import patch, MagicMock +import os +from modules.config import load_config, DEFAULT_CONFIG, load_config_file, load_env_variable + + +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): + config = load_config() + assert config == DEFAULT_CONFIG + assert config["templates_config_context"] is False + assert config["create_hostgroups"] is True + + +def test_load_config_file(): + """Test that load_config properly loads values from config file""" + mock_config = DEFAULT_CONFIG.copy() + 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): + config = load_config() + assert config["templates_config_context"] is True + assert config["sync_vms"] is True + # Unchanged values should remain as defaults + assert config["create_journal"] is False + + +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": + return True + if key == "create_journal": + 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): + config = load_config() + assert config["sync_vms"] is True + assert config["create_journal"] is True + # Unchanged values should remain as defaults + assert config["templates_config_context"] is False + + +def test_env_vars_override_config_file(): + """Test that environment variables override values from config file""" + mock_config = DEFAULT_CONFIG.copy() + mock_config["templates_config_context"] = True + mock_config["sync_vms"] = False + + # Mock env variable that will override the config file value + def mock_load_env(key): + if key == "sync_vms": + 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): + config = load_config() + # This should be overridden by the env var + assert config["sync_vms"] is True + # This should remain from the config file + assert config["templates_config_context"] is True + + +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: + # Setup the mock module with attributes + mock_module = MagicMock() + mock_module.templates_config_context = True + mock_module.sync_vms = True + + # Setup the mock spec + mock_spec_instance = MagicMock() + mock_spec.return_value = mock_spec_instance + 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): + config = load_config_file(DEFAULT_CONFIG.copy()) + assert config["templates_config_context"] is True + assert config["sync_vms"] is True + + +def test_load_config_file_not_found(): + """Test load_config_file when the config file doesn't exist""" + # Instead of trying to assert on the logger call, we'll just check the return value + # and verify the function works as expected in this case + with patch('pathlib.Path.exists', return_value=False): + result = load_config_file(DEFAULT_CONFIG.copy()) + assert result is None + + +def test_load_env_variable_function(): + """Test the load_env_variable function directly""" + # Create a real environment variable for testing with correct prefix and uppercase + test_var = "NBZX_TEMPLATES_CONFIG_CONTEXT" + original_env = os.environ.get(test_var, None) + try: + # Set the environment variable with the proper prefix and case + os.environ[test_var] = "True" + + # Test that it's properly read (using lowercase in the function call) + value = load_env_variable("templates_config_context") + assert value == "True" + + # Test when the environment variable doesn't exist + value = load_env_variable("nonexistent_variable") + assert value is None + finally: + # Clean up - restore original environment + if original_env is not None: + os.environ[test_var] = original_env + else: + os.environ.pop(test_var, None) + + +def test_load_config_file_exception_handling(): + """Test that load_config_file handles exceptions gracefully""" + # This test requires modifying the load_config_file function to handle exceptions + # For now, we're just checking that an exception is raised + with patch('pathlib.Path.exists', return_value=True), \ + patch('importlib.util.spec_from_file_location', side_effect=Exception("Import error")): + # Since the current implementation doesn't handle exceptions, we should + # expect an exception to be raised + try: + load_config_file(DEFAULT_CONFIG.copy()) + assert False, "An exception should have been raised" + except Exception: # pylint: disable=broad-except + # This is expected + pass diff --git a/tests/test_device_deletion.py b/tests/test_device_deletion.py new file mode 100644 index 0000000..392ba1a --- /dev/null +++ b/tests/test_device_deletion.py @@ -0,0 +1,166 @@ +"""Tests for device deletion functionality in the PhysicalDevice class.""" +import unittest +from unittest.mock import MagicMock, patch +from zabbix_utils import APIRequestError +from modules.device import PhysicalDevice +from modules.exceptions import SyncExternalError + + +class TestDeviceDeletion(unittest.TestCase): + """Test class for device deletion functionality.""" + + def setUp(self): + """Set up test fixtures.""" + # Create mock NetBox device + self.mock_nb_device = MagicMock() + self.mock_nb_device.id = 123 + self.mock_nb_device.name = "test-device" + self.mock_nb_device.status.label = "Decommissioning" + self.mock_nb_device.custom_fields = {"zabbix_hostid": "456"} + self.mock_nb_device.config_context = {} + + # Set up a primary IP + primary_ip = MagicMock() + primary_ip.address = "192.168.1.1/24" + self.mock_nb_device.primary_ip = primary_ip + + # Create mock Zabbix API + self.mock_zabbix = MagicMock() + self.mock_zabbix.version = "6.0" + + # Set up mock host.get response + self.mock_zabbix.host.get.return_value = [{"hostid": "456"}] + + # Mock NetBox journal class + self.mock_nb_journal = MagicMock() + + # Create logger mock + self.mock_logger = MagicMock() + + # Create PhysicalDevice instance with mocks + 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 + ) + + def test_cleanup_successful_deletion(self): + """Test successful device deletion from Zabbix.""" + # Setup + self.mock_zabbix.host.get.return_value = [{"hostid": "456"}] + self.mock_zabbix.host.delete.return_value = {"hostids": ["456"]} + + # Execute + 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_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.") + + def test_cleanup_device_already_deleted(self): + """Test cleanup when device is already deleted from Zabbix.""" + # Setup + self.mock_zabbix.host.get.return_value = [] # Empty list means host not found + + # Execute + self.device.cleanup() + + # Verify + 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.") + + def test_cleanup_api_error(self): + """Test cleanup when Zabbix API returns an error.""" + # Setup + self.mock_zabbix.host.get.return_value = [{"hostid": "456"}] + self.mock_zabbix.host.delete.side_effect = APIRequestError("API Error") + + # Execute and verify + with self.assertRaises(SyncExternalError): + 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_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 + + # Verify + self.assertIsNone(self.mock_nb_device.custom_fields["zabbix_hostid"]) + self.mock_nb_device.save.assert_called_once() + + def test_create_journal_entry(self): + """Test create_journal_entry method.""" + # Setup + test_message = "Test journal entry" + + # Execute + result = self.device.create_journal_entry("info", test_message) + + # Verify + self.assertTrue(result) + self.mock_nb_journal.create.assert_called_once() + journal_entry = self.mock_nb_journal.create.call_args[0][0] + self.assertEqual(journal_entry["assigned_object_type"], "dcim.device") + self.assertEqual(journal_entry["assigned_object_id"], 123) + self.assertEqual(journal_entry["kind"], "info") + self.assertEqual(journal_entry["comments"], test_message) + + def test_create_journal_entry_invalid_severity(self): + """Test create_journal_entry with invalid severity.""" + # Execute + result = self.device.create_journal_entry("invalid", "Test message") + + # Verify + self.assertFalse(result) + self.mock_nb_journal.create.assert_not_called() + self.mock_logger.warning.assert_called() + + 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"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + journal=False, # Disable journaling + logger=self.mock_logger + ) + + # Execute + result = device.create_journal_entry("info", "Test message") + + # Verify + self.assertFalse(result) + self.mock_nb_journal.create.assert_not_called() + + def test_cleanup_updates_journal(self): + """Test that cleanup method creates a journal entry.""" + # Setup + self.mock_zabbix.host.get.return_value = [{"hostid": "456"}] + + # Execute + 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") diff --git a/tests/test_interface.py b/tests/test_interface.py new file mode 100644 index 0000000..ff55218 --- /dev/null +++ b/tests/test_interface.py @@ -0,0 +1,247 @@ +"""Tests for the ZabbixInterface class in the interface module.""" +import unittest +from modules.interface import ZabbixInterface +from modules.exceptions import InterfaceConfigError + + +class TestZabbixInterface(unittest.TestCase): + """Test class for ZabbixInterface functionality.""" + + def setUp(self): + """Set up test fixtures.""" + self.test_ip = "192.168.1.1" + self.empty_context = {} + self.default_interface = ZabbixInterface(self.empty_context, self.test_ip) + + # Create some test contexts for different scenarios + self.snmpv2_context = { + "zabbix": { + "interface_type": 2, + "interface_port": "161", + "snmp": { + "version": 2, + "community": "public", + "bulk": 1 + } + } + } + + self.snmpv3_context = { + "zabbix": { + "interface_type": 2, + "snmp": { + "version": 3, + "securityname": "snmpuser", + "securitylevel": "authPriv", + "authprotocol": "SHA", + "authpassphrase": "authpass123", + "privprotocol": "AES", + "privpassphrase": "privpass123", + "contextname": "context1" + } + } + } + + self.agent_context = { + "zabbix": { + "interface_type": 1, + "interface_port": "10050" + } + } + + def test_init(self): + """Test initialization of ZabbixInterface.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + + # Check basic properties + self.assertEqual(interface.ip, self.test_ip) + self.assertEqual(interface.context, self.empty_context) + self.assertEqual(interface.interface["ip"], self.test_ip) + self.assertEqual(interface.interface["main"], "1") + self.assertEqual(interface.interface["useip"], "1") + self.assertEqual(interface.interface["dns"], "") + + def test_get_context_empty(self): + """Test get_context with empty context.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + result = interface.get_context() + self.assertFalse(result) + + def test_get_context_with_interface_type(self): + """Test get_context with interface_type but no port.""" + context = {"zabbix": {"interface_type": 2}} + interface = ZabbixInterface(context, self.test_ip) + + # Should set type and default port + result = interface.get_context() + self.assertTrue(result) + self.assertEqual(interface.interface["type"], 2) + self.assertEqual(interface.interface["port"], "161") # Default port for SNMP + + def test_get_context_with_interface_type_and_port(self): + """Test get_context with both interface_type and port.""" + context = {"zabbix": {"interface_type": 1, "interface_port": "12345"}} + interface = ZabbixInterface(context, self.test_ip) + + # Should set type and specified port + result = interface.get_context() + self.assertTrue(result) + self.assertEqual(interface.interface["type"], 1) + self.assertEqual(interface.interface["port"], "12345") + + def test_set_default_port(self): + """Test _set_default_port for different interface types.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + + # Test for agent type (1) + interface.interface["type"] = 1 + 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 + self.assertEqual(interface.interface["port"], "161") + + # Test for IPMI type (3) + interface.interface["type"] = 3 + 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 + self.assertEqual(interface.interface["port"], "12345") + + # Test for unsupported type + interface.interface["type"] = 99 + result = interface._set_default_port() # pylint: disable=protected-access + self.assertFalse(result) + + def test_set_snmp_v2(self): + """Test set_snmp with SNMPv2 configuration.""" + interface = ZabbixInterface(self.snmpv2_context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp + 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") + + def test_set_snmp_v3(self): + """Test set_snmp with SNMPv3 configuration.""" + interface = ZabbixInterface(self.snmpv3_context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp + 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") + + def test_set_snmp_no_snmp_config(self): + """Test set_snmp with missing SNMP configuration.""" + # Create context with interface type but no SNMP config + context = {"zabbix": {"interface_type": 2}} + interface = ZabbixInterface(context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp - should raise exception + with self.assertRaises(InterfaceConfigError): + interface.set_snmp() + + def test_set_snmp_unsupported_version(self): + """Test set_snmp with unsupported SNMP version.""" + # Create context with invalid SNMP version + context = { + "zabbix": { + "interface_type": 2, + "snmp": { + "version": 4 # Invalid version + } + } + } + interface = ZabbixInterface(context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp - should raise exception + with self.assertRaises(InterfaceConfigError): + interface.set_snmp() + + def test_set_snmp_no_version(self): + """Test set_snmp with missing SNMP version.""" + # Create context without SNMP version + context = { + "zabbix": { + "interface_type": 2, + "snmp": { + "community": "public" # No version specified + } + } + } + interface = ZabbixInterface(context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp - should raise exception + with self.assertRaises(InterfaceConfigError): + interface.set_snmp() + + def test_set_snmp_non_snmp_interface(self): + """Test set_snmp with non-SNMP interface type.""" + interface = ZabbixInterface(self.agent_context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp - should raise exception + with self.assertRaises(InterfaceConfigError): + interface.set_snmp() + + def test_set_default_snmp(self): + """Test set_default_snmp method.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + interface.set_default_snmp() + + # 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") + + def test_set_default_agent(self): + """Test set_default_agent method.""" + interface = ZabbixInterface(self.empty_context, self.test_ip) + interface.set_default_agent() + + # Check interface properties + self.assertEqual(interface.interface["type"], "1") + self.assertEqual(interface.interface["port"], "10050") + + 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 + } + } + } + interface = ZabbixInterface(context, self.test_ip) + interface.get_context() # Set the interface type + + # Call set_snmp + interface.set_snmp() + + # Should use default community string + self.assertEqual(interface.interface["details"]["community"], "{$SNMP_COMMUNITY}") diff --git a/tests/test_physical_device.py b/tests/test_physical_device.py new file mode 100644 index 0000000..1b79ad8 --- /dev/null +++ b/tests/test_physical_device.py @@ -0,0 +1,429 @@ +"""Tests for the PhysicalDevice class in the device module.""" +import unittest +from unittest.mock import MagicMock, patch +from modules.device import PhysicalDevice +from modules.exceptions import TemplateError, SyncInventoryError + + +class TestPhysicalDevice(unittest.TestCase): + """Test class for PhysicalDevice functionality.""" + + def setUp(self): + """Set up test fixtures.""" + # Create mock NetBox device + self.mock_nb_device = MagicMock() + self.mock_nb_device.id = 123 + self.mock_nb_device.name = "test-device" + self.mock_nb_device.status.label = "Active" + self.mock_nb_device.custom_fields = {"zabbix_hostid": None} + self.mock_nb_device.config_context = {} + + # Set up a primary IP + primary_ip = MagicMock() + primary_ip.address = "192.168.1.1/24" + self.mock_nb_device.primary_ip = primary_ip + + # Create mock Zabbix API + self.mock_zabbix = MagicMock() + self.mock_zabbix.version = "6.0" + + # Mock NetBox journal class + self.mock_nb_journal = MagicMock() + + # Create logger mock + 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": {} + }): + 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.""" + # Check that basic properties are set correctly + self.assertEqual(self.device.name, "test-device") + self.assertEqual(self.device.id, 123) + self.assertEqual(self.device.status, "Active") + self.assertEqual(self.device.ip, "192.168.1.1") + self.assertEqual(self.device.cidr, "192.168.1.1/24") + + def test_init_no_primary_ip(self): + """Test initialization when device has no primary IP.""" + # Set primary_ip to None + self.mock_nb_device.primary_ip = None + + # Creating device should raise SyncInventoryError + with patch('modules.device.config', {"device_cf": "zabbix_hostid"}): + with self.assertRaises(SyncInventoryError): + PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + def test_set_basics_with_special_characters(self): + """Test _setBasics when device name contains special characters.""" + # Set name with special characters that + # will actually trigger the special character detection + 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"}): + # Make the search function return True to simulate special characters + mock_search.return_value = True + + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # With the mocked search function, the name should be changed to NETBOX_ID format + self.assertEqual(device.name, f"NETBOX_ID{self.mock_nb_device.id}") + # And visible_name should be set to the original name + self.assertEqual(device.visible_name, "test-devïce") + # use_visible_name flag should be set + self.assertTrue(device.use_visible_name) + + def test_get_templates_context(self): + """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"] + } + } + + # Create device with the updated mock + 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 + ) + + # Test that templates are returned correctly + templates = device.get_templates_context() + self.assertEqual(templates, ["Template1", "Template2"]) + + 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" + } + } + + # Create device with the updated mock + 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 + ) + + # Test that template is wrapped in a list + templates = device.get_templates_context() + self.assertEqual(templates, ["Template1"]) + + def test_get_templates_context_no_zabbix_key(self): + """Test get_templates_context when zabbix key is missing.""" + # Set up config_context without zabbix key + self.mock_nb_device.config_context = {} + + # Create device with the updated mock + 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 + ) + + # Test that TemplateError is raised + with self.assertRaises(TemplateError): + device.get_templates_context() + + def test_get_templates_context_no_templates_key(self): + """Test get_templates_context when templates key is missing.""" + # Set up config_context without templates key + self.mock_nb_device.config_context = {"zabbix": {}} + + # Create device with the updated mock + 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 + ) + + # Test that TemplateError is raised + with self.assertRaises(TemplateError): + device.get_templates_context() + + 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"] + } + } + + # 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"}): + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Call set_template with prefer_config_context=True + result = device.set_template(prefer_config_context=True, overrule_custom=False) + + # Check result and template names + self.assertTrue(result) + self.assertEqual(device.zbx_template_names, ["Template1"]) + + def test_set_inventory_disabled_mode(self): + """Test set_inventory with inventory_mode=disabled.""" + # Configure with disabled inventory mode + config_patch = { + "device_cf": "zabbix_hostid", + "inventory_mode": "disabled", + "inventory_sync": False + } + + 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 + ) + + # Call set_inventory with the config patch still active + with patch('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) + + def test_set_inventory_manual_mode(self): + """Test set_inventory with inventory_mode=manual.""" + # Configure with manual inventory mode + config_patch = { + "device_cf": "zabbix_hostid", + "inventory_mode": "manual", + "inventory_sync": False + } + + 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 + ) + + # Call set_inventory with the config patch still active + with patch('modules.device.config', config_patch): + result = device.set_inventory({}) + + # 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.""" + # Configure with automatic inventory mode + config_patch = { + "device_cf": "zabbix_hostid", + "inventory_mode": "automatic", + "inventory_sync": False + } + + 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 + ) + + # Call set_inventory with the config patch still active + with patch('modules.device.config', config_patch): + result = device.set_inventory({}) + + # 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.""" + # Configure with inventory sync enabled + config_patch = { + "device_cf": "zabbix_hostid", + "inventory_mode": "manual", + "inventory_sync": True, + "device_inventory_map": { + "name": "name", + "serial": "serialno_a" + } + } + + 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 + ) + + # Create a mock device with the required attributes + mock_device_data = { + "name": "test-device", + "serial": "ABC123" + } + + # Call set_inventory with the config patch still active + 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" + }) + + def test_iscluster_true(self): + """Test isCluster when device is part of a cluster.""" + # Set up virtual_chassis + self.mock_nb_device.virtual_chassis = MagicMock() + + # Create device with the updated mock + 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 + ) + + # Check isCluster result + self.assertTrue(device.isCluster()) + + def test_is_cluster_false(self): + """Test isCluster when device is not part of a cluster.""" + # Set virtual_chassis to None + self.mock_nb_device.virtual_chassis = None + + # Create device with the updated mock + 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 + ) + + # Check isCluster result + self.assertFalse(device.isCluster()) + + + 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_vc.master = mock_master + self.mock_nb_device.virtual_chassis = mock_vc + + # Create device with the updated mock + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Call promoteMasterDevice and check the result + result = device.promoteMasterDevice() + + # Should return True for primary device + self.assertTrue(result) + # 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_vc.master = mock_master + self.mock_nb_device.virtual_chassis = mock_vc + + # Create device with the updated mock + device = PhysicalDevice( + self.mock_nb_device, + self.mock_zabbix, + self.mock_nb_journal, + "3.0", + logger=self.mock_logger + ) + + # Call promoteMasterDevice and check the result + result = device.promoteMasterDevice() + + # Should return False for secondary device + self.assertFalse(result) + # Device name should not be modified + self.assertEqual(device.name, "test-device")