nicolargo /
glances
| 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
Loading history...
|
|||
| 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 -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 | 'diskio', |
||
| 108 | 'hddtemp', |
||
| 109 | 'batpercent', |
||
| 110 | 'network', |
||
| 111 | 'folders', |
||
| 112 | 'amps', |
||
| 113 | 'ports', |
||
| 114 | 'irq', |
||
| 115 | 'wifi', |
||
| 116 | 'gpu', |
||
| 117 | 'containers', |
||
| 118 | ): |
||
| 119 | self.assertIsInstance(req.json(), list) |
||
| 120 | if len(req.json()) > 0: |
||
| 121 | self.assertIsInstance(req.json()[0], dict) |
||
| 122 | else: |
||
| 123 | self.assertIsInstance(req.json(), dict) |
||
| 124 | |||
| 125 | def test_004_items(self): |
||
| 126 | """Items.""" |
||
| 127 | method = "cpu" |
||
| 128 | print('INFO: [TEST_004] Items for the CPU method') |
||
| 129 | ilist = self.http_get(f"{URL}/{method}") |
||
| 130 | |||
| 131 | for i in ilist.json(): |
||
| 132 | print(f"HTTP RESTful request: {URL}/{method}/{i}") |
||
| 133 | req = self.http_get(f"{URL}/{method}/{i}") |
||
| 134 | self.assertTrue(req.ok) |
||
| 135 | self.assertIsInstance(req.json(), dict) |
||
| 136 | print(req.json()[i]) |
||
| 137 | self.assertIsInstance(req.json()[i], numbers.Number) |
||
| 138 | |||
| 139 | def test_005_values(self): |
||
| 140 | """Values.""" |
||
| 141 | method = "processlist" |
||
| 142 | print('INFO: [TEST_005] Item=Value for the PROCESSLIST method') |
||
| 143 | print(f"{URL}/{method}/pid/0") |
||
| 144 | req = self.http_get(f"{URL}/{method}/pid/0") |
||
| 145 | |||
| 146 | self.assertTrue(req.ok) |
||
| 147 | self.assertIsInstance(req.json(), dict) |
||
| 148 | |||
| 149 | def test_006_all_limits(self): |
||
| 150 | """All limits.""" |
||
| 151 | method = "all/limits" |
||
| 152 | print('INFO: [TEST_006] Get all limits') |
||
| 153 | print(f"HTTP RESTful request: {URL}/{method}") |
||
| 154 | req = self.http_get(f"{URL}/{method}") |
||
| 155 | |||
| 156 | self.assertTrue(req.ok) |
||
| 157 | self.assertIsInstance(req.json(), dict) |
||
| 158 | |||
| 159 | def test_007_all_views(self): |
||
| 160 | """All views.""" |
||
| 161 | method = "all/views" |
||
| 162 | print('INFO: [TEST_007] Get all views') |
||
| 163 | print(f"HTTP RESTful request: {URL}/{method}") |
||
| 164 | req = self.http_get(f"{URL}/{method}") |
||
| 165 | |||
| 166 | self.assertTrue(req.ok) |
||
| 167 | self.assertIsInstance(req.json(), dict) |
||
| 168 | |||
| 169 | def test_008_plugins_limits(self): |
||
| 170 | """Plugins limits.""" |
||
| 171 | method = "pluginslist" |
||
| 172 | print('INFO: [TEST_008] Plugins limits') |
||
| 173 | plist = self.http_get(f"{URL}/{method}") |
||
| 174 | |||
| 175 | for p in plist.json(): |
||
| 176 | print(f"HTTP RESTful request: {URL}/{p}/limits") |
||
| 177 | req = self.http_get(f"{URL}/{p}/limits") |
||
| 178 | self.assertTrue(req.ok) |
||
| 179 | self.assertIsInstance(req.json(), dict) |
||
| 180 | |||
| 181 | def test_009_plugins_views(self): |
||
| 182 | """Plugins views.""" |
||
| 183 | method = "pluginslist" |
||
| 184 | print('INFO: [TEST_009] Plugins views') |
||
| 185 | plist = self.http_get(f"{URL}/{method}") |
||
| 186 | |||
| 187 | for p in plist.json(): |
||
| 188 | print(f"HTTP RESTful request: {URL}/{p}/views") |
||
| 189 | req = self.http_get(f"{URL}/{p}/views") |
||
| 190 | self.assertTrue(req.ok) |
||
| 191 | self.assertIsInstance(req.json(), dict) |
||
| 192 | |||
| 193 | def test_010_history(self): |
||
| 194 | """History.""" |
||
| 195 | method = "history" |
||
| 196 | print('INFO: [TEST_010] History') |
||
| 197 | print(f"HTTP RESTful request: {URL}/cpu/{method}") |
||
| 198 | req = self.http_get(f"{URL}/cpu/{method}") |
||
| 199 | self.assertIsInstance(req.json(), dict) |
||
| 200 | self.assertIsInstance(req.json()['user'], list) |
||
| 201 | self.assertTrue(len(req.json()['user']) > 0) |
||
| 202 | print(f"HTTP RESTful request: {URL}/cpu/{method}/3") |
||
| 203 | req = self.http_get(f"{URL}/cpu/{method}/3") |
||
| 204 | self.assertIsInstance(req.json(), dict) |
||
| 205 | self.assertIsInstance(req.json()['user'], list) |
||
| 206 | self.assertTrue(len(req.json()['user']) > 1) |
||
| 207 | print(f"HTTP RESTful request: {URL}/cpu/system/{method}") |
||
| 208 | req = self.http_get(f"{URL}/cpu/system/{method}") |
||
| 209 | self.assertIsInstance(req.json(), list) |
||
| 210 | self.assertIsInstance(req.json()[0], list) |
||
| 211 | print(f"HTTP RESTful request: {URL}/cpu/system/{method}/3") |
||
| 212 | req = self.http_get(f"{URL}/cpu/system/{method}/3") |
||
| 213 | self.assertIsInstance(req.json(), list) |
||
| 214 | self.assertIsInstance(req.json()[0], list) |
||
| 215 | |||
| 216 | def test_011_issue1401(self): |
||
| 217 | """Check issue #1401.""" |
||
| 218 | method = "network/interface_name" |
||
| 219 | print('INFO: [TEST_011] Issue #1401') |
||
| 220 | req = self.http_get(f"{URL}/{method}") |
||
| 221 | self.assertTrue(req.ok) |
||
| 222 | self.assertIsInstance(req.json(), dict) |
||
| 223 | self.assertIsInstance(req.json()['interface_name'], list) |
||
| 224 | |||
| 225 | def test_012_status(self): |
||
| 226 | """Check status endpoint.""" |
||
| 227 | method = "status" |
||
| 228 | print('INFO: [TEST_012] Status') |
||
| 229 | print(f"HTTP RESTful request: {URL}/{method}") |
||
| 230 | req = self.http_get(f"{URL}/{method}") |
||
| 231 | |||
| 232 | self.assertTrue(req.ok) |
||
| 233 | self.assertEqual(req.json()['version'], __version__) |
||
| 234 | |||
| 235 | def test_013_top(self): |
||
| 236 | """Values.""" |
||
| 237 | method = "processlist" |
||
| 238 | request = f"{URL}/{method}/top/2" |
||
| 239 | print('INFO: [TEST_013] Top nb item of PROCESSLIST') |
||
| 240 | print(request) |
||
| 241 | req = self.http_get(request) |
||
| 242 | |||
| 243 | self.assertTrue(req.ok) |
||
| 244 | self.assertIsInstance(req.json(), list) |
||
| 245 | self.assertEqual(len(req.json()), 2) |
||
| 246 | |||
| 247 | def test_014_config(self): |
||
| 248 | """Test API request to get Glances configuration.""" |
||
| 249 | method = "config" |
||
| 250 | print('INFO: [TEST_014] Get config') |
||
| 251 | |||
| 252 | req = self.http_get(f"{URL}/{method}") |
||
| 253 | self.assertTrue(req.ok) |
||
| 254 | self.assertIsInstance(req.json(), dict) |
||
| 255 | |||
| 256 | req = self.http_get(f"{URL}/{method}/global/refresh") |
||
| 257 | self.assertTrue(req.ok) |
||
| 258 | self.assertEqual(req.json(), "2") |
||
| 259 | |||
| 260 | def test_015_all_gzip(self): |
||
| 261 | """All with Gzip.""" |
||
| 262 | method = "all" |
||
| 263 | print('INFO: [TEST_015] Get all stats (with GZip compression)') |
||
| 264 | print(f"HTTP RESTful request: {URL}/{method}") |
||
| 265 | req = self.http_get(f"{URL}/{method}", gzip=True) |
||
| 266 | |||
| 267 | self.assertTrue(req.ok) |
||
| 268 | self.assertTrue(req.headers['Content-Encoding'] == 'gzip') |
||
| 269 | self.assertTrue(req.json(), dict) |
||
| 270 | |||
| 271 | def test_016_fields_description(self): |
||
| 272 | """Fields description.""" |
||
| 273 | print('INFO: [TEST_016] Get fields description and unit') |
||
| 274 | |||
| 275 | print(f"HTTP RESTful request: {URL}/cpu/total/description") |
||
| 276 | req = self.http_get(f"{URL}/cpu/total/description") |
||
| 277 | self.assertTrue(req.ok) |
||
| 278 | self.assertTrue(req.json(), str) |
||
| 279 | |||
| 280 | print(f"HTTP RESTful request: {URL}/cpu/total/unit") |
||
| 281 | req = self.http_get(f"{URL}/cpu/total/unit") |
||
| 282 | self.assertTrue(req.ok) |
||
| 283 | self.assertTrue(req.json(), str) |
||
| 284 | |||
| 285 | def test_999_stop_server(self): |
||
| 286 | """Stop the Glances Web Server.""" |
||
| 287 | print('INFO: [TEST_999] Stop the Glances Web Server') |
||
| 288 | |||
| 289 | print("Stop the Glances Web Server") |
||
| 290 | pid.terminate() |
||
| 291 | time.sleep(1) |
||
| 292 | |||
| 293 | self.assertTrue(True) |
||
| 294 | |||
| 295 | |||
| 296 | if __name__ == '__main__': |
||
| 297 | unittest.main() |
||
| 298 |