Test Failed
Push — develop ( a3895e...b9843a )
by Nicolas
02:27 queued 13s
created

unittest-core.TestGlances.test_005_mem()   A

Complexity

Conditions 2

Size

Total Lines 12
Code Lines 9

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 2
eloc 9
nop 1
dl 0
loc 12
rs 9.95
c 0
b 0
f 0
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
# Refactor by @ariel-anieli in 2024
10
11
"""Glances unitary tests suite."""
12
13
import functools
14
import json
15
import time
16
import unittest
17
18
from glances import __version__
19
from glances.events_list import GlancesEventsList
20
from glances.filter import GlancesFilter, GlancesFilterList
21
from glances.globals import LINUX, WINDOWS, string_value_to_float, subsample
22
from glances.main import GlancesMain
23
from glances.outputs.glances_bars import Bar
24
from glances.plugins.plugin.model import GlancesPluginModel
25
from glances.programs import processes_to_programs
26
from glances.stats import GlancesStats
27
from glances.thresholds import (
28
    GlancesThresholdCareful,
29
    GlancesThresholdCritical,
30
    GlancesThresholdOk,
31
    GlancesThresholds,
32
    GlancesThresholdWarning,
33
)
34
35
# Global variables
36
# =================
37
38
# Init Glances core
39
core = GlancesMain()
40
test_config = core.get_config()
41
test_args = core.get_args()
42
43
# Init Glances stats
44
stats = GlancesStats(config=test_config, args=test_args)
45
46
# Unitest class
47
# ==============
48
print(f'Unitary tests for Glances {__version__}')
49
50
51
class TestGlances(unittest.TestCase):
52
    """Test Glances class."""
53
54
    def setUp(self):
55
        """The function is called *every time* before test_*."""
56
        print('\n' + '=' * 78)
57
58
    def reset_stats_history_and_views(self, plugin):
59
        plugin_instance = stats.get_plugin(plugin)
60
        plugin_instance.reset()  # reset stats
61
        plugin_instance.reset_views()  # reset views
62
        plugin_instance.reset_stats_history()  # reset history
63
64
        return plugin_instance
65
66
    def zip_with(self, method, elems, values):
67
        [method(elem, value) for elem, value in zip(elems, values)]
68
69
    def do_checks_before_update(self, plugin_instance):
70
        elems = [
71
            plugin_instance.get_raw(),
72
            plugin_instance.is_enabled(),
73
            plugin_instance.is_disabled(),
74
            plugin_instance.get_views(),
75
        ]
76
77
        values = [plugin_instance.stats_init_value, True, False, {}]
78
79
        self.zip_with(self.assertEqual, elems, values)
80
81
        self.assertIsInstance(plugin_instance.get_raw(), (dict, list))
82
83
        if plugin_instance.history_enable() and isinstance(plugin_instance.get_raw(), dict):
84
            self.assertEqual(plugin_instance.get_key(), None)
85
            self.assertTrue(
86
                all(
87
                    f in [h['name'] for h in plugin_instance.items_history_list]
88
                    for f in plugin_instance.get_raw_history()
89
                )
90
            )
91
        elif plugin_instance.history_enable() and isinstance(plugin_instance.get_raw(), list):
92
            self.assertNotEqual(plugin_instance.get_key(), None)
93
94
    def update_stats(self, plugin_instance):
95
        plugin_instance.update()
96
        plugin_instance.update_stats_history()
97
        plugin_instance.update_views()
98
99
        return plugin_instance
100
101
    def is_field_in_plugin(self, plugin_instance):
102
        def is_in_description(result, f):
103
            return True if f in plugin_instance.fields_description else result
104
105
        def is_field_in_this_description(result, i):
106
            return functools.reduce(is_in_description, i, False)
107
108
        return is_field_in_this_description
109
110
    def is_field_in_stats(self, plugin_instance, plugin):
111
        def is_field_in_this_plugin(result, f):
112
            if f not in plugin_instance.get_raw():
113
                print(f"WARNING: {f} field not found in {plugin} plugin stats")
114
            else:
115
                result = True
116
117
            return result
118
119
        return is_field_in_this_plugin
120
121
    def check_stats(self, plugin_instance, plugin):
122
        self.assertIsInstance(plugin_instance.get_raw(), (dict, list))
123
        if isinstance(plugin_instance.get_raw(), dict) and plugin_instance.get_raw() != {}:
124
            init = False
125
            is_field_present = self.is_field_in_stats(plugin_instance, plugin)
126
            field_description = plugin_instance.fields_description
127
            result = functools.reduce(is_field_present, field_description, init)
128
129
            self.assertTrue(result)
130
131
        elif isinstance(plugin_instance.get_raw(), list) and len(plugin_instance.get_raw()) > 0:
132
            init = False
133
            is_field_present = self.is_field_in_plugin(plugin_instance)
134
            raw_from_instance = plugin_instance.get_raw()
135
            result = functools.reduce(is_field_present, raw_from_instance, init)
136
137
            self.assertTrue(result)
138
139
        elems = [plugin_instance.get_raw(), plugin_instance.get_stats(), json.loads(plugin_instance.get_stats())]
140
141
        values = [plugin_instance.get_export(), plugin_instance.get_json(), plugin_instance.get_raw()]
142
143
        self.zip_with(self.assertEqual, elems, values)
144
145
        if len(plugin_instance.fields_description) > 0:
146
            # Get first item of the fields_description
147
            first_field = list(plugin_instance.fields_description.keys())[0]
148
149
            elems = [
150
                plugin_instance.get_raw_stats_item(first_field),
151
                json.loads(plugin_instance.get_stats_item(first_field)),
152
                plugin_instance.get_item_info(first_field, 'description'),
153
            ]
154
155
            values = [dict, dict, str]
156
157
            self.zip_with(self.assertIsInstance, elems, values)
158
159
    def filter_stats(self, plugin_instance):
160
        current_stats = plugin_instance.get_raw()
161
        if isinstance(plugin_instance.get_raw(), dict):
162
            current_stats['foo'] = 'bar'
163
            current_stats = plugin_instance.filter_stats(current_stats)
164
            self.assertTrue('foo' not in current_stats)
165
        elif isinstance(plugin_instance.get_raw(), list) and len(plugin_instance.get_raw()) > 0:
166
            current_stats[0]['foo'] = 'bar'
167
            current_stats = plugin_instance.filter_stats(current_stats)
168
            self.assertTrue('foo' not in current_stats[0])
169
170
    def get_first_history_field(self, plugin_instance):
171
        if isinstance(plugin_instance.get_raw(), dict):
172
            first_history_field = plugin_instance.get_items_history_list()[0]['name']
173
        elif isinstance(plugin_instance.get_raw(), list) and len(plugin_instance.get_raw()) > 0:
174
            first_history_field = '_'.join(
175
                [
176
                    plugin_instance.get_raw()[0][plugin_instance.get_key()],
177
                    plugin_instance.get_items_history_list()[0]['name'],
178
                ]
179
            )
180
181
        return first_history_field
0 ignored issues
show
introduced by
The variable first_history_field does not seem to be defined for all execution paths.
Loading history...
182
183
    def maybe_assert_first_for_raw(self, plugin_instance, first_history_field):
184
        if len(plugin_instance.get_raw()) > 0:
185
            self.assertEqual(len(plugin_instance.get_raw_history(first_history_field)), 2)
186
            self.assertGreater(
187
                plugin_instance.get_raw_history(first_history_field)[1][0],
188
                plugin_instance.get_raw_history(first_history_field)[0][0],
189
            )
190
191
    def maybe_assert_first_for_stats(self, plugin_instance, first_history_field):
192
        if len(plugin_instance.get_raw()) > 0:
193
            self.assertEqual(len(plugin_instance.get_raw_history(first_history_field)), 3)
194
            self.assertEqual(len(plugin_instance.get_raw_history(first_history_field, 2)), 2)
195
            self.assertIsInstance(json.loads(plugin_instance.get_stats_history()), dict)
196
197
    def chk_hist_maybe_add_third_elem(self, plugin_instance):
198
        if plugin_instance.history_enable():
199
            first_history_field = self.get_first_history_field(plugin_instance)
200
            self.maybe_assert_first_for_raw(plugin_instance, first_history_field)
201
202
            # Update stats (add third element)
203
            plugin_instance = self.update_stats(plugin_instance)
204
205
            self.maybe_assert_first_for_stats(plugin_instance, first_history_field)
206
207
        return first_history_field
0 ignored issues
show
introduced by
The variable first_history_field does not seem to be defined in case plugin_instance.history_enable() on line 198 is False. Are you sure this can never be the case?
Loading history...
208
209
    def check_views(self, plugin_instance, first_history_field):
210
        self.assertIsInstance(plugin_instance.get_views(), dict)
211
        if isinstance(plugin_instance.get_raw(), dict):
212
            self.assertIsInstance(plugin_instance.get_views(first_history_field), dict)
213
            self.assertTrue('decoration' in plugin_instance.get_views(first_history_field))
214
        elif isinstance(plugin_instance.get_raw(), list) and len(plugin_instance.get_raw()) > 0:
215
            first_history_field = plugin_instance.get_items_history_list()[0]['name']
216
            first_item = plugin_instance.get_raw()[0][plugin_instance.get_key()]
217
            self.assertIsInstance(plugin_instance.get_views(item=first_item, key=first_history_field), dict)
218
            self.assertTrue('decoration' in plugin_instance.get_views(item=first_item, key=first_history_field))
219
        self.assertIsInstance(json.loads(plugin_instance.get_json_views()), dict)
220
        self.assertEqual(json.loads(plugin_instance.get_json_views()), plugin_instance.get_views())
221
222
    def _common_plugin_tests(self, plugin):
223
        """Common method to test a Glances plugin
224
        This method is called multiple time by test 100 to 1xx"""
225
226
        # Reset all the stats, history and views
227
        plugin_instance = self.reset_stats_history_and_views(plugin)
228
229
        # Check before update
230
        self.do_checks_before_update(plugin_instance)
231
232
        # Update stats (add first element)
233
        plugin_instance = self.update_stats(plugin_instance)
234
235
        # Check stats
236
        self.check_stats(plugin_instance, plugin)
237
238
        # Filter stats
239
        self.filter_stats(plugin_instance)
240
241
        # Update stats (add second element)
242
        plugin_instance = self.update_stats(plugin_instance)
243
244
        # Check history
245
        first_history_field = self.chk_hist_maybe_add_third_elem(plugin_instance)
246
247
        # Check views
248
        self.check_views(plugin_instance, first_history_field)
249
250
    def test_000_update(self):
251
        """Update stats (mandatory step for all the stats).
252
253
        The update is made twice (for rate computation).
254
        """
255
        print('INFO: [TEST_000] Test the stats update function')
256
        try:
257
            stats.update()
258
        except Exception as e:
259
            print(f'ERROR: Stats update failed: {e}')
260
            self.assertTrue(False)
261
        time.sleep(1)
262
        try:
263
            stats.update()
264
        except Exception as e:
265
            print(f'ERROR: Stats update failed: {e}')
266
            self.assertTrue(False)
267
268
        self.assertTrue(True)
269
270
    def test_001_plugins(self):
271
        """Check mandatory plugins."""
272
        plugins_to_check = ['system', 'cpu', 'load', 'mem', 'memswap', 'network', 'diskio', 'fs']
273
        print('INFO: [TEST_001] Check the mandatory plugins list: {}'.format(', '.join(plugins_to_check)))
274
        plugins_list = stats.getPluginsList()
275
        for plugin in plugins_to_check:
276
            self.assertTrue(plugin in plugins_list)
277
278
    def test_002_system(self):
279
        """Check SYSTEM plugin."""
280
        stats_to_check = ['hostname', 'os_name']
281
        print('INFO: [TEST_002] Check SYSTEM stats: {}'.format(', '.join(stats_to_check)))
282
        stats_grab = stats.get_plugin('system').get_raw()
283
        for stat in stats_to_check:
284
            # Check that the key exist
285
            self.assertTrue(stat in stats_grab, msg=f'Cannot find key: {stat}')
286
        print(f'INFO: SYSTEM stats: {stats_grab}')
287
288
    def test_003_cpu(self):
289
        """Check CPU plugin."""
290
        stats_to_check = ['system', 'user', 'idle']
291
        print('INFO: [TEST_003] Check mandatory CPU stats: {}'.format(', '.join(stats_to_check)))
292
        stats_grab = stats.get_plugin('cpu').get_raw()
293
        for stat in stats_to_check:
294
            # Check that the key exist
295
            self.assertTrue(stat in stats_grab, msg=f'Cannot find key: {stat}')
296
            # Check that % is > 0 and < 100
297
            self.assertGreaterEqual(stats_grab[stat], 0)
298
            self.assertLessEqual(stats_grab[stat], 100)
299
        print(f'INFO: CPU stats: {stats_grab}')
300
301
    @unittest.skipIf(WINDOWS, "Load average not available on Windows")
302
    def test_004_load(self):
303
        """Check LOAD plugin."""
304
        stats_to_check = ['cpucore', 'min1', 'min5', 'min15']
305
        print('INFO: [TEST_004] Check LOAD stats: {}'.format(', '.join(stats_to_check)))
306
        stats_grab = stats.get_plugin('load').get_raw()
307
        for stat in stats_to_check:
308
            # Check that the key exist
309
            self.assertTrue(stat in stats_grab, msg=f'Cannot find key: {stat}')
310
            # Check that % is > 0
311
            self.assertGreaterEqual(stats_grab[stat], 0)
312
        print(f'INFO: LOAD stats: {stats_grab}')
313
314
    def test_005_mem(self):
315
        """Check MEM plugin."""
316
        plugin_name = 'mem'
317
        stats_to_check = ['available', 'used', 'free', 'total']
318
        print('INFO: [TEST_005] Check {} stats: {}'.format(plugin_name, ', '.join(stats_to_check)))
319
        stats_grab = stats.get_plugin('mem').get_raw()
320
        for stat in stats_to_check:
321
            # Check that the key exist
322
            self.assertTrue(stat in stats_grab, msg=f'Cannot find key: {stat}')
323
            # Check that % is > 0
324
            self.assertGreaterEqual(stats_grab[stat], 0)
325
        print(f'INFO: MEM stats: {stats_grab}')
326
327
    def test_006_memswap(self):
328
        """Check MEMSWAP plugin."""
329
        stats_to_check = ['used', 'free', 'total']
330
        print('INFO: [TEST_006] Check MEMSWAP stats: {}'.format(', '.join(stats_to_check)))
331
        stats_grab = stats.get_plugin('memswap').get_raw()
332
        for stat in stats_to_check:
333
            # Check that the key exist
334
            self.assertTrue(stat in stats_grab, msg=f'Cannot find key: {stat}')
335
            # Check that % is > 0
336
            self.assertGreaterEqual(stats_grab[stat], 0)
337
        print(f'INFO: MEMSWAP stats: {stats_grab}')
338
339
    def test_007_network(self):
340
        """Check NETWORK plugin."""
341
        print('INFO: [TEST_007] Check NETWORK stats')
342
        stats_grab = stats.get_plugin('network').get_raw()
343
        self.assertTrue(isinstance(stats_grab, list), msg='Network stats is not a list')
344
        print(f'INFO: NETWORK stats: {stats_grab}')
345
346
    def test_008_diskio(self):
347
        """Check DISKIO plugin."""
348
        print('INFO: [TEST_008] Check DISKIO stats')
349
        stats_grab = stats.get_plugin('diskio').get_raw()
350
        self.assertTrue(isinstance(stats_grab, list), msg='DiskIO stats is not a list')
351
        print(f'INFO: diskio stats: {stats_grab}')
352
353
    def test_009_fs(self):
354
        """Check File System plugin."""
355
        # stats_to_check = [ ]
356
        print('INFO: [TEST_009] Check FS stats')
357
        stats_grab = stats.get_plugin('fs').get_raw()
358
        self.assertTrue(isinstance(stats_grab, list), msg='FileSystem stats is not a list')
359
        print(f'INFO: FS stats: {stats_grab}')
360
361
    def test_010_processes(self):
362
        """Check Process plugin."""
363
        # stats_to_check = [ ]
364
        print('INFO: [TEST_010] Check PROCESS stats')
365
        stats_grab = stats.get_plugin('processcount').get_raw()
366
        # total = stats_grab['total']
367
        self.assertTrue(isinstance(stats_grab, dict), msg='Process count stats is not a dict')
368
        print(f'INFO: PROCESS count stats: {stats_grab}')
369
        stats_grab = stats.get_plugin('processlist').get_raw()
370
        self.assertTrue(isinstance(stats_grab, list), msg='Process count stats is not a list')
371
        print(f'INFO: PROCESS list stats: {len(stats_grab)} items in the list')
372
        # Check if number of processes in the list equal counter
373
        # self.assertEqual(total, len(stats_grab))
374
375
    def test_011_folders(self):
376
        """Check File System plugin."""
377
        # stats_to_check = [ ]
378
        print('INFO: [TEST_011] Check FOLDER stats')
379
        stats_grab = stats.get_plugin('folders').get_raw()
380
        self.assertTrue(isinstance(stats_grab, list), msg='Folders stats is not a list')
381
        print(f'INFO: Folders stats: {stats_grab}')
382
383
    def test_012_ip(self):
384
        """Check IP plugin."""
385
        print('INFO: [TEST_012] Check IP stats')
386
        stats_grab = stats.get_plugin('ip').get_raw()
387
        self.assertTrue(isinstance(stats_grab, dict), msg='IP stats is not a dict')
388
        print(f'INFO: IP stats: {stats_grab}')
389
390
    @unittest.skipIf(not LINUX, "IRQs available only on Linux")
391
    def test_013_irq(self):
392
        """Check IRQ plugin."""
393
        print('INFO: [TEST_013] Check IRQ stats')
394
        stats_grab = stats.get_plugin('irq').get_raw()
395
        self.assertTrue(isinstance(stats_grab, list), msg='IRQ stats is not a list')
396
        print(f'INFO: IRQ stats: {stats_grab}')
397
398
    @unittest.skipIf(not LINUX, "GPU available only on Linux")
399
    def test_014_gpu(self):
400
        """Check GPU plugin."""
401
        print('INFO: [TEST_014] Check GPU stats')
402
        stats_grab = stats.get_plugin('gpu').get_raw()
403
        self.assertTrue(isinstance(stats_grab, list), msg='GPU stats is not a list')
404
        print(f'INFO: GPU stats: {stats_grab}')
405
406
    def test_015_sorted_stats(self):
407
        """Check sorted stats method."""
408
        print('INFO: [TEST_015] Check sorted stats method')
409
        aliases = {
410
            "key2": "alias11",
411
            "key5": "alias2",
412
        }
413
        unsorted_stats = [
414
            {"key": "key4"},
415
            {"key": "key2"},
416
            {"key": "key5"},
417
            {"key": "key21"},
418
            {"key": "key3"},
419
        ]
420
421
        gp = GlancesPluginModel()
422
        gp.get_key = lambda: "key"
423
        gp.has_alias = aliases.get
424
        gp.stats = unsorted_stats
425
426
        sorted_stats = gp.sorted_stats()
427
        self.assertEqual(len(sorted_stats), 5)
428
        self.assertEqual(sorted_stats[0]["key"], "key5")
429
        self.assertEqual(sorted_stats[1]["key"], "key2")
430
        self.assertEqual(sorted_stats[2]["key"], "key3")
431
        self.assertEqual(sorted_stats[3]["key"], "key4")
432
        self.assertEqual(sorted_stats[4]["key"], "key21")
433
434
    def test_016_subsample(self):
435
        """Test subsampling function."""
436
        print('INFO: [TEST_016] Subsampling')
437
        for l_test in [
438
            ([1, 2, 3], 4),
439
            ([1, 2, 3, 4], 4),
440
            ([1, 2, 3, 4, 5, 6, 7], 4),
441
            ([1, 2, 3, 4, 5, 6, 7, 8], 4),
442
            (list(range(1, 800)), 4),
443
            (list(range(1, 8000)), 800),
444
        ]:
445
            l_subsample = subsample(l_test[0], l_test[1])
446
            self.assertLessEqual(len(l_subsample), l_test[1])
447
448
    def test_017_hddsmart(self):
449
        """Check hard disk SMART data plugin."""
450
        try:
451
            from glances.globals import is_admin
452
        except ImportError:
453
            print("INFO: [TEST_017] pySMART not found, not running SMART plugin test")
454
            return
455
456
        stat = 'DeviceName'
457
        print(f'INFO: [TEST_017] Check SMART stats: {stat}')
458
        stats_grab = stats.get_plugin('smart').get_raw()
459
        if not is_admin():
460
            print("INFO: Not admin, SMART list should be empty")
461
            assert len(stats_grab) == 0
462
        elif stats_grab == {}:
463
            print("INFO: Admin but SMART list is empty")
464
            assert len(stats_grab) == 0
465
        else:
466
            print(stats_grab)
467
            self.assertTrue(stat in stats_grab[0].keys(), msg=f'Cannot find key: {stat}')
468
469
        print(f'INFO: SMART stats: {stats_grab}')
470
471
    def test_017_programs(self):
472
        """Check Programs function (it's not a plugin)."""
473
        # stats_to_check = [ ]
474
        print('INFO: [TEST_017] Check PROGRAM stats')
475
        stats_grab = processes_to_programs(stats.get_plugin('processlist').get_raw())
476
        self.assertIsInstance(stats_grab, list, msg='Programs stats list is not a list')
477
        self.assertIsInstance(stats_grab[0], dict, msg='First item should be a dict')
478
479
    def test_018_string_value_to_float(self):
480
        """Check string_value_to_float function"""
481
        print('INFO: [TEST_018] Check string_value_to_float function')
482
        self.assertEqual(string_value_to_float('32kB'), 32000.0)
483
        self.assertEqual(string_value_to_float('32 KB'), 32000.0)
484
        self.assertEqual(string_value_to_float('15.5MB'), 15500000.0)
485
        self.assertEqual(string_value_to_float('25.9'), 25.9)
486
        self.assertEqual(string_value_to_float('12'), 12)
487
        self.assertEqual(string_value_to_float('--'), None)
488
489
    def test_019_events(self):
490
        """Test events class"""
491
        print('INFO: [TEST_019] Test events')
492
        # Init events
493
        events = GlancesEventsList(max_events=5, min_duration=1, min_interval=3)
494
        # Minimal event duration not reached
495
        events.add('WARNING', 'LOAD', 4)
496
        events.add('CRITICAL', 'LOAD', 5)
497
        events.add('OK', 'LOAD', 1)
498
        self.assertEqual(len(events.get()), 0)
499
        # Minimal event duration LOAD reached
500
        events.add('WARNING', 'LOAD', 4)
501
        time.sleep(1)
502
        events.add('CRITICAL', 'LOAD', 5)
503
        time.sleep(1)
504
        events.add('OK', 'LOAD', 1)
505
        self.assertEqual(len(events.get()), 1)
506
        self.assertEqual(events.get()[0]['type'], 'LOAD')
507
        self.assertEqual(events.get()[0]['state'], 'CRITICAL')
508
        self.assertEqual(events.get()[0]['max'], 5)
509
        # Minimal event duration CPU reached
510
        events.add('WARNING', 'CPU', 60)
511
        time.sleep(1)
512
        events.add('WARNING', 'CPU', 70)
513
        time.sleep(1)
514
        events.add('OK', 'CPU', 10)
515
        self.assertEqual(len(events.get()), 2)
516
        self.assertEqual(events.get()[0]['type'], 'CPU')
517
        self.assertEqual(events.get()[0]['state'], 'WARNING')
518
        self.assertEqual(events.get()[0]['min'], 60)
519
        self.assertEqual(events.get()[0]['max'], 70)
520
        self.assertEqual(events.get()[0]['count'], 2)
521
        # Minimal event duration CPU reached (again)
522
        # but time between two events (min_interval) is too short
523
        # a merge will be done
524
        time.sleep(0.5)
525
        events.add('WARNING', 'CPU', 60)
526
        time.sleep(1)
527
        events.add('WARNING', 'CPU', 80)
528
        time.sleep(1)
529
        events.add('OK', 'CPU', 10)
530
        self.assertEqual(len(events.get()), 2)
531
        self.assertEqual(events.get()[0]['type'], 'CPU')
532
        self.assertEqual(events.get()[0]['state'], 'WARNING')
533
        self.assertEqual(events.get()[0]['min'], 60)
534
        self.assertEqual(events.get()[0]['max'], 80)
535
        self.assertEqual(events.get()[0]['count'], 4)
536
        # Clean WARNING events
537
        events.clean()
538
        self.assertEqual(len(events.get()), 1)
539
540
    def test_020_filter(self):
541
        """Test filter classes"""
542
        print('INFO: [TEST_020] Test filter')
543
        gf = GlancesFilter()
544
        gf.filter = '.*python.*'
545
        self.assertEqual(gf.filter, '.*python.*')
546
        self.assertEqual(gf.filter_key, None)
547
        self.assertTrue(gf.is_filtered({'name': 'python'}))
548
        self.assertTrue(gf.is_filtered({'name': '/usr/bin/python -m glances'}))
549
        self.assertFalse(gf.is_filtered({'noname': 'python'}))
550
        self.assertFalse(gf.is_filtered({'name': 'snake'}))
551
        gf.filter = 'username:nicolargo'
552
        self.assertEqual(gf.filter, 'nicolargo')
553
        self.assertEqual(gf.filter_key, 'username')
554
        self.assertTrue(gf.is_filtered({'username': 'nicolargo'}))
555
        self.assertFalse(gf.is_filtered({'username': 'notme'}))
556
        self.assertFalse(gf.is_filtered({'notuser': 'nicolargo'}))
557
        gfl = GlancesFilterList()
558
        gfl.filter = '.*python.*,username:nicolargo'
559
        self.assertTrue(gfl.is_filtered({'name': 'python is in the place'}))
560
        self.assertFalse(gfl.is_filtered({'name': 'snake is in the place'}))
561
        self.assertTrue(gfl.is_filtered({'name': 'snake is in the place', 'username': 'nicolargo'}))
562
        self.assertFalse(gfl.is_filtered({'name': 'snake is in the place', 'username': 'notme'}))
563
564
    def test_094_thresholds(self):
565
        """Test thresholds classes"""
566
        print('INFO: [TEST_094] Thresholds')
567
        ok = GlancesThresholdOk()
568
        careful = GlancesThresholdCareful()
569
        warning = GlancesThresholdWarning()
570
        critical = GlancesThresholdCritical()
571
        self.assertTrue(ok < careful)
572
        self.assertTrue(careful < warning)
573
        self.assertTrue(warning < critical)
574
        self.assertFalse(ok > careful)
575
        self.assertEqual(ok, ok)
576
        self.assertEqual(str(ok), 'OK')
577
        thresholds = GlancesThresholds()
578
        thresholds.add('cpu_percent', 'OK')
579
        self.assertEqual(thresholds.get(stat_name='cpu_percent').description(), 'OK')
580
581
    def test_095_methods(self):
582
        """Test mandatories methods"""
583
        print('INFO: [TEST_095] Mandatories methods')
584
        mandatories_methods = ['reset', 'update']
585
        plugins_list = stats.getPluginsList()
586
        for plugin in plugins_list:
587
            for method in mandatories_methods:
588
                self.assertTrue(hasattr(stats.get_plugin(plugin), method), msg=f'{plugin} has no method {method}()')
589
590
    def test_096_views(self):
591
        """Test get_views method"""
592
        print('INFO: [TEST_096] Test views')
593
        plugins_list = stats.getPluginsList()
594
        for plugin in plugins_list:
595
            stats.get_plugin(plugin).get_raw()
596
            views_grab = stats.get_plugin(plugin).get_views()
597
            self.assertTrue(isinstance(views_grab, dict), msg=f'{plugin} view is not a dict')
598
599
    def test_097_attribute(self):
600
        """Test GlancesAttribute classes"""
601
        print('INFO: [TEST_097] Test attribute')
602
        # GlancesAttribute
603
        from glances.attribute import GlancesAttribute
604
605
        a = GlancesAttribute('a', description='ad', history_max_size=3)
606
        self.assertEqual(a.name, 'a')
607
        self.assertEqual(a.description, 'ad')
608
        a.description = 'adn'
609
        self.assertEqual(a.description, 'adn')
610
        a.value = 1
611
        a.value = 2
612
        self.assertEqual(len(a.history), 2)
613
        a.value = 3
614
        self.assertEqual(len(a.history), 3)
615
        a.value = 4
616
        # Check if history_max_size=3 is OK
617
        self.assertEqual(len(a.history), 3)
618
        self.assertEqual(a.history_size(), 3)
619
        self.assertEqual(a.history_len(), 3)
620
        self.assertEqual(a.history_value()[1], 4)
621
        self.assertEqual(a.history_mean(nb=3), 4.5)
622
623
    def test_098_history(self):
624
        """Test GlancesHistory classes"""
625
        print('INFO: [TEST_098] Test history')
626
        # GlancesHistory
627
        from glances.history import GlancesHistory
628
629
        h = GlancesHistory()
630
        h.add('a', 1, history_max_size=100)
631
        h.add('a', 2, history_max_size=100)
632
        h.add('a', 3, history_max_size=100)
633
        h.add('b', 10, history_max_size=100)
634
        h.add('b', 20, history_max_size=100)
635
        h.add('b', 30, history_max_size=100)
636
        self.assertEqual(len(h.get()), 2)
637
        self.assertEqual(len(h.get()['a']), 3)
638
        h.reset()
639
        self.assertEqual(len(h.get()), 2)
640
        self.assertEqual(len(h.get()['a']), 0)
641
642
    def test_099_output_bars(self):
643
        """Test quick look plugin.
644
645
        > bar.min_value
646
        0
647
        > bar.max_value
648
        100
649
        > bar.percent = -1
650
        > bar.percent
651
        0
652
        """
653
        print('INFO: [TEST_099] Test progress bar')
654
655
        bar = Bar(size=1)
656
        # Percent value can not be lower than min_value
657
        bar.percent = -1
658
        self.assertLessEqual(bar.percent, bar.min_value)
659
        # but... percent value can be higher than max_value
660
        bar.percent = 101
661
        self.assertLessEqual(bar.percent, 101)
662
663
        # Test display
664
        bar = Bar(size=50)
665
        bar.percent = 0
666
        self.assertEqual(bar.get(), '                                              0.0%')
667
        bar.percent = 70
668
        self.assertEqual(bar.get(), '|||||||||||||||||||||||||||||||              70.0%')
669
        bar.percent = 100
670
        self.assertEqual(bar.get(), '||||||||||||||||||||||||||||||||||||||||||||  100%')
671
        bar.percent = 110
672
        self.assertEqual(bar.get(), '|||||||||||||||||||||||||||||||||||||||||||| >100%')
673
674
    # Error in Github Action. Do not remove the comment.
675
    # def test_100_system_plugin_method(self):
676
    #     """Test system plugin methods"""
677
    #     print('INFO: [TEST_100] Test system plugin methods')
678
    #     self._common_plugin_tests('system')
679
680
    def test_101_cpu_plugin_method(self):
681
        """Test cpu plugin methods"""
682
        print('INFO: [TEST_100] Test cpu plugin methods')
683
        self._common_plugin_tests('cpu')
684
685
    @unittest.skipIf(WINDOWS, "Load average not available on Windows")
686
    def test_102_load_plugin_method(self):
687
        """Test load plugin methods"""
688
        print('INFO: [TEST_102] Test load plugin methods')
689
        self._common_plugin_tests('load')
690
691
    def test_103_mem_plugin_method(self):
692
        """Test mem plugin methods"""
693
        print('INFO: [TEST_103] Test mem plugin methods')
694
        self._common_plugin_tests('mem')
695
696
    def test_104_memswap_plugin_method(self):
697
        """Test memswap plugin methods"""
698
        print('INFO: [TEST_104] Test memswap plugin methods')
699
        self._common_plugin_tests('memswap')
700
701
    def test_105_network_plugin_method(self):
702
        """Test network plugin methods"""
703
        print('INFO: [TEST_105] Test network plugin methods')
704
        self._common_plugin_tests('network')
705
706
    # Error in Github Action. Do not remove the comment.
707
    # def test_106_diskio_plugin_method(self):
708
    #     """Test diskio plugin methods"""
709
    #     print('INFO: [TEST_106] Test diskio plugin methods')
710
    #     self._common_plugin_tests('diskio')
711
712
    def test_107_fs_plugin_method(self):
713
        """Test fs plugin methods"""
714
        print('INFO: [TEST_107] Test fs plugin methods')
715
        self._common_plugin_tests('fs')
716
717
    def test_200_views_hidden(self):
718
        """Test hide feature"""
719
        print('INFO: [TEST_200] Test views hidden feature')
720
        # Test will be done with the diskio plugin, first available interface (read_bytes fields)
721
        plugin = 'diskio'
722
        field = 'read_bytes_rate_per_sec'
723
        plugin_instance = stats.get_plugin(plugin)
724
        if len(plugin_instance.get_views()) == 0 or not test_config.get_bool_value(plugin, 'hide_zero', False):
725
            # No diskIO interface, test can not be done
726
            return
727
        # Get first disk interface
728
        key = list(plugin_instance.get_views().keys())[0]
729
        # Test
730
        ######
731
        # Init the stats
732
        plugin_instance.update()
733
        raw_stats = plugin_instance.get_raw()
734
        # Reset the views
735
        plugin_instance.set_views({})
736
        # Set field to 0 (should be hidden)
737
        raw_stats[0][field] = 0
738
        plugin_instance.set_stats(raw_stats)
739
        self.assertEqual(plugin_instance.get_raw()[0][field], 0)
740
        plugin_instance.update_views()
741
        self.assertTrue(plugin_instance.get_views()[key][field]['hidden'])
742
        # Set field to 0 (should be hidden)
743
        raw_stats[0][field] = 0
744
        plugin_instance.set_stats(raw_stats)
745
        self.assertEqual(plugin_instance.get_raw()[0][field], 0)
746
        plugin_instance.update_views()
747
        self.assertTrue(plugin_instance.get_views()[key][field]['hidden'])
748
        # Set field to 1 (should not be hidden)
749
        raw_stats[0][field] = 1
750
        plugin_instance.set_stats(raw_stats)
751
        self.assertEqual(plugin_instance.get_raw()[0][field], 1)
752
        plugin_instance.update_views()
753
        self.assertFalse(plugin_instance.get_views()[key][field]['hidden'])
754
        # Set field back to 0 (should not be hidden)
755
        raw_stats[0][field] = 0
756
        plugin_instance.set_stats(raw_stats)
757
        self.assertEqual(plugin_instance.get_raw()[0][field], 0)
758
        plugin_instance.update_views()
759
        self.assertFalse(plugin_instance.get_views()[key][field]['hidden'])
760
761
    # def test_700_secure(self):
762
    #     """Test secure functions"""
763
    #     print('INFO: [TEST_700] Secure functions')
764
765
    #     if WINDOWS:
766
    #         self.assertIn(secure_popen('echo TEST'), ['TEST\n', 'TEST\r\n'])
767
    #         self.assertIn(secure_popen('echo TEST1 && echo TEST2'), ['TEST1\nTEST2\n', 'TEST1\r\nTEST2\r\n'])
768
    #     else:
769
    #         self.assertEqual(secure_popen('echo -n TEST'), 'TEST')
770
    #         self.assertEqual(secure_popen('echo -n TEST1 && echo -n TEST2'), 'TEST1TEST2')
771
    #         # Make the test failed on Github (AssertionError: '' != 'FOO\n')
772
    #         # but not on my localLinux computer...
773
    #         # self.assertEqual(secure_popen('echo FOO | grep FOO'), 'FOO\n')
774
775
    # def test_800_memory_leak(self):
776
    #     """Memory leak check"""
777
    #     import tracemalloc
778
779
    #     print('INFO: [TEST_800] Memory leak check')
780
    #     tracemalloc.start()
781
    #     # 3 iterations just to init the stats and fill the memory
782
    #     for _ in range(3):
783
    #         stats.update()
784
785
    #     # Start the memory leak check
786
    #     snapshot_begin = tracemalloc.take_snapshot()
787
    #     for _ in range(3):
788
    #         stats.update()
789
    #     snapshot_end = tracemalloc.take_snapshot()
790
    #     snapshot_diff = snapshot_end.compare_to(snapshot_begin, 'filename')
791
    #     memory_leak = sum([s.size_diff for s in snapshot_diff])
792
    #     print(f'INFO: Memory leak: {memory_leak} bytes')
793
794
    #     # snapshot_begin = tracemalloc.take_snapshot()
795
    #     for _ in range(30):
796
    #         stats.update()
797
    #     snapshot_end = tracemalloc.take_snapshot()
798
    #     snapshot_diff = snapshot_end.compare_to(snapshot_begin, 'filename')
799
    #     memory_leak = sum([s.size_diff for s in snapshot_diff])
800
    #     print(f'INFO: Memory leak: {memory_leak} bytes')
801
802
    #     # snapshot_begin = tracemalloc.take_snapshot()
803
    #     for _ in range(300):
804
    #         stats.update()
805
    #     snapshot_end = tracemalloc.take_snapshot()
806
    #     snapshot_diff = snapshot_end.compare_to(snapshot_begin, 'filename')
807
    #     memory_leak = sum([s.size_diff for s in snapshot_diff])
808
    #     print(f'INFO: Memory leak: {memory_leak} bytes')
809
    #     snapshot_top = snapshot_end.compare_to(snapshot_begin, 'traceback')
810
    #     print("Memory consumption (top 5):")
811
    #     for stat in snapshot_top[:5]:
812
    #         print(stat)
813
    #         for line in stat.traceback.format():
814
    #             print(line)
815
816
    def test_999_the_end(self):
817
        """Free all the stats"""
818
        print('INFO: [TEST_999] Free the stats')
819
        stats.end()
820
        self.assertTrue(True)
821
822
823
if __name__ == '__main__':
824
    unittest.main()
825