Release v2.0.0

Author: aiantsen

.github/scripts/additional_api_tests.py

@@ -9,24 +9,32 @@
 import unittest
 
 sys.path.append('.')
-from zabbix_utils.api import ZabbixAPI, APIVersion
+from zabbix_utils.api import ZabbixAPI
+from zabbix_utils.types import APIVersion
+from zabbix_utils.aioapi import AsyncZabbixAPI
+
+ZABBIX_URL = 'https://127.0.0.1:443'
+ZABBIX_USER = 'Admin'
+ZABBIX_PASSWORD = 'zabbix'
+HTTP_USER = 'http_user'
+HTTP_PASSWORD = 'http_pass'
 
 
 class IntegrationAPITest(unittest.TestCase):
-    """Test working with a real Zabbix API instance"""
+    """Test working with a real Zabbix API instance synchronously"""
 
     def setUp(self):
-        self.url = 'https://127.0.0.1:443'
-        self.user = 'Admin'
-        self.password = 'zabbix'
+        self.url = ZABBIX_URL
+        self.user = ZABBIX_USER
+        self.password = ZABBIX_PASSWORD
         self.api = ZabbixAPI(
             url=self.url,
             user=self.user,
             password=self.password,
             skip_version_check=True,
             validate_certs=False,
-            http_user='http_user',
-            http_password='http_pass'
+            http_user=HTTP_USER,
+            http_password=HTTP_PASSWORD
         )
 
     def tearDown(self):
@@ -81,5 +89,79 @@ def test_user_get(self):
         self.assertEqual(type(users), list, "Request user.get was going wrong")
 
 
+class IntegrationAsyncAPITest(unittest.IsolatedAsyncioTestCase):
+    """Test working with a real Zabbix API instance asynchronously"""
+
+    async def asyncSetUp(self):
+        self.url = ZABBIX_URL
+        self.user = ZABBIX_USER
+        self.password = ZABBIX_PASSWORD
+        self.api = AsyncZabbixAPI(
+            url=self.url,
+            skip_version_check=True,
+            validate_certs=False,
+            http_user=HTTP_USER,
+            http_password=HTTP_PASSWORD
+        )
+        await self.api.login(
+            user=self.user,
+            password=self.password
+        )
+
+    async def asyncTearDown(self):
+        if self.api:
+            await self.api.logout()
+
+    async def test_login(self):
+        """Tests login function works properly"""
+
+        self.assertEqual(
+            type(self.api), AsyncZabbixAPI, "Login was going wrong")
+        self.assertEqual(
+            type(self.api.api_version()), APIVersion, "Version getting was going wrong")
+
+    async def test_basic_auth(self):
+        """Tests __basic_auth function works properly"""
+
+        basic_auth = self.api.client_session._default_auth
+
+        self.assertEqual(
+            base64.b64encode(f"{basic_auth.login}:{basic_auth.password}".encode()).decode(),
+            base64.b64encode(f"{HTTP_USER}:{HTTP_PASSWORD}".encode()).decode(),
+            "Basic auth credentials generation was going wrong"
+        )
+
+    async def test_version_get(self):
+        """Tests getting version info works properly"""
+
+        version = None
+        if self.api:
+            version = await self.api.apiinfo.version()
+        self.assertEqual(
+            version, str(self.api.api_version()), "Request apiinfo.version was going wrong")
+
+    async def test_check_auth(self):
+        """Tests checking authentication state works properly"""
+
+        resp = None
+        if self.api:
+            if self.api._AsyncZabbixAPI__session_id == self.api._AsyncZabbixAPI__token:
+                resp = await self.api.user.checkAuthentication(token=(self.api._AsyncZabbixAPI__session_id or ''))
+            else:
+                resp = await self.api.user.checkAuthentication(sessionid=(self.api._AsyncZabbixAPI__session_id or ''))
+        self.assertEqual(
+            type(resp), dict, "Request user.checkAuthentication was going wrong")
+
+    async def test_user_get(self):
+        """Tests getting users info works properly"""
+
+        users = None
+        if self.api:
+            users = await self.api.user.get(
+                output=['userid', 'name']
+            )
+        self.assertEqual(type(users), list, "Request user.get was going wrong")
+
+
 if __name__ == '__main__':
     unittest.main()

.github/scripts/compatibility_api_test_5.py

@@ -9,23 +9,27 @@
 import unittest
 
 sys.path.append('.')
-from zabbix_utils.getter import Getter, AgentResponse
-from zabbix_utils.api import ZabbixAPI, APIVersion
-from zabbix_utils.sender import ItemValue, Sender, TrapperResponse
+from zabbix_utils.api import ZabbixAPI
+from zabbix_utils.sender import Sender
+from zabbix_utils.getter import Getter
+from zabbix_utils.aioapi import AsyncZabbixAPI
+from zabbix_utils.aiosender import AsyncSender
+from zabbix_utils.aiogetter import AsyncGetter
 from zabbix_utils.exceptions import APIRequestError, APINotSupported
+from zabbix_utils.types import AgentResponse, ItemValue, TrapperResponse, APIVersion
 
-ZABBIX_URL = 'localhost'
+ZABBIX_URL = '127.0.0.1'
 ZABBIX_USER = 'Admin'
 ZABBIX_PASSWORD = 'zabbix'
 
 
 class CompatibilityAPITest(unittest.TestCase):
-    """Compatibility test with Zabbix API version 5.0"""
+    """Compatibility synchronous test with Zabbix API version 5.0"""
 
     def setUp(self):
-        self.url = 'localhost'
-        self.user = 'Admin'
-        self.password = 'zabbix'
+        self.url = ZABBIX_URL
+        self.user = ZABBIX_USER
+        self.password = ZABBIX_PASSWORD
         self.token = 'token'
         self.zapi = ZabbixAPI(
             url=self.url
@@ -63,7 +67,7 @@ def test_classic_auth(self):
 
         with self.assertRaises(APIRequestError,
                                msg="Request user.checkAuthentication after logout was going wrong"):
-            resp = self.zapi.user.checkAuthentication(sessionid=self.zapi._ZabbixAPI__session_id)
+            resp = self.zapi.user.checkAuthentication(sessionid=(self.zapi._ZabbixAPI__session_id or ''))
 
     def test_token_auth(self):
         """Tests auth using token"""
@@ -74,10 +78,10 @@ def test_token_auth(self):
 
 
 class CompatibilitySenderTest(unittest.TestCase):
-    """Compatibility test with Zabbix sender version 5.0"""
+    """Compatibility synchronous test with Zabbix sender version 5.0"""
 
     def setUp(self):
-        self.ip = '127.0.0.1'
+        self.ip = ZABBIX_URL
         self.port = 10051
         self.chunk_size = 10
         self.sender = Sender(
@@ -143,7 +147,7 @@ def prepare_items(self):
                 value_type=3
             )['itemids'][0]
 
-            time.sleep(2)
+        time.sleep(2)
 
         self.assertIsNotNone(hostid, "Creating test item was going wrong")
 
@@ -174,10 +178,10 @@ def test_send_values(self):
 
 
 class CompatibilityGetTest(unittest.TestCase):
-    """Compatibility test with Zabbix get version 5.0"""
+    """Compatibility synchronous test with Zabbix get version 5.0"""
 
     def setUp(self):
-        self.host = 'localhost'
+        self.host = ZABBIX_URL
         self.port = 10050
         self.agent = Getter(
             host=self.host,
@@ -194,5 +198,188 @@ def test_get_values(self):
         self.assertEqual(type(resp.value), str, "Got value is unexpected")
 
 
+class CompatibilityAsyncAPITest(unittest.IsolatedAsyncioTestCase):
+    """Compatibility asynchronous test with Zabbix API version 5.0"""
+
+    async def asyncSetUp(self):
+        self.url = ZABBIX_URL
+        self.user = ZABBIX_USER
+        self.password = ZABBIX_PASSWORD
+        self.token = 'token'
+        self.zapi = AsyncZabbixAPI(
+            url=self.url
+        )
+
+    async def asyncTearDown(self):
+        if self.zapi:
+            await self.zapi.logout()
+
+    async def test_classic_auth(self):
+        """Tests auth using username and password"""
+
+        self.assertEqual(
+            type(self.zapi), AsyncZabbixAPI, "Creating AsyncZabbixAPI object was going wrong")
+
+        self.assertEqual(
+            type(self.zapi.api_version()), APIVersion, "Version getting was going wrong")
+
+        await self.zapi.login(
+            user=self.user,
+            password=self.password
+        )
+
+        self.assertIsNotNone(self.zapi._AsyncZabbixAPI__session_id, "Login by user and password was going wrong")
+
+        resp = await self.zapi.user.checkAuthentication(sessionid=self.zapi._AsyncZabbixAPI__session_id)
+
+        self.assertEqual(
+            type(resp), dict, "Request user.checkAuthentication was going wrong")
+
+        users = await self.zapi.user.get(
+            output=['userid', 'name']
+        )
+        self.assertEqual(type(users), list, "Request user.get was going wrong")
+
+        await self.zapi.logout()
+
+        self.assertIsNone(self.zapi._AsyncZabbixAPI__session_id, "Logout was going wrong")
+
+        with self.assertRaises(RuntimeError,
+                               msg="Request user.checkAuthentication after logout was going wrong"):
+            resp = await self.zapi.user.checkAuthentication(sessionid=(self.zapi._AsyncZabbixAPI__session_id or ''))
+
+    async def test_token_auth(self):
+        """Tests auth using token"""
+
+        with self.assertRaises(APINotSupported,
+                               msg="Login by token should be not supported"):
+            await self.zapi.login(token=self.token)
+
+
+class CompatibilityAsyncSenderTest(unittest.IsolatedAsyncioTestCase):
+    """Compatibility asynchronous test with Zabbix sender version 5.0"""
+
+    async def asyncSetUp(self):
+        self.ip = ZABBIX_URL
+        self.port = 10051
+        self.chunk_size = 10
+        self.sender = AsyncSender(
+            server=self.ip,
+            port=self.port,
+            chunk_size=self.chunk_size
+        )
+        self.hostname = f"{self.__class__.__name__}_host"
+        self.itemname = f"{self.__class__.__name__}_item"
+        self.itemkey = f"{self.__class__.__name__}"
+        await self.prepare_items()
+
+    async def prepare_items(self):
+        """Creates host and items for sending values later"""
+
+        zapi = AsyncZabbixAPI(
+            url=ZABBIX_URL,
+            skip_version_check=True
+        )
+        await zapi.login(
+            user=ZABBIX_USER,
+            password=ZABBIX_PASSWORD
+        )
+
+        hosts = await zapi.host.get(
+            filter={'host': self.hostname},
+            output=['hostid']
+        )
+
+        hostid = None
+        if len(hosts) > 0:
+            hostid = hosts[0].get('hostid')
+
+        if not hostid:
+            created_host = await zapi.host.create(
+                host=self.hostname,
+                interfaces=[{
+                    "type": 1,
+                    "main": 1,
+                    "useip": 1,
+                    "ip": "127.0.0.1",
+                    "dns": "",
+                    "port": "10050"
+                }],
+                groups=[{"groupid": "2"}]
+            )
+            hostid = created_host['hostids'][0]
+
+        self.assertIsNotNone(hostid, "Creating test host was going wrong")
+
+        items = await zapi.item.get(
+            filter={'key_': self.itemkey},
+            output=['itemid']
+        )
+
+        itemid = None
+        if len(items) > 0:
+            itemid = items[0].get('itemid')
+
+        if not itemid:
+            created_item = await zapi.item.create(
+                name=self.itemname,
+                key_=self.itemkey,
+                hostid=hostid,
+                type=2,
+                value_type=3
+            )
+            itemid = created_item['itemids'][0]
+
+        self.assertIsNotNone(hostid, "Creating test item was going wrong")
+
+        await zapi.logout()
+
+    async def test_send_values(self):
+        """Tests sending item values"""
+
+        time.sleep(2)
+
+        items = [
+            ItemValue(self.hostname, self.itemkey, 10),
+            ItemValue(self.hostname, self.itemkey, 'test message'),
+            ItemValue(self.hostname, 'item_key1', -1, 1695713666),
+            ItemValue(self.hostname, 'item_key2', '{"msg":"test message"}'),
+            ItemValue(self.hostname, self.itemkey, 0, 1695713666, 100),
+            ItemValue(self.hostname, self.itemkey, 5.5, 1695713666)
+        ]
+        resp = await self.sender.send(items)
+        self.assertEqual(type(resp), TrapperResponse, "Sending item values was going wrong")
+        self.assertEqual(resp.total, len(items), "Total number of the sent values is unexpected")
+        self.assertEqual(resp.processed, 4, "Number of the processed values is unexpected")
+        self.assertEqual(resp.failed, (resp.total - resp.processed), "Number of the failed values is unexpected")
+
+        first_chunk = list(resp.details.values())[0][0]
+        self.assertEqual(type(first_chunk), TrapperResponse, "Sending item values was going wrong")
+        self.assertEqual(first_chunk.total, len(items), "Total number of the sent values is unexpected")
+        self.assertEqual(first_chunk.processed, 4, "Number of the processed values is unexpected")
+        self.assertEqual(first_chunk.failed, (first_chunk.total - first_chunk.processed), "Number of the failed values is unexpected")
+
+
+class CompatibilityAsyncGetTest(unittest.IsolatedAsyncioTestCase):
+    """Compatibility asynchronous test with Zabbix get version 5.0"""
+
+    async def asyncSetUp(self):
+        self.host = ZABBIX_URL
+        self.port = 10050
+        self.agent = AsyncGetter(
+            host=self.host,
+            port=self.port
+        )
+
+    async def test_get_values(self):
+        """Tests getting item values"""
+
+        resp = await self.agent.get('system.uname')
+
+        self.assertIsNotNone(resp, "Getting item values was going wrong")
+        self.assertEqual(type(resp), AgentResponse, "Got value is unexpected")
+        self.assertEqual(type(resp.value), str, "Got value is unexpected")
+
+
 if __name__ == '__main__':
     unittest.main()