diff --git a/tests/testing.py b/tests/testing.py new file mode 100644 index 0000000..8590d90 --- /dev/null +++ b/tests/testing.py @@ -0,0 +1,152 @@ +import logging +import os +import unittest +from os import getenv + +from zabbixci import ZabbixCI +from zabbixci.settings import Settings + +DEV_ZABBIX_URL = getenv("ZABBIX_URL") +DEV_ZABBIX_TOKEN = getenv("ZABBIX_TOKEN") +DEV_GIT_REMOTE = getenv("REMOTE") + + +class TestPushFunctions(unittest.IsolatedAsyncioTestCase): + def setUp(self): + if os.path.exists(".cache"): + ZabbixCI.cleanup_cache(full=True) + + logging.basicConfig( + level=logging.ERROR, + format="%(asctime)s - %(name)s - %(message)s", + ) + + Settings.ZABBIX_URL = DEV_ZABBIX_URL + Settings.ZABBIX_TOKEN = DEV_ZABBIX_TOKEN + Settings.REMOTE = DEV_GIT_REMOTE + Settings.TEMPLATE_WHITELIST = "Linux by Zabbix agent,Linux by Zabbix 00000,Windows by Zabbix agent,Acronis Cyber Protect Cloud by HTTP,Kubernetes API server by HTTP,Kubernetes cluster state by HTTP" + + self.zci = ZabbixCI() + + async def restoreState(self): + ZabbixCI.cleanup_cache() + + # Restore Zabbix to initial testing state + Settings.PULL_BRANCH = "test" + await self.zci.pull() + + Settings.PULL_BRANCH = "main" + await self.zci.push() + + async def asyncSetUp(self): + await self.zci.create_zabbix() + + await self.restoreState() + + async def test_push_pull_remote_defaults(self): + # Push default Zabbix templates to remote + await self.zci.push() + + # No changed when we have just pushed + changed = await self.zci.pull() + self.assertFalse(changed) + + async def test_template_change(self): + # Rename a template + id = self.zci._zabbix.get_templates_name(["Windows by Zabbix agent"])[0][ + "templateid" + ] + self.zci._zabbix.set_template(id, {"name": "Windows by Zabbix agent (renamed)"}) + + # Push changes to git + changed = await self.zci.push() + self.assertTrue(changed, "Template change not detected") + + # Revert changes in Zabbix + self.zci._zabbix.set_template(id, {"name": "Windows by Zabbix agent"}) + + # Restore to Git version + changed = await self.zci.pull() + self.assertTrue(changed, "Template was not restored") + + # Assert Git version is imported back into Zabbix + matches = self.zci._zabbix.get_templates_filtered( + [Settings.ROOT_TEMPLATE_GROUP], ["Windows by Zabbix agent"] + ) + self.assertEqual(len(matches), 1, "Template not found") + self.assertEqual( + matches[0]["name"], + "Windows by Zabbix agent (renamed)", + "Template name not restored", + ) + + await self.restoreState() + + async def test_template_rename(self): + # Rename a template + id = self.zci._zabbix.get_templates_name(["Linux by Zabbix agent"])[0][ + "templateid" + ] + self.zci._zabbix.set_template(id, {"host": "Linux by Zabbix 00000"}) + + # Push changes to git + changed = await self.zci.push() + self.assertTrue(changed, "Renaming not detected") + + # Make changes in Zabbix + self.zci._zabbix.set_template(id, {"host": "Linux by Zabbix agent"}) + + # Restore to Git version + changed = await self.zci.pull() + self.assertTrue(changed, "Template was not restored") + + # Assert Git version is restored + matches = self.zci._zabbix.get_templates_filtered( + [Settings.ROOT_TEMPLATE_GROUP], ["Linux by Zabbix 00000"] + ) + self.assertEqual(len(matches), 1, "Template not found") + + await self.restoreState() + + async def test_template_delete(self): + # Delete a template + id = self.zci._zabbix.get_templates_name( + ["Acronis Cyber Protect Cloud by HTTP"] + )[0]["templateid"] + self.zci._zabbix.delete_template([id]) + + # Push changes to git + changed = await self.zci.push() + self.assertTrue(changed, "Deletion from Zabbix not detected") + + Settings.PULL_BRANCH = "test" + + changed = await self.zci.pull() + self.assertTrue(changed, "Template was not restored") + + Settings.PULL_BRANCH = "main" + changed = await self.zci.pull() + self.assertTrue(changed, "Template deletion from Git was not detected") + + await self.restoreState() + + async def test_push_to_new_branch(self): + Settings.PUSH_BRANCH = "new-branch" + + # Push default Zabbix templates to remote + await self.zci.push() + + Settings.PUSH_BRANCH = "main" + + await self.restoreState() + + async def asyncTearDown(self): + await self.zci._zabbix.zapi.logout() + + # Close custom session, if it exists + if self.zci._zabbix._client_session: + await self.zci._zabbix._client_session.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/zabbixci/settings.py b/zabbixci/settings.py index b676fb1..aea911f 100644 --- a/zabbixci/settings.py +++ b/zabbixci/settings.py @@ -17,8 +17,8 @@ class Settings: PULL_BRANCH = "main" PUSH_BRANCH = "main" TEMPLATE_PREFIX_PATH = "" - TEMPLATE_WHITELIST = [] - TEMPLATE_BLACKLIST = [] + TEMPLATE_WHITELIST = str + TEMPLATE_BLACKLIST = str CACHE_PATH = "./cache" BATCH_SIZE = 5 IGNORE_TEMPLATE_VERSION = False @@ -33,6 +33,14 @@ class Settings: VENDOR = None SET_VERSION = False + @classmethod + def get_template_whitelist(cls): + return cls.TEMPLATE_WHITELIST.split(",") + + @classmethod + def get_template_blacklist(cls): + return cls.TEMPLATE_BLACKLIST.split(",") + @classmethod def from_env(cls): for key in cls.__dict__.keys(): diff --git a/zabbixci/utils/zabbix/zabbix.py b/zabbixci/utils/zabbix/zabbix.py index ca065d1..11fc358 100644 --- a/zabbixci/utils/zabbix/zabbix.py +++ b/zabbixci/utils/zabbix/zabbix.py @@ -40,6 +40,17 @@ def get_templates(self, template_group_names: list[dict]): "template.get", {"groupids": template_group_ids} )["result"] + def get_templates_filtered( + self, template_group_names: list[dict], filter: list[str] + ): + ids = self._get_template_group(template_group_names) + + template_group_ids = [group["groupid"] for group in ids] + + return self.zapi.send_sync_request( + "template.get", {"groupids": template_group_ids, "filter": {"host": filter}} + )["result"] + def set_template(self, template_id: int, dict: dict): return self.zapi.send_sync_request( "template.update", {"templateid": template_id, **dict} diff --git a/zabbixci/zabbixci.py b/zabbixci/zabbixci.py index 3d8eac3..83f6da1 100644 --- a/zabbixci/zabbixci.py +++ b/zabbixci/zabbixci.py @@ -178,7 +178,7 @@ async def push(self): # Check if there are any changes to commit if not self._git.has_changes and not self._git.ahead_of_remote: self.logger.info("No changes detected") - return + return False self.logger.info("Remote differs from local state, preparing to push") change_amount = len(self._git.status()) @@ -252,6 +252,8 @@ async def push(self): f"Dry run enabled, would have committed {change_amount} new changes to {Settings.REMOTE}:{Settings.PUSH_BRANCH}" ) + return change_amount > 0 + async def pull(self): """ Pull current state from git remote and update Zabbix @@ -418,6 +420,8 @@ async def pull(self): # clean local changes self._git.clean() + return len(templates) > 0 or len(deletion_queue) > 0 + async def zabbix_export(self, templates: list[dict]): batches = [ templates[i : i + Settings.BATCH_SIZE] @@ -466,7 +470,13 @@ async def zabbix_to_file(self) -> list[str]: """ Export Zabbix templates to the cache """ - templates = self._zabbix.get_templates([Settings.ROOT_TEMPLATE_GROUP]) + templates = [] + if Settings.get_template_whitelist(): + templates = self._zabbix.get_templates_filtered( + [Settings.ROOT_TEMPLATE_GROUP], Settings.get_template_whitelist() + ) + else: + templates = self._zabbix.get_templates([Settings.ROOT_TEMPLATE_GROUP]) self.logger.info(f"Found {len(templates)} templates in Zabbix") self.logger.debug(f"Found Zabbix templates: {[t['host'] for t in templates]}") @@ -508,13 +518,13 @@ def ignore_template(cls, template_name: str) -> bool: """ Returns true if template should be ignored because of the blacklist or whitelist """ - if template_name in Settings.TEMPLATE_BLACKLIST: + if template_name in Settings.get_template_blacklist(): cls.logger.debug(f"Skipping blacklisted template {template_name}") return True if ( - len(Settings.TEMPLATE_WHITELIST) - and template_name not in Settings.TEMPLATE_WHITELIST + len(Settings.get_template_whitelist()) + and template_name not in Settings.get_template_whitelist() ): cls.logger.debug(f"Skipping non whitelisted template {template_name}") return True