55# See the LICENSE file in the project root for more information.
66
77import sys
8+ import ssl
89import base64
910import unittest
11+ from aiohttp import ClientSession , TCPConnector
1012
1113sys .path .append ('.' )
1214from zabbix_utils .api import ZabbixAPI
@@ -24,9 +26,9 @@ class IntegrationAPITest(unittest.TestCase):
2426 """Test working with a real Zabbix API instance synchronously"""
2527
2628 def setUp (self ):
27- self .url = ZABBIX_URL
2829 self .user = ZABBIX_USER
2930 self .password = ZABBIX_PASSWORD
31+ self .url = ZABBIX_URL + '/http_auth/'
3032 self .api = ZabbixAPI (
3133 url = self .url ,
3234 user = self .user ,
@@ -89,13 +91,76 @@ def test_user_get(self):
8991 self .assertEqual (type (users ), list , "Request user.get was going wrong" )
9092
9193
94+ class CustomCertAPITest (unittest .TestCase ):
95+ """Test working with a real Zabbix API instance synchronously"""
96+
97+ def setUp (self ):
98+ self .user = ZABBIX_USER
99+ self .password = ZABBIX_PASSWORD
100+ self .url = ZABBIX_URL + '/ssl_context/'
101+
102+ context = ssl .create_default_context ()
103+ context .load_verify_locations ('/etc/nginx/ssl/nginx.crt' )
104+
105+ self .api = ZabbixAPI (
106+ url = self .url ,
107+ user = self .user ,
108+ password = self .password ,
109+ skip_version_check = True ,
110+ ssl_context = context
111+ )
112+
113+ def tearDown (self ):
114+ if self .api :
115+ self .api .logout ()
116+
117+ def test_login (self ):
118+ """Tests login function works properly"""
119+
120+ self .assertEqual (
121+ type (self .api ), ZabbixAPI , "Login was going wrong" )
122+ self .assertEqual (
123+ type (self .api .api_version ()), APIVersion , "Version getting was going wrong" )
124+
125+ def test_version_get (self ):
126+ """Tests getting version info works properly"""
127+
128+ version = None
129+ if self .api :
130+ version = self .api .apiinfo .version ()
131+ self .assertEqual (
132+ version , str (self .api .api_version ()), "Request apiinfo.version was going wrong" )
133+
134+ def test_check_auth (self ):
135+ """Tests checking authentication state works properly"""
136+
137+ resp = None
138+ if self .api :
139+ if self .api ._ZabbixAPI__session_id == self .api ._ZabbixAPI__token :
140+ resp = self .api .user .checkAuthentication (token = self .api ._ZabbixAPI__session_id )
141+ else :
142+ resp = self .api .user .checkAuthentication (sessionid = self .api ._ZabbixAPI__session_id )
143+ self .assertEqual (
144+ type (resp ), dict , "Request user.checkAuthentication was going wrong" )
145+
146+ def test_user_get (self ):
147+ """Tests getting users info works properly"""
148+
149+ users = None
150+ if self .api :
151+ users = self .api .user .get (
152+ output = ['userid' , 'name' ]
153+ )
154+ self .assertEqual (type (users ), list , "Request user.get was going wrong" )
155+
156+
92157class IntegrationAsyncAPITest (unittest .IsolatedAsyncioTestCase ):
93158 """Test working with a real Zabbix API instance asynchronously"""
94159
95160 async def asyncSetUp (self ):
96- self .url = ZABBIX_URL
97161 self .user = ZABBIX_USER
98162 self .password = ZABBIX_PASSWORD
163+ self .url = ZABBIX_URL + '/http_auth/'
99164 self .api = AsyncZabbixAPI (
100165 url = self .url ,
101166 skip_version_check = True ,
@@ -163,5 +228,72 @@ async def test_user_get(self):
163228 self .assertEqual (type (users ), list , "Request user.get was going wrong" )
164229
165230
231+ class CustomCertAsyncAPITest (unittest .IsolatedAsyncioTestCase ):
232+ """Test working with a real Zabbix API instance asynchronously"""
233+
234+ async def asyncSetUp (self ):
235+ self .user = ZABBIX_USER
236+ self .password = ZABBIX_PASSWORD
237+ self .url = ZABBIX_URL + '/ssl_context/'
238+
239+ context = ssl .create_default_context ()
240+ context .load_verify_locations ('/etc/nginx/ssl/nginx.crt' )
241+ session = ClientSession (
242+ connector = TCPConnector (ssl = context )
243+ )
244+
245+ self .api = AsyncZabbixAPI (
246+ url = self .url ,
247+ skip_version_check = True ,
248+ client_session = session
249+ )
250+ await self .api .login (
251+ user = self .user ,
252+ password = self .password
253+ )
254+
255+ async def asyncTearDown (self ):
256+ if self .api :
257+ await self .api .logout ()
258+
259+ async def test_login (self ):
260+ """Tests login function works properly"""
261+
262+ self .assertEqual (
263+ type (self .api ), AsyncZabbixAPI , "Login was going wrong" )
264+ self .assertEqual (
265+ type (self .api .api_version ()), APIVersion , "Version getting was going wrong" )
266+
267+ async def test_version_get (self ):
268+ """Tests getting version info works properly"""
269+
270+ version = None
271+ if self .api :
272+ version = await self .api .apiinfo .version ()
273+ self .assertEqual (
274+ version , str (self .api .api_version ()), "Request apiinfo.version was going wrong" )
275+
276+ async def test_check_auth (self ):
277+ """Tests checking authentication state works properly"""
278+
279+ resp = None
280+ if self .api :
281+ if self .api ._AsyncZabbixAPI__session_id == self .api ._AsyncZabbixAPI__token :
282+ resp = await self .api .user .checkAuthentication (token = (self .api ._AsyncZabbixAPI__session_id or '' ))
283+ else :
284+ resp = await self .api .user .checkAuthentication (sessionid = (self .api ._AsyncZabbixAPI__session_id or '' ))
285+ self .assertEqual (
286+ type (resp ), dict , "Request user.checkAuthentication was going wrong" )
287+
288+ async def test_user_get (self ):
289+ """Tests getting users info works properly"""
290+
291+ users = None
292+ if self .api :
293+ users = await self .api .user .get (
294+ output = ['userid' , 'name' ]
295+ )
296+ self .assertEqual (type (users ), list , "Request user.get was going wrong" )
297+
166298if __name__ == '__main__' :
167299 unittest .main ()
0 commit comments