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