1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
|
#!/usr/bin/env python
#
# Glances - An eye on your system
#
# SPDX-FileCopyrightText: 2024 Nicolas Hennion <nicolas@nicolargo.com>
#
# SPDX-License-Identifier: LGPL-3.0-only
#
"""Tests for the CPU plugin."""
import json
import time
import pytest
from glances.globals import WINDOWS
@pytest.fixture
def cpu_plugin(glances_stats):
"""Return the CPU plugin instance from glances_stats."""
return glances_stats.get_plugin('cpu')
class TestCpuPluginBasics:
"""Test basic CPU plugin functionality."""
def test_plugin_name(self, cpu_plugin):
"""Test plugin name is correctly set."""
assert cpu_plugin.plugin_name == 'cpu'
def test_plugin_is_enabled(self, cpu_plugin):
"""Test that the plugin is enabled by default."""
assert cpu_plugin.is_enabled() is True
def test_display_curse_enabled(self, cpu_plugin):
"""Test that curse display is enabled."""
assert cpu_plugin.display_curse is True
def test_history_items_defined(self, cpu_plugin):
"""Test that history items are properly defined."""
items = cpu_plugin.get_items_history_list()
assert items is not None
assert len(items) >= 2
item_names = [item['name'] for item in items]
assert 'user' in item_names
assert 'system' in item_names
class TestCpuPluginUpdate:
"""Test CPU plugin update functionality."""
def test_update_returns_dict(self, cpu_plugin):
"""Test that update returns a dictionary."""
cpu_plugin.update()
stats = cpu_plugin.get_raw()
assert isinstance(stats, dict)
def test_update_contains_mandatory_keys(self, cpu_plugin):
"""Test that stats contain mandatory keys."""
cpu_plugin.update()
stats = cpu_plugin.get_raw()
mandatory_keys = ['total', 'system', 'user', 'idle']
for key in mandatory_keys:
assert key in stats, f"Missing mandatory key: {key}"
def test_cpu_percentages_in_valid_range(self, cpu_plugin):
"""Test that CPU percentages are within valid range."""
cpu_plugin.update()
stats = cpu_plugin.get_raw()
percentage_keys = ['total', 'system', 'user', 'idle']
for key in percentage_keys:
if key in stats and stats[key] is not None:
assert 0 <= stats[key] <= 100, f"{key} percentage out of range: {stats[key]}"
def test_total_cpu_calculation(self, cpu_plugin):
"""Test that total CPU is correctly calculated."""
cpu_plugin.update()
stats = cpu_plugin.get_raw()
assert 'total' in stats
assert stats['total'] >= 0
def test_cpucore_count(self, cpu_plugin):
"""Test that CPU core count is reported."""
cpu_plugin.update()
stats = cpu_plugin.get_raw()
assert 'cpucore' in stats
assert stats['cpucore'] >= 1
class TestCpuPluginContextSwitches:
"""Test CPU context switches and interrupts stats."""
def test_ctx_switches_present(self, cpu_plugin):
"""Test that context switches stat is present."""
cpu_plugin.update()
time.sleep(0.1)
cpu_plugin.update()
stats = cpu_plugin.get_raw()
assert 'ctx_switches' in stats
def test_interrupts_present(self, cpu_plugin):
"""Test that interrupts stat is present."""
cpu_plugin.update()
time.sleep(0.1)
cpu_plugin.update()
stats = cpu_plugin.get_raw()
assert 'interrupts' in stats
@pytest.mark.skipif(WINDOWS, reason="soft_interrupts not available on Windows")
def test_soft_interrupts_present(self, cpu_plugin):
"""Test that soft interrupts stat is present on non-Windows."""
cpu_plugin.update()
time.sleep(0.1)
cpu_plugin.update()
stats = cpu_plugin.get_raw()
assert 'soft_interrupts' in stats
class TestCpuPluginViews:
"""Test CPU plugin views functionality."""
def test_update_views_creates_views(self, cpu_plugin):
"""Test that update_views creates views dictionary."""
cpu_plugin.update()
cpu_plugin.update_views()
views = cpu_plugin.get_views()
assert isinstance(views, dict)
def test_views_contain_decoration(self, cpu_plugin):
"""Test that views contain decoration for total CPU."""
cpu_plugin.update()
cpu_plugin.update_views()
views = cpu_plugin.get_views()
assert 'total' in views
assert 'decoration' in views['total']
class TestCpuPluginJSON:
"""Test CPU plugin JSON serialization."""
def test_get_stats_returns_json(self, cpu_plugin):
"""Test that get_stats returns valid JSON."""
cpu_plugin.update()
stats_json = cpu_plugin.get_stats()
parsed = json.loads(stats_json)
assert isinstance(parsed, dict)
def test_json_contains_expected_fields(self, cpu_plugin):
"""Test that JSON output contains expected fields."""
cpu_plugin.update()
stats_json = cpu_plugin.get_stats()
parsed = json.loads(stats_json)
assert 'total' in parsed
assert 'user' in parsed
assert 'system' in parsed
class TestCpuPluginHistory:
"""Test CPU plugin history functionality."""
def test_history_enable_check(self, cpu_plugin):
"""Test that history_enable returns a boolean."""
result = cpu_plugin.history_enable()
assert isinstance(result, bool)
def test_get_items_history_list(self, cpu_plugin):
"""Test that get_items_history_list returns the history items."""
items = cpu_plugin.get_items_history_list()
if items is not None:
assert isinstance(items, list)
assert len(items) >= 2
class TestCpuPluginReset:
"""Test CPU plugin reset functionality."""
def test_reset_clears_stats(self, cpu_plugin):
"""Test that reset clears stats."""
cpu_plugin.update()
cpu_plugin.reset()
stats = cpu_plugin.get_raw()
assert stats == cpu_plugin.get_init_value()
def test_reset_views(self, cpu_plugin):
"""Test that reset_views clears views."""
cpu_plugin.update()
cpu_plugin.update_views()
cpu_plugin.reset_views()
assert cpu_plugin.get_views() == {}
class TestCpuPluginFieldsDescription:
"""Test CPU plugin fields description."""
def test_fields_description_exists(self, cpu_plugin):
"""Test that fields_description is defined."""
assert cpu_plugin.fields_description is not None
def test_mandatory_fields_described(self, cpu_plugin):
"""Test that mandatory fields have descriptions."""
mandatory_fields = ['total', 'system', 'user', 'idle']
for field in mandatory_fields:
assert field in cpu_plugin.fields_description
def test_field_has_description(self, cpu_plugin):
"""Test that each field has a description."""
for field, info in cpu_plugin.fields_description.items():
assert 'description' in info, f"Field {field} missing description"
class TestCpuPluginAlerts:
"""Test CPU plugin alert functionality."""
def test_get_alert_returns_valid_status(self, cpu_plugin):
"""Test that get_alert returns a valid status."""
cpu_plugin.update()
alert = cpu_plugin.get_alert(50, minimum=0, maximum=100, header='total')
valid_statuses = [
'OK',
'OK_LOG',
'CAREFUL',
'CAREFUL_LOG',
'WARNING',
'WARNING_LOG',
'CRITICAL',
'CRITICAL_LOG',
'DEFAULT',
'MAX',
]
assert any(alert.startswith(status) for status in valid_statuses)
def test_alert_levels(self, cpu_plugin):
"""Test different alert levels based on CPU usage."""
cpu_plugin.update()
# Low usage should be OK
alert_low = cpu_plugin.get_alert(10, minimum=0, maximum=100, header='total')
# The result depends on config, but should not be CRITICAL for 10%
assert not alert_low.startswith('CRITICAL')
class TestCpuPluginMsgCurse:
"""Test CPU plugin curse message generation."""
def test_msg_curse_returns_list(self, cpu_plugin):
"""Test that msg_curse returns a list."""
cpu_plugin.update()
msg = cpu_plugin.msg_curse()
assert isinstance(msg, list)
def test_msg_curse_format(self, cpu_plugin):
"""Test that msg_curse returns properly formatted entries."""
cpu_plugin.update()
# Ensure args.percpu is not set to get output
if hasattr(cpu_plugin.args, 'percpu'):
original_percpu = cpu_plugin.args.percpu
cpu_plugin.args.percpu = False
msg = cpu_plugin.msg_curse()
if msg: # May be empty if percpu mode or disabled
for entry in msg:
assert isinstance(entry, dict)
assert 'msg' in entry
if hasattr(cpu_plugin.args, 'percpu'):
cpu_plugin.args.percpu = original_percpu
def test_msg_curse_has_title_when_enabled(self, cpu_plugin):
"""Test that msg_curse contains CPU title when not in percpu mode."""
cpu_plugin.update()
if hasattr(cpu_plugin.args, 'percpu'):
original_percpu = cpu_plugin.args.percpu
cpu_plugin.args.percpu = False
msg = cpu_plugin.msg_curse()
if msg:
messages = [m.get('msg', '') for m in msg]
assert any('CPU' in str(m) for m in messages)
if hasattr(cpu_plugin.args, 'percpu'):
cpu_plugin.args.percpu = original_percpu
|
