Merge pull request #173 from TheNetworkGuy/develop

Package code to main

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
 

netbox_zabbix_sync.py

@@ -1,6 +1,6 @@
-#!/usr/bin/env python3
-
-from netbox_zabbix_sync.modules.cli import parse_cli
-
-if __name__ == "__main__":
-    parse_cli()
+#!/usr/bin/env python3
+
+from netbox_zabbix_sync.modules.cli import parse_cli
+
+if __name__ == "__main__":
+    parse_cli()

netbox_zabbix_sync/__init__.py

@@ -2,4 +2,4 @@
 Makes core module sync function available at package level for easier imports.
 """
 
-from netbox_zabbix_sync.modules.core import sync as sync
+from netbox_zabbix_sync.modules.core import Sync as Sync

netbox_zabbix_sync/modules/cli.py

@@ -2,16 +2,88 @@ import argparse
 import logging
 from os import environ
 
-from netbox_zabbix_sync.modules.core import sync
+from netbox_zabbix_sync.modules.core import Sync
 from netbox_zabbix_sync.modules.exceptions import EnvironmentVarError
-from netbox_zabbix_sync.modules.logging import get_logger, set_log_levels
+from netbox_zabbix_sync.modules.logging import get_logger, set_log_levels, setup_logger
+from netbox_zabbix_sync.modules.settings import load_config
 
-logger = get_logger()
+# Boolean settings that can be toggled via --flag / --no-flag
+_BOOL_ARGS = [
+    ("clustering", "Enable clustering of devices with virtual chassis setup."),
+    ("create_hostgroups", "Enable hostgroup generation (requires Zabbix permissions)."),
+    ("create_journal", "Create NetBox journal entries on changes."),
+    ("sync_vms", "Enable virtual machine sync."),
+    (
+        "full_proxy_sync",
+        "Enable full proxy sync (removes proxies not in config context).",
+    ),
+    (
+        "templates_config_context",
+        "Use config context as the template source instead of a custom field.",
+    ),
+    (
+        "templates_config_context_overrule",
+        "Give config context templates higher priority than custom field templates.",
+    ),
+    ("traverse_regions", "Use the full parent-region path in hostgroup names."),
+    ("traverse_site_groups", "Use the full parent-site-group path in hostgroup names."),
+    (
+        "extended_site_properties",
+        "Fetch additional site info from NetBox (increases API queries).",
+    ),
+    ("inventory_sync", "Sync NetBox device properties to Zabbix inventory."),
+    ("usermacro_sync", "Sync usermacros from NetBox to Zabbix."),
+    ("tag_sync", "Sync host tags to Zabbix."),
+    ("tag_lower", "Lowercase tag names and values before syncing."),
+]
+
+# String settings that can be set via --option VALUE
+_STR_ARGS = [
+    ("template_cf", "NetBox custom field name for the Zabbix template.", "FIELD"),
+    ("device_cf", "NetBox custom field name for the Zabbix host ID.", "FIELD"),
+    (
+        "hostgroup_format",
+        "Hostgroup path pattern for physical devices (e.g. site/manufacturer/role).",
+        "PATTERN",
+    ),
+    (
+        "vm_hostgroup_format",
+        "Hostgroup path pattern for virtual machines (e.g. cluster_type/cluster/role).",
+        "PATTERN",
+    ),
+    (
+        "inventory_mode",
+        "Zabbix inventory mode: disabled, manual, or automatic.",
+        "MODE",
+    ),
+    ("tag_name", "Zabbix tag name used when syncing NetBox tags.", "NAME"),
+    (
+        "tag_value",
+        "NetBox tag property to use as the Zabbix tag value (name, slug, or display).",
+        "PROPERTY",
+    ),
+]
+
+
+def _apply_cli_overrides(config: dict, arguments: argparse.Namespace) -> dict:
+    """Override loaded config with any values explicitly provided on the CLI."""
+    for key, _help in _BOOL_ARGS:
+        cli_val = getattr(arguments, key, None)
+        if cli_val is not None:
+            config[key] = cli_val
+    for key, _help, _meta in _STR_ARGS:
+        cli_val = getattr(arguments, key, None)
+        if cli_val is not None:
+            config[key] = cli_val
+    return config
 
 
 def main(arguments):
     """Run the sync process."""
-    # set environment variables
+    # Set 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:
@@ -46,8 +118,13 @@ def main(arguments):
     netbox_host = environ.get("NETBOX_HOST")
     netbox_token = environ.get("NETBOX_TOKEN")
 
+    # Load config (defaults → config.py → env vars), then apply CLI overrides
+    config = load_config(config_file=arguments.config)
+    config = _apply_cli_overrides(config, arguments)
+
     # Run main sync process
-    sync(
+    syncer = Sync(config=config)
+    syncer.connect(
         nb_host=netbox_host,
         nb_token=netbox_token,
         zbx_host=zabbix_host,
@@ -55,12 +132,18 @@ def main(arguments):
         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="A script to sync Zabbix with NetBox device data."
+        description="Synchronise NetBox device data to Zabbix."
     )
+
+    # ── Verbosity ──────────────────────────────────────────────────────────────
     parser.add_argument(
         "-v", "--verbose", help="Turn on verbose logging.", action="store_true"
     )
@@ -74,5 +157,49 @@ def parse_cli():
         action="store_true",
     )
     parser.add_argument("-q", "--quiet", help="Turn off warnings.", action="store_true")
+    parser.add_argument(
+        "-c",
+        "--config",
+        help="Path to the config file (default: config.py next to the script or in the current directory).",
+        metavar="FILE",
+        default=None,
+    )
+    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)

netbox_zabbix_sync/modules/core.py

@@ -1,18 +1,18 @@
 """Core component of the sync process"""
 
 import ssl
-import sys
 from os import environ
+from typing import Any
 
-from pynetbox import api
-from pynetbox.core.query import RequestError as NBRequestError
+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.config import load_config
 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, setup_logger
+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,
@@ -20,259 +20,386 @@ from netbox_zabbix_sync.modules.tools import (
 )
 from netbox_zabbix_sync.modules.virtual_machine import VirtualMachine
 
-# Import configuration settings
-config = load_config()
-
-
-setup_logger()
 logger = get_logger()
 
 
-def sync(nb_host, nb_token, zbx_host, zbx_user, zbx_pass, zbx_token):
+class Sync:
     """
-    Run the NetBox to Zabbix sync process.
+    Class that hosts the main sync process.
+    This class is used to connect to NetBox and Zabbix and run the sync process.
     """
-    # Set NetBox API
-    netbox = api(nb_host, token=nb_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.",
-            nb_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"])
+    def __init__(self, config: dict[str, Any] | None = None):
+        """
+        Docstring for __init__
 
-        if not zbx_token:
-            zabbix = ZabbixAPI(
-                zbx_host, user=zbx_user, password=zbx_pass, ssl_context=ssl_ctx
+        :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:
-            zabbix = ZabbixAPI(zbx_host, token=zbx_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"])
+            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"
+            )
         )
-    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)
+        # 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,
-                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)
+        # 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
-                # 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
+                vm.set_hostgroup(
+                    self.config["vm_hostgroup_format"],
+                    netbox_site_groups,
+                    netbox_regions,
                 )
-                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
+                # 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)
                     )
-                else:
-                    # Device is secondary in cluster.
-                    # Don't continue with this device.
+                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(
-                        "Device %s: Is part of cluster but not primary. Skipping this host...",
+                        "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
-            # 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)
+                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
-                # 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,
+                # 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
                 )
-                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()
+            except SyncError:
+                pass
+        self.zabbix.logout()
+        return True

netbox_zabbix_sync/modules/device.py

@@ -11,15 +11,16 @@ from typing import Any
 from pynetbox import RequestError as NetboxRequestError
 from zabbix_utils import APIRequestError
 
-from netbox_zabbix_sync.modules.config import load_config
 from netbox_zabbix_sync.modules.exceptions import (
     InterfaceConfigError,
     SyncExternalError,
     SyncInventoryError,
     TemplateError,
 )
+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,
@@ -29,8 +30,6 @@ from netbox_zabbix_sync.modules.tools import (
 )
 from netbox_zabbix_sync.modules.usermacros import ZabbixUsermacros
 
-config = load_config()
-
 
 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(

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"]

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)
@@ -108,8 +114,9 @@ def load_env_variable(config_environvar):
 
 def load_config_file(config_default, config_file="config.py"):
     """Returns config from config.py file"""
+
     # Find the script path and config file next to it.
-    script_dir = path.dirname(path.dirname(path.abspath(__file__)))
+    script_dir = path.dirname(path.dirname(path.dirname(path.abspath(__file__))))
     config_path = Path(path.join(script_dir, config_file))
 
     # If the script directory is not found, try the current working directory

netbox_zabbix_sync/modules/virtual_machine.py

@@ -1,6 +1,5 @@
 """Module that hosts all functions for virtual machine processing"""
 
-from netbox_zabbix_sync.modules.config import load_config
 from netbox_zabbix_sync.modules.device import PhysicalDevice
 from netbox_zabbix_sync.modules.exceptions import (
     InterfaceConfigError,
@@ -9,9 +8,6 @@ from netbox_zabbix_sync.modules.exceptions import (
 )
 from netbox_zabbix_sync.modules.interface import ZabbixInterface
 
-# Load config
-config = load_config()
-
 
 class VirtualMachine(PhysicalDevice):
     """Model for virtual machines"""
@@ -24,15 +20,15 @@ class VirtualMachine(PhysicalDevice):
 
     def _inventory_map(self):
         """use VM inventory maps"""
-        return config["vm_inventory_map"]
+        return self.config["vm_inventory_map"]
 
     def _usermacro_map(self):
         """use VM usermacro maps"""
-        return config["vm_usermacro_map"]
+        return self.config["vm_usermacro_map"]
 
     def _tag_map(self):
         """use VM tag maps"""
-        return config["vm_tag_map"]
+        return self.config["vm_tag_map"]
 
     def set_vm_template(self):
         """Set Template for VMs. Overwrites default class

pyproject.toml

@@ -80,6 +80,8 @@ select = [
 "tests/*" = [
     # Ignore use of assert
     "S101",
+    # Ignore hardcoded passwords / tokens
+    "S106",
 ]
 
 

tests/test_configuration_parsing.py

@@ -3,7 +3,7 @@
 import os
 from unittest.mock import MagicMock, patch
 
-from netbox_zabbix_sync.modules.config import (
+from netbox_zabbix_sync.modules.settings import (
     DEFAULT_CONFIG,
     load_config,
     load_config_file,
@@ -15,10 +15,12 @@ def test_load_config_defaults():
     """Test that load_config returns default values when no config file or env vars are present"""
     with (
         patch(
-            "netbox_zabbix_sync.modules.config.load_config_file",
+            "netbox_zabbix_sync.modules.settings.load_config_file",
             return_value=DEFAULT_CONFIG.copy(),
         ),
-        patch("netbox_zabbix_sync.modules.config.load_env_variable", return_value=None),
+        patch(
+            "netbox_zabbix_sync.modules.settings.load_env_variable", return_value=None
+        ),
     ):
         config = load_config()
         assert config == DEFAULT_CONFIG
@@ -34,10 +36,12 @@ def test_load_config_file():
 
     with (
         patch(
-            "netbox_zabbix_sync.modules.config.load_config_file",
+            "netbox_zabbix_sync.modules.settings.load_config_file",
             return_value=mock_config,
         ),
-        patch("netbox_zabbix_sync.modules.config.load_env_variable", return_value=None),
+        patch(
+            "netbox_zabbix_sync.modules.settings.load_env_variable", return_value=None
+        ),
     ):
         config = load_config()
         assert config["templates_config_context"] is True
@@ -59,11 +63,11 @@ def test_load_env_variables():
 
     with (
         patch(
-            "netbox_zabbix_sync.modules.config.load_config_file",
+            "netbox_zabbix_sync.modules.settings.load_config_file",
             return_value=DEFAULT_CONFIG.copy(),
         ),
         patch(
-            "netbox_zabbix_sync.modules.config.load_env_variable",
+            "netbox_zabbix_sync.modules.settings.load_env_variable",
             side_effect=mock_load_env,
         ),
     ):
@@ -88,11 +92,11 @@ def test_env_vars_override_config_file():
 
     with (
         patch(
-            "netbox_zabbix_sync.modules.config.load_config_file",
+            "netbox_zabbix_sync.modules.settings.load_config_file",
             return_value=mock_config,
         ),
         patch(
-            "netbox_zabbix_sync.modules.config.load_env_variable",
+            "netbox_zabbix_sync.modules.settings.load_env_variable",
             side_effect=mock_load_env,
         ),
     ):

tests/test_device_deletion.py

@@ -41,17 +41,15 @@ class TestDeviceDeletion(unittest.TestCase):
         self.mock_logger = MagicMock()
 
         # Create PhysicalDevice instance with mocks
-        with patch(
-            "netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"}
-        ):
-            self.device = PhysicalDevice(
-                self.mock_nb_device,
-                self.mock_zabbix,
-                self.mock_nb_journal,
-                "3.0",
-                journal=True,
-                logger=self.mock_logger,
-            )
+        self.device = PhysicalDevice(
+            self.mock_nb_device,
+            self.mock_zabbix,
+            self.mock_nb_journal,
+            "3.0",
+            journal=True,
+            logger=self.mock_logger,
+            config={"device_cf": "zabbix_hostid"},
+        )
 
     def test_cleanup_successful_deletion(self):
         """Test successful device deletion from Zabbix."""
@@ -149,17 +147,15 @@ class TestDeviceDeletion(unittest.TestCase):
     def test_create_journal_entry_when_disabled(self):
         """Test create_journal_entry when journaling is disabled."""
         # Setup - create device with journal=False
-        with patch(
-            "netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"}
-        ):
-            device = PhysicalDevice(
-                self.mock_nb_device,
-                self.mock_zabbix,
-                self.mock_nb_journal,
-                "3.0",
-                journal=False,  # Disable journaling
-                logger=self.mock_logger,
-            )
+        device = PhysicalDevice(
+            self.mock_nb_device,
+            self.mock_zabbix,
+            self.mock_nb_journal,
+            "3.0",
+            journal=False,  # Disable journaling
+            logger=self.mock_logger,
+            config={"device_cf": "zabbix_hostid"},
+        )
 
         # Execute
         result = device.create_journal_entry("info", "Test message")

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()

tests/test_hostgroups.py

@@ -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)
@@ -361,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()

tests/test_physical_device.py

@@ -36,9 +36,14 @@ class TestPhysicalDevice(unittest.TestCase):
         self.mock_logger = MagicMock()
 
         # Create PhysicalDevice instance with mocks
-        with patch(
-            "netbox_zabbix_sync.modules.device.config",
-            {
+        self.device = PhysicalDevice(
+            self.mock_nb_device,
+            self.mock_zabbix,
+            self.mock_nb_journal,
+            "3.0",
+            journal=True,
+            logger=self.mock_logger,
+            config={
                 "device_cf": "zabbix_hostid",
                 "template_cf": "zabbix_template",
                 "templates_config_context": False,
@@ -49,15 +54,7 @@ class TestPhysicalDevice(unittest.TestCase):
                 "inventory_sync": False,
                 "device_inventory_map": {},
             },
-        ):
-            self.device = PhysicalDevice(
-                self.mock_nb_device,
-                self.mock_zabbix,
-                self.mock_nb_journal,
-                "3.0",
-                journal=True,
-                logger=self.mock_logger,
-            )
+        )
 
     def test_init(self):
         """Test the initialization of the PhysicalDevice class."""
@@ -75,13 +72,7 @@ class TestPhysicalDevice(unittest.TestCase):
         self.mock_nb_device.name = "test-devïce"
 
         # We need to patch the search function to simulate finding special characters
-        with (
-            patch("netbox_zabbix_sync.modules.device.search") as mock_search,
-            patch(
-                "netbox_zabbix_sync.modules.device.config",
-                {"device_cf": "zabbix_hostid"},
-            ),
-        ):
+        with patch("netbox_zabbix_sync.modules.device.search") as mock_search:
             # Make the search function return True to simulate special characters
             mock_search.return_value = True
 
@@ -91,6 +82,7 @@ class TestPhysicalDevice(unittest.TestCase):
                 self.mock_nb_journal,
                 "3.0",
                 logger=self.mock_logger,
+                config={"device_cf": "zabbix_hostid"},
             )
 
         # With the mocked search function, the name should be changed to NETBOX_ID format
@@ -108,16 +100,14 @@ class TestPhysicalDevice(unittest.TestCase):
         }
 
         # Create device with the updated mock
-        with patch(
-            "netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"}
-        ):
-            device = PhysicalDevice(
-                self.mock_nb_device,
-                self.mock_zabbix,
-                self.mock_nb_journal,
-                "3.0",
-                logger=self.mock_logger,
-            )
+        device = PhysicalDevice(
+            self.mock_nb_device,
+            self.mock_zabbix,
+            self.mock_nb_journal,
+            "3.0",
+            logger=self.mock_logger,
+            config={"device_cf": "zabbix_hostid"},
+        )
 
         # Test that templates are returned correctly
         templates = device.get_templates_context()
@@ -129,16 +119,14 @@ class TestPhysicalDevice(unittest.TestCase):
         self.mock_nb_device.config_context = {"zabbix": {"templates": "Template1"}}
 
         # Create device with the updated mock
-        with patch(
-            "netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"}
-        ):
-            device = PhysicalDevice(
-                self.mock_nb_device,
-                self.mock_zabbix,
-                self.mock_nb_journal,
-                "3.0",
-                logger=self.mock_logger,
-            )
+        device = PhysicalDevice(
+            self.mock_nb_device,
+            self.mock_zabbix,
+            self.mock_nb_journal,
+            "3.0",
+            logger=self.mock_logger,
+            config={"device_cf": "zabbix_hostid"},
+        )
 
         # Test that template is wrapped in a list
         templates = device.get_templates_context()
@@ -150,16 +138,14 @@ class TestPhysicalDevice(unittest.TestCase):
         self.mock_nb_device.config_context = {}
 
         # Create device with the updated mock
-        with patch(
-            "netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"}
-        ):
-            device = PhysicalDevice(
-                self.mock_nb_device,
-                self.mock_zabbix,
-                self.mock_nb_journal,
-                "3.0",
-                logger=self.mock_logger,
-            )
+        device = PhysicalDevice(
+            self.mock_nb_device,
+            self.mock_zabbix,
+            self.mock_nb_journal,
+            "3.0",
+            logger=self.mock_logger,
+            config={"device_cf": "zabbix_hostid"},
+        )
 
         # Test that TemplateError is raised
         with self.assertRaises(TemplateError):
@@ -171,16 +157,14 @@ class TestPhysicalDevice(unittest.TestCase):
         self.mock_nb_device.config_context = {"zabbix": {}}
 
         # Create device with the updated mock
-        with patch(
-            "netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"}
-        ):
-            device = PhysicalDevice(
-                self.mock_nb_device,
-                self.mock_zabbix,
-                self.mock_nb_journal,
-                "3.0",
-                logger=self.mock_logger,
-            )
+        device = PhysicalDevice(
+            self.mock_nb_device,
+            self.mock_zabbix,
+            self.mock_nb_journal,
+            "3.0",
+            logger=self.mock_logger,
+            config={"device_cf": "zabbix_hostid"},
+        )
 
         # Test that TemplateError is raised
         with self.assertRaises(TemplateError):
@@ -195,17 +179,14 @@ class TestPhysicalDevice(unittest.TestCase):
         with patch.object(
             PhysicalDevice, "get_templates_context", return_value=["Template1"]
         ):
-            with patch(
-                "netbox_zabbix_sync.modules.device.config",
-                {"device_cf": "zabbix_hostid"},
-            ):
-                device = PhysicalDevice(
-                    self.mock_nb_device,
-                    self.mock_zabbix,
-                    self.mock_nb_journal,
-                    "3.0",
-                    logger=self.mock_logger,
-                )
+            device = PhysicalDevice(
+                self.mock_nb_device,
+                self.mock_zabbix,
+                self.mock_nb_journal,
+                "3.0",
+                logger=self.mock_logger,
+                config={"device_cf": "zabbix_hostid"},
+            )
 
             # Call set_template with prefer_config_context=True
             result = device.set_template(
@@ -225,23 +206,20 @@ class TestPhysicalDevice(unittest.TestCase):
             "inventory_sync": False,
         }
 
-        with patch("netbox_zabbix_sync.modules.device.config", config_patch):
-            device = PhysicalDevice(
-                self.mock_nb_device,
-                self.mock_zabbix,
-                self.mock_nb_journal,
-                "3.0",
-                logger=self.mock_logger,
-            )
+        device = PhysicalDevice(
+            self.mock_nb_device,
+            self.mock_zabbix,
+            self.mock_nb_journal,
+            "3.0",
+            logger=self.mock_logger,
+            config=config_patch,
+        )
+        result = device.set_inventory({})
 
-            # Call set_inventory with the config patch still active
-            with patch("netbox_zabbix_sync.modules.device.config", config_patch):
-                result = device.set_inventory({})
-
-            # Check result
-            self.assertTrue(result)
-            # Default value for disabled inventory
-            self.assertEqual(device.inventory_mode, -1)
+        # Check result
+        self.assertTrue(result)
+        # Default value for disabled inventory
+        self.assertEqual(device.inventory_mode, -1)
 
     def test_set_inventory_manual_mode(self):
         """Test set_inventory with inventory_mode=manual."""
@@ -252,22 +230,19 @@ class TestPhysicalDevice(unittest.TestCase):
             "inventory_sync": False,
         }
 
-        with patch("netbox_zabbix_sync.modules.device.config", config_patch):
-            device = PhysicalDevice(
-                self.mock_nb_device,
-                self.mock_zabbix,
-                self.mock_nb_journal,
-                "3.0",
-                logger=self.mock_logger,
-            )
+        device = PhysicalDevice(
+            self.mock_nb_device,
+            self.mock_zabbix,
+            self.mock_nb_journal,
+            "3.0",
+            logger=self.mock_logger,
+            config=config_patch,
+        )
+        result = device.set_inventory({})
 
-            # Call set_inventory with the config patch still active
-            with patch("netbox_zabbix_sync.modules.device.config", config_patch):
-                result = device.set_inventory({})
-
-            # Check result
-            self.assertTrue(result)
-            self.assertEqual(device.inventory_mode, 0)  # Manual mode
+        # Check result
+        self.assertTrue(result)
+        self.assertEqual(device.inventory_mode, 0)  # Manual mode
 
     def test_set_inventory_automatic_mode(self):
         """Test set_inventory with inventory_mode=automatic."""
@@ -278,22 +253,19 @@ class TestPhysicalDevice(unittest.TestCase):
             "inventory_sync": False,
         }
 
-        with patch("netbox_zabbix_sync.modules.device.config", config_patch):
-            device = PhysicalDevice(
-                self.mock_nb_device,
-                self.mock_zabbix,
-                self.mock_nb_journal,
-                "3.0",
-                logger=self.mock_logger,
-            )
+        device = PhysicalDevice(
+            self.mock_nb_device,
+            self.mock_zabbix,
+            self.mock_nb_journal,
+            "3.0",
+            logger=self.mock_logger,
+            config=config_patch,
+        )
+        result = device.set_inventory({})
 
-            # Call set_inventory with the config patch still active
-            with patch("netbox_zabbix_sync.modules.device.config", config_patch):
-                result = device.set_inventory({})
-
-            # Check result
-            self.assertTrue(result)
-            self.assertEqual(device.inventory_mode, 1)  # Automatic mode
+        # Check result
+        self.assertTrue(result)
+        self.assertEqual(device.inventory_mode, 1)  # Automatic mode
 
     def test_set_inventory_with_inventory_sync(self):
         """Test set_inventory with inventory_sync=True."""
@@ -305,28 +277,25 @@ class TestPhysicalDevice(unittest.TestCase):
             "device_inventory_map": {"name": "name", "serial": "serialno_a"},
         }
 
-        with patch("netbox_zabbix_sync.modules.device.config", config_patch):
-            device = PhysicalDevice(
-                self.mock_nb_device,
-                self.mock_zabbix,
-                self.mock_nb_journal,
-                "3.0",
-                logger=self.mock_logger,
-            )
+        device = PhysicalDevice(
+            self.mock_nb_device,
+            self.mock_zabbix,
+            self.mock_nb_journal,
+            "3.0",
+            logger=self.mock_logger,
+            config=config_patch,
+        )
 
-            # Create a mock device with the required attributes
-            mock_device_data = {"name": "test-device", "serial": "ABC123"}
+        # Create a mock device with the required attributes
+        mock_device_data = {"name": "test-device", "serial": "ABC123"}
+        result = device.set_inventory(mock_device_data)
 
-            # Call set_inventory with the config patch still active
-            with patch("netbox_zabbix_sync.modules.device.config", config_patch):
-                result = device.set_inventory(mock_device_data)
-
-            # Check result
-            self.assertTrue(result)
-            self.assertEqual(device.inventory_mode, 0)  # Manual mode
-            self.assertEqual(
-                device.inventory, {"name": "test-device", "serialno_a": "ABC123"}
-            )
+        # Check result
+        self.assertTrue(result)
+        self.assertEqual(device.inventory_mode, 0)  # Manual mode
+        self.assertEqual(
+            device.inventory, {"name": "test-device", "serialno_a": "ABC123"}
+        )
 
     def test_iscluster_true(self):
         """Test isCluster when device is part of a cluster."""
@@ -334,16 +303,14 @@ class TestPhysicalDevice(unittest.TestCase):
         self.mock_nb_device.virtual_chassis = MagicMock()
 
         # Create device with the updated mock
-        with patch(
-            "netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"}
-        ):
-            device = PhysicalDevice(
-                self.mock_nb_device,
-                self.mock_zabbix,
-                self.mock_nb_journal,
-                "3.0",
-                logger=self.mock_logger,
-            )
+        device = PhysicalDevice(
+            self.mock_nb_device,
+            self.mock_zabbix,
+            self.mock_nb_journal,
+            "3.0",
+            logger=self.mock_logger,
+            config={"device_cf": "zabbix_hostid"},
+        )
 
         # Check isCluster result
         self.assertTrue(device.is_cluster())
@@ -354,16 +321,14 @@ class TestPhysicalDevice(unittest.TestCase):
         self.mock_nb_device.virtual_chassis = None
 
         # Create device with the updated mock
-        with patch(
-            "netbox_zabbix_sync.modules.device.config", {"device_cf": "zabbix_hostid"}
-        ):
-            device = PhysicalDevice(
-                self.mock_nb_device,
-                self.mock_zabbix,
-                self.mock_nb_journal,
-                "3.0",
-                logger=self.mock_logger,
-            )
+        device = PhysicalDevice(
+            self.mock_nb_device,
+            self.mock_zabbix,
+            self.mock_nb_journal,
+            "3.0",
+            logger=self.mock_logger,
+            config={"device_cf": "zabbix_hostid"},
+        )
 
         # Check isCluster result
         self.assertFalse(device.is_cluster())

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()

tests/test_usermacros.py

@@ -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(
-        "netbox_zabbix_sync.modules.device.config",
-        {"usermacro_sync": False, "device_cf": "zabbix_hostid", "tag_sync": False},
-    )
     @patch.object(PhysicalDevice, "_usermacro_map")
     def test_usermacro_sync_false(self, mock_usermacro_map):
         mock_usermacro_map.return_value = self.usermacro_map
-        device = self.create_mock_device()
+        device = self.create_mock_device(
+            config={
+                "usermacro_sync": False,
+                "device_cf": "zabbix_hostid",
+                "tag_sync": False,
+            }
+        )
 
         # Call set_usermacros
         result = device.set_usermacros()
@@ -65,10 +65,6 @@ class TestUsermacroSync(unittest.TestCase):
         self.assertEqual(device.usermacros, [])
         self.assertTrue(result is True or result is None)
 
-    @patch(
-        "netbox_zabbix_sync.modules.device.config",
-        {"usermacro_sync": True, "device_cf": "zabbix_hostid", "tag_sync": False},
-    )
     @patch("netbox_zabbix_sync.modules.device.ZabbixUsermacros")
     @patch.object(PhysicalDevice, "_usermacro_map")
     def test_usermacro_sync_true(self, mock_usermacro_map, mock_usermacros_class):
@@ -81,7 +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,10 +91,6 @@ class TestUsermacroSync(unittest.TestCase):
         self.assertIsInstance(device.usermacros, list)
         self.assertGreater(len(device.usermacros), 0)
 
-    @patch(
-        "netbox_zabbix_sync.modules.device.config",
-        {"usermacro_sync": "full", "device_cf": "zabbix_hostid", "tag_sync": False},
-    )
     @patch("netbox_zabbix_sync.modules.device.ZabbixUsermacros")
     @patch.object(PhysicalDevice, "_usermacro_map")
     def test_usermacro_sync_full(self, mock_usermacro_map, mock_usermacros_class):
@@ -105,7 +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()