1 | #!/usr/bin/env python |
||
2 | # |
||
3 | # Glances - An eye on your system |
||
4 | # |
||
5 | # SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]> |
||
6 | # |
||
7 | # SPDX-License-Identifier: LGPL-3.0-only |
||
8 | # |
||
9 | |||
10 | """Glances unitary tests suite for the RESTful API.""" |
||
11 | |||
12 | import numbers |
||
13 | import os |
||
14 | import shlex |
||
15 | import subprocess |
||
16 | import time |
||
17 | import unittest |
||
18 | |||
19 | import requests |
||
20 | |||
21 | from glances import __version__ |
||
22 | from glances.globals import text_type |
||
23 | from glances.outputs.glances_restful_api import GlancesRestfulApi |
||
24 | |||
25 | SERVER_PORT = 61234 |
||
26 | API_VERSION = GlancesRestfulApi.API_VERSION |
||
27 | URL = f"http://localhost:{SERVER_PORT}/api/{API_VERSION}" |
||
28 | pid = None |
||
29 | |||
30 | # Unitest class |
||
31 | # ============== |
||
32 | print(f'RESTful API unitary tests for Glances {__version__}') |
||
33 | |||
34 | |||
35 | class TestGlances(unittest.TestCase): |
||
36 | """Test Glances class.""" |
||
37 | |||
38 | def setUp(self): |
||
39 | """The function is called *every time* before test_*.""" |
||
40 | print('\n' + '=' * 78) |
||
41 | |||
42 | def http_get(self, url, gzip=False): |
||
43 | """Make the request""" |
||
44 | if gzip: |
||
45 | ret = requests.get(url, stream=True, headers={'Accept-encoding': 'gzip'}) |
||
46 | else: |
||
47 | ret = requests.get(url, headers={'Accept-encoding': 'identity'}) |
||
48 | return ret |
||
49 | |||
50 | View Code Duplication | def test_000_start_server(self): |
|
0 ignored issues
–
show
Duplication
introduced
by
![]() |
|||
51 | """Start the Glances Web Server.""" |
||
52 | global pid |
||
53 | |||
54 | print('INFO: [TEST_000] Start the Glances Web Server API') |
||
55 | if os.path.isfile('./venv/bin/python'): |
||
56 | cmdline = "./venv/bin/python" |
||
57 | else: |
||
58 | cmdline = "python" |
||
59 | cmdline += f" -m glances -B 0.0.0.0 -w --browser -p {SERVER_PORT} --disable-webui -C ./conf/glances.conf" |
||
60 | print(f"Run the Glances Web Server on port {SERVER_PORT}") |
||
61 | args = shlex.split(cmdline) |
||
62 | pid = subprocess.Popen(args) |
||
63 | print("Please wait 5 seconds...") |
||
64 | time.sleep(5) |
||
65 | |||
66 | self.assertTrue(pid is not None) |
||
67 | |||
68 | def test_001_all(self): |
||
69 | """All.""" |
||
70 | method = "all" |
||
71 | print('INFO: [TEST_001] Get all stats') |
||
72 | print(f"HTTP RESTful request: {URL}/{method}") |
||
73 | req = self.http_get(f"{URL}/{method}") |
||
74 | |||
75 | self.assertTrue(req.ok) |
||
76 | self.assertTrue(req.json(), dict) |
||
77 | |||
78 | def test_002_pluginslist(self): |
||
79 | """Plugins list.""" |
||
80 | method = "pluginslist" |
||
81 | print('INFO: [TEST_002] Plugins list') |
||
82 | print(f"HTTP RESTful request: {URL}/{method}") |
||
83 | req = self.http_get(f"{URL}/{method}") |
||
84 | |||
85 | self.assertTrue(req.ok) |
||
86 | self.assertIsInstance(req.json(), list) |
||
87 | self.assertIn('cpu', req.json()) |
||
88 | |||
89 | def test_003_plugins(self): |
||
90 | """Plugins.""" |
||
91 | method = "pluginslist" |
||
92 | print('INFO: [TEST_003] Plugins') |
||
93 | plist = self.http_get(f"{URL}/{method}") |
||
94 | |||
95 | for p in plist.json(): |
||
96 | print(f"HTTP RESTful request: {URL}/{p}") |
||
97 | req = self.http_get(f"{URL}/{p}") |
||
98 | self.assertTrue(req.ok) |
||
99 | if p in ('uptime', 'version', 'psutilversion'): |
||
100 | self.assertIsInstance(req.json(), text_type) |
||
101 | elif p in ( |
||
102 | 'fs', |
||
103 | 'percpu', |
||
104 | 'sensors', |
||
105 | 'alert', |
||
106 | 'processlist', |
||
107 | 'programlist', |
||
108 | 'diskio', |
||
109 | 'hddtemp', |
||
110 | 'batpercent', |
||
111 | 'network', |
||
112 | 'folders', |
||
113 | 'amps', |
||
114 | 'ports', |
||
115 | 'irq', |
||
116 | 'wifi', |
||
117 | 'gpu', |
||
118 | 'containers', |
||
119 | 'vms', |
||
120 | ): |
||
121 | self.assertIsInstance(req.json(), list) |
||
122 | if len(req.json()) > 0: |
||
123 | self.assertIsInstance(req.json()[0], dict) |
||
124 | else: |
||
125 | self.assertIsInstance(req.json(), dict) |
||
126 | |||
127 | def test_004_items(self): |
||
128 | """Items.""" |
||
129 | method = "cpu" |
||
130 | print('INFO: [TEST_004] Items for the CPU method') |
||
131 | ilist = self.http_get(f"{URL}/{method}") |
||
132 | |||
133 | for i in ilist.json(): |
||
134 | print(f"HTTP RESTful request: {URL}/{method}/{i}") |
||
135 | req = self.http_get(f"{URL}/{method}/{i}") |
||
136 | self.assertTrue(req.ok) |
||
137 | self.assertIsInstance(req.json(), dict) |
||
138 | print(req.json()[i]) |
||
139 | self.assertIsInstance(req.json()[i], numbers.Number) |
||
140 | |||
141 | def test_005_values(self): |
||
142 | """Values.""" |
||
143 | method = "processlist" |
||
144 | print('INFO: [TEST_005] Item=Value for the PROCESSLIST method') |
||
145 | req = self.http_get(f"{URL}/{method}/pid/value/0") |
||
146 | |||
147 | self.assertTrue(req.ok) |
||
148 | self.assertIsInstance(req.json(), dict) |
||
149 | |||
150 | def test_006_all_limits(self): |
||
151 | """All limits.""" |
||
152 | method = "all/limits" |
||
153 | print('INFO: [TEST_006] Get all limits') |
||
154 | print(f"HTTP RESTful request: {URL}/{method}") |
||
155 | req = self.http_get(f"{URL}/{method}") |
||
156 | |||
157 | self.assertTrue(req.ok) |
||
158 | self.assertIsInstance(req.json(), dict) |
||
159 | |||
160 | def test_007_all_views(self): |
||
161 | """All views.""" |
||
162 | method = "all/views" |
||
163 | print('INFO: [TEST_007] Get all views') |
||
164 | print(f"HTTP RESTful request: {URL}/{method}") |
||
165 | req = self.http_get(f"{URL}/{method}") |
||
166 | |||
167 | self.assertTrue(req.ok) |
||
168 | self.assertIsInstance(req.json(), dict) |
||
169 | |||
170 | def test_008_plugins_limits(self): |
||
171 | """Plugins limits.""" |
||
172 | method = "pluginslist" |
||
173 | print('INFO: [TEST_008] Plugins limits') |
||
174 | plist = self.http_get(f"{URL}/{method}") |
||
175 | |||
176 | for p in plist.json(): |
||
177 | print(f"HTTP RESTful request: {URL}/{p}/limits") |
||
178 | req = self.http_get(f"{URL}/{p}/limits") |
||
179 | self.assertTrue(req.ok) |
||
180 | self.assertIsInstance(req.json(), dict) |
||
181 | |||
182 | def test_009_plugins_views(self): |
||
183 | """Plugins views.""" |
||
184 | method = "pluginslist" |
||
185 | print('INFO: [TEST_009] Plugins views') |
||
186 | plist = self.http_get(f"{URL}/{method}") |
||
187 | |||
188 | for p in plist.json(): |
||
189 | print(f"HTTP RESTful request: {URL}/{p}/views") |
||
190 | req = self.http_get(f"{URL}/{p}/views") |
||
191 | self.assertTrue(req.ok) |
||
192 | self.assertIsInstance(req.json(), dict) |
||
193 | |||
194 | def test_010_history(self): |
||
195 | """History.""" |
||
196 | method = "history" |
||
197 | print('INFO: [TEST_010] History') |
||
198 | print(f"HTTP RESTful request: {URL}/cpu/{method}") |
||
199 | req = self.http_get(f"{URL}/cpu/{method}") |
||
200 | self.assertIsInstance(req.json(), dict) |
||
201 | self.assertIsInstance(req.json()['user'], list) |
||
202 | self.assertTrue(len(req.json()['user']) > 0) |
||
203 | print(f"HTTP RESTful request: {URL}/cpu/{method}/3") |
||
204 | req = self.http_get(f"{URL}/cpu/{method}/3") |
||
205 | self.assertIsInstance(req.json(), dict) |
||
206 | self.assertIsInstance(req.json()['user'], list) |
||
207 | self.assertTrue(len(req.json()['user']) > 1) |
||
208 | print(f"HTTP RESTful request: {URL}/cpu/system/{method}") |
||
209 | req = self.http_get(f"{URL}/cpu/system/{method}") |
||
210 | self.assertIsInstance(req.json(), list) |
||
211 | self.assertIsInstance(req.json()[0], list) |
||
212 | print(f"HTTP RESTful request: {URL}/cpu/system/{method}/3") |
||
213 | req = self.http_get(f"{URL}/cpu/system/{method}/3") |
||
214 | self.assertIsInstance(req.json(), list) |
||
215 | self.assertIsInstance(req.json()[0], list) |
||
216 | |||
217 | def test_011_issue1401(self): |
||
218 | """Check issue #1401.""" |
||
219 | method = "network/interface_name" |
||
220 | print('INFO: [TEST_011] Issue #1401') |
||
221 | req = self.http_get(f"{URL}/{method}") |
||
222 | self.assertTrue(req.ok) |
||
223 | self.assertIsInstance(req.json(), dict) |
||
224 | self.assertIsInstance(req.json()['interface_name'], list) |
||
225 | |||
226 | def test_012_status(self): |
||
227 | """Check status endpoint.""" |
||
228 | method = "status" |
||
229 | print('INFO: [TEST_012] Status') |
||
230 | print(f"HTTP RESTful request: {URL}/{method}") |
||
231 | req = self.http_get(f"{URL}/{method}") |
||
232 | |||
233 | self.assertTrue(req.ok) |
||
234 | self.assertEqual(req.json()['version'], __version__) |
||
235 | |||
236 | def test_013_top(self): |
||
237 | """Values.""" |
||
238 | method = "processlist" |
||
239 | request = f"{URL}/{method}/top/2" |
||
240 | print('INFO: [TEST_013] Top nb item of PROCESSLIST') |
||
241 | print(request) |
||
242 | req = self.http_get(request) |
||
243 | |||
244 | self.assertTrue(req.ok) |
||
245 | self.assertIsInstance(req.json(), list) |
||
246 | self.assertEqual(len(req.json()), 2) |
||
247 | |||
248 | def test_014_config(self): |
||
249 | """Test API request to get Glances configuration.""" |
||
250 | method = "config" |
||
251 | print('INFO: [TEST_014] Get config') |
||
252 | |||
253 | req = self.http_get(f"{URL}/{method}") |
||
254 | self.assertTrue(req.ok) |
||
255 | self.assertIsInstance(req.json(), dict) |
||
256 | |||
257 | req = self.http_get(f"{URL}/{method}/global/refresh") |
||
258 | self.assertTrue(req.ok) |
||
259 | self.assertEqual(req.json(), "2") |
||
260 | |||
261 | def test_015_all_gzip(self): |
||
262 | """All with Gzip.""" |
||
263 | method = "all" |
||
264 | print('INFO: [TEST_015] Get all stats (with GZip compression)') |
||
265 | print(f"HTTP RESTful request: {URL}/{method}") |
||
266 | req = self.http_get(f"{URL}/{method}", gzip=True) |
||
267 | |||
268 | self.assertTrue(req.ok) |
||
269 | self.assertTrue(req.headers['Content-Encoding'] == 'gzip') |
||
270 | self.assertTrue(req.json(), dict) |
||
271 | |||
272 | def test_016_fields_description(self): |
||
273 | """Fields description.""" |
||
274 | print('INFO: [TEST_016] Get fields description and unit') |
||
275 | |||
276 | print(f"HTTP RESTful request: {URL}/cpu/total/description") |
||
277 | req = self.http_get(f"{URL}/cpu/total/description") |
||
278 | self.assertTrue(req.ok) |
||
279 | self.assertTrue(req.json(), str) |
||
280 | |||
281 | print(f"HTTP RESTful request: {URL}/cpu/total/unit") |
||
282 | req = self.http_get(f"{URL}/cpu/total/unit") |
||
283 | self.assertTrue(req.ok) |
||
284 | self.assertTrue(req.json(), str) |
||
285 | |||
286 | def test_017_item_key(self): |
||
287 | """Get /plugin/item/key value.""" |
||
288 | print('INFO: [TEST_017] Get /plugin/item/key value') |
||
289 | plugin = "network" |
||
290 | item = "interface_name" |
||
291 | req = self.http_get(f"{URL}/{plugin}/{item}") |
||
292 | self.assertTrue(req.ok) |
||
293 | self.assertIsInstance(req.json(), dict) |
||
294 | self.assertIsInstance(req.json()[item], list) |
||
295 | if len(req.json()[item]) > 0: |
||
296 | key = req.json()[item][0] |
||
297 | item = "bytes_all" |
||
298 | req = self.http_get(f"{URL}/{plugin}/{item}/{key}") |
||
299 | self.assertTrue(req.ok) |
||
300 | self.assertIsInstance(req.json(), dict) |
||
301 | self.assertIsInstance(req.json()[item], int) |
||
302 | |||
303 | def test_100_browser(self): |
||
304 | """Get /serverslist (for Glances Central Browser).""" |
||
305 | print('INFO: [TEST_100] Get /serverslist (for Glances Central Browser)') |
||
306 | req = self.http_get(f"{URL}/serverslist") |
||
307 | self.assertTrue(req.ok) |
||
308 | self.assertIsInstance(req.json(), list) |
||
309 | if len(req.json()) > 0: |
||
310 | self.assertIsInstance(req.json()[0], dict) |
||
311 | |||
312 | def test_999_stop_server(self): |
||
313 | """Stop the Glances Web Server.""" |
||
314 | print('INFO: [TEST_999] Stop the Glances Web Server') |
||
315 | |||
316 | print("Stop the Glances Web Server") |
||
317 | pid.terminate() |
||
318 | time.sleep(1) |
||
319 | |||
320 | self.assertTrue(True) |
||
321 | |||
322 | |||
323 | if __name__ == '__main__': |
||
324 | unittest.main() |
||
325 |