diff --git a/.github/workflows/publish-package.yml b/.github/workflows/publish-package.yml new file mode 100644 index 0000000..008a9a0 --- /dev/null +++ b/.github/workflows/publish-package.yml @@ -0,0 +1,33 @@ +name: Upload Python Package to PyPI when a Release is Created + +permissions: + contents: read + +on: + release: + types: [published] + +jobs: + pypi-publish: + name: Publish release to PyPI + runs-on: ubuntu-latest + environment: + name: release + url: https://pypi.org/p/netbox-zabbix-sync + permissions: + id-token: write + steps: + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 + - name: Set up Python + uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 + with: + python-version: "3.x" + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install setuptools wheel build + - name: Build package + run: | + python -m build + - name: Publish package distributions to PyPI + uses: pypa/gh-action-pypi-publish@76f52bc884231f62b9a034ebfe128415bbaabdfc diff --git a/.gitignore b/.gitignore index 0cf4953..40b2d6a 100644 --- a/.gitignore +++ b/.gitignore @@ -1,7 +1,7 @@ *.log .venv .env -config.py +/config.py Pipfile Pipfile.lock # Byte-compiled / optimized / DLL files @@ -9,4 +9,10 @@ __pycache__/ *.py[cod] .vscode .flake -.coverage \ No newline at end of file +.coverage + +*.egg-info +dist +build + +netbox_zabbix_sync/_version.py \ No newline at end of file diff --git a/config.py.example b/config.py.example index acc1629..5dbe28e 100644 --- a/config.py.example +++ b/config.py.example @@ -12,6 +12,15 @@ templates_config_context_overrule = False template_cf = "zabbix_template" device_cf = "zabbix_hostid" +# Zabbix host description +# The following options are available for the description of all created hosts in Zabbix +# static: Uses the default static string "Host added by NetBox sync script." +# dynamic: "Uses a predefined dynamic string which resolves the owner of an object and datetime. Recommended for users who use Netbox 4.5+ +# custom: Use a custom string such as "This host was created by Zabbix-sync on machine MGMT01.internal". It is also posible to resolve dynamic values in this string using {} markers. +description = "static" +# The timedate format which is used for generating the datetime macro when used in the dynamic description type or custom type. +description_dt_format = "%Y-%m-%d %H:%M:%S" + ## Enable clustering of devices with virtual chassis setup clustering = False diff --git a/netbox_zabbix_sync.py b/netbox_zabbix_sync.py index 4a5c880..89d2677 100755 --- a/netbox_zabbix_sync.py +++ b/netbox_zabbix_sync.py @@ -1,329 +1,6 @@ #!/usr/bin/env python3 -"""NetBox to Zabbix sync script.""" - -import argparse -import logging -import ssl -import sys -from os import environ - -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, SyncError -from modules.logging import get_logger, set_log_levels, setup_logger -from modules.tools import convert_recordset, proxy_prepper, verify_hg_format -from modules.virtual_machine import VirtualMachine - -config = load_config() - - -setup_logger() -logger = get_logger() - - -def main(arguments): - """Run the sync process.""" - # set environment variables - if arguments.verbose: - set_log_levels(logging.WARNING, logging.INFO) - if arguments.debug: - set_log_levels(logging.WARNING, logging.DEBUG) - if arguments.debug_all: - set_log_levels(logging.DEBUG, logging.DEBUG) - if arguments.quiet: - set_log_levels(logging.ERROR, logging.ERROR) - - env_vars = ["ZABBIX_HOST", "NETBOX_HOST", "NETBOX_TOKEN"] - if "ZABBIX_TOKEN" in environ: - env_vars.append("ZABBIX_TOKEN") - else: - env_vars.append("ZABBIX_USER") - env_vars.append("ZABBIX_PASS") - for var in env_vars: - if var not in environ: - e = f"Environment variable {var} has not been defined." - logger.error(e) - raise EnvironmentVarError(e) - # Get all virtual environment variables - if "ZABBIX_TOKEN" in env_vars: - zabbix_user = None - zabbix_pass = None - zabbix_token = environ.get("ZABBIX_TOKEN") - else: - zabbix_user = environ.get("ZABBIX_USER") - zabbix_pass = environ.get("ZABBIX_PASS") - zabbix_token = None - zabbix_host = environ.get("ZABBIX_HOST") - netbox_host = environ.get("NETBOX_HOST") - netbox_token = environ.get("NETBOX_TOKEN") - # Set NetBox API - netbox = api(netbox_host, token=netbox_token, threading=True) - # Create API call to get all custom fields which are on the device objects - try: - # Get NetBox version - nb_version = netbox.version - logger.debug("NetBox version is %s.", nb_version) - except RequestsConnectionError: - logger.error( - "Unable to connect to NetBox with URL %s. Please check the URL and status of NetBox.", - netbox_host, - ) - sys.exit(1) - except NBRequestError as e: - logger.error("NetBox error: %s", e) - sys.exit(1) - # Check if the provided Hostgroup layout is valid - device_cfs = [] - vm_cfs = [] - device_cfs = list( - netbox.extras.custom_fields.filter( - type=["text", "object", "select"], content_types="dcim.device" - ) - ) - verify_hg_format( - config["hostgroup_format"], device_cfs=device_cfs, hg_type="dev", logger=logger - ) - if config["sync_vms"]: - vm_cfs = list( - netbox.extras.custom_fields.filter( - type=["text", "object", "select"], - content_types="virtualization.virtualmachine", - ) - ) - verify_hg_format( - config["vm_hostgroup_format"], vm_cfs=vm_cfs, hg_type="vm", logger=logger - ) - # Set Zabbix API - try: - ssl_ctx = ssl.create_default_context() - - # If a custom CA bundle is set for pynetbox (requests), also use it for the Zabbix API - if environ.get("REQUESTS_CA_BUNDLE", None): - ssl_ctx.load_verify_locations(environ["REQUESTS_CA_BUNDLE"]) - - if not zabbix_token: - zabbix = ZabbixAPI( - zabbix_host, user=zabbix_user, password=zabbix_pass, ssl_context=ssl_ctx - ) - else: - zabbix = ZabbixAPI(zabbix_host, token=zabbix_token, ssl_context=ssl_ctx) - zabbix.check_auth() - except (APIRequestError, ProcessingError) as zbx_error: - e = f"Zabbix returned the following error: {zbx_error}." - logger.error(e) - sys.exit(1) - # Set API parameter mapping based on API version - proxy_name = "host" if not str(zabbix.version).startswith("7") else "name" - # Get all Zabbix and NetBox data - netbox_devices = list(netbox.dcim.devices.filter(**config["nb_device_filter"])) - netbox_vms = [] - 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 - zabbix_groups = zabbix.hostgroup.get(output=["groupid", "name"]) # type: ignore[attr-defined] - zabbix_templates = zabbix.template.get(output=["templateid", "name"]) # type: ignore[attr-defined] - zabbix_proxies = zabbix.proxy.get(output=["proxyid", proxy_name]) # type: ignore[attr-defined] - # Set empty list for proxy processing Zabbix <= 6 - zabbix_proxygroups = [] - if str(zabbix.version).startswith("7"): - zabbix_proxygroups = zabbix.proxygroup.get(output=["proxy_groupid", "name"]) # type: ignore[attr-defined] - # Sanitize proxy data - if proxy_name == "host": - for proxy in zabbix_proxies: - proxy["name"] = proxy.pop("host") - # Prepare list of all proxy and proxy_groups - zabbix_proxy_list = proxy_prepper(zabbix_proxies, zabbix_proxygroups) - - # Go through all NetBox devices - for nb_vm in netbox_vms: - try: - vm = VirtualMachine( - nb_vm, - zabbix, - netbox_journals, - nb_version, - config["create_journal"], - logger, - ) - logger.debug("Host %s: Started operations on VM.", vm.name) - 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( - config["vm_hostgroup_format"], netbox_site_groups, netbox_regions - ) - # Check if a valid hostgroup has been found for this VM. - if not vm.hostgroups: - continue - if config["extended_site_properties"] and nb_vm.site: - logger.debug("VM %s: extending site information.", vm.name) - vm.site = convert_recordset(netbox.dcim.sites.filter(id=nb_vm.site.id)) # type: ignore[attr-defined] - vm.set_inventory(nb_vm) - vm.set_usermacros() - vm.set_tags() - # Checks if device is in cleanup state - if vm.status in config["zabbix_device_removal"]: - if vm.zabbix_id: - # Delete device from Zabbix - # and remove hostID from NetBox. - vm.cleanup() - logger.info("VM %s: cleanup complete", vm.name) - continue - # Device has been added to NetBox - # but is not in Activate state - logger.info( - "VM %s: Skipping since this VM is not in the active state.", vm.name - ) - continue - # Check if the VM is in the disabled state - if vm.status in config["zabbix_device_disable"]: - vm.zabbix_state = 1 - # Add hostgroup if config is set - if config["create_hostgroups"]: - # Create new hostgroup. Potentially multiple groups if nested - hostgroups = vm.create_zbx_hostgroup(zabbix_groups) - # go through all newly created hostgroups - for group in hostgroups: - # Add new hostgroups to zabbix group list - zabbix_groups.append(group) - # Check if VM is already in Zabbix - if vm.zabbix_id: - vm.consistency_check( - zabbix_groups, - zabbix_templates, - zabbix_proxy_list, - config["full_proxy_sync"], - config["create_hostgroups"], - ) - continue - # Add VM to Zabbix - vm.create_in_zabbix(zabbix_groups, zabbix_templates, zabbix_proxy_list) - except SyncError: - pass - - 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, - config["create_journal"], - logger, - ) - logger.debug("Host %s: Started operations on device.", device.name) - 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( - config["hostgroup_format"], netbox_site_groups, netbox_regions - ) - # Check if a valid hostgroup has been found for this VM. - if not device.hostgroups: - logger.warning( - "Host %s: Host has no valid hostgroups, Skipping this host...", - device.name, - ) - continue - if config["extended_site_properties"] and nb_device.site: - logger.debug("Device %s: extending site information.", device.name) - device.site = convert_recordset( # type: ignore[attr-defined] - netbox.dcim.sites.filter(id=nb_device.site.id) - ) - device.set_inventory(nb_device) - device.set_usermacros() - device.set_tags() - # Checks if device is part of cluster. - # Requires clustering variable - if device.is_cluster() and config["clustering"]: - # Check if device is primary or secondary - if device.promote_primary_device(): - logger.info( - "Device %s: is part of cluster and primary.", device.name - ) - else: - # Device is secondary in cluster. - # Don't continue with this device. - logger.info( - "Device %s: Is part of cluster but not primary. Skipping this host...", - device.name, - ) - continue - # Checks if device is in cleanup state - if device.status in config["zabbix_device_removal"]: - if device.zabbix_id: - # Delete device from Zabbix - # and remove hostID from NetBox. - device.cleanup() - logger.info("Device %s: cleanup complete", device.name) - continue - # Device has been added to NetBox - # but is not in Activate state - logger.info( - "Device %s: Skipping since this device is not in the active state.", - device.name, - ) - continue - # Check if the device is in the disabled state - if device.status in config["zabbix_device_disable"]: - device.zabbix_state = 1 - # Add hostgroup is config is set - if config["create_hostgroups"]: - # Create new hostgroup. Potentially multiple groups if nested - hostgroups = device.create_zbx_hostgroup(zabbix_groups) - # go through all newly created hostgroups - for group in hostgroups: - # Add new hostgroups to zabbix group list - zabbix_groups.append(group) - # Check if device is already in Zabbix - if device.zabbix_id: - device.consistency_check( - zabbix_groups, - zabbix_templates, - zabbix_proxy_list, - config["full_proxy_sync"], - config["create_hostgroups"], - ) - continue - # Add device to Zabbix - device.create_in_zabbix(zabbix_groups, zabbix_templates, zabbix_proxy_list) - except SyncError: - pass - zabbix.logout() - +from netbox_zabbix_sync.modules.cli import parse_cli if __name__ == "__main__": - parser = argparse.ArgumentParser( - description="A script to sync Zabbix with NetBox device data." - ) - parser.add_argument( - "-v", "--verbose", help="Turn on verbose logging.", action="store_true" - ) - parser.add_argument( - "-vv", "--debug", help="Turn on debugging.", action="store_true" - ) - parser.add_argument( - "-vvv", - "--debug-all", - help="Turn on debugging for all modules.", - action="store_true", - ) - parser.add_argument("-q", "--quiet", help="Turn off warnings.", action="store_true") - args = parser.parse_args() - main(args) + parse_cli() diff --git a/netbox_zabbix_sync/__init__.py b/netbox_zabbix_sync/__init__.py new file mode 100644 index 0000000..7517e89 --- /dev/null +++ b/netbox_zabbix_sync/__init__.py @@ -0,0 +1,5 @@ +""" +Makes core module sync function available at package level for easier imports. +""" + +from netbox_zabbix_sync.modules.core import Sync as Sync diff --git a/modules/__init__.py b/netbox_zabbix_sync/modules/__init__.py similarity index 100% rename from modules/__init__.py rename to netbox_zabbix_sync/modules/__init__.py diff --git a/netbox_zabbix_sync/modules/cli.py b/netbox_zabbix_sync/modules/cli.py new file mode 100644 index 0000000..afc3495 --- /dev/null +++ b/netbox_zabbix_sync/modules/cli.py @@ -0,0 +1,205 @@ +import argparse +import logging +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 + +# 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 logging + setup_logger() + logger = get_logger() + # Set log levels based on verbosity flags + if arguments.verbose: + set_log_levels(logging.WARNING, logging.INFO) + if arguments.debug: + set_log_levels(logging.WARNING, logging.DEBUG) + if arguments.debug_all: + set_log_levels(logging.DEBUG, logging.DEBUG) + if arguments.quiet: + set_log_levels(logging.ERROR, logging.ERROR) + + # Gather environment variables for Zabbix and Netbox communication + env_vars = ["ZABBIX_HOST", "NETBOX_HOST", "NETBOX_TOKEN"] + if "ZABBIX_TOKEN" in environ: + env_vars.append("ZABBIX_TOKEN") + else: + env_vars.append("ZABBIX_USER") + env_vars.append("ZABBIX_PASS") + for var in env_vars: + if var not in environ: + e = f"Environment variable {var} has not been defined." + logger.error(e) + raise EnvironmentVarError(e) + # Get all virtual environment variables + if "ZABBIX_TOKEN" in env_vars: + zabbix_user = None + zabbix_pass = None + zabbix_token = environ.get("ZABBIX_TOKEN") + else: + zabbix_user = environ.get("ZABBIX_USER") + zabbix_pass = environ.get("ZABBIX_PASS") + zabbix_token = None + zabbix_host = environ.get("ZABBIX_HOST") + 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(config=config) + syncer.connect( + nb_host=netbox_host, + nb_token=netbox_token, + zbx_host=zabbix_host, + zbx_user=zabbix_user, + zbx_pass=zabbix_pass, + zbx_token=zabbix_token, + ) + syncer.start() + + +def parse_cli(): + """ + Parse command-line arguments and run the main function. + """ + parser = argparse.ArgumentParser( + description="Synchronise NetBox device data to Zabbix." + ) + + # ── Verbosity ────────────────────────────────────────────────────────────── + parser.add_argument( + "-v", "--verbose", help="Turn on verbose logging.", action="store_true" + ) + parser.add_argument( + "-vv", "--debug", help="Turn on debugging.", action="store_true" + ) + parser.add_argument( + "-vvv", + "--debug-all", + help="Turn on debugging for all modules.", + 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, + ) + parser.add_argument( + "--version", action="version", version="NetBox-Zabbix Sync 3.4.0" + ) + + # ── 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 new file mode 100644 index 0000000..d5470e5 --- /dev/null +++ b/netbox_zabbix_sync/modules/core.py @@ -0,0 +1,405 @@ +"""Core component of the sync process""" + +import ssl +from os import environ +from typing import Any + +from pynetbox import api as nbapi +from pynetbox.core.query import RequestError as NetBoxRequestError +from requests.exceptions import ConnectionError as RequestsConnectionError +from zabbix_utils import APIRequestError, ProcessingError, ZabbixAPI + +from netbox_zabbix_sync.modules.device import PhysicalDevice +from netbox_zabbix_sync.modules.exceptions import SyncError +from netbox_zabbix_sync.modules.logging import get_logger +from netbox_zabbix_sync.modules.settings import DEFAULT_CONFIG +from netbox_zabbix_sync.modules.tools import ( + convert_recordset, + proxy_prepper, + verify_hg_format, +) +from netbox_zabbix_sync.modules.virtual_machine import VirtualMachine + +logger = get_logger() + + +class Sync: + """ + Class that hosts the main sync process. + This class is used to connect to NetBox and Zabbix and run the sync process. + """ + + def __init__(self, config: dict[str, Any] | None = None): + """ + Docstring for __init__ + + :param self: Description + :param config: Description + """ + self.netbox = None + self.zabbix = None + self.nb_version = None + + default_config = DEFAULT_CONFIG.copy() + + combined_config = { + **default_config, + **(config if config else {}), + } + + self.config: dict[str, Any] = combined_config + + def connect( + self, nb_host, nb_token, zbx_host, zbx_user=None, zbx_pass=None, zbx_token=None + ): + """ + Docstring for connect + + :param self: Description + :param nb_host: Description + :param nb_token: Description + :param zbx_host: Description + :param zbx_user: Description + :param zbx_pass: Description + :param zbx_token: Description + """ + # Initialize Netbox API connection + netbox = nbapi(nb_host, token=nb_token, threading=True) + try: + # Get NetBox version + nb_version = netbox.version + # Test API access by attempting to access a basic endpoint + # This will catch authorization errors early + netbox.dcim.devices.count() + logger.debug("NetBox version is %s.", nb_version) + self.netbox = netbox + self.nb_version = nb_version + except RequestsConnectionError: + logger.error( + "Unable to connect to NetBox with URL %s. Please check the URL and status of NetBox.", + nb_host, + ) + return False + except NetBoxRequestError as nb_error: + e = f"NetBox returned the following error: {nb_error}." + logger.error(e) + return False + # Check Netbox API token format based on NetBox version + if not self._validate_netbox_token(nb_token, self.nb_version): + return False + # Set Zabbix API + if (zbx_pass or zbx_user) and zbx_token: + e = ( + "Both ZABBIX_PASS, ZABBIX_USER and ZABBIX_TOKEN environment variables are set. " + "Please choose between token or password based authentication." + ) + logger.error(e) + return False + try: + ssl_ctx = ssl.create_default_context() + + # If a custom CA bundle is set for pynetbox (requests), also use it for the Zabbix API + if environ.get("REQUESTS_CA_BUNDLE", None): + ssl_ctx.load_verify_locations(environ["REQUESTS_CA_BUNDLE"]) + if not zbx_token: + logger.debug("Using user/password authentication for Zabbix API.") + self.zabbix = ZabbixAPI( + zbx_host, user=zbx_user, password=zbx_pass, ssl_context=ssl_ctx + ) + else: + logger.debug("Using token authentication for Zabbix API.") + self.zabbix = ZabbixAPI(zbx_host, token=zbx_token, ssl_context=ssl_ctx) + self.zabbix.check_auth() + logger.debug("Zabbix version is %s.", self.zabbix.version) + except (APIRequestError, ProcessingError) as zbx_error: + e = f"Zabbix returned the following error: {zbx_error}." + logger.error(e) + return False + return True + + def _validate_netbox_token(self, token: str, nb_version: str) -> bool: + """Validate the format of the NetBox token based on the NetBox version. + :param token: The NetBox token to validate. + :param nb_version: The version of NetBox being used. + :return: True if the token format is valid for the given NetBox version, False otherwise. + """ + support_token_url = ( + "https://netboxlabs.com/docs/netbox/integrations/rest-api/#v1-and-v2-tokens" # noqa: S105 + ) + token_prefix = "nbt_" # noqa: S105 + nb_v2_support_version = "4.5" + v2_token = bool(token.startswith(token_prefix) and "." in token) + v2_error_token = bool(token.startswith(token_prefix) and "." not in token) + # Check if the token is passed without a proper key.token format + if v2_error_token: + logger.error( + "It looks like an invalid v2 token was passed. For more info, see %s", + support_token_url, + ) + return False + # Warning message for Netbox token v1 with Netbox v4.5 and higher + if not v2_token and nb_version >= nb_v2_support_version: + logger.warning( + "Using Netbox v1 token format. " + "Consider updating to a v2 token. For more info, see %s", + support_token_url, + ) + elif v2_token and nb_version < nb_v2_support_version: + logger.error( + "Using Netbox v2 token format with Netbox version lower than 4.5. " + "Revert to v1 token or upgrade Netbox to 4.5 or higher. For more info, see %s", + support_token_url, + ) + return False + elif v2_token and nb_version >= nb_v2_support_version: + logger.debug("Using NetBox v2 token format.") + else: + logger.debug("Using NetBox v1 token format.") + return True + + def start(self): + """ + Run the NetBox to Zabbix sync process. + """ + if not self.netbox or not self.zabbix: + e = "Not able to start sync: No connection to NetBox or Zabbix API." + logger.error(e) + return False + device_cfs = [] + vm_cfs = [] + # Create API call to get all custom fields which are on the device objects + device_cfs = list( + self.netbox.extras.custom_fields.filter( + type=["text", "object", "select"], content_types="dcim.device" + ) + ) + # Check if the provided Hostgroup layout is valid + verify_hg_format( + self.config["hostgroup_format"], + device_cfs=device_cfs, + hg_type="dev", + logger=logger, + ) + if self.config["sync_vms"]: + vm_cfs = list( + self.netbox.extras.custom_fields.filter( + type=["text", "object", "select"], + content_types="virtualization.virtualmachine", + ) + ) + verify_hg_format( + self.config["vm_hostgroup_format"], + vm_cfs=vm_cfs, + hg_type="vm", + logger=logger, + ) + # Set API parameter mapping based on API version + proxy_name = "host" if str(self.zabbix.version) < "7" else "name" + # Get all Zabbix and NetBox data + netbox_devices = list( + self.netbox.dcim.devices.filter(**self.config["nb_device_filter"]) + ) + netbox_vms = [] + if self.config["sync_vms"]: + netbox_vms = list( + self.netbox.virtualization.virtual_machines.filter( + **self.config["nb_vm_filter"] + ) + ) + netbox_site_groups = convert_recordset(self.netbox.dcim.site_groups.all()) + netbox_regions = convert_recordset(self.netbox.dcim.regions.all()) + netbox_journals = self.netbox.extras.journal_entries + zabbix_groups = self.zabbix.hostgroup.get( # type: ignore + output=["groupid", "name"] + ) + zabbix_templates = self.zabbix.template.get( # type: ignore + output=["templateid", "name"] + ) + zabbix_proxies = self.zabbix.proxy.get( # type: ignore + output=["proxyid", proxy_name] + ) + # Set empty list for proxy processing Zabbix <= 6 + zabbix_proxygroups = [] + if str(self.zabbix.version) >= "7": + zabbix_proxygroups = self.zabbix.proxygroup.get( # type: ignore + output=["proxy_groupid", "name"] + ) + # Sanitize proxy data + if proxy_name == "host": + for proxy in zabbix_proxies: + proxy["name"] = proxy.pop("host") + # Prepare list of all proxy and proxy_groups + zabbix_proxy_list = proxy_prepper(zabbix_proxies, zabbix_proxygroups) + + # Go through all NetBox devices + for nb_vm in netbox_vms: + try: + vm = VirtualMachine( + nb_vm, + self.zabbix, + netbox_journals, + 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() + # Check if a valid template has been found for this VM. + if not vm.zbx_template_names: + continue + vm.set_hostgroup( + self.config["vm_hostgroup_format"], + netbox_site_groups, + netbox_regions, + ) + # Check if a valid hostgroup has been found for this VM. + if not vm.hostgroups: + continue + if self.config["extended_site_properties"] and nb_vm.site: + logger.debug("Host %s: extending site information.", vm.name) + vm.site = convert_recordset( + self.netbox.dcim.sites.filter(id=nb_vm.site.id) + ) + vm.set_inventory(nb_vm) + vm.set_usermacros() + vm.set_tags() + # Checks if device is in cleanup state + if vm.status in self.config["zabbix_device_removal"]: + if vm.zabbix_id: + # Delete device from Zabbix + # and remove hostID from self.netbox. + vm.cleanup() + logger.info("Host %s: cleanup complete", vm.name) + continue + # Device has been added to NetBox + # but is not in Activate state + logger.info( + "Host %s: Skipping since this host is not in the active state.", + vm.name, + ) + continue + # Check if the VM is in the disabled state + if vm.status in self.config["zabbix_device_disable"]: + vm.zabbix_state = 1 + # Add hostgroup if config is set + if self.config["create_hostgroups"]: + # Create new hostgroup. Potentially multiple groups if nested + hostgroups = vm.create_zbx_hostgroup(zabbix_groups) + # go through all newly created hostgroups + for group in hostgroups: + # Add new hostgroups to zabbix group list + zabbix_groups.append(group) + # Check if VM is already in Zabbix + if vm.zabbix_id: + vm.consistency_check( + zabbix_groups, + zabbix_templates, + zabbix_proxy_list, + self.config["full_proxy_sync"], + self.config["create_hostgroups"], + ) + continue + # Add VM to Zabbix + vm.create_in_zabbix(zabbix_groups, zabbix_templates, zabbix_proxy_list) + except SyncError: + pass + + for nb_device in netbox_devices: + try: + # Set device instance set data such as hostgroup and template information. + device = PhysicalDevice( + nb_device, + self.zabbix, + netbox_journals, + self.nb_version, + self.config["create_journal"], + logger, + config=self.config, + ) + logger.debug("Host %s: Started operations on device.", device.name) + device.set_template( + self.config["templates_config_context"], + self.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( + self.config["hostgroup_format"], netbox_site_groups, netbox_regions + ) + # Check if a valid hostgroup has been found for this VM. + if not device.hostgroups: + logger.warning( + "Host %s: has no valid hostgroups, Skipping this host...", + device.name, + ) + continue + if self.config["extended_site_properties"] and nb_device.site: + logger.debug("Host %s: extending site information.", device.name) + device.site = convert_recordset( + self.netbox.dcim.sites.filter(id=nb_device.site.id) + ) + device.set_inventory(nb_device) + device.set_usermacros() + device.set_tags() + # Checks if device is part of cluster. + # Requires clustering variable + if device.is_cluster() and self.config["clustering"]: + # Check if device is primary or secondary + if device.promote_primary_device(): + logger.info( + "Host %s: is part of cluster and primary.", device.name + ) + else: + # Device is secondary in cluster. + # Don't continue with this device. + logger.info( + "Host %s: Is part of cluster but not primary. Skipping this host...", + device.name, + ) + continue + # Checks if device is in cleanup state + if device.status in self.config["zabbix_device_removal"]: + if device.zabbix_id: + # Delete device from Zabbix + # and remove hostID from NetBox. + device.cleanup() + logger.info("Host %s: cleanup complete", device.name) + continue + # Device has been added to NetBox + # but is not in Activate state + logger.info( + "Host %s: Skipping since this host is not in the active state.", + device.name, + ) + continue + # Check if the device is in the disabled state + if device.status in self.config["zabbix_device_disable"]: + device.zabbix_state = 1 + # Add hostgroup is config is set + if self.config["create_hostgroups"]: + # Create new hostgroup. Potentially multiple groups if nested + hostgroups = device.create_zbx_hostgroup(zabbix_groups) + # go through all newly created hostgroups + for group in hostgroups: + # Add new hostgroups to zabbix group list + zabbix_groups.append(group) + # Check if device is already in Zabbix + if device.zabbix_id: + device.consistency_check( + zabbix_groups, + zabbix_templates, + zabbix_proxy_list, + self.config["full_proxy_sync"], + self.config["create_hostgroups"], + ) + continue + # Add device to Zabbix + device.create_in_zabbix( + zabbix_groups, zabbix_templates, zabbix_proxy_list + ) + except SyncError: + pass + self.zabbix.logout() + return True diff --git a/modules/device.py b/netbox_zabbix_sync/modules/device.py similarity index 91% rename from modules/device.py rename to netbox_zabbix_sync/modules/device.py index 7d60142..3a80898 100644 --- a/modules/device.py +++ b/netbox_zabbix_sync/modules/device.py @@ -11,25 +11,24 @@ from typing import Any from pynetbox import RequestError as NetboxRequestError from zabbix_utils import APIRequestError -from modules.config import load_config -from modules.exceptions import ( +from netbox_zabbix_sync.modules.exceptions import ( InterfaceConfigError, SyncExternalError, SyncInventoryError, TemplateError, ) -from modules.hostgroups import Hostgroup -from modules.interface import ZabbixInterface -from modules.tags import ZabbixTags -from modules.tools import ( +from netbox_zabbix_sync.modules.host_description import Description +from netbox_zabbix_sync.modules.hostgroups import Hostgroup +from netbox_zabbix_sync.modules.interface import ZabbixInterface +from netbox_zabbix_sync.modules.settings import load_config +from netbox_zabbix_sync.modules.tags import ZabbixTags +from netbox_zabbix_sync.modules.tools import ( cf_to_string, field_mapper, remove_duplicates, sanatize_log_output, ) -from modules.usermacros import ZabbixUsermacros - -config = load_config() +from netbox_zabbix_sync.modules.usermacros import ZabbixUsermacros class PhysicalDevice: @@ -39,8 +38,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 @@ -55,6 +62,7 @@ class PhysicalDevice: self.hostgroups = [] self.hostgroup_type = "dev" self.tenant = nb.tenant + self.site = nb.site self.config_context = nb.config_context self.zbxproxy = None self.zabbix_state = 0 @@ -75,15 +83,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): """ @@ -99,10 +107,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) @@ -133,8 +141,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, ) @@ -181,12 +189,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}." ) @@ -215,27 +223,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 @@ -381,7 +389,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): @@ -426,7 +434,7 @@ class PhysicalDevice: macros = ZabbixUsermacros( self.nb, self._usermacro_map(), - config["usermacro_sync"], + self.config["usermacro_sync"], logger=self.logger, host=self.name, ) @@ -444,14 +452,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() @@ -481,20 +489,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 @@ -514,7 +522,10 @@ class PhysicalDevice: # If the proxy name matches if proxy["name"] == proxy_name: self.logger.debug( - "Host %s: using {proxy['type']} '%s'", self.name, proxy_name + "Host %s: using %s '%s'", + self.name, + proxy["type"], + proxy_name, ) self.zbxproxy = proxy return True @@ -524,13 +535,7 @@ class PhysicalDevice: ) return False - def create_in_zabbix( - self, - groups, - templates, - proxies, - description="Host added by NetBox sync script.", - ): + def create_in_zabbix(self, groups, templates, proxies): """ Creates Zabbix host object with parameters from NetBox object. """ @@ -552,6 +557,14 @@ class PhysicalDevice: interfaces = self.set_interface_details() # Set Zabbix proxy if defined self._set_proxy(proxies) + # Set description + description_handler = Description( + self.nb, + self.config, + logger=self.logger, + nb_version=self.nb_api_version, + ) + description = description_handler.generate() # Set basic data for host creation create_data = { "host": self.name, @@ -585,7 +598,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) @@ -659,7 +672,8 @@ class PhysicalDevice: Checks if Zabbix object is still valid with NetBox parameters. """ # If group is found or if the hostgroup is nested - if not self.set_zbx_groupid(groups): # or len(self.hostgroups.split("/")) > 1: + # or len(self.hostgroups.split("/")) > 1: + if not self.set_zbx_groupid(groups): if create_hostgroups: # Script is allowed to create a new hostgroup new_groups = self.create_zbx_hostgroup(groups) @@ -826,7 +840,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) @@ -835,12 +849,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: @@ -863,7 +877,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/modules/exceptions.py b/netbox_zabbix_sync/modules/exceptions.py similarity index 100% rename from modules/exceptions.py rename to netbox_zabbix_sync/modules/exceptions.py diff --git a/netbox_zabbix_sync/modules/host_description.py b/netbox_zabbix_sync/modules/host_description.py new file mode 100644 index 0000000..40d8a7b --- /dev/null +++ b/netbox_zabbix_sync/modules/host_description.py @@ -0,0 +1,125 @@ +""" +Modules that set description of a host in Zabbix +""" + +from datetime import datetime +from logging import getLogger +from re import findall as re_findall + + +class Description: + """ + Class that generates the description for a host in Zabbix based on the configuration provided. + + INPUT: + - netbox_object: The NetBox object that is being synced. + - configuration: configuration of the syncer. + Required keys in configuration: + description: Can be "static", "dynamic" or a custom description with macros. + - nb_version: The version of NetBox that is being used. + """ + + def __init__(self, netbox_object, configuration, nb_version, logger=None): + self.netbox_object = netbox_object + self.name = self.netbox_object.name + self.configuration = configuration + self.nb_version = nb_version + self.logger = logger or getLogger(__name__) + self._set_default_macro_values() + self._set_defaults() + + def _set_default_macro_values(self): + """ + Sets the default macro values for the description. + """ + # Get the datetime format from the configuration, + # or use the default format if not provided + dt_format = self.configuration.get("description_dt_format", "%Y-%m-%d %H:%M:%S") + # Set the datetime macro + try: + datetime_value = datetime.now().strftime(dt_format) + except (ValueError, TypeError) as e: + self.logger.warning( + "Host %s: invalid datetime format '%s': %s. Using default format.", + self.name, + dt_format, + e, + ) + datetime_value = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + # Set the owner macro + owner = self.netbox_object.owner if self.nb_version >= "4.5" else "" + # Set the macro list + self.macros = {"{datetime}": datetime_value, "{owner}": owner} + + def _resolve_macros(self, description): + """ + Takes a description and resolves the macro's in it. + Returns the description with the macro's resolved. + """ + # Find all macros in the description + provided_macros = re_findall(r"\{\w+\}", description) + # Go through all macros provided in the NB description + for macro in provided_macros: + # If the macro is in the list of default macro values + if macro in self.macros: + # Replace the macro in the description with the value of the macro + description = description.replace(macro, str(self.macros[macro])) + else: + # One of the macro's is invalid. + self.logger.warning( + "Host %s: macro %s is not valid. Failing back to default.", + self.name, + macro, + ) + return False + return description + + def _set_defaults(self): + """ + Sets the default descriptions for the host. + """ + self.defaults = { + "static": "Host added by NetBox sync script.", + "dynamic": ( + "Host by owner {owner} added by NetBox sync script on {datetime}." + ), + } + + def _custom_override(self): + """ + Checks if the description is mentioned in the config context. + """ + zabbix_config = self.netbox_object.config_context.get("zabbix") + if zabbix_config and "description" in zabbix_config: + return zabbix_config["description"] + return False + + def generate(self): + """ + Generates the description for the host. + """ + # First: check if an override is present. + config_context_description = self._custom_override() + if config_context_description is not False: + resolved = self._resolve_macros(config_context_description) + return resolved if resolved else self.defaults["static"] + # Override is not present: continue with config description + description = "" + if "description" not in self.configuration: + # If no description config is provided, use default static + return self.defaults["static"] + if not self.configuration["description"]: + # The configuration is set to False, meaning an empty description + return description + if self.configuration["description"] in self.defaults: + # The description is one of the default options + description = self.defaults[self.configuration["description"]] + else: + # The description is set to a custom description + description = self.configuration["description"] + # Resolve the macro's in the description + final_description = self._resolve_macros(description) + if final_description: + return final_description + return self.defaults["static"] diff --git a/modules/hostgroups.py b/netbox_zabbix_sync/modules/hostgroups.py similarity index 98% rename from modules/hostgroups.py rename to netbox_zabbix_sync/modules/hostgroups.py index c78c00c..6a647cd 100644 --- a/modules/hostgroups.py +++ b/netbox_zabbix_sync/modules/hostgroups.py @@ -2,8 +2,8 @@ from logging import getLogger -from modules.exceptions import HostgroupError -from modules.tools import build_path, cf_to_string +from netbox_zabbix_sync.modules.exceptions import HostgroupError +from netbox_zabbix_sync.modules.tools import build_path, cf_to_string class Hostgroup: diff --git a/modules/interface.py b/netbox_zabbix_sync/modules/interface.py similarity index 98% rename from modules/interface.py rename to netbox_zabbix_sync/modules/interface.py index 1bb5502..cea0eda 100644 --- a/modules/interface.py +++ b/netbox_zabbix_sync/modules/interface.py @@ -2,7 +2,7 @@ All of the Zabbix interface related configuration """ -from modules.exceptions import InterfaceConfigError +from netbox_zabbix_sync.modules.exceptions import InterfaceConfigError class ZabbixInterface: diff --git a/modules/logging.py b/netbox_zabbix_sync/modules/logging.py similarity index 100% rename from modules/logging.py rename to netbox_zabbix_sync/modules/logging.py diff --git a/modules/config.py b/netbox_zabbix_sync/modules/settings.py similarity index 91% rename from modules/config.py rename to netbox_zabbix_sync/modules/settings.py index e5509c6..f44930a 100644 --- a/modules/config.py +++ b/netbox_zabbix_sync/modules/settings.py @@ -82,13 +82,19 @@ DEFAULT_CONFIG = { "cluster/name": "cluster", "platform/name": "target", }, + "description_dt_format": "%Y-%m-%d %H:%M:%S", + "description": "static", } -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/modules/tags.py b/netbox_zabbix_sync/modules/tags.py similarity index 98% rename from modules/tags.py rename to netbox_zabbix_sync/modules/tags.py index 6ed17be..b05e599 100644 --- a/modules/tags.py +++ b/netbox_zabbix_sync/modules/tags.py @@ -4,7 +4,7 @@ All of the Zabbix Usermacro related configuration from logging import getLogger -from modules.tools import field_mapper, remove_duplicates +from netbox_zabbix_sync.modules.tools import field_mapper, remove_duplicates class ZabbixTags: diff --git a/modules/tools.py b/netbox_zabbix_sync/modules/tools.py similarity index 99% rename from modules/tools.py rename to netbox_zabbix_sync/modules/tools.py index a6b0a22..50a9781 100644 --- a/modules/tools.py +++ b/netbox_zabbix_sync/modules/tools.py @@ -3,7 +3,7 @@ from collections.abc import Callable from typing import Any, cast, overload -from modules.exceptions import HostgroupError +from netbox_zabbix_sync.modules.exceptions import HostgroupError def convert_recordset(recordset): diff --git a/modules/usermacros.py b/netbox_zabbix_sync/modules/usermacros.py similarity index 98% rename from modules/usermacros.py rename to netbox_zabbix_sync/modules/usermacros.py index 3258dde..05a8778 100644 --- a/modules/usermacros.py +++ b/netbox_zabbix_sync/modules/usermacros.py @@ -5,7 +5,7 @@ All of the Zabbix Usermacro related configuration from logging import getLogger from re import match -from modules.tools import field_mapper, sanatize_log_output +from netbox_zabbix_sync.modules.tools import field_mapper, sanatize_log_output class ZabbixUsermacros: diff --git a/modules/virtual_machine.py b/netbox_zabbix_sync/modules/virtual_machine.py similarity index 83% rename from modules/virtual_machine.py rename to netbox_zabbix_sync/modules/virtual_machine.py index 2ee70ce..a9f57bf 100644 --- a/modules/virtual_machine.py +++ b/netbox_zabbix_sync/modules/virtual_machine.py @@ -1,12 +1,12 @@ """Module that hosts all functions for virtual machine processing""" -from modules.config import load_config -from modules.device import PhysicalDevice -from modules.exceptions import InterfaceConfigError, SyncInventoryError, TemplateError -from modules.interface import ZabbixInterface - -# Load config -config = load_config() +from netbox_zabbix_sync.modules.device import PhysicalDevice +from netbox_zabbix_sync.modules.exceptions import ( + InterfaceConfigError, + SyncInventoryError, + TemplateError, +) +from netbox_zabbix_sync.modules.interface import ZabbixInterface class VirtualMachine(PhysicalDevice): @@ -20,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/pyproject.toml b/pyproject.toml index 77282b9..df55916 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,12 +4,25 @@ description = "Python script to synchronize Netbox devices to Zabbix." readme = "README.md" requires-python = ">=3.12" dependencies = ["igraph>=1.0.0", "pynetbox>=7.6.1", "zabbix-utils>=2.0.4"] -version = "3.3.0" +dynamic = ["version"] [project.urls] "Homepage" = "https://github.com/TheNetworkGuy/netbox-zabbix-sync" "Issues" = "https://github.com/TheNetworkGuy/netbox-zabbix-sync/issues" +[project.scripts] +netbox-zabbix-sync = "netbox_zabbix_sync.modules.cli:parse_cli" + +[build-system] +requires = ["setuptools>=64", "setuptools_scm>=8"] +build-backend = "setuptools.build_meta" + +[tool.setuptools.packages.find] +include = ["netbox_zabbix_sync*"] + +[tool.setuptools_scm] +version_file = "netbox_zabbix_sync/_version.py" + [tool.ruff.lint] ignore = [ # Ignore line-length @@ -67,6 +80,8 @@ select = [ "tests/*" = [ # Ignore use of assert "S101", + # Ignore hardcoded passwords / tokens + "S106", ] diff --git a/tests/test_configuration_parsing.py b/tests/test_configuration_parsing.py index d6186e9..845ef52 100644 --- a/tests/test_configuration_parsing.py +++ b/tests/test_configuration_parsing.py @@ -3,7 +3,7 @@ import os from unittest.mock import MagicMock, patch -from modules.config import ( +from netbox_zabbix_sync.modules.settings import ( DEFAULT_CONFIG, load_config, load_config_file, @@ -14,8 +14,13 @@ from modules.config import ( def test_load_config_defaults(): """Test that load_config returns default values when no config file or env vars are present""" with ( - patch("modules.config.load_config_file", return_value=DEFAULT_CONFIG.copy()), - patch("modules.config.load_env_variable", return_value=None), + patch( + "netbox_zabbix_sync.modules.settings.load_config_file", + return_value=DEFAULT_CONFIG.copy(), + ), + patch( + "netbox_zabbix_sync.modules.settings.load_env_variable", return_value=None + ), ): config = load_config() assert config == DEFAULT_CONFIG @@ -30,8 +35,13 @@ def test_load_config_file(): 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), + patch( + "netbox_zabbix_sync.modules.settings.load_config_file", + return_value=mock_config, + ), + patch( + "netbox_zabbix_sync.modules.settings.load_env_variable", return_value=None + ), ): config = load_config() assert config["templates_config_context"] is True @@ -52,8 +62,14 @@ def test_load_env_variables(): 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), + patch( + "netbox_zabbix_sync.modules.settings.load_config_file", + return_value=DEFAULT_CONFIG.copy(), + ), + patch( + "netbox_zabbix_sync.modules.settings.load_env_variable", + side_effect=mock_load_env, + ), ): config = load_config() assert config["sync_vms"] is True @@ -75,8 +91,14 @@ def test_env_vars_override_config_file(): return None with ( - patch("modules.config.load_config_file", return_value=mock_config), - patch("modules.config.load_env_variable", side_effect=mock_load_env), + patch( + "netbox_zabbix_sync.modules.settings.load_config_file", + return_value=mock_config, + ), + patch( + "netbox_zabbix_sync.modules.settings.load_env_variable", + side_effect=mock_load_env, + ), ): config = load_config() # This should be overridden by the env var diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..d8dc573 --- /dev/null +++ b/tests/test_core.py @@ -0,0 +1,1583 @@ +"""Tests for the core sync module.""" + +import unittest +from typing import ClassVar +from unittest.mock import MagicMock, patch + +from requests.exceptions import ConnectionError as RequestsConnectionError +from zabbix_utils import APIRequestError + +from netbox_zabbix_sync.modules.core import Sync + + +class MockNetboxDevice: + """Mock NetBox device object.""" + + def __init__( + self, + device_id=1, + name="test-device", + status_label="Active", + zabbix_hostid=None, + config_context=None, + site=None, + primary_ip=None, + virtual_chassis=None, + device_type=None, + tenant=None, + device_role=None, + role=None, + platform=None, + serial="", + tags=None, + ): + self.id = device_id + self.name = name + self.status = MagicMock() + self.status.label = status_label + self.status.value = status_label.lower() + self.custom_fields = { + "zabbix_hostid": zabbix_hostid, + "zabbix_template": "TestTemplate", + } + self.config_context = config_context or {} + self.tenant = tenant + self.platform = platform + self.serial = serial + self.asset_tag = None + self.location = None + self.rack = None + self.position = None + self.face = None + self.latitude = None + self.longitude = None + self.parent_device = None + self.airflow = None + self.cluster = None + self.vc_position = None + self.vc_priority = None + self.description = "" + self.comments = "" + self.tags = tags or [] + self.oob_ip = None + + # Setup site with proper structure + if site is None: + self.site = MagicMock() + self.site.name = "TestSite" + self.site.slug = "testsite" + else: + self.site = site + + # Setup primary IP with proper structure + if primary_ip is None: + self.primary_ip = MagicMock() + self.primary_ip.address = "192.168.1.1/24" + else: + self.primary_ip = primary_ip + + self.primary_ip4 = self.primary_ip + self.primary_ip6 = None + + # Setup device type with proper structure + if device_type is None: + self.device_type = MagicMock() + self.device_type.custom_fields = {"zabbix_template": "TestTemplate"} + self.device_type.manufacturer = MagicMock() + self.device_type.manufacturer.name = "TestManufacturer" + self.device_type.display = "Test Device Type" + self.device_type.model = "Test Model" + self.device_type.slug = "test-model" + else: + self.device_type = device_type + + if device_role is None and role is None: + # Create default role + mock_role = MagicMock() + mock_role.name = "Switch" + mock_role.slug = "switch" + self.device_role = mock_role # NetBox 2/3 + self.role = mock_role # NetBox 4+ + else: + self.device_role = device_role or role + self.role = role or device_role + + self.virtual_chassis = virtual_chassis + + def save(self): + """Mock save method for NetBox device.""" + + +class MockNetboxVM: + """Mock NetBox virtual machine object. + + Mirrors the real NetBox API response structure so the full VirtualMachine + pipeline runs without mocking the class itself. + """ + + def __init__( + self, + vm_id=1, + name="test-vm", + status_label="Active", + zabbix_hostid=None, + config_context=None, + site=None, + primary_ip=None, + role=None, + cluster=None, + tenant=None, + platform=None, + tags=None, + ): + self.id = vm_id + self.name = name + self.status = MagicMock() + self.status.label = status_label + self.status.value = status_label.lower() + self.custom_fields = {"zabbix_hostid": zabbix_hostid} + # Default config_context includes a template so the VM is not skipped + self.config_context = ( + config_context + if config_context is not None + else {"zabbix": {"templates": ["TestTemplate"]}} + ) + self.tenant = tenant + self.platform = platform + self.serial = "" + self.description = "" + self.comments = "" + self.vcpus = None + self.memory = None + self.disk = None + self.virtual_chassis = None + self.tags = tags or [] + self.oob_ip = None + + # Setup site + if site is None: + self.site = MagicMock() + self.site.name = "TestSite" + self.site.slug = "testsite" + self.site.region = None + self.site.group = None + else: + self.site = site + + # Setup primary IP + if primary_ip is None: + self.primary_ip = MagicMock() + self.primary_ip.address = "192.168.1.1/24" + else: + self.primary_ip = primary_ip + self.primary_ip4 = self.primary_ip + self.primary_ip6 = None + + # Setup role + if role is None: + mock_role = MagicMock() + mock_role.name = "Switch" + mock_role.slug = "switch" + self.role = mock_role + else: + self.role = role + + # Setup cluster + if cluster is None: + mock_cluster = MagicMock() + mock_cluster.name = "TestCluster" + mock_cluster_type = MagicMock() + mock_cluster_type.name = "TestClusterType" + mock_cluster.type = mock_cluster_type + self.cluster = mock_cluster + else: + self.cluster = cluster + + def save(self): + """Mock save method.""" + + +class TestNetboxTokenHandling(unittest.TestCase): + """Test that sync properly handles NetBox token authentication.""" + + def test_v1_token_with_netbox_45(self): + """Test that v1 token with NetBox 4.5+ logs warning but returns True.""" + syncer = Sync() + + with self.assertLogs("NetBox-Zabbix-sync", level="WARNING") as log_context: + result = syncer._validate_netbox_token("token123", "4.5") + + self.assertTrue(result) + self.assertTrue( + any("v1 token format" in record.message for record in log_context.records) + ) + + def test_v2_token_with_netbox_35(self): + """Test that v2 token with NetBox < 4.5 logs error and returns False.""" + syncer = Sync() + + with self.assertLogs("NetBox-Zabbix-sync", level="ERROR") as log_context: + result = syncer._validate_netbox_token("nbt_key123.token123", "3.5") + + self.assertFalse(result) + self.assertTrue( + any( + "v2 token format with Netbox version lower than 4.5" in record.message + for record in log_context.records + ) + ) + + def test_v2_token_with_netbox_45(self): + """Test that v2 token with NetBox 4.5+ logs debug and returns True.""" + syncer = Sync() + + with self.assertLogs("NetBox-Zabbix-sync", level="DEBUG") as log_context: + result = syncer._validate_netbox_token("nbt_key123.token123", "4.5") + + self.assertTrue(result) + self.assertTrue( + any("v2 token format" in record.message for record in log_context.records) + ) + + def test_v1_token_with_netbox_35(self): + """Test that v1 token with NetBox < 4.5 logs debug and returns True.""" + syncer = Sync() + + with self.assertLogs("NetBox-Zabbix-sync", level="DEBUG") as log_context: + result = syncer._validate_netbox_token("token123", "3.5") + + self.assertTrue(result) + self.assertTrue( + any("v1 token format" in record.message for record in log_context.records) + ) + + +class TestSyncNetboxConnection(unittest.TestCase): + """Test NetBox connection handling in sync function.""" + + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_error_on_netbox_connection_error(self, mock_api): + """Test that sync returns False when NetBox connection fails.""" + mock_netbox = MagicMock() + mock_api.return_value = mock_netbox + # Simulate connection error when accessing version + type(mock_netbox).version = property( + lambda self: (_ for _ in ()).throw(RequestsConnectionError()) + ) + + syncer = Sync() + result = syncer.connect( + nb_host="http://netbox.local", + nb_token="token", + zbx_host="http://zabbix.local", + zbx_user="user", + zbx_pass="pass", + zbx_token=None, + ) + + self.assertFalse(result) + + +class TestZabbixUserTokenConflict(unittest.TestCase): + """Test that sync returns False when both ZABBIX_USER/PASS and ZABBIX_TOKEN are set.""" + + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_error_on_user_token_conflict(self, mock_api): + """Test that sync returns False when both user/pass and token are provided.""" + mock_netbox = MagicMock() + mock_api.return_value = mock_netbox + mock_netbox.version = "3.5" + + syncer = Sync() + result = syncer.connect( + nb_host="http://netbox.local", + nb_token="token", + zbx_host="http://zabbix.local", + zbx_user="user", + zbx_pass="pass", + zbx_token="token", # Both token and user/pass provided + ) + + self.assertFalse(result) + + +class TestSyncZabbixConnection(unittest.TestCase): + """Test Zabbix connection handling in sync function.""" + + def _setup_netbox_mock(self, mock_api): + """Helper to setup a working NetBox mock.""" + mock_netbox = MagicMock() + mock_api.return_value = mock_netbox + mock_netbox.version = "3.5" + return mock_netbox + + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_exits_on_zabbix_api_error(self, mock_api, mock_zabbix_api): + """Test that sync exits when Zabbix API authentication fails.""" + # Simulate Netbox API + self._setup_netbox_mock(mock_api) + # Simulate Zabbix API error + mock_zabbix_api.return_value.check_auth.side_effect = APIRequestError( + "Invalid credentials" + ) + # Start syncer and set connection details + syncer = Sync() + result = syncer.connect( + nb_host="http://netbox.local", + nb_token="token", + zbx_host="http://zabbix.local", + zbx_user="user", + zbx_pass="pass", + zbx_token=None, + ) + # Should return False due to Zabbix API error + self.assertFalse(result) + result = syncer.connect( + "http://netbox.local", + "token", + "http://zabbix.local", + "user", + "pass", + None, + ) + # Validate that result is False due to Zabbix API error + self.assertFalse(result) + + +class TestSyncZabbixAuthentication(unittest.TestCase): + """Test Zabbix authentication methods.""" + + def _setup_netbox_mock(self, mock_api): + """Helper to setup a working NetBox mock.""" + mock_netbox = MagicMock() + mock_api.return_value = mock_netbox + mock_netbox.version = "3.5" + mock_netbox.extras.custom_fields.filter.return_value = [] + mock_netbox.dcim.devices.filter.return_value = [] + mock_netbox.virtualization.virtual_machines.filter.return_value = [] + mock_netbox.dcim.site_groups.all.return_value = [] + mock_netbox.dcim.regions.all.return_value = [] + return mock_netbox + + def _setup_zabbix_mock(self, mock_zabbix_api, version="7.0"): + """Helper to setup a working Zabbix mock.""" + mock_zabbix = MagicMock() + mock_zabbix_api.return_value = mock_zabbix + mock_zabbix.version = version + mock_zabbix.hostgroup.get.return_value = [] + mock_zabbix.template.get.return_value = [] + mock_zabbix.proxy.get.return_value = [] + mock_zabbix.proxygroup.get.return_value = [] + return mock_zabbix + + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_uses_user_password_when_no_token(self, mock_api, mock_zabbix_api): + """Test that sync uses user/password auth when no token is provided.""" + self._setup_netbox_mock(mock_api) + + syncer = Sync() + syncer.connect( + nb_host="http://netbox.local", + nb_token="nb_token", + zbx_host="http://zabbix.local", + zbx_user="zbx_user", + zbx_pass="zbx_pass", + ) + + # Verify ZabbixAPI was called with user/password and without token + mock_zabbix_api.assert_called_once() + call_kwargs = mock_zabbix_api.call_args.kwargs + self.assertEqual(call_kwargs["user"], "zbx_user") + self.assertEqual(call_kwargs["password"], "zbx_pass") + self.assertNotIn("token", call_kwargs) + + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_uses_token_when_provided(self, mock_api, mock_zabbix_api): + """Test that sync uses token auth when token is provided.""" + self._setup_netbox_mock(mock_api) + self._setup_zabbix_mock(mock_zabbix_api) + + syncer = Sync() + syncer.connect( + nb_host="http://netbox.local", + nb_token="nb_token", + zbx_host="http://zabbix.local", + zbx_token="zbx_token", + ) + + # Verify ZabbixAPI was called with token and without user/password + mock_zabbix_api.assert_called_once() + call_kwargs = mock_zabbix_api.call_args.kwargs + self.assertEqual(call_kwargs["token"], "zbx_token") + self.assertNotIn("user", call_kwargs) + self.assertNotIn("password", call_kwargs) + + +class TestSyncDeviceProcessing(unittest.TestCase): + """Test device processing in sync function.""" + + def _setup_netbox_mock(self, mock_api, devices=None, vms=None): + """Helper to setup a working NetBox mock.""" + mock_netbox = MagicMock() + mock_api.return_value = mock_netbox + mock_netbox.version = "3.5" + mock_netbox.extras.custom_fields.filter.return_value = [] + mock_netbox.dcim.devices.filter.return_value = devices or [] + mock_netbox.virtualization.virtual_machines.filter.return_value = vms or [] + mock_netbox.dcim.site_groups.all.return_value = [] + mock_netbox.dcim.regions.all.return_value = [] + mock_netbox.extras.journal_entries = MagicMock() + return mock_netbox + + def _setup_zabbix_mock(self, mock_zabbix_api, version="6.0"): + """Helper to setup a working Zabbix mock.""" + mock_zabbix = MagicMock() + mock_zabbix_api.return_value = mock_zabbix + mock_zabbix.version = version + mock_zabbix.hostgroup.get.return_value = [{"groupid": "1", "name": "TestGroup"}] + mock_zabbix.template.get.return_value = [ + {"templateid": "1", "name": "TestTemplate"} + ] + mock_zabbix.proxy.get.return_value = [] + mock_zabbix.proxygroup.get.return_value = [] + return mock_zabbix + + @patch("netbox_zabbix_sync.modules.core.PhysicalDevice") + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_processes_devices_from_netbox( + self, mock_api, mock_zabbix_api, mock_physical_device + ): + """Test that sync creates PhysicalDevice instances for NetBox devices.""" + device1 = MockNetboxDevice(device_id=1, name="device1") + device2 = MockNetboxDevice(device_id=2, name="device2") + + self._setup_netbox_mock(mock_api, devices=[device1, device2]) + self._setup_zabbix_mock(mock_zabbix_api) + + # Mock PhysicalDevice to have no template (skip further processing) + mock_device_instance = MagicMock() + mock_device_instance.zbx_template_names = [] + mock_physical_device.return_value = mock_device_instance + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + # Verify PhysicalDevice was instantiated for each device + self.assertEqual(mock_physical_device.call_count, 2) + + @patch("netbox_zabbix_sync.modules.core.VirtualMachine") + @patch("netbox_zabbix_sync.modules.core.PhysicalDevice") + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_processes_vms_when_enabled( + self, mock_api, mock_zabbix_api, mock_physical_device, mock_virtual_machine + ): + """Test that sync processes VMs when sync_vms is enabled.""" + vm1 = MockNetboxVM(vm_id=1, name="vm1") + vm2 = MockNetboxVM(vm_id=2, name="vm2") + + self._setup_netbox_mock(mock_api, vms=[vm1, vm2]) + self._setup_zabbix_mock(mock_zabbix_api) + + # Mock VM to have no template (skip further processing) + mock_vm_instance = MagicMock() + mock_vm_instance.zbx_template_names = [] + mock_virtual_machine.return_value = mock_vm_instance + + syncer = Sync({"sync_vms": True}) + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + # Verify VirtualMachine was instantiated for each VM + self.assertEqual(mock_virtual_machine.call_count, 2) + + @patch("netbox_zabbix_sync.modules.core.VirtualMachine") + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_skips_vms_when_disabled( + self, mock_api, mock_zabbix_api, mock_virtual_machine + ): + """Test that sync does NOT process VMs when sync_vms is disabled.""" + vm1 = MockNetboxVM(vm_id=1, name="vm1") + + self._setup_netbox_mock(mock_api, vms=[vm1]) + self._setup_zabbix_mock(mock_zabbix_api) + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + # Verify VirtualMachine was never called + mock_virtual_machine.assert_not_called() + + +class TestSyncZabbixVersionHandling(unittest.TestCase): + """Test Zabbix version-specific handling.""" + + def _setup_netbox_mock(self, mock_api): + """Helper to setup a working NetBox mock.""" + mock_netbox = MagicMock() + mock_api.return_value = mock_netbox + mock_netbox.version = "3.5" + mock_netbox.extras.custom_fields.filter.return_value = [] + mock_netbox.dcim.devices.filter.return_value = [] + mock_netbox.virtualization.virtual_machines.filter.return_value = [] + mock_netbox.dcim.site_groups.all.return_value = [] + mock_netbox.dcim.regions.all.return_value = [] + return mock_netbox + + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_uses_host_proxy_name_for_zabbix_6(self, mock_api, mock_zabbix_api): + """Test that sync uses 'host' as proxy name field for Zabbix 6.""" + self._setup_netbox_mock(mock_api) + + mock_zabbix = MagicMock() + mock_zabbix_api.return_value = mock_zabbix + mock_zabbix.version = "6.0" + mock_zabbix.hostgroup.get.return_value = [] + mock_zabbix.template.get.return_value = [] + mock_zabbix.proxy.get.return_value = [{"proxyid": "1", "host": "proxy1"}] + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + # Verify proxy.get was called with 'host' field + mock_zabbix.proxy.get.assert_called_with(output=["proxyid", "host"]) + + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_uses_name_proxy_field_for_zabbix_7(self, mock_api, mock_zabbix_api): + """Test that sync uses 'name' as proxy name field for Zabbix 7.""" + self._setup_netbox_mock(mock_api) + + mock_zabbix = MagicMock() + mock_zabbix_api.return_value = mock_zabbix + mock_zabbix.version = "7.0" + mock_zabbix.hostgroup.get.return_value = [] + mock_zabbix.template.get.return_value = [] + mock_zabbix.proxy.get.return_value = [{"proxyid": "1", "name": "proxy1"}] + mock_zabbix.proxygroup.get.return_value = [] + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + # Verify proxy.get was called with 'name' field + mock_zabbix.proxy.get.assert_called_with(output=["proxyid", "name"]) + + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_fetches_proxygroups_for_zabbix_7(self, mock_api, mock_zabbix_api): + """Test that sync fetches proxy groups for Zabbix 7.""" + self._setup_netbox_mock(mock_api) + + mock_zabbix = MagicMock() + mock_zabbix_api.return_value = mock_zabbix + mock_zabbix.version = "7.0" + mock_zabbix.hostgroup.get.return_value = [] + mock_zabbix.template.get.return_value = [] + mock_zabbix.proxy.get.return_value = [] + mock_zabbix.proxygroup.get.return_value = [] + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + # Verify proxygroup.get was called for Zabbix 7 + mock_zabbix.proxygroup.get.assert_called_once() + + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_skips_proxygroups_for_zabbix_6(self, mock_api, mock_zabbix_api): + """Test that sync does NOT fetch proxy groups for Zabbix 6.""" + self._setup_netbox_mock(mock_api) + + mock_zabbix = MagicMock() + mock_zabbix_api.return_value = mock_zabbix + mock_zabbix.version = "6.0" + mock_zabbix.hostgroup.get.return_value = [] + mock_zabbix.template.get.return_value = [] + mock_zabbix.proxy.get.return_value = [] + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + # Verify proxygroup.get was NOT called for Zabbix 6 + mock_zabbix.proxygroup.get.assert_not_called() + + +class TestSyncLogout(unittest.TestCase): + """Test that sync properly logs out from Zabbix.""" + + def _setup_netbox_mock(self, mock_api): + """Helper to setup a working NetBox mock.""" + mock_netbox = MagicMock() + mock_api.return_value = mock_netbox + mock_netbox.version = "3.5" + mock_netbox.extras.custom_fields.filter.return_value = [] + mock_netbox.dcim.devices.filter.return_value = [] + mock_netbox.virtualization.virtual_machines.filter.return_value = [] + mock_netbox.dcim.site_groups.all.return_value = [] + mock_netbox.dcim.regions.all.return_value = [] + return mock_netbox + + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_logs_out_from_zabbix(self, mock_api, mock_zabbix_api): + """Test that sync calls logout on Zabbix API after completion.""" + self._setup_netbox_mock(mock_api) + + mock_zabbix = MagicMock() + mock_zabbix_api.return_value = mock_zabbix + mock_zabbix.version = "6.0" + mock_zabbix.hostgroup.get.return_value = [] + mock_zabbix.template.get.return_value = [] + mock_zabbix.proxy.get.return_value = [] + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + # Verify logout was called + mock_zabbix.logout.assert_called_once() + + +class TestSyncProxyNameSanitization(unittest.TestCase): + """Test proxy name field sanitization for Zabbix 6.""" + + def _setup_netbox_mock(self, mock_api): + """Helper to setup a working NetBox mock.""" + mock_netbox = MagicMock() + mock_api.return_value = mock_netbox + mock_netbox.version = "3.5" + mock_netbox.extras.custom_fields.filter.return_value = [] + mock_netbox.dcim.devices.filter.return_value = [] + mock_netbox.virtualization.virtual_machines.filter.return_value = [] + mock_netbox.dcim.site_groups.all.return_value = [] + mock_netbox.dcim.regions.all.return_value = [] + return mock_netbox + + @patch("netbox_zabbix_sync.modules.core.proxy_prepper") + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_renames_host_to_name_for_zabbix_6_proxies( + self, mock_api, mock_zabbix_api, mock_proxy_prepper + ): + """Test that for Zabbix 6, proxy 'host' field is renamed to 'name'.""" + self._setup_netbox_mock(mock_api) + + mock_zabbix = MagicMock() + mock_zabbix_api.return_value = mock_zabbix + mock_zabbix.version = "6.0" + mock_zabbix.hostgroup.get.return_value = [] + mock_zabbix.template.get.return_value = [] + # Zabbix 6 returns 'host' field + mock_zabbix.proxy.get.return_value = [ + {"proxyid": "1", "host": "proxy1"}, + {"proxyid": "2", "host": "proxy2"}, + ] + mock_proxy_prepper.return_value = [] + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + # Verify proxy_prepper was called with sanitized proxy list + call_args = mock_proxy_prepper.call_args[0] + proxies = call_args[0] + # Check that 'host' was renamed to 'name' + for proxy in proxies: + self.assertIn("name", proxy) + self.assertNotIn("host", proxy) + + +class TestDeviceHandeling(unittest.TestCase): + """ + Tests several devices which can be synced to Zabbix. + This class contains a lot of data in order to validate proper handling of different device types and configurations. + """ + + def _setup_netbox_mock(self, mock_api): + """Helper to setup a working NetBox mock.""" + mock_netbox = MagicMock() + mock_api.return_value = mock_netbox + mock_netbox.version = "3.5" + mock_netbox.extras.custom_fields.filter.return_value = [] + mock_netbox.dcim.devices.filter.return_value = [] + mock_netbox.virtualization.virtual_machines.filter.return_value = [] + mock_netbox.dcim.site_groups.all.return_value = [] + mock_netbox.dcim.regions.all.return_value = [] + return mock_netbox + + def _setup_zabbix_mock(self, mock_zabbix_api, version=7.0): + """Helper to setup a working Zabbix mock.""" + mock_zabbix = MagicMock() + mock_zabbix_api.return_value = mock_zabbix + mock_zabbix.version = version + mock_zabbix.hostgroup.get.return_value = [{"groupid": "1", "name": "TestGroup"}] + mock_zabbix.template.get.return_value = [ + {"templateid": "1", "name": "TestTemplate"} + ] + mock_zabbix.proxy.get.return_value = [] + mock_zabbix.proxygroup.get.return_value = [] + mock_zabbix.logout = MagicMock() + # Mock host.get to return empty (host doesn't exist yet) + mock_zabbix.host.get.return_value = [] + # Mock host.create to return success + mock_zabbix.host.create.return_value = {"hostids": ["1"]} + return mock_zabbix + + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_cluster_where_device_is_primary(self, mock_api, mock_zabbix_api): + """Test that sync properly handles a device that is the primary in a virtual chassis.""" + # Create a device that is part of a virtual chassis and is the primary + # Setup virtual chassis mock + vc_master = MagicMock() + vc_master.id = 1 # Same as device ID - device is primary + + virtual_chassis = MagicMock() + virtual_chassis.master = vc_master + virtual_chassis.name = "SW01" + + device = MockNetboxDevice( + device_id=1, + name="SW01N0", + virtual_chassis=virtual_chassis, + ) + + # Setup NetBox mock with a site for hostgroup + mock_netbox = self._setup_netbox_mock(mock_api) + mock_netbox.dcim.devices.filter.return_value = [device] + + # Create a mock site for hostgroup generation + mock_site = MagicMock() + mock_site.name = "TestSite" + device.site = mock_site + + # Setup Zabbix mock + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + + # Run the sync with clustering enabled + syncer = Sync({"clustering": True}) + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + # Verify that host.create was called with the cluster name "SW01", not "SW01N0" + mock_zabbix.host.create.assert_called_once() + create_call_kwargs = mock_zabbix.host.create.call_args.kwargs + + # The host should be created with the virtual chassis name, not the device name + self.assertEqual(create_call_kwargs["host"], "SW01") + + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_sync_cluster_where_device_is_not_primary(self, mock_api, mock_zabbix_api): + """Test that a non-primary cluster member is skipped and not created in Zabbix.""" + # vc_master.id (2) differs from device.id (1) → device is secondary + vc_master = MagicMock() + vc_master.id = 2 # Different from device ID → device is NOT primary + + virtual_chassis = MagicMock() + virtual_chassis.master = vc_master + virtual_chassis.name = "SW01" + + device = MockNetboxDevice( + device_id=1, + name="SW01N1", + virtual_chassis=virtual_chassis, + ) + + mock_netbox = self._setup_netbox_mock(mock_api) + mock_netbox.dcim.devices.filter.return_value = [device] + + mock_site = MagicMock() + mock_site.name = "TestSite" + device.site = mock_site + + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + + syncer = Sync({"clustering": True}) + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + # Secondary cluster member must be skipped — no host should be created + mock_zabbix.host.create.assert_not_called() + + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_templates_from_config_context(self, mock_api, mock_zabbix_api): + """Test that templates_config_context=True uses the config context template.""" + device = MockNetboxDevice( + device_id=1, + name="Router01", + config_context={ + "zabbix": { + "templates": ["ContextTemplate"], + } + }, + ) + + mock_netbox = self._setup_netbox_mock(mock_api) + mock_netbox.dcim.devices.filter.return_value = [device] + + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + # Both templates exist in Zabbix + mock_zabbix.template.get.return_value = [ + {"templateid": "1", "name": "TestTemplate"}, + {"templateid": "2", "name": "ContextTemplate"}, + ] + + syncer = Sync({"templates_config_context": True}) + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + # Verify host was created with the config context template, not the custom field one + mock_zabbix.host.create.assert_called_once() + create_call_kwargs = mock_zabbix.host.create.call_args.kwargs + self.assertEqual(create_call_kwargs["templates"], [{"templateid": "2"}]) + + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_templates_config_context_overrule(self, mock_api, mock_zabbix_api): + """Test that templates_config_context_overrule=True prefers config context over custom field. + + The device has: + - Custom field template (device type): "TestTemplate" + - Config context template (device): "ContextTemplate" + + With overrule enabled the config context should win and the host should + be created with "ContextTemplate" only. + """ + device = MockNetboxDevice( + device_id=1, + name="Router01", + config_context={ + "zabbix": { + "templates": ["ContextTemplate"], + } + }, + ) + + mock_netbox = self._setup_netbox_mock(mock_api) + mock_netbox.dcim.devices.filter.return_value = [device] + + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + # Both templates exist in Zabbix + mock_zabbix.template.get.return_value = [ + {"templateid": "1", "name": "TestTemplate"}, + {"templateid": "2", "name": "ContextTemplate"}, + ] + + syncer = Sync({"templates_config_context_overrule": True}) + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + # Config context overrides the custom field - only "ContextTemplate" should be used + mock_zabbix.host.create.assert_called_once() + create_call_kwargs = mock_zabbix.host.create.call_args.kwargs + self.assertEqual(create_call_kwargs["templates"], [{"templateid": "2"}]) + # Verify the custom field template was NOT used + self.assertNotIn({"templateid": "1"}, create_call_kwargs["templates"]) + + +class TestDeviceStatusHandling(unittest.TestCase): + """ + Tests device status handling during NetBox to Zabbix synchronization. + + Validates the correct sync behavior for various combinations of NetBox device + status, Zabbix host state, and the 'zabbix_device_removal' / 'zabbix_device_disable' + configuration settings. + + Scenarios: + 1. Active, not in Zabbix → created enabled + 2. Active, already in Zabbix → consistency check passes, no update + 3. Staged, not in Zabbix → created disabled + 4. Staged, already in Zabbix → consistency check passes, no update + 5. Decommissioning, not in Zabbix → skipped entirely + 6. Decommissioning, in Zabbix → host deleted from Zabbix (cleanup) + 7. Active, in Zabbix but disabled → host re-enabled via consistency check + 8. Failed, in Zabbix but enabled → host disabled via consistency check + """ + + # Hostgroup produced by the default "site/manufacturer/role" format + # for the default MockNetboxDevice attributes. + EXPECTED_HOSTGROUP = "TestSite/TestManufacturer/Switch" + + def _setup_netbox_mock(self, mock_api, devices=None): + """Helper to setup a working NetBox mock.""" + mock_netbox = MagicMock() + mock_api.return_value = mock_netbox + mock_netbox.version = "3.5" + mock_netbox.extras.custom_fields.filter.return_value = [] + mock_netbox.dcim.devices.filter.return_value = devices or [] + mock_netbox.virtualization.virtual_machines.filter.return_value = [] + mock_netbox.dcim.site_groups.all.return_value = [] + mock_netbox.dcim.regions.all.return_value = [] + mock_netbox.extras.journal_entries = MagicMock() + return mock_netbox + + def _setup_zabbix_mock(self, mock_zabbix_api, version=7.0): + """Helper to setup a working Zabbix mock.""" + mock_zabbix = MagicMock() + mock_zabbix_api.return_value = mock_zabbix + mock_zabbix.version = version + mock_zabbix.hostgroup.get.return_value = [ + {"groupid": "1", "name": self.EXPECTED_HOSTGROUP} + ] + mock_zabbix.hostgroup.create.return_value = {"groupids": ["2"]} + mock_zabbix.template.get.return_value = [ + {"templateid": "1", "name": "TestTemplate"} + ] + mock_zabbix.proxy.get.return_value = [] + mock_zabbix.proxygroup.get.return_value = [] + mock_zabbix.logout = MagicMock() + mock_zabbix.host.get.return_value = [] + mock_zabbix.host.create.return_value = {"hostids": ["1"]} + mock_zabbix.host.update.return_value = {"hostids": ["42"]} + mock_zabbix.host.delete.return_value = [42] + return mock_zabbix + + def _make_zabbix_host(self, hostname="test-device", status="0"): + """Build a minimal but complete Zabbix host response for consistency_check.""" + return [ + { + "hostid": "42", + "host": hostname, + "name": hostname, + "parentTemplates": [{"templateid": "1"}], + "hostgroups": [{"groupid": "1"}], + "groups": [{"groupid": "1"}], + "status": status, + # Single empty-dict interface: len==1 avoids SyncInventoryError, + # empty keys prevent any spurious interface-update calls. + "interfaces": [{}], + "inventory_mode": "-1", + "inventory": {}, + "macros": [], + "tags": [], + "proxy_hostid": "0", + "proxyid": "0", + "proxy_groupid": "0", + } + ] + + # ------------------------------------------------------------------ + # Scenario 1: Active device, not yet in Zabbix → created enabled (status=0) + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_active_device_not_in_zabbix_is_created(self, mock_api, mock_zabbix_api): + """Active device not yet synced to Zabbix should be created with status enabled (0).""" + device = MockNetboxDevice( + name="test-device", status_label="Active", zabbix_hostid=None + ) + self._setup_netbox_mock(mock_api, devices=[device]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.create.assert_called_once() + create_kwargs = mock_zabbix.host.create.call_args.kwargs + self.assertEqual(create_kwargs["host"], "test-device") + self.assertEqual(create_kwargs["status"], 0) + + # ------------------------------------------------------------------ + # Scenario 2: Active device, already in Zabbix → consistency check, + # Zabbix status matches → no updates + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_active_device_in_zabbix_is_consistent(self, mock_api, mock_zabbix_api): + """Active device already in Zabbix with matching status should require no updates.""" + device = MockNetboxDevice( + name="test-device", status_label="Active", zabbix_hostid=42 + ) + self._setup_netbox_mock(mock_api, devices=[device]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + mock_zabbix.host.get.return_value = self._make_zabbix_host(status="0") + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.create.assert_not_called() + mock_zabbix.host.update.assert_not_called() + + # ------------------------------------------------------------------ + # Scenario 3: Staged device, not yet in Zabbix → created disabled (status=1) + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_staged_device_not_in_zabbix_is_created_disabled( + self, mock_api, mock_zabbix_api + ): + """Staged device not yet in Zabbix should be created with status disabled (1).""" + device = MockNetboxDevice( + name="test-device", status_label="Staged", zabbix_hostid=None + ) + self._setup_netbox_mock(mock_api, devices=[device]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.create.assert_called_once() + create_kwargs = mock_zabbix.host.create.call_args.kwargs + self.assertEqual(create_kwargs["status"], 1) + + # ------------------------------------------------------------------ + # Scenario 4: Staged device, already in Zabbix as disabled → no update needed + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_staged_device_in_zabbix_is_consistent(self, mock_api, mock_zabbix_api): + """Staged device already in Zabbix as disabled should pass consistency check with no updates.""" + device = MockNetboxDevice( + name="test-device", status_label="Staged", zabbix_hostid=42 + ) + self._setup_netbox_mock(mock_api, devices=[device]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + mock_zabbix.host.get.return_value = self._make_zabbix_host(status="1") + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.create.assert_not_called() + mock_zabbix.host.update.assert_not_called() + + # ------------------------------------------------------------------ + # Scenario 5: Decommissioning device, not in Zabbix → skipped (no create, no delete) + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_decommissioning_device_not_in_zabbix_is_skipped( + self, mock_api, mock_zabbix_api + ): + """Decommissioning device with no Zabbix ID should be skipped entirely.""" + device = MockNetboxDevice( + name="test-device", status_label="Decommissioning", zabbix_hostid=None + ) + self._setup_netbox_mock(mock_api, devices=[device]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.create.assert_not_called() + mock_zabbix.host.delete.assert_not_called() + + # ------------------------------------------------------------------ + # Scenario 6: Decommissioning device, already in Zabbix → cleanup (host deleted) + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_decommissioning_device_in_zabbix_is_deleted( + self, mock_api, mock_zabbix_api + ): + """Decommissioning device with a Zabbix ID should be deleted from Zabbix.""" + device = MockNetboxDevice( + name="test-device", status_label="Decommissioning", zabbix_hostid=42 + ) + self._setup_netbox_mock(mock_api, devices=[device]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + # Zabbix still has the host → it should be deleted + mock_zabbix.host.get.return_value = [{"hostid": "42"}] + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.delete.assert_called_once_with(42) + + # ------------------------------------------------------------------ + # Scenario 7: Active device, Zabbix host is disabled → re-enable via consistency check + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_active_device_disabled_in_zabbix_is_enabled( + self, mock_api, mock_zabbix_api + ): + """Active device whose Zabbix host is disabled should be re-enabled by consistency check.""" + device = MockNetboxDevice( + name="test-device", status_label="Active", zabbix_hostid=42 + ) + self._setup_netbox_mock(mock_api, devices=[device]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + # Zabbix host currently disabled; device is Active → status out-of-sync + mock_zabbix.host.get.return_value = self._make_zabbix_host(status="1") + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.update.assert_called_once_with(hostid=42, status="0") + + # ------------------------------------------------------------------ + # Scenario 8: Failed device, Zabbix host is enabled → disable via consistency check + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_failed_device_enabled_in_zabbix_is_disabled( + self, mock_api, mock_zabbix_api + ): + """Failed device whose Zabbix host is enabled should be disabled by consistency check.""" + device = MockNetboxDevice( + name="test-device", status_label="Failed", zabbix_hostid=42 + ) + self._setup_netbox_mock(mock_api, devices=[device]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + # Zabbix host currently enabled; device is Failed → status out-of-sync + mock_zabbix.host.get.return_value = self._make_zabbix_host(status="0") + + syncer = Sync() + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.update.assert_called_once_with(hostid=42, status="1") + + +class TestVMStatusHandling(unittest.TestCase): + """ + Mirrors TestDeviceStatusHandling for VirtualMachine objects. + + Validates the VM sync loop in core.py using real VirtualMachine instances + (not mocked) for the same 8 status scenarios. + """ + + # Hostgroup produced by vm_hostgroup_format "site/role" with default MockNetboxVM values. + EXPECTED_HOSTGROUP = "TestSite/Switch" + + def _setup_netbox_mock(self, mock_api, vms=None): + """Helper to setup a working NetBox mock.""" + mock_netbox = MagicMock() + mock_api.return_value = mock_netbox + mock_netbox.version = "3.5" + mock_netbox.extras.custom_fields.filter.return_value = [] + mock_netbox.dcim.devices.filter.return_value = [] + mock_netbox.virtualization.virtual_machines.filter.return_value = vms or [] + mock_netbox.dcim.site_groups.all.return_value = [] + mock_netbox.dcim.regions.all.return_value = [] + mock_netbox.extras.journal_entries = MagicMock() + return mock_netbox + + def _setup_zabbix_mock(self, mock_zabbix_api, version=7.0): + """Helper to setup a working Zabbix mock.""" + mock_zabbix = MagicMock() + mock_zabbix_api.return_value = mock_zabbix + mock_zabbix.version = version + mock_zabbix.hostgroup.get.return_value = [ + {"groupid": "1", "name": self.EXPECTED_HOSTGROUP} + ] + mock_zabbix.hostgroup.create.return_value = {"groupids": ["2"]} + mock_zabbix.template.get.return_value = [ + {"templateid": "1", "name": "TestTemplate"} + ] + mock_zabbix.proxy.get.return_value = [] + mock_zabbix.proxygroup.get.return_value = [] + mock_zabbix.logout = MagicMock() + mock_zabbix.host.get.return_value = [] + mock_zabbix.host.create.return_value = {"hostids": ["1"]} + mock_zabbix.host.update.return_value = {"hostids": ["42"]} + mock_zabbix.host.delete.return_value = [42] + return mock_zabbix + + def _make_zabbix_host(self, hostname="test-vm", status="0"): + """Build a minimal Zabbix host response for consistency_check.""" + return [ + { + "hostid": "42", + "host": hostname, + "name": hostname, + "parentTemplates": [{"templateid": "1"}], + "hostgroups": [{"groupid": "1"}], + "groups": [{"groupid": "1"}], + "status": status, + # Single empty-dict interface: len==1 avoids SyncInventoryError, + # empty keys mean no spurious interface-update calls. + "interfaces": [{}], + "inventory_mode": "-1", + "inventory": {}, + "macros": [], + "tags": [], + "proxy_hostid": "0", + "proxyid": "0", + "proxy_groupid": "0", + } + ] + + # Simple Sync config that enables VM sync with a flat hostgroup format + _SYNC_CFG: ClassVar[dict] = {"sync_vms": True, "vm_hostgroup_format": "site/role"} + + # ------------------------------------------------------------------ + # Scenario 1: Active VM, not yet in Zabbix → created enabled (status=0) + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_active_vm_not_in_zabbix_is_created(self, mock_api, mock_zabbix_api): + """Active VM not yet synced to Zabbix should be created with status enabled (0).""" + vm = MockNetboxVM(name="test-vm", status_label="Active", zabbix_hostid=None) + self._setup_netbox_mock(mock_api, vms=[vm]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + + syncer = Sync(self._SYNC_CFG) + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.create.assert_called_once() + create_kwargs = mock_zabbix.host.create.call_args.kwargs + self.assertEqual(create_kwargs["host"], "test-vm") + self.assertEqual(create_kwargs["status"], 0) + + # ------------------------------------------------------------------ + # Scenario 2: Active VM, already in Zabbix → consistency check, no update + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_active_vm_in_zabbix_is_consistent(self, mock_api, mock_zabbix_api): + """Active VM already in Zabbix with matching status should require no updates.""" + vm = MockNetboxVM(name="test-vm", status_label="Active", zabbix_hostid=42) + self._setup_netbox_mock(mock_api, vms=[vm]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + mock_zabbix.host.get.return_value = self._make_zabbix_host(status="0") + + syncer = Sync(self._SYNC_CFG) + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.create.assert_not_called() + mock_zabbix.host.update.assert_not_called() + + # ------------------------------------------------------------------ + # Scenario 3: Staged VM, not yet in Zabbix → created disabled (status=1) + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_staged_vm_not_in_zabbix_is_created_disabled( + self, mock_api, mock_zabbix_api + ): + """Staged VM not yet in Zabbix should be created with status disabled (1).""" + vm = MockNetboxVM(name="test-vm", status_label="Staged", zabbix_hostid=None) + self._setup_netbox_mock(mock_api, vms=[vm]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + + syncer = Sync(self._SYNC_CFG) + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.create.assert_called_once() + create_kwargs = mock_zabbix.host.create.call_args.kwargs + self.assertEqual(create_kwargs["status"], 1) + + # ------------------------------------------------------------------ + # Scenario 4: Staged VM, already in Zabbix as disabled → no update needed + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_staged_vm_in_zabbix_is_consistent(self, mock_api, mock_zabbix_api): + """Staged VM already in Zabbix as disabled should pass consistency check with no updates.""" + vm = MockNetboxVM(name="test-vm", status_label="Staged", zabbix_hostid=42) + self._setup_netbox_mock(mock_api, vms=[vm]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + mock_zabbix.host.get.return_value = self._make_zabbix_host(status="1") + + syncer = Sync(self._SYNC_CFG) + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.create.assert_not_called() + mock_zabbix.host.update.assert_not_called() + + # ------------------------------------------------------------------ + # Scenario 5: Decommissioning VM, not in Zabbix → skipped (no create, no delete) + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_decommissioning_vm_not_in_zabbix_is_skipped( + self, mock_api, mock_zabbix_api + ): + """Decommissioning VM with no Zabbix ID should be skipped entirely.""" + vm = MockNetboxVM( + name="test-vm", status_label="Decommissioning", zabbix_hostid=None + ) + self._setup_netbox_mock(mock_api, vms=[vm]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + + syncer = Sync(self._SYNC_CFG) + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.create.assert_not_called() + mock_zabbix.host.delete.assert_not_called() + + # ------------------------------------------------------------------ + # Scenario 6: Decommissioning VM, already in Zabbix → cleanup (host deleted) + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_decommissioning_vm_in_zabbix_is_deleted(self, mock_api, mock_zabbix_api): + """Decommissioning VM with a Zabbix ID should be deleted from Zabbix.""" + vm = MockNetboxVM( + name="test-vm", status_label="Decommissioning", zabbix_hostid=42 + ) + self._setup_netbox_mock(mock_api, vms=[vm]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + mock_zabbix.host.get.return_value = [{"hostid": "42"}] + + syncer = Sync(self._SYNC_CFG) + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.delete.assert_called_once_with(42) + + # ------------------------------------------------------------------ + # Scenario 7: Active VM, Zabbix host is disabled → re-enable via consistency check + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_active_vm_disabled_in_zabbix_is_enabled(self, mock_api, mock_zabbix_api): + """Active VM whose Zabbix host is disabled should be re-enabled by consistency check.""" + vm = MockNetboxVM(name="test-vm", status_label="Active", zabbix_hostid=42) + self._setup_netbox_mock(mock_api, vms=[vm]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + mock_zabbix.host.get.return_value = self._make_zabbix_host(status="1") + + syncer = Sync(self._SYNC_CFG) + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.update.assert_called_once_with(hostid=42, status="0") + + # ------------------------------------------------------------------ + # Scenario 8: Failed VM, Zabbix host is enabled → disable via consistency check + # ------------------------------------------------------------------ + @patch("netbox_zabbix_sync.modules.core.ZabbixAPI") + @patch("netbox_zabbix_sync.modules.core.nbapi") + def test_failed_vm_enabled_in_zabbix_is_disabled(self, mock_api, mock_zabbix_api): + """Failed VM whose Zabbix host is enabled should be disabled by consistency check.""" + vm = MockNetboxVM(name="test-vm", status_label="Failed", zabbix_hostid=42) + self._setup_netbox_mock(mock_api, vms=[vm]) + mock_zabbix = self._setup_zabbix_mock(mock_zabbix_api) + mock_zabbix.host.get.return_value = self._make_zabbix_host(status="0") + + syncer = Sync(self._SYNC_CFG) + syncer.connect( + "http://netbox.local", + "nb_token", + "http://zabbix.local", + "user", + "pass", + None, + ) + syncer.start() + + mock_zabbix.host.update.assert_called_once_with(hostid=42, status="1") diff --git a/tests/test_device_deletion.py b/tests/test_device_deletion.py index 918ab33..2ee154f 100644 --- a/tests/test_device_deletion.py +++ b/tests/test_device_deletion.py @@ -5,8 +5,8 @@ from unittest.mock import MagicMock, patch from zabbix_utils import APIRequestError -from modules.device import PhysicalDevice -from modules.exceptions import SyncExternalError +from netbox_zabbix_sync.modules.device import PhysicalDevice +from netbox_zabbix_sync.modules.exceptions import SyncExternalError class TestDeviceDeletion(unittest.TestCase): @@ -41,15 +41,15 @@ class TestDeviceDeletion(unittest.TestCase): 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, - ) + 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.""" @@ -147,15 +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("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_host_description.py b/tests/test_host_description.py new file mode 100644 index 0000000..a358481 --- /dev/null +++ b/tests/test_host_description.py @@ -0,0 +1,157 @@ +"""Tests for the Description class in the host_description module.""" + +import unittest +from unittest.mock import MagicMock, patch + +from netbox_zabbix_sync.modules.host_description import Description + + +class TestDescription(unittest.TestCase): + """Test class for Description functionality.""" + + def setUp(self): + """Set up test fixtures.""" + # Create mock NetBox object + self.mock_nb_object = MagicMock() + self.mock_nb_object.name = "test-host" + self.mock_nb_object.owner = "admin" + self.mock_nb_object.config_context = {} + + # Create logger mock + self.mock_logger = MagicMock() + + # Base configuration + self.base_config = {} + + # Test 1: Config context description override + @patch("netbox_zabbix_sync.modules.host_description.datetime") + def test_1_config_context_override_value(self, mock_datetime): + """Test 1: User that provides a config context description value should get this override value back.""" + mock_now = MagicMock() + mock_now.strftime.return_value = "2026-02-25 10:30:00" + mock_datetime.now.return_value = mock_now + + # Set config context with description + self.mock_nb_object.config_context = { + "zabbix": {"description": "Custom override for {owner}"} + } + + config = {"description": "static"} + desc = Description(self.mock_nb_object, config, "4.5", logger=self.mock_logger) + + result = desc.generate() + # Should use config context, not config + self.assertEqual(result, "Custom override for admin") + + # Test 2: Static description + def test_2_static_description( + self, + ): + """Test 2: User that provides static as description should get the default static value.""" + config = {"description": "static"} + desc = Description(self.mock_nb_object, config, "4.5", logger=self.mock_logger) + + result = desc.generate() + self.assertEqual(result, "Host added by NetBox sync script.") + + # Test 3: Dynamic description + @patch("netbox_zabbix_sync.modules.host_description.datetime") + def test_3_dynamic_description(self, mock_datetime): + """Test 3: User that provides 'dynamic' should get the resolved description string back.""" + mock_now = MagicMock() + mock_now.strftime.return_value = "2026-02-25 10:30:00" + mock_datetime.now.return_value = mock_now + + config = {"description": "dynamic"} + desc = Description(self.mock_nb_object, config, "4.5", logger=self.mock_logger) + + result = desc.generate() + expected = ( + "Host by owner admin added by NetBox sync script on 2026-02-25 10:30:00." + ) + self.assertEqual(result, expected) + + # Test 4: Invalid macro fallback + def test_4_invalid_macro_fallback_to_static(self): + """Test 4: Users who provide invalid macros should fallback to the static variant.""" + config = {"description": "Host {owner} with {invalid_macro}"} + desc = Description(self.mock_nb_object, config, "4.5", logger=self.mock_logger) + + result = desc.generate() + # Should fall back to static default + self.assertEqual(result, "Host added by NetBox sync script.") + # Verify warning was logged + self.mock_logger.warning.assert_called_once() + + # Test 5: Custom time format + @patch("netbox_zabbix_sync.modules.host_description.datetime") + def test_5_custom_datetime_format(self, mock_datetime): + """Test 5: Users who change the time format.""" + mock_now = MagicMock() + # Will be called twice: once with custom format, once for string + mock_now.strftime.side_effect = ["25/02/2026", "25/02/2026"] + mock_datetime.now.return_value = mock_now + + config = { + "description": "Updated on {datetime}", + "description_dt_format": "%d/%m/%Y", + } + desc = Description(self.mock_nb_object, config, "4.5", logger=self.mock_logger) + + result = desc.generate() + self.assertEqual(result, "Updated on 25/02/2026") + + # Test 6: Custom description format in config + @patch("netbox_zabbix_sync.modules.host_description.datetime") + def test_6_custom_description_format(self, mock_datetime): + """Test 6: Users who provide a custom description format in the config.""" + mock_now = MagicMock() + mock_now.strftime.return_value = "2026-02-25 10:30:00" + mock_datetime.now.return_value = mock_now + + config = {"description": "Server {owner} managed at {datetime}"} + desc = Description(self.mock_nb_object, config, "4.5", logger=self.mock_logger) + + result = desc.generate() + self.assertEqual(result, "Server admin managed at 2026-02-25 10:30:00") + + # Test 7: Owner on lower NetBox version + @patch("netbox_zabbix_sync.modules.host_description.datetime") + def test_7_owner_on_lower_netbox_version(self, mock_datetime): + """Test 7: Users who try to resolve the owner property on a lower NetBox version (3.2).""" + mock_now = MagicMock() + mock_now.strftime.return_value = "2026-02-25 10:30:00" + mock_datetime.now.return_value = mock_now + + config = {"description": "Device owned by {owner}"} + desc = Description( + self.mock_nb_object, + config, + "3.2", # Lower NetBox version + logger=self.mock_logger, + ) + + result = desc.generate() + # Owner should be empty string on version < 4.5 + self.assertEqual(result, "Device owned by ") + + # Test 8: Missing or False description returns static + def test_8a_missing_description_returns_static(self): + """Test 8a: When description option is not found, script should return the static variant.""" + config = {} # No description key + desc = Description(self.mock_nb_object, config, "4.5", logger=self.mock_logger) + + result = desc.generate() + self.assertEqual(result, "Host added by NetBox sync script.") + + def test_8b_false_description_returns_empty(self): + """Test 8b: When description is set to False, script should return empty string.""" + config = {"description": False} + desc = Description(self.mock_nb_object, config, "4.5", logger=self.mock_logger) + + result = desc.generate() + self.assertEqual(result, "") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_hostgroups.py b/tests/test_hostgroups.py index d9c0a69..05e1ad1 100644 --- a/tests/test_hostgroups.py +++ b/tests/test_hostgroups.py @@ -3,8 +3,8 @@ import unittest from unittest.mock import MagicMock, patch -from modules.exceptions import HostgroupError -from modules.hostgroups import Hostgroup +from netbox_zabbix_sync.modules.exceptions import HostgroupError +from netbox_zabbix_sync.modules.hostgroups import Hostgroup class TestHostgroups(unittest.TestCase): @@ -78,8 +78,12 @@ class TestHostgroups(unittest.TestCase): location.__str__.return_value = "TestLocation" self.mock_device.location = location - # Custom fields - self.mock_device.custom_fields = {"test_cf": "TestCF"} + rack = MagicMock() + rack.name = "TestRack" + self.mock_device.rack = rack + + # Custom fields — empty_cf is intentionally None to test the empty CF path + self.mock_device.custom_fields = {"test_cf": "TestCF", "empty_cf": None} # *** Mock NetBox VM setup *** # Create mock VM with all properties @@ -137,6 +141,7 @@ class TestHostgroups(unittest.TestCase): self.assertEqual(hostgroup.format_options["platform"], "TestPlatform") self.assertEqual(hostgroup.format_options["manufacturer"], "TestManufacturer") self.assertEqual(hostgroup.format_options["location"], "TestLocation") + self.assertEqual(hostgroup.format_options["rack"], "TestRack") def test_vm_hostgroup_creation(self): """Test basic VM hostgroup creation.""" @@ -192,16 +197,40 @@ class TestHostgroups(unittest.TestCase): self.assertEqual(complex_result, "TestCluster/TestClusterType/TestPlatform") def test_device_netbox_version_differences(self): - """Test hostgroup generation with different NetBox versions.""" - # NetBox v2.x - hostgroup_v2 = Hostgroup("dev", self.mock_device, "2.11", self.mock_logger) - self.assertEqual(hostgroup_v2.format_options["role"], "TestRole") + """Test hostgroup generation with different NetBox versions. - # NetBox v3.x - hostgroup_v3 = Hostgroup("dev", self.mock_device, "3.5", self.mock_logger) - self.assertEqual(hostgroup_v3.format_options["role"], "TestRole") + device_role (v2/v3) and role (v4+) are set to different values so the + test can verify that the correct attribute is read for each version. + """ + # Build a device with deliberately different names on each role attribute + versioned_device = MagicMock() + versioned_device.name = "versioned-device" + versioned_device.site = self.mock_device.site + versioned_device.tenant = self.mock_device.tenant + versioned_device.platform = self.mock_device.platform + versioned_device.location = self.mock_device.location + versioned_device.rack = self.mock_device.rack + versioned_device.device_type = self.mock_device.device_type + versioned_device.custom_fields = self.mock_device.custom_fields - # NetBox v4.x (already tested in other methods) + old_role = MagicMock() + old_role.name = "OldRole" + new_role = MagicMock() + new_role.name = "NewRole" + versioned_device.device_role = old_role # read by NetBox v2 / v3 code path + versioned_device.role = new_role # read by NetBox v4+ code path + + # v2 must use device_role + hostgroup_v2 = Hostgroup("dev", versioned_device, "2.11", self.mock_logger) + self.assertEqual(hostgroup_v2.format_options["role"], "OldRole") + + # v3 must also use device_role + hostgroup_v3 = Hostgroup("dev", versioned_device, "3.5", self.mock_logger) + self.assertEqual(hostgroup_v3.format_options["role"], "OldRole") + + # v4+ must use role + hostgroup_v4 = Hostgroup("dev", versioned_device, "4.0", self.mock_logger) + self.assertEqual(hostgroup_v4.format_options["role"], "NewRole") def test_custom_field_lookup(self): """Test custom field lookup functionality.""" @@ -217,6 +246,11 @@ class TestHostgroups(unittest.TestCase): self.assertFalse(cf_result["result"]) self.assertIsNone(cf_result["cf"]) + # Test custom field exists but has no value (None) + cf_result = hostgroup.custom_field_lookup("empty_cf") + self.assertTrue(cf_result["result"]) # key is present + self.assertIsNone(cf_result["cf"]) # value is empty + def test_hostgroup_with_custom_field(self): """Test hostgroup generation including a custom field.""" hostgroup = Hostgroup("dev", self.mock_device, "4.0", self.mock_logger) @@ -262,7 +296,9 @@ class TestHostgroups(unittest.TestCase): def test_nested_region_hostgroups(self): """Test hostgroup generation with nested regions.""" # Mock the build_path function to return a predictable result - with patch("modules.hostgroups.build_path") as mock_build_path: + with patch( + "netbox_zabbix_sync.modules.hostgroups.build_path" + ) as mock_build_path: # Configure the mock to return a list of regions in the path mock_build_path.return_value = ["ParentRegion", "TestRegion"] @@ -284,7 +320,9 @@ class TestHostgroups(unittest.TestCase): def test_nested_sitegroup_hostgroups(self): """Test hostgroup generation with nested site groups.""" # Mock the build_path function to return a predictable result - with patch("modules.hostgroups.build_path") as mock_build_path: + with patch( + "netbox_zabbix_sync.modules.hostgroups.build_path" + ) as mock_build_path: # Configure the mock to return a list of site groups in the path mock_build_path.return_value = ["ParentSiteGroup", "TestSiteGroup"] @@ -357,6 +395,69 @@ class TestHostgroups(unittest.TestCase): self.assertEqual(results["platform/location"], "TestPlatform/TestLocation") self.assertEqual(results["tenant_group/tenant"], "TestTenantGroup/TestTenant") + def test_literal_string_in_format(self): + """Test that quoted literal strings in a format are used verbatim.""" + hostgroup = Hostgroup("dev", self.mock_device, "4.0", self.mock_logger) + + # Single-quoted literal + result = hostgroup.generate("'MyDevices'/role") + self.assertEqual(result, "MyDevices/TestRole") + + # Double-quoted literal + result = hostgroup.generate('"MyDevices"/role') + self.assertEqual(result, "MyDevices/TestRole") + + def test_generate_returns_none_when_all_fields_empty(self): + """Test that generate() returns None when every format field resolves to no value.""" + empty_device = MagicMock() + empty_device.name = "empty-device" + empty_device.site = None + empty_device.tenant = None + empty_device.platform = None + empty_device.role = None + empty_device.location = None + empty_device.rack = None + empty_device.custom_fields = {} + device_type = MagicMock() + manufacturer = MagicMock() + manufacturer.name = "SomeManufacturer" + device_type.manufacturer = manufacturer + empty_device.device_type = device_type + + hostgroup = Hostgroup("dev", empty_device, "4.0", self.mock_logger) + # site, tenant and platform all have no value → hg_output stays empty → None + result = hostgroup.generate("site/tenant/platform") + self.assertIsNone(result) + + def test_vm_without_cluster(self): + """Test that cluster/cluster_type are absent from format_options when VM has no cluster.""" + clusterless_vm = MagicMock() + clusterless_vm.name = "clusterless-vm" + clusterless_vm.site = self.mock_vm.site + clusterless_vm.tenant = self.mock_vm.tenant + clusterless_vm.platform = self.mock_vm.platform + clusterless_vm.role = self.mock_device_role + clusterless_vm.cluster = None + clusterless_vm.custom_fields = {} + + hostgroup = Hostgroup("vm", clusterless_vm, "4.0", self.mock_logger) + + # cluster and cluster_type must not appear in format_options + self.assertNotIn("cluster", hostgroup.format_options) + self.assertNotIn("cluster_type", hostgroup.format_options) + + # Requesting cluster in a format must raise HostgroupError + with self.assertRaises(HostgroupError): + hostgroup.generate("cluster/role") + + def test_empty_custom_field_skipped_in_format(self): + """Test that an empty (None) custom field is silently omitted from the hostgroup name.""" + hostgroup = Hostgroup("dev", self.mock_device, "4.0", self.mock_logger) + + # empty_cf has no value → it is skipped; only site and role appear + result = hostgroup.generate("site/empty_cf/role") + self.assertEqual(result, "TestSite/TestRole") + if __name__ == "__main__": unittest.main() diff --git a/tests/test_interface.py b/tests/test_interface.py index ee6b4a7..97131ec 100644 --- a/tests/test_interface.py +++ b/tests/test_interface.py @@ -3,8 +3,8 @@ import unittest from typing import cast -from modules.exceptions import InterfaceConfigError -from modules.interface import ZabbixInterface +from netbox_zabbix_sync.modules.exceptions import InterfaceConfigError +from netbox_zabbix_sync.modules.interface import ZabbixInterface class TestZabbixInterface(unittest.TestCase): diff --git a/tests/test_list_hostgroup_formats.py b/tests/test_list_hostgroup_formats.py index aeaa181..47f3fc5 100644 --- a/tests/test_list_hostgroup_formats.py +++ b/tests/test_list_hostgroup_formats.py @@ -3,9 +3,9 @@ import unittest from unittest.mock import MagicMock -from modules.exceptions import HostgroupError -from modules.hostgroups import Hostgroup -from modules.tools import verify_hg_format +from netbox_zabbix_sync.modules.exceptions import HostgroupError +from netbox_zabbix_sync.modules.hostgroups import Hostgroup +from netbox_zabbix_sync.modules.tools import verify_hg_format class TestListHostgroupFormats(unittest.TestCase): diff --git a/tests/test_physical_device.py b/tests/test_physical_device.py index 2c9843f..133de1d 100644 --- a/tests/test_physical_device.py +++ b/tests/test_physical_device.py @@ -3,8 +3,8 @@ import unittest from unittest.mock import MagicMock, patch -from modules.device import PhysicalDevice -from modules.exceptions import TemplateError +from netbox_zabbix_sync.modules.device import PhysicalDevice +from netbox_zabbix_sync.modules.exceptions import TemplateError class TestPhysicalDevice(unittest.TestCase): @@ -36,9 +36,14 @@ class TestPhysicalDevice(unittest.TestCase): self.mock_logger = MagicMock() # Create PhysicalDevice instance with mocks - with patch( - "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,10 +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("modules.device.search") as mock_search, - patch("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 @@ -88,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 @@ -105,14 +100,14 @@ class TestPhysicalDevice(unittest.TestCase): } # 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, - ) + 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() @@ -124,14 +119,14 @@ class TestPhysicalDevice(unittest.TestCase): 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, - ) + 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() @@ -143,14 +138,14 @@ class TestPhysicalDevice(unittest.TestCase): self.mock_nb_device.config_context = {} # Create device with the updated mock - with patch("modules.device.config", {"device_cf": "zabbix_hostid"}): - 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): @@ -162,14 +157,14 @@ class TestPhysicalDevice(unittest.TestCase): self.mock_nb_device.config_context = {"zabbix": {}} # Create device with the updated mock - with patch("modules.device.config", {"device_cf": "zabbix_hostid"}): - 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): @@ -184,14 +179,14 @@ class TestPhysicalDevice(unittest.TestCase): 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, - ) + 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( @@ -211,23 +206,20 @@ class TestPhysicalDevice(unittest.TestCase): "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, - ) + 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("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.""" @@ -238,22 +230,19 @@ class TestPhysicalDevice(unittest.TestCase): "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, - ) + 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("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.""" @@ -264,22 +253,19 @@ class TestPhysicalDevice(unittest.TestCase): "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, - ) + 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("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.""" @@ -291,28 +277,25 @@ class TestPhysicalDevice(unittest.TestCase): "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, - ) + 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("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.""" @@ -320,14 +303,14 @@ class TestPhysicalDevice(unittest.TestCase): self.mock_nb_device.virtual_chassis = MagicMock() # Create device with the updated mock - with patch("modules.device.config", {"device_cf": "zabbix_hostid"}): - 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()) @@ -338,14 +321,14 @@ class TestPhysicalDevice(unittest.TestCase): self.mock_nb_device.virtual_chassis = None # Create device with the updated mock - with patch("modules.device.config", {"device_cf": "zabbix_hostid"}): - 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_tags.py b/tests/test_tags.py new file mode 100644 index 0000000..3e6c416 --- /dev/null +++ b/tests/test_tags.py @@ -0,0 +1,284 @@ +"""Tests for the ZabbixTags class in the tags module.""" + +import unittest +from unittest.mock import MagicMock + +from netbox_zabbix_sync.modules.tags import ZabbixTags + + +class DummyNBForTags: + """Minimal NetBox object that supports field_mapper's dict-style access.""" + + def __init__(self, name="test-host", config_context=None, tags=None, site=None): + self.name = name + self.config_context = config_context or {} + self.tags = tags or [] + # Stored as a plain dict so field_mapper can traverse "site/name" + self.site = site if site is not None else {"name": "TestSite"} + + def __getitem__(self, key): + return getattr(self, key) + + +class TestZabbixTagsInit(unittest.TestCase): + """Tests for ZabbixTags initialisation.""" + + def test_sync_true_when_tag_sync_enabled(self): + """sync flag should be True when tag_sync=True.""" + nb = DummyNBForTags() + tags = ZabbixTags(nb, tag_map={}, tag_sync=True, logger=MagicMock()) + self.assertTrue(tags.sync) + + def test_sync_false_when_tag_sync_disabled(self): + """sync flag should be False when tag_sync=False (default).""" + nb = DummyNBForTags() + tags = ZabbixTags(nb, tag_map={}, logger=MagicMock()) + self.assertFalse(tags.sync) + + def test_repr_and_str_return_host_name(self): + nb = DummyNBForTags(name="my-host") + tags = ZabbixTags(nb, tag_map={}, host="my-host", logger=MagicMock()) + self.assertEqual(repr(tags), "my-host") + self.assertEqual(str(tags), "my-host") + + +class TestRenderTag(unittest.TestCase): + """Tests for ZabbixTags.render_tag().""" + + def setUp(self): + nb = DummyNBForTags() + self.logger = MagicMock() + self.tags = ZabbixTags( + nb, tag_map={}, tag_sync=True, tag_lower=True, logger=self.logger + ) + + def test_valid_tag_lowercased(self): + """Valid name+value with tag_lower=True should produce lowercase keys.""" + result = self.tags.render_tag("Site", "Production") + self.assertEqual(result, {"tag": "site", "value": "production"}) + + def test_valid_tag_not_lowercased(self): + """tag_lower=False should preserve original case.""" + nb = DummyNBForTags() + tags = ZabbixTags( + nb, tag_map={}, tag_sync=True, tag_lower=False, logger=self.logger + ) + result = tags.render_tag("Site", "Production") + self.assertEqual(result, {"tag": "Site", "value": "Production"}) + + def test_invalid_name_none_returns_false(self): + """None as tag name should return False.""" + result = self.tags.render_tag(None, "somevalue") + self.assertFalse(result) + + def test_invalid_name_too_long_returns_false(self): + """Name exceeding 256 characters should return False.""" + long_name = "x" * 257 + result = self.tags.render_tag(long_name, "somevalue") + self.assertFalse(result) + + def test_invalid_value_none_returns_false(self): + """None as tag value should return False.""" + result = self.tags.render_tag("site", None) + self.assertFalse(result) + + def test_invalid_value_empty_string_returns_false(self): + """Empty string as tag value should return False.""" + result = self.tags.render_tag("site", "") + self.assertFalse(result) + + def test_invalid_value_too_long_returns_false(self): + """Value exceeding 256 characters should return False.""" + long_value = "x" * 257 + result = self.tags.render_tag("site", long_value) + self.assertFalse(result) + + +class TestGenerateFromTagMap(unittest.TestCase): + """Tests for the field_mapper-driven tag generation path.""" + + def setUp(self): + self.logger = MagicMock() + + def test_generate_tag_from_field_map(self): + """Tags derived from tag_map fields are lowercased and returned correctly.""" + nb = DummyNBForTags(name="router01") + # "site/name" → nb["site"]["name"] → "TestSite", mapped to tag name "site" + tag_map = {"site/name": "site"} + tags = ZabbixTags( + nb, + tag_map=tag_map, + tag_sync=True, + tag_lower=True, + logger=self.logger, + ) + result = tags.generate() + self.assertEqual(len(result), 1) + self.assertEqual(result[0]["tag"], "site") + self.assertEqual(result[0]["value"], "testsite") + + def test_generate_empty_field_map_produces_no_tags(self): + """An empty tag_map with no context or NB tags should return an empty list.""" + nb = DummyNBForTags() + tags = ZabbixTags(nb, tag_map={}, tag_sync=True, logger=self.logger) + result = tags.generate() + self.assertEqual(result, []) + + def test_generate_deduplicates_tags(self): + """Duplicate tags produced by the map should be deduplicated.""" + # Two map entries that resolve to the same tag/value pair + nb = DummyNBForTags(name="router01") + tag_map = {"site/name": "site", "site/name": "site"} # noqa: F601 + tags = ZabbixTags( + nb, + tag_map=tag_map, + tag_sync=True, + tag_lower=True, + logger=self.logger, + ) + result = tags.generate() + self.assertEqual(len(result), 1) + + +class TestGenerateFromConfigContext(unittest.TestCase): + """Tests for the config_context-driven tag generation path.""" + + def setUp(self): + self.logger = MagicMock() + + def test_generates_tags_from_config_context(self): + """Tags listed in config_context['zabbix']['tags'] are added correctly.""" + nb = DummyNBForTags( + config_context={ + "zabbix": { + "tags": [ + {"environment": "production"}, + {"location": "DC1"}, + ] + } + } + ) + tags = ZabbixTags( + nb, tag_map={}, tag_sync=True, tag_lower=True, logger=self.logger + ) + result = tags.generate() + self.assertEqual(len(result), 2) + tag_names = [t["tag"] for t in result] + self.assertIn("environment", tag_names) + self.assertIn("location", tag_names) + + def test_skips_config_context_tags_with_invalid_values(self): + """Config context tags with None value should be silently dropped.""" + nb = DummyNBForTags( + config_context={ + "zabbix": { + "tags": [ + {"environment": None}, # invalid value + {"location": "DC1"}, + ] + } + } + ) + tags = ZabbixTags( + nb, tag_map={}, tag_sync=True, tag_lower=True, logger=self.logger + ) + result = tags.generate() + self.assertEqual(len(result), 1) + self.assertEqual(result[0]["tag"], "location") + + def test_ignores_zabbix_tags_key_missing(self): + """Missing 'tags' key inside config_context['zabbix'] produces no tags.""" + nb = DummyNBForTags(config_context={"zabbix": {"templates": ["T1"]}}) + tags = ZabbixTags(nb, tag_map={}, tag_sync=True, logger=self.logger) + result = tags.generate() + self.assertEqual(result, []) + + def test_ignores_config_context_tags_not_a_list(self): + """Non-list value for config_context['zabbix']['tags'] produces no tags.""" + nb = DummyNBForTags(config_context={"zabbix": {"tags": "not-a-list"}}) + tags = ZabbixTags(nb, tag_map={}, tag_sync=True, logger=self.logger) + result = tags.generate() + self.assertEqual(result, []) + + +class TestGenerateFromNetboxTags(unittest.TestCase): + """Tests for the NetBox device tags forwarding path.""" + + def setUp(self): + self.logger = MagicMock() + # Simulate a list of NetBox tag objects (as dicts, matching real API shape) + self.nb_tags = [ + {"name": "ping", "slug": "ping", "display": "ping"}, + {"name": "snmp", "slug": "snmp", "display": "snmp"}, + ] + + def test_generates_tags_from_netbox_tags_using_name(self): + """NetBox device tags are forwarded using tag_name label and tag_value='name'.""" + nb = DummyNBForTags(tags=self.nb_tags) + tags = ZabbixTags( + nb, + tag_map={}, + tag_sync=True, + tag_lower=True, + tag_name="NetBox", + tag_value="name", + logger=self.logger, + ) + result = tags.generate() + self.assertEqual(len(result), 2) + for t in result: + self.assertEqual(t["tag"], "netbox") + values = {t["value"] for t in result} + self.assertIn("ping", values) + self.assertIn("snmp", values) + + def test_generates_tags_from_netbox_tags_using_slug(self): + """tag_value='slug' should use the slug field from each NetBox tag.""" + nb = DummyNBForTags(tags=self.nb_tags) + tags = ZabbixTags( + nb, + tag_map={}, + tag_sync=True, + tag_lower=False, + tag_name="NetBox", + tag_value="slug", + logger=self.logger, + ) + result = tags.generate() + values = {t["value"] for t in result} + self.assertIn("ping", values) + self.assertIn("snmp", values) + + def test_generates_tags_from_netbox_tags_default_value_field(self): + """When tag_value is not a recognised field name, falls back to 'name'.""" + nb = DummyNBForTags(tags=self.nb_tags) + tags = ZabbixTags( + nb, + tag_map={}, + tag_sync=True, + tag_lower=True, + tag_name="NetBox", + tag_value="invalid_field", # not display/name/slug → fall back to "name" + logger=self.logger, + ) + result = tags.generate() + values = {t["value"] for t in result} + self.assertIn("ping", values) + + def test_skips_netbox_tags_when_tag_name_not_set(self): + """NetBox tag forwarding is skipped when tag_name is not configured.""" + nb = DummyNBForTags(tags=self.nb_tags) + tags = ZabbixTags( + nb, + tag_map={}, + tag_sync=True, + tag_lower=True, + tag_name=None, + logger=self.logger, + ) + result = tags.generate() + self.assertEqual(result, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tools.py b/tests/test_tools.py index 5361743..1199cf5 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -1,4 +1,4 @@ -from modules.tools import sanatize_log_output +from netbox_zabbix_sync.modules.tools import sanatize_log_output def test_sanatize_log_output_secrets(): diff --git a/tests/test_usermacros.py b/tests/test_usermacros.py index dbcfaed..82dbf75 100644 --- a/tests/test_usermacros.py +++ b/tests/test_usermacros.py @@ -1,8 +1,8 @@ import unittest from unittest.mock import MagicMock, patch -from modules.device import PhysicalDevice -from modules.usermacros import ZabbixUsermacros +from netbox_zabbix_sync.modules.device import PhysicalDevice +from netbox_zabbix_sync.modules.usermacros import ZabbixUsermacros class DummyNB: @@ -13,12 +13,7 @@ class DummyNB: setattr(self, k, v) def __getitem__(self, key): - # Allow dict-style access for test compatibility - if hasattr(self, key): - return getattr(self, key) - if key in self.config_context: - return self.config_context[key] - raise KeyError(key) + return getattr(self, key) class TestUsermacroSync(unittest.TestCase): @@ -27,7 +22,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 +34,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 +43,21 @@ class TestUsermacroSync(unittest.TestCase): nb_journal_class=MagicMock(), nb_version="3.0", logger=self.logger, + config=device_config, ) return device - @patch( - "modules.device.config", - {"usermacro_sync": False, "device_cf": "zabbix_hostid", "tag_sync": False}, - ) @patch.object(PhysicalDevice, "_usermacro_map") def test_usermacro_sync_false(self, mock_usermacro_map): mock_usermacro_map.return_value = self.usermacro_map - device = self.create_mock_device() + device = self.create_mock_device( + config={ + "usermacro_sync": False, + "device_cf": "zabbix_hostid", + "tag_sync": False, + } + ) # Call set_usermacros result = device.set_usermacros() @@ -65,11 +65,7 @@ class TestUsermacroSync(unittest.TestCase): self.assertEqual(device.usermacros, []) self.assertTrue(result is True or result is None) - @patch( - "modules.device.config", - {"usermacro_sync": True, "device_cf": "zabbix_hostid", "tag_sync": False}, - ) - @patch("modules.device.ZabbixUsermacros") + @patch("netbox_zabbix_sync.modules.device.ZabbixUsermacros") @patch.object(PhysicalDevice, "_usermacro_map") def test_usermacro_sync_true(self, mock_usermacro_map, mock_usermacros_class): mock_usermacro_map.return_value = self.usermacro_map @@ -81,7 +77,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,11 +91,7 @@ class TestUsermacroSync(unittest.TestCase): self.assertIsInstance(device.usermacros, list) self.assertGreater(len(device.usermacros), 0) - @patch( - "modules.device.config", - {"usermacro_sync": "full", "device_cf": "zabbix_hostid", "tag_sync": False}, - ) - @patch("modules.device.ZabbixUsermacros") + @patch("netbox_zabbix_sync.modules.device.ZabbixUsermacros") @patch.object(PhysicalDevice, "_usermacro_map") def test_usermacro_sync_full(self, mock_usermacro_map, mock_usermacros_class): mock_usermacro_map.return_value = self.usermacro_map @@ -105,7 +103,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() diff --git a/uv.lock b/uv.lock index d6885eb..746fcfa 100644 --- a/uv.lock +++ b/uv.lock @@ -191,8 +191,7 @@ wheels = [ [[package]] name = "netbox-zabbix-sync" -version = "3.3.0" -source = { virtual = "." } +source = { editable = "." } dependencies = [ { name = "igraph" }, { name = "pynetbox" },