5
5
# See the LICENSE file in the project root for more information.
6
6
7
7
import sys
8
+ import ssl
8
9
import base64
9
10
import unittest
11
+ from aiohttp import ClientSession , TCPConnector
10
12
11
13
sys .path .append ('.' )
12
14
from zabbix_utils .api import ZabbixAPI
@@ -24,9 +26,9 @@ class IntegrationAPITest(unittest.TestCase):
24
26
"""Test working with a real Zabbix API instance synchronously"""
25
27
26
28
def setUp (self ):
27
- self .url = ZABBIX_URL
28
29
self .user = ZABBIX_USER
29
30
self .password = ZABBIX_PASSWORD
31
+ self .url = ZABBIX_URL + '/http_auth/'
30
32
self .api = ZabbixAPI (
31
33
url = self .url ,
32
34
user = self .user ,
@@ -89,13 +91,76 @@ def test_user_get(self):
89
91
self .assertEqual (type (users ), list , "Request user.get was going wrong" )
90
92
91
93
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
+
92
157
class IntegrationAsyncAPITest (unittest .IsolatedAsyncioTestCase ):
93
158
"""Test working with a real Zabbix API instance asynchronously"""
94
159
95
160
async def asyncSetUp (self ):
96
- self .url = ZABBIX_URL
97
161
self .user = ZABBIX_USER
98
162
self .password = ZABBIX_PASSWORD
163
+ self .url = ZABBIX_URL + '/http_auth/'
99
164
self .api = AsyncZabbixAPI (
100
165
url = self .url ,
101
166
skip_version_check = True ,
@@ -163,5 +228,72 @@ async def test_user_get(self):
163
228
self .assertEqual (type (users ), list , "Request user.get was going wrong" )
164
229
165
230
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
+
166
298
if __name__ == '__main__' :
167
299
unittest .main ()
0 commit comments