Mercurial > repos > guerler > springsuite

comparison planemo/lib/python3.7/site-packages/psutil/tests/test_bsd.py @ 1:56ad4e20f292 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author guerler
date Fri, 31 Jul 2020 00:32:28 -0400
parents
children
comparison
equal deleted inserted replaced
0:d30785e31577 1:56ad4e20f292
1 #!/usr/bin/env python3
2
3 # Copyright (c) 2009, Giampaolo Rodola'. All rights reserved.
4 # Use of this source code is governed by a BSD-style license that can be
5 # found in the LICENSE file.
6
7 # TODO: (FreeBSD) add test for comparing connections with 'sockstat' cmd.
8
9
10 """Tests specific to all BSD platforms."""
11
12
13 import datetime
14 import os
15 import re
16 import time
17
18 import psutil
19 from psutil import BSD
20 from psutil import FREEBSD
21 from psutil import NETBSD
22 from psutil import OPENBSD
23 from psutil.tests import spawn_testproc
24 from psutil.tests import HAS_BATTERY
25 from psutil.tests import PsutilTestCase
26 from psutil.tests import retry_on_failure
27 from psutil.tests import sh
28 from psutil.tests import TOLERANCE_SYS_MEM
29 from psutil.tests import terminate
30 from psutil.tests import unittest
31 from psutil.tests import which
32
33
34 if BSD:
35 PAGESIZE = os.sysconf("SC_PAGE_SIZE")
36 if os.getuid() == 0: # muse requires root privileges
37 MUSE_AVAILABLE = which('muse')
38 else:
39 MUSE_AVAILABLE = False
40 else:
41 MUSE_AVAILABLE = False
42
43
44 def sysctl(cmdline):
45 """Expects a sysctl command with an argument and parse the result
46 returning only the value of interest.
47 """
48 result = sh("sysctl " + cmdline)
49 if FREEBSD:
50 result = result[result.find(": ") + 2:]
51 elif OPENBSD or NETBSD:
52 result = result[result.find("=") + 1:]
53 try:
54 return int(result)
55 except ValueError:
56 return result
57
58
59 def muse(field):
60 """Thin wrapper around 'muse' cmdline utility."""
61 out = sh('muse')
62 for line in out.split('\n'):
63 if line.startswith(field):
64 break
65 else:
66 raise ValueError("line not found")
67 return int(line.split()[1])
68
69
70 # =====================================================================
71 # --- All BSD*
72 # =====================================================================
73
74
75 @unittest.skipIf(not BSD, "BSD only")
76 class BSDTestCase(PsutilTestCase):
77 """Generic tests common to all BSD variants."""
78
79 @classmethod
80 def setUpClass(cls):
81 cls.pid = spawn_testproc().pid
82
83 @classmethod
84 def tearDownClass(cls):
85 terminate(cls.pid)
86
87 @unittest.skipIf(NETBSD, "-o lstart doesn't work on NETBSD")
88 def test_process_create_time(self):
89 output = sh("ps -o lstart -p %s" % self.pid)
90 start_ps = output.replace('STARTED', '').strip()
91 start_psutil = psutil.Process(self.pid).create_time()
92 start_psutil = time.strftime("%a %b %e %H:%M:%S %Y",
93 time.localtime(start_psutil))
94 self.assertEqual(start_ps, start_psutil)
95
96 def test_disks(self):
97 # test psutil.disk_usage() and psutil.disk_partitions()
98 # against "df -a"
99 def df(path):
100 out = sh('df -k "%s"' % path).strip()
101 lines = out.split('\n')
102 lines.pop(0)
103 line = lines.pop(0)
104 dev, total, used, free = line.split()[:4]
105 if dev == 'none':
106 dev = ''
107 total = int(total) * 1024
108 used = int(used) * 1024
109 free = int(free) * 1024
110 return dev, total, used, free
111
112 for part in psutil.disk_partitions(all=False):
113 usage = psutil.disk_usage(part.mountpoint)
114 dev, total, used, free = df(part.mountpoint)
115 self.assertEqual(part.device, dev)
116 self.assertEqual(usage.total, total)
117 # 10 MB tollerance
118 if abs(usage.free - free) > 10 * 1024 * 1024:
119 self.fail("psutil=%s, df=%s" % (usage.free, free))
120 if abs(usage.used - used) > 10 * 1024 * 1024:
121 self.fail("psutil=%s, df=%s" % (usage.used, used))
122
123 @unittest.skipIf(not which('sysctl'), "sysctl cmd not available")
124 def test_cpu_count_logical(self):
125 syst = sysctl("hw.ncpu")
126 self.assertEqual(psutil.cpu_count(logical=True), syst)
127
128 @unittest.skipIf(not which('sysctl'), "sysctl cmd not available")
129 def test_virtual_memory_total(self):
130 num = sysctl('hw.physmem')
131 self.assertEqual(num, psutil.virtual_memory().total)
132
133 def test_net_if_stats(self):
134 for name, stats in psutil.net_if_stats().items():
135 try:
136 out = sh("ifconfig %s" % name)
137 except RuntimeError:
138 pass
139 else:
140 self.assertEqual(stats.isup, 'RUNNING' in out, msg=out)
141 if "mtu" in out:
142 self.assertEqual(stats.mtu,
143 int(re.findall(r'mtu (\d+)', out)[0]))
144
145
146 # =====================================================================
147 # --- FreeBSD
148 # =====================================================================
149
150
151 @unittest.skipIf(not FREEBSD, "FREEBSD only")
152 class FreeBSDPsutilTestCase(PsutilTestCase):
153
154 @classmethod
155 def setUpClass(cls):
156 cls.pid = spawn_testproc().pid
157
158 @classmethod
159 def tearDownClass(cls):
160 terminate(cls.pid)
161
162 @retry_on_failure()
163 def test_memory_maps(self):
164 out = sh('procstat -v %s' % self.pid)
165 maps = psutil.Process(self.pid).memory_maps(grouped=False)
166 lines = out.split('\n')[1:]
167 while lines:
168 line = lines.pop()
169 fields = line.split()
170 _, start, stop, perms, res = fields[:5]
171 map = maps.pop()
172 self.assertEqual("%s-%s" % (start, stop), map.addr)
173 self.assertEqual(int(res), map.rss)
174 if not map.path.startswith('['):
175 self.assertEqual(fields[10], map.path)
176
177 def test_exe(self):
178 out = sh('procstat -b %s' % self.pid)
179 self.assertEqual(psutil.Process(self.pid).exe(),
180 out.split('\n')[1].split()[-1])
181
182 def test_cmdline(self):
183 out = sh('procstat -c %s' % self.pid)
184 self.assertEqual(' '.join(psutil.Process(self.pid).cmdline()),
185 ' '.join(out.split('\n')[1].split()[2:]))
186
187 def test_uids_gids(self):
188 out = sh('procstat -s %s' % self.pid)
189 euid, ruid, suid, egid, rgid, sgid = out.split('\n')[1].split()[2:8]
190 p = psutil.Process(self.pid)
191 uids = p.uids()
192 gids = p.gids()
193 self.assertEqual(uids.real, int(ruid))
194 self.assertEqual(uids.effective, int(euid))
195 self.assertEqual(uids.saved, int(suid))
196 self.assertEqual(gids.real, int(rgid))
197 self.assertEqual(gids.effective, int(egid))
198 self.assertEqual(gids.saved, int(sgid))
199
200 @retry_on_failure()
201 def test_ctx_switches(self):
202 tested = []
203 out = sh('procstat -r %s' % self.pid)
204 p = psutil.Process(self.pid)
205 for line in out.split('\n'):
206 line = line.lower().strip()
207 if ' voluntary context' in line:
208 pstat_value = int(line.split()[-1])
209 psutil_value = p.num_ctx_switches().voluntary
210 self.assertEqual(pstat_value, psutil_value)
211 tested.append(None)
212 elif ' involuntary context' in line:
213 pstat_value = int(line.split()[-1])
214 psutil_value = p.num_ctx_switches().involuntary
215 self.assertEqual(pstat_value, psutil_value)
216 tested.append(None)
217 if len(tested) != 2:
218 raise RuntimeError("couldn't find lines match in procstat out")
219
220 @retry_on_failure()
221 def test_cpu_times(self):
222 tested = []
223 out = sh('procstat -r %s' % self.pid)
224 p = psutil.Process(self.pid)
225 for line in out.split('\n'):
226 line = line.lower().strip()
227 if 'user time' in line:
228 pstat_value = float('0.' + line.split()[-1].split('.')[-1])
229 psutil_value = p.cpu_times().user
230 self.assertEqual(pstat_value, psutil_value)
231 tested.append(None)
232 elif 'system time' in line:
233 pstat_value = float('0.' + line.split()[-1].split('.')[-1])
234 psutil_value = p.cpu_times().system
235 self.assertEqual(pstat_value, psutil_value)
236 tested.append(None)
237 if len(tested) != 2:
238 raise RuntimeError("couldn't find lines match in procstat out")
239
240
241 @unittest.skipIf(not FREEBSD, "FREEBSD only")
242 class FreeBSDSystemTestCase(PsutilTestCase):
243
244 @staticmethod
245 def parse_swapinfo():
246 # the last line is always the total
247 output = sh("swapinfo -k").splitlines()[-1]
248 parts = re.split(r'\s+', output)
249
250 if not parts:
251 raise ValueError("Can't parse swapinfo: %s" % output)
252
253 # the size is in 1k units, so multiply by 1024
254 total, used, free = (int(p) * 1024 for p in parts[1:4])
255 return total, used, free
256
257 def test_cpu_frequency_against_sysctl(self):
258 # Currently only cpu 0 is frequency is supported in FreeBSD
259 # All other cores use the same frequency.
260 sensor = "dev.cpu.0.freq"
261 try:
262 sysctl_result = int(sysctl(sensor))
263 except RuntimeError:
264 self.skipTest("frequencies not supported by kernel")
265 self.assertEqual(psutil.cpu_freq().current, sysctl_result)
266
267 sensor = "dev.cpu.0.freq_levels"
268 sysctl_result = sysctl(sensor)
269 # sysctl returns a string of the format:
270 # <freq_level_1>/<voltage_level_1> <freq_level_2>/<voltage_level_2>...
271 # Ordered highest available to lowest available.
272 max_freq = int(sysctl_result.split()[0].split("/")[0])
273 min_freq = int(sysctl_result.split()[-1].split("/")[0])
274 self.assertEqual(psutil.cpu_freq().max, max_freq)
275 self.assertEqual(psutil.cpu_freq().min, min_freq)
276
277 # --- virtual_memory(); tests against sysctl
278
279 @retry_on_failure()
280 def test_vmem_active(self):
281 syst = sysctl("vm.stats.vm.v_active_count") * PAGESIZE
282 self.assertAlmostEqual(psutil.virtual_memory().active, syst,
283 delta=TOLERANCE_SYS_MEM)
284
285 @retry_on_failure()
286 def test_vmem_inactive(self):
287 syst = sysctl("vm.stats.vm.v_inactive_count") * PAGESIZE
288 self.assertAlmostEqual(psutil.virtual_memory().inactive, syst,
289 delta=TOLERANCE_SYS_MEM)
290
291 @retry_on_failure()
292 def test_vmem_wired(self):
293 syst = sysctl("vm.stats.vm.v_wire_count") * PAGESIZE
294 self.assertAlmostEqual(psutil.virtual_memory().wired, syst,
295 delta=TOLERANCE_SYS_MEM)
296
297 @retry_on_failure()
298 def test_vmem_cached(self):
299 syst = sysctl("vm.stats.vm.v_cache_count") * PAGESIZE
300 self.assertAlmostEqual(psutil.virtual_memory().cached, syst,
301 delta=TOLERANCE_SYS_MEM)
302
303 @retry_on_failure()
304 def test_vmem_free(self):
305 syst = sysctl("vm.stats.vm.v_free_count") * PAGESIZE
306 self.assertAlmostEqual(psutil.virtual_memory().free, syst,
307 delta=TOLERANCE_SYS_MEM)
308
309 @retry_on_failure()
310 def test_vmem_buffers(self):
311 syst = sysctl("vfs.bufspace")
312 self.assertAlmostEqual(psutil.virtual_memory().buffers, syst,
313 delta=TOLERANCE_SYS_MEM)
314
315 # --- virtual_memory(); tests against muse
316
317 @unittest.skipIf(not MUSE_AVAILABLE, "muse not installed")
318 def test_muse_vmem_total(self):
319 num = muse('Total')
320 self.assertEqual(psutil.virtual_memory().total, num)
321
322 @unittest.skipIf(not MUSE_AVAILABLE, "muse not installed")
323 @retry_on_failure()
324 def test_muse_vmem_active(self):
325 num = muse('Active')
326 self.assertAlmostEqual(psutil.virtual_memory().active, num,
327 delta=TOLERANCE_SYS_MEM)
328
329 @unittest.skipIf(not MUSE_AVAILABLE, "muse not installed")
330 @retry_on_failure()
331 def test_muse_vmem_inactive(self):
332 num = muse('Inactive')
333 self.assertAlmostEqual(psutil.virtual_memory().inactive, num,
334 delta=TOLERANCE_SYS_MEM)
335
336 @unittest.skipIf(not MUSE_AVAILABLE, "muse not installed")
337 @retry_on_failure()
338 def test_muse_vmem_wired(self):
339 num = muse('Wired')
340 self.assertAlmostEqual(psutil.virtual_memory().wired, num,
341 delta=TOLERANCE_SYS_MEM)
342
343 @unittest.skipIf(not MUSE_AVAILABLE, "muse not installed")
344 @retry_on_failure()
345 def test_muse_vmem_cached(self):
346 num = muse('Cache')
347 self.assertAlmostEqual(psutil.virtual_memory().cached, num,
348 delta=TOLERANCE_SYS_MEM)
349
350 @unittest.skipIf(not MUSE_AVAILABLE, "muse not installed")
351 @retry_on_failure()
352 def test_muse_vmem_free(self):
353 num = muse('Free')
354 self.assertAlmostEqual(psutil.virtual_memory().free, num,
355 delta=TOLERANCE_SYS_MEM)
356
357 @unittest.skipIf(not MUSE_AVAILABLE, "muse not installed")
358 @retry_on_failure()
359 def test_muse_vmem_buffers(self):
360 num = muse('Buffer')
361 self.assertAlmostEqual(psutil.virtual_memory().buffers, num,
362 delta=TOLERANCE_SYS_MEM)
363
364 def test_cpu_stats_ctx_switches(self):
365 self.assertAlmostEqual(psutil.cpu_stats().ctx_switches,
366 sysctl('vm.stats.sys.v_swtch'), delta=1000)
367
368 def test_cpu_stats_interrupts(self):
369 self.assertAlmostEqual(psutil.cpu_stats().interrupts,
370 sysctl('vm.stats.sys.v_intr'), delta=1000)
371
372 def test_cpu_stats_soft_interrupts(self):
373 self.assertAlmostEqual(psutil.cpu_stats().soft_interrupts,
374 sysctl('vm.stats.sys.v_soft'), delta=1000)
375
376 @retry_on_failure()
377 def test_cpu_stats_syscalls(self):
378 # pretty high tolerance but it looks like it's OK.
379 self.assertAlmostEqual(psutil.cpu_stats().syscalls,
380 sysctl('vm.stats.sys.v_syscall'), delta=200000)
381
382 # def test_cpu_stats_traps(self):
383 # self.assertAlmostEqual(psutil.cpu_stats().traps,
384 # sysctl('vm.stats.sys.v_trap'), delta=1000)
385
386 # --- swap memory
387
388 def test_swapmem_free(self):
389 total, used, free = self.parse_swapinfo()
390 self.assertAlmostEqual(
391 psutil.swap_memory().free, free, delta=TOLERANCE_SYS_MEM)
392
393 def test_swapmem_used(self):
394 total, used, free = self.parse_swapinfo()
395 self.assertAlmostEqual(
396 psutil.swap_memory().used, used, delta=TOLERANCE_SYS_MEM)
397
398 def test_swapmem_total(self):
399 total, used, free = self.parse_swapinfo()
400 self.assertAlmostEqual(
401 psutil.swap_memory().total, total, delta=TOLERANCE_SYS_MEM)
402
403 # --- others
404
405 def test_boot_time(self):
406 s = sysctl('sysctl kern.boottime')
407 s = s[s.find(" sec = ") + 7:]
408 s = s[:s.find(',')]
409 btime = int(s)
410 self.assertEqual(btime, psutil.boot_time())
411
412 # --- sensors_battery
413
414 @unittest.skipIf(not HAS_BATTERY, "no battery")
415 def test_sensors_battery(self):
416 def secs2hours(secs):
417 m, s = divmod(secs, 60)
418 h, m = divmod(m, 60)
419 return "%d:%02d" % (h, m)
420
421 out = sh("acpiconf -i 0")
422 fields = dict([(x.split('\t')[0], x.split('\t')[-1])
423 for x in out.split("\n")])
424 metrics = psutil.sensors_battery()
425 percent = int(fields['Remaining capacity:'].replace('%', ''))
426 remaining_time = fields['Remaining time:']
427 self.assertEqual(metrics.percent, percent)
428 if remaining_time == 'unknown':
429 self.assertEqual(metrics.secsleft, psutil.POWER_TIME_UNLIMITED)
430 else:
431 self.assertEqual(secs2hours(metrics.secsleft), remaining_time)
432
433 @unittest.skipIf(not HAS_BATTERY, "no battery")
434 def test_sensors_battery_against_sysctl(self):
435 self.assertEqual(psutil.sensors_battery().percent,
436 sysctl("hw.acpi.battery.life"))
437 self.assertEqual(psutil.sensors_battery().power_plugged,
438 sysctl("hw.acpi.acline") == 1)
439 secsleft = psutil.sensors_battery().secsleft
440 if secsleft < 0:
441 self.assertEqual(sysctl("hw.acpi.battery.time"), -1)
442 else:
443 self.assertEqual(secsleft, sysctl("hw.acpi.battery.time") * 60)
444
445 @unittest.skipIf(HAS_BATTERY, "has battery")
446 def test_sensors_battery_no_battery(self):
447 # If no battery is present one of these calls is supposed
448 # to fail, see:
449 # https://github.com/giampaolo/psutil/issues/1074
450 with self.assertRaises(RuntimeError):
451 sysctl("hw.acpi.battery.life")
452 sysctl("hw.acpi.battery.time")
453 sysctl("hw.acpi.acline")
454 self.assertIsNone(psutil.sensors_battery())
455
456 # --- sensors_temperatures
457
458 def test_sensors_temperatures_against_sysctl(self):
459 num_cpus = psutil.cpu_count(True)
460 for cpu in range(num_cpus):
461 sensor = "dev.cpu.%s.temperature" % cpu
462 # sysctl returns a string in the format 46.0C
463 try:
464 sysctl_result = int(float(sysctl(sensor)[:-1]))
465 except RuntimeError:
466 self.skipTest("temperatures not supported by kernel")
467 self.assertAlmostEqual(
468 psutil.sensors_temperatures()["coretemp"][cpu].current,
469 sysctl_result, delta=10)
470
471 sensor = "dev.cpu.%s.coretemp.tjmax" % cpu
472 sysctl_result = int(float(sysctl(sensor)[:-1]))
473 self.assertEqual(
474 psutil.sensors_temperatures()["coretemp"][cpu].high,
475 sysctl_result)
476
477 # =====================================================================
478 # --- OpenBSD
479 # =====================================================================
480
481
482 @unittest.skipIf(not OPENBSD, "OPENBSD only")
483 class OpenBSDTestCase(PsutilTestCase):
484
485 def test_boot_time(self):
486 s = sysctl('kern.boottime')
487 sys_bt = datetime.datetime.strptime(s, "%a %b %d %H:%M:%S %Y")
488 psutil_bt = datetime.datetime.fromtimestamp(psutil.boot_time())
489 self.assertEqual(sys_bt, psutil_bt)
490
491
492 # =====================================================================
493 # --- NetBSD
494 # =====================================================================
495
496
497 @unittest.skipIf(not NETBSD, "NETBSD only")
498 class NetBSDTestCase(PsutilTestCase):
499
500 @staticmethod
501 def parse_meminfo(look_for):
502 with open('/proc/meminfo', 'rt') as f:
503 for line in f:
504 if line.startswith(look_for):
505 return int(line.split()[1]) * 1024
506 raise ValueError("can't find %s" % look_for)
507
508 def test_vmem_total(self):
509 self.assertEqual(
510 psutil.virtual_memory().total, self.parse_meminfo("MemTotal:"))
511
512 def test_vmem_free(self):
513 self.assertAlmostEqual(
514 psutil.virtual_memory().free, self.parse_meminfo("MemFree:"),
515 delta=TOLERANCE_SYS_MEM)
516
517 def test_vmem_buffers(self):
518 self.assertAlmostEqual(
519 psutil.virtual_memory().buffers, self.parse_meminfo("Buffers:"),
520 delta=TOLERANCE_SYS_MEM)
521
522 def test_vmem_shared(self):
523 self.assertAlmostEqual(
524 psutil.virtual_memory().shared, self.parse_meminfo("MemShared:"),
525 delta=TOLERANCE_SYS_MEM)
526
527 def test_swapmem_total(self):
528 self.assertAlmostEqual(
529 psutil.swap_memory().total, self.parse_meminfo("SwapTotal:"),
530 delta=TOLERANCE_SYS_MEM)
531
532 def test_swapmem_free(self):
533 self.assertAlmostEqual(
534 psutil.swap_memory().free, self.parse_meminfo("SwapFree:"),
535 delta=TOLERANCE_SYS_MEM)
536
537 def test_swapmem_used(self):
538 smem = psutil.swap_memory()
539 self.assertEqual(smem.used, smem.total - smem.free)
540
541 def test_cpu_stats_interrupts(self):
542 with open('/proc/stat', 'rb') as f:
543 for line in f:
544 if line.startswith(b'intr'):
545 interrupts = int(line.split()[1])
546 break
547 else:
548 raise ValueError("couldn't find line")
549 self.assertAlmostEqual(
550 psutil.cpu_stats().interrupts, interrupts, delta=1000)
551
552 def test_cpu_stats_ctx_switches(self):
553 with open('/proc/stat', 'rb') as f:
554 for line in f:
555 if line.startswith(b'ctxt'):
556 ctx_switches = int(line.split()[1])
557 break
558 else:
559 raise ValueError("couldn't find line")
560 self.assertAlmostEqual(
561 psutil.cpu_stats().ctx_switches, ctx_switches, delta=1000)
562
563
564 if __name__ == '__main__':
565 from psutil.tests.runner import run_from_name
566 run_from_name(__file__)