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
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
|
"""Runtime disruption tests -- simulating mid-run sensor/core changes.
These tests mock psutil to return normal data during __init__, then change
the mock before calling update() to simulate real-world events like CPU
hotplug, sensor disconnection, or permission changes.
All tests are marked @pytest.mark.xfail(strict=True) to document current
crashes without blocking CI. When each crash is fixed, remove the xfail
marker so the test becomes a regression guard.
"""
from collections import OrderedDict
from unittest.mock import MagicMock
import pytest
from s_tui.sources.fan_source import FanSource
from s_tui.sources.freq_source import FreqSource
from s_tui.sources.rapl_power_source import RaplPowerSource
from s_tui.sources.temp_source import TempSource
from s_tui.sources.util_source import UtilSource
from tests.conftest import (
RaplStats,
SensorTemperature,
make_cpu_freq_list,
make_cpu_freq_overall,
make_fans_dict,
)
# =====================================================================
# UtilSource -- core count changes
# =====================================================================
class TestUtilCoreCountChanges:
def test_core_count_grows(self, mocker):
"""Simulate a core coming online mid-run (e.g. CPU hotplug)."""
mocker.patch("psutil.cpu_count", return_value=4)
mocker.patch("psutil.cpu_percent", return_value=[25.0, 30.0, 20.0, 15.0])
mocker.patch(
"s_tui.sources.source.Source._get_total_core_count", return_value=5
)
mocker.patch(
"s_tui.sources.source.Source._get_online_cpu_ids",
return_value=[0, 1, 2, 3],
)
src = UtilSource()
assert len(src.get_sensor_list()) == 6 # Avg + 5 cores
# Core comes online: 5 values now returned
mocker.patch("psutil.cpu_percent", return_value=[25.0, 30.0, 20.0, 15.0, 10.0])
mocker.patch(
"s_tui.sources.source.Source._get_online_cpu_ids",
return_value=[0, 1, 2, 3, 4],
)
src.update()
# This should not crash
summary = src.get_sensors_summary()
assert summary is not None
def test_core_count_shrinks(self, mocker):
"""Simulate a core going offline mid-run.
The code marks offline cores as N/A and keeps the sensor list
at its original length.
"""
mocker.patch("psutil.cpu_count", return_value=4)
mocker.patch("psutil.cpu_percent", return_value=[25.0, 30.0, 20.0, 15.0])
mocker.patch(
"s_tui.sources.source.Source._get_total_core_count", return_value=4
)
mocker.patch(
"s_tui.sources.source.Source._get_online_cpu_ids",
return_value=[0, 1, 2, 3],
)
src = UtilSource()
assert len(src.get_sensor_list()) == 5
# Core goes offline: only 3 values, online_ids shrinks
mocker.patch("psutil.cpu_percent", return_value=[25.0, 30.0, 20.0])
mocker.patch(
"s_tui.sources.source.Source._get_online_cpu_ids",
return_value=[0, 1, 2],
)
src.update()
# Does not crash; sensor_list keeps original length
summary = src.get_sensors_summary()
assert summary is not None
# =====================================================================
# FreqSource -- core count changes
# =====================================================================
class TestFreqCoreCountChanges:
def test_freq_core_count_shrinks(self, mocker):
"""cpu_freq(percpu=True) returns fewer cores after init.
The code marks disappeared cores as N/A without crashing.
"""
per_cpu_4 = make_cpu_freq_list(4)
overall = make_cpu_freq_overall()
def _freq_init(percpu=False):
return per_cpu_4 if percpu else overall
mocker.patch("psutil.cpu_freq", side_effect=_freq_init)
mocker.patch(
"s_tui.sources.source.Source._get_total_core_count", return_value=4
)
mocker.patch(
"s_tui.sources.source.Source._get_online_cpu_ids",
return_value=[0, 1, 2, 3],
)
src = FreqSource()
assert len(src.get_sensor_list()) == 5
# Now fewer cores
per_cpu_2 = make_cpu_freq_list(2)
def _freq_shrunk(percpu=False):
return per_cpu_2 if percpu else overall
mocker.patch("psutil.cpu_freq", side_effect=_freq_shrunk)
mocker.patch(
"s_tui.sources.source.Source._get_online_cpu_ids",
return_value=[0, 1],
)
src.update()
summary = src.get_sensors_summary()
assert summary is not None
# =====================================================================
# TempSource -- sensor appear / disappear
# =====================================================================
class TestTempSensorChanges:
def test_temp_sensor_disappears(self, mocker):
"""sensors_temperatures() returns fewer sensors mid-run.
Disappeared sensor is marked N/A via sensor_available.
"""
sensors_2 = [
SensorTemperature(label="Core 0", current=55.0, high=80.0, critical=100.0),
SensorTemperature(label="Core 1", current=60.0, high=80.0, critical=100.0),
]
temps_2 = OrderedDict([("coretemp", sensors_2)])
mocker.patch("psutil.sensors_temperatures", return_value=temps_2)
src = TempSource()
assert len(src.get_sensor_list()) == 2
# Sensor disappears
sensors_1 = [
SensorTemperature(label="Core 0", current=55.0, high=80.0, critical=100.0),
]
temps_1 = OrderedDict([("coretemp", sensors_1)])
mocker.patch("psutil.sensors_temperatures", return_value=temps_1)
src.update()
summary = src.get_sensors_summary()
assert summary is not None
assert len(src.get_sensor_list()) == 2
# Core 0 still available, Core 1 disappeared
assert src.sensor_available[0] is True
assert src.sensor_available[1] is False
# Summary shows N/A for disappeared sensor
values = list(summary.values())
assert values[1] == "N/A"
def test_temp_sensor_appears(self, mocker):
"""New sensor shows up in sensors_temperatures() mid-run."""
sensors_1 = [
SensorTemperature(label="Core 0", current=55.0, high=80.0, critical=100.0),
]
temps_1 = OrderedDict([("coretemp", sensors_1)])
mocker.patch("psutil.sensors_temperatures", return_value=temps_1)
src = TempSource()
assert len(src.get_sensor_list()) == 1
# New sensor appears
sensors_2 = [
SensorTemperature(label="Core 0", current=55.0, high=80.0, critical=100.0),
SensorTemperature(label="Core 1", current=60.0, high=80.0, critical=100.0),
]
temps_2 = OrderedDict([("coretemp", sensors_2)])
mocker.patch("psutil.sensors_temperatures", return_value=temps_2)
src.update()
summary = src.get_sensors_summary()
assert len(summary) == len(src.get_reading_list())
# =====================================================================
# FanSource -- sensor disappear / None
# =====================================================================
class TestFanSensorChanges:
def test_fan_returns_none_during_update(self, mocker):
"""sensors_fans() returns None mid-run — keeps stale data (GH-256)."""
fans = make_fans_dict(count=1)
mocker.patch("psutil.sensors_fans", return_value=fans)
src = FanSource()
assert src.get_is_available() is True
src.update()
prev_measurement = list(src.last_measurement)
# Mid-run: returns None
mocker.patch("psutil.sensors_fans", return_value=None)
src.update() # should not crash
# Stale data preserved
assert src.last_measurement == prev_measurement
def test_fan_typeerror_during_update(self, mocker):
"""sensors_fans() raises TypeError mid-run — keeps stale data (GH-256)."""
fans = make_fans_dict(count=1)
mocker.patch("psutil.sensors_fans", return_value=fans)
src = FanSource()
assert src.get_is_available() is True
src.update()
prev_measurement = list(src.last_measurement)
# Mid-run: psutil raises TypeError (sysfs None bug)
mocker.patch("psutil.sensors_fans", side_effect=TypeError)
src.update() # should not crash
# Stale data preserved
assert src.last_measurement == prev_measurement
def test_fan_sensor_disappears(self, mocker):
"""sensors_fans() returns empty dict mid-run.
Disappeared fan sensor is marked N/A via sensor_available.
"""
fans = make_fans_dict(count=1)
mocker.patch("psutil.sensors_fans", return_value=fans)
src = FanSource()
assert len(src.get_sensor_list()) == 1
# Sensor disappears
mocker.patch("psutil.sensors_fans", return_value={})
src.update()
summary = src.get_sensors_summary()
assert summary is not None
assert len(src.get_sensor_list()) == 1
assert src.sensor_available[0] is False
# Summary shows N/A for disappeared sensor
values = list(summary.values())
assert values[0] == "N/A"
# =====================================================================
# RaplPowerSource -- reader failure mid-run
# =====================================================================
class TestRaplReaderFailsMidRun:
def test_rapl_reader_fails_during_update(self, mocker):
"""RAPL energy file becomes unreadable mid-run."""
reader = MagicMock()
reader.read_power.return_value = [
RaplStats(label="pkg", current=1000000.0, max=0.0),
]
mocker.patch(
"s_tui.sources.rapl_power_source.get_power_reader", return_value=reader
)
src = RaplPowerSource()
assert src.get_is_available() is True
# File disappears
reader.read_power.side_effect = OSError("permission denied")
src.update() # should not crash
assert src.get_is_available() is True # or False, but no crash
# =====================================================================
# Source.update() exception propagation
# =====================================================================
class TestSourceUpdateExceptionPropagation:
def test_source_update_raises_oserror(self, mocker):
"""Any psutil call raises OSError during update() — should not propagate."""
mocker.patch("psutil.cpu_count", return_value=4)
mocker.patch("psutil.cpu_percent", return_value=[25.0, 30.0, 20.0, 15.0])
mocker.patch(
"s_tui.sources.source.Source._get_total_core_count", return_value=4
)
mocker.patch(
"s_tui.sources.source.Source._get_online_cpu_ids",
return_value=[0, 1, 2, 3],
)
src = UtilSource()
mocker.patch("psutil.cpu_percent", side_effect=OSError("device gone"))
# In current code this propagates unhandled
# The test verifies we want it to NOT propagate
try:
src.update()
except OSError:
pytest.fail("update() should not propagate OSError to caller")
|
