From 8df17f208ce8494469bc4a079c83f5cd4e7cc788 Mon Sep 17 00:00:00 2001 From: TheNetworkGuy Date: Wed, 11 Jun 2025 20:09:53 +0000 Subject: [PATCH] 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 | 22 ++++++ README.md | 2 +- modules/device.py | 34 ++++++--- modules/usermacros.py | 10 ++- requirements.txt | 4 +- tests/test_usermacros.py | 125 ++++++++++++++++++++++++++++++++ 6 files changed, 178 insertions(+), 19 deletions(-) create mode 100644 .devcontainer/devcontainer.json create mode 100644 tests/test_usermacros.py diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json new file mode 100644 index 0000000..99322f6 --- /dev/null +++ b/.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" +} diff --git a/README.md b/README.md index 62a6673..86bb532 100644 --- a/README.md +++ b/README.md @@ -720,7 +720,7 @@ I would recommend using usermacros for sensitive data such as community strings since the data in NetBox is plain-text. > **_NOTE:_** Not all SNMP data is required for a working configuration. -> [The following parameters are allowed](https://www.zabbix.com/documentation/current/manual/api/reference/hostinterface/object#details_tag "The following parameters are allowed")but +> [The following parameters are allowed](https://www.zabbix.com/documentation/current/manual/api/reference/hostinterface/object#details_tag "The following parameters are allowed") but > are not all required, depending on your environment. diff --git a/modules/device.py b/modules/device.py index 971a5b3..b6903fe 100644 --- a/modules/device.py +++ b/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 = [] - # Do not re-sync secret usermacros unless sync is set to 'full' - if str(config['usermacro_sync']).lower() != "full": - for m in deepcopy(self.usermacros): - if m["type"] == str(1): - # Remove the value as the api doesn't return it - # this will allow us to only update usermacros that don't exist - m.pop("value") - macros_filtered.append(m) - if host["macros"] == self.usermacros or host["macros"] == macros_filtered: + # 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") + 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: + # Remove the value as the Zabbix api does not return the value key + # 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.") diff --git a/modules/usermacros.py b/modules/usermacros.py index c1d783b..6d396c8 100644 --- a/modules/usermacros.py +++ b/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 diff --git a/requirements.txt b/requirements.txt index 8da5ce5..295b59f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,2 @@ -pynetbox -zabbix-utils==2.0.1 \ No newline at end of file +pynetbox==7.4.1 +zabbix-utils==2.0.2 diff --git a/tests/test_usermacros.py b/tests/test_usermacros.py new file mode 100644 index 0000000..28305af --- /dev/null +++ b/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()