|
12 | 12 | import getpass
|
13 | 13 | import io
|
14 | 14 | import itertools
|
| 15 | +import multiprocessing |
15 | 16 | import os
|
16 | 17 | import signal
|
17 | 18 | import socket
|
@@ -269,17 +270,33 @@ def test_cpu_times(self):
|
269 | 270 | time.strftime("%H:%M:%S", time.localtime(getattr(times, name)))
|
270 | 271 |
|
271 | 272 | def test_cpu_times_2(self):
|
272 |
| - user_time, kernel_time = psutil.Process().cpu_times()[:2] |
273 |
| - utime, ktime = os.times()[:2] |
274 |
| - |
275 |
| - # Use os.times()[:2] as base values to compare our results |
276 |
| - # using a tolerance of +/- 0.1 seconds. |
277 |
| - # It will fail if the difference between the values is > 0.1s. |
278 |
| - if (max([user_time, utime]) - min([user_time, utime])) > 0.1: |
279 |
| - raise self.fail(f"expected: {utime}, found: {user_time}") |
280 |
| - |
281 |
| - if (max([kernel_time, ktime]) - min([kernel_time, ktime])) > 0.1: |
282 |
| - raise self.fail(f"expected: {ktime}, found: {kernel_time}") |
| 273 | + def waste_cpu(): |
| 274 | + for x in range(100000): |
| 275 | + x **= 2 |
| 276 | + |
| 277 | + while os.times().user < 0.2: |
| 278 | + waste_cpu() |
| 279 | + a = psutil.Process().cpu_times() |
| 280 | + b = os.times() |
| 281 | + self.assertAlmostEqual(a.user, b.user, delta=0.1) |
| 282 | + self.assertAlmostEqual(a.system, b.system, delta=0.1) |
| 283 | + |
| 284 | + def test_cpu_times_3(self): |
| 285 | + # same as above but for process children |
| 286 | + def waste_cpu(): |
| 287 | + while os.times().user < 0.2: |
| 288 | + for x in range(100000): |
| 289 | + x **= 2 |
| 290 | + |
| 291 | + proc = multiprocessing.Process(target=waste_cpu) |
| 292 | + proc.start() |
| 293 | + proc.join() |
| 294 | + |
| 295 | + a = psutil.Process().cpu_times() |
| 296 | + b = os.times() |
| 297 | + assert b.children_user >= 0.2 |
| 298 | + self.assertAlmostEqual(a.children_user, b.children_user, delta=0.1) |
| 299 | + self.assertAlmostEqual(a.children_system, b.children_system, delta=0.1) |
283 | 300 |
|
284 | 301 | @pytest.mark.skipif(not HAS_PROC_CPU_NUM, reason="not supported")
|
285 | 302 | def test_cpu_num(self):
|
|
0 commit comments