Fixed small typo in Readme, Updated zabbix-utils, Added Devcontainer, Fixed logging and class description in usermacros module, fixed Zabbix consistencycheck for Usermacros and added unit tests for usermacros.

.devcontainer/devcontainer.json

@@ -0,0 +1,22 @@
+// For format details, see https://aka.ms/devcontainer.json. For config options, see the
+// README at: https://github.com/devcontainers/templates/tree/main/src/python
+{
+	"name": "Python 3",
+	// Or use a Dockerfile or Docker Compose file. More info: https://containers.dev/guide/dockerfile
+	"image": "mcr.microsoft.com/devcontainers/python:1-3.12-bullseye",
+
+	// Features to add to the dev container. More info: https://containers.dev/features.
+	// "features": {},
+
+	// Use 'forwardPorts' to make a list of ports inside the container available locally.
+	// "forwardPorts": [],
+
+	// Use 'postCreateCommand' to run commands after the container is created.
+	"postCreateCommand": "pip3 install --user -r requirements.txt && pip3 install --user pylint pytest"
+
+	// Configure tool-specific properties.
+	// "customizations": {},
+
+	// Uncomment to connect as root instead. More info: https://aka.ms/dev-containers-non-root.
+	// "remoteUser": "root"
+}

modules/device.py

@@ -55,7 +55,7 @@ class PhysicalDevice:
         self.nb_journals = nb_journal_class
         self.inventory_mode = -1
         self.inventory = {}
-        self.usermacros = {}
+        self.usermacros = []
         self.tags = {}
         self.logger = logger if logger else getLogger(__name__)
         self._setBasics()
@@ -400,6 +400,7 @@ class PhysicalDevice:
         )
         if macros.sync is False:
             self.usermacros = []
+            return True
 
         self.usermacros = macros.generate()
         return True
@@ -772,22 +773,31 @@ class PhysicalDevice:
 
         # Check host usermacros
         if config['usermacro_sync']:
-            macros_filtered = []
+            # Make a full copy synce we dont want to lose the original value
-            # Do not re-sync secret usermacros unless sync is set to 'full'
+            # of secret type macros from Netbox
-            if str(config['usermacro_sync']).lower() != "full":
+            netbox_macros = deepcopy(self.usermacros)
-                for m in deepcopy(self.usermacros):
+            # Set the sync bit
-                    if m["type"] == str(1):
+            full_sync_bit = bool(str(config['usermacro_sync']).lower() == "full")
-                        # Remove the value as the api doesn't return it
+            for macro in netbox_macros:
-                        # this will allow us to only update usermacros that don't exist
+                # If the Macro is a secret and full sync is NOT activated
-                        m.pop("value")
+                if macro["type"] == str(1) and not full_sync_bit:
-                    macros_filtered.append(m)
+                    # Remove the value as the Zabbix api does not return the value key
-            if host["macros"] == self.usermacros or host["macros"] == macros_filtered:
+                    # This is required when you want to do a diff between both lists
+                    macro.pop("value")
+            # Sort all lists
+            def filter_with_macros(macro):
+                return macro["macro"]
+            host["macros"].sort(key=filter_with_macros)
+            netbox_macros.sort(key=filter_with_macros)
+            # Check if both lists are the same
+            if host["macros"] == netbox_macros:
                 self.logger.debug(f"Host {self.name}: usermacros in-sync.")
             else:
                 self.logger.warning(f"Host {self.name}: usermacros OUT of sync.")
+                # Update Zabbix with NetBox usermacros
                 self.updateZabbixHost(macros=self.usermacros)
 
-        # Check host usermacros
+        # Check host tags
         if config['tag_sync']:
             if remove_duplicates(host["tags"], sortkey="tag") == self.tags:
                 self.logger.debug(f"Host {self.name}: tags in-sync.")

modules/usermacros.py

@@ -10,7 +10,7 @@ from modules.tools import field_mapper
 
 
 class ZabbixUsermacros:
-    """Class that represents a Zabbix interface."""
+    """Class that represents Zabbix usermacros."""
 
     def __init__(self, nb, usermacro_map, usermacro_sync, logger=None, host=None):
         self.nb = nb
@@ -57,7 +57,8 @@ class ZabbixUsermacros:
             macro["macro"] = str(macro_name)
             if isinstance(macro_properties, dict):
                 if not "value" in macro_properties:
-                    self.logger.warning(f"Usermacro {macro_name} has no value, skipping.")
+                    self.logger.warning(f"Host {self.name}: Usermacro {macro_name} has "
+                                        "no value in Netbox, skipping.")
                     return False
                 macro["value"] = macro_properties["value"]
 
@@ -82,11 +83,12 @@ class ZabbixUsermacros:
                 macro["description"] = ""
 
             else:
-                self.logger.warning(f"Usermacro {macro_name} has no value, skipping.")
+                self.logger.warning(f"Host {self.name}: Usermacro {macro_name} "
+                                    "has no value, skipping.")
                 return False
         else:
             self.logger.error(
-                f"Usermacro {macro_name} is not a valid usermacro name, skipping."
+                f"Host {self.name}: Usermacro {macro_name} is not a valid usermacro name, skipping."
             )
             return False
         return macro

requirements.txt

@@ -1,2 +1,2 @@
-pynetbox
+pynetbox==7.4.1
-zabbix-utils==2.0.1
+zabbix-utils==2.0.2

tests/test_usermacros.py

@@ -0,0 +1,125 @@
+import unittest
+from unittest.mock import MagicMock, patch
+from modules.device import PhysicalDevice
+from modules.usermacros import ZabbixUsermacros
+
+class DummyNB:
+    def __init__(self, name="dummy", config_context=None, **kwargs):
+        self.name = name
+        self.config_context = config_context or {}
+        for k, v in kwargs.items():
+            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)
+
+class TestUsermacroSync(unittest.TestCase):
+    def setUp(self):
+        self.nb = DummyNB(serial="1234")
+        self.logger = MagicMock()
+        self.usermacro_map = {"serial": "{$HW_SERIAL}"}
+
+    @patch("modules.device.config", {"usermacro_sync": False})
+    def test_usermacro_sync_false(self):
+        device = PhysicalDevice.__new__(PhysicalDevice)
+        device.nb = self.nb
+        device.logger = self.logger
+        device.name = "dummy"
+        device._usermacro_map = MagicMock(return_value=self.usermacro_map)
+        # call set_usermacros
+        result = device.set_usermacros()
+        self.assertEqual(device.usermacros, [])
+        self.assertTrue(result is True or result is None)
+
+    @patch("modules.device.config", {"usermacro_sync": True})
+    def test_usermacro_sync_true(self):
+        device = PhysicalDevice.__new__(PhysicalDevice)
+        device.nb = self.nb
+        device.logger = self.logger
+        device.name = "dummy"
+        device._usermacro_map = MagicMock(return_value=self.usermacro_map)
+        result = device.set_usermacros()
+        self.assertIsInstance(device.usermacros, list)
+        self.assertGreater(len(device.usermacros), 0)
+
+    @patch("modules.device.config", {"usermacro_sync": "full"})
+    def test_usermacro_sync_full(self):
+        device = PhysicalDevice.__new__(PhysicalDevice)
+        device.nb = self.nb
+        device.logger = self.logger
+        device.name = "dummy"
+        device._usermacro_map = MagicMock(return_value=self.usermacro_map)
+        result = device.set_usermacros()
+        self.assertIsInstance(device.usermacros, list)
+        self.assertGreater(len(device.usermacros), 0)
+
+class TestZabbixUsermacros(unittest.TestCase):
+    def setUp(self):
+        self.nb = DummyNB()
+        self.logger = MagicMock()
+
+    def test_validate_macro_valid(self):
+        macros = ZabbixUsermacros(self.nb, {}, False, logger=self.logger)
+        self.assertTrue(macros.validate_macro("{$TEST_MACRO}"))
+        self.assertTrue(macros.validate_macro("{$A1_2.3}"))
+        self.assertTrue(macros.validate_macro("{$FOO:bar}"))
+
+    def test_validate_macro_invalid(self):
+        macros = ZabbixUsermacros(self.nb, {}, False, logger=self.logger)
+        self.assertFalse(macros.validate_macro("$TEST_MACRO"))
+        self.assertFalse(macros.validate_macro("{TEST_MACRO}"))
+        self.assertFalse(macros.validate_macro("{$test}"))  # lower-case not allowed
+        self.assertFalse(macros.validate_macro(""))
+
+    def test_render_macro_dict(self):
+        macros = ZabbixUsermacros(self.nb, {}, False, logger=self.logger)
+        macro = macros.render_macro("{$FOO}", {"value": "bar", "type": "secret", "description": "desc"})
+        self.assertEqual(macro["macro"], "{$FOO}")
+        self.assertEqual(macro["value"], "bar")
+        self.assertEqual(macro["type"], "1")
+        self.assertEqual(macro["description"], "desc")
+
+    def test_render_macro_dict_missing_value(self):
+        macros = ZabbixUsermacros(self.nb, {}, False, logger=self.logger)
+        result = macros.render_macro("{$FOO}", {"type": "text"})
+        self.assertFalse(result)
+        self.logger.warning.assert_called()
+
+    def test_render_macro_str(self):
+        macros = ZabbixUsermacros(self.nb, {}, False, logger=self.logger)
+        macro = macros.render_macro("{$FOO}", "bar")
+        self.assertEqual(macro["macro"], "{$FOO}")
+        self.assertEqual(macro["value"], "bar")
+        self.assertEqual(macro["type"], "0")
+        self.assertEqual(macro["description"], "")
+
+    def test_render_macro_invalid_name(self):
+        macros = ZabbixUsermacros(self.nb, {}, False, logger=self.logger)
+        result = macros.render_macro("FOO", "bar")
+        self.assertFalse(result)
+        self.logger.error.assert_called()
+
+    def test_generate_from_map(self):
+        nb = DummyNB(memory="bar", role="baz")
+        usermacro_map = {"memory": "{$FOO}", "role": "{$BAR}"}
+        macros = ZabbixUsermacros(nb, usermacro_map, True, logger=self.logger)
+        result = macros.generate()
+        self.assertEqual(len(result), 2)
+        self.assertEqual(result[0]["macro"], "{$FOO}")
+        self.assertEqual(result[1]["macro"], "{$BAR}")
+
+    def test_generate_from_config_context(self):
+        config_context = {"zabbix": {"usermacros": {"{$FOO}": {"value": "bar"}}}}
+        nb = DummyNB(config_context=config_context)
+        macros = ZabbixUsermacros(nb, {}, True, logger=self.logger)
+        result = macros.generate()
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0]["macro"], "{$FOO}")
+
+if __name__ == "__main__":
+    unittest.main()