✅ Added tests for primary project functions

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

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

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}

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