python/psutil/test/_bsd.py@b8a032363ba2 (annotated)
python/psutil/test/_bsd.py
Thu, 22 Jan 2015 13:21:57 +0100
- author
- Michael Schloh von Bennewitz <michael@schloh.com>
- date
- Thu, 22 Jan 2015 13:21:57 +0100
- branch
- TOR_BUG_9701
- changeset 15
- b8a032363ba2
- permissions
- -rw-r--r--
Incorporate requested changes from Mozilla in review:
https://bugzilla.mozilla.org/show_bug.cgi?id=1123480#c6
| michael@0 | 1 | #!/usr/bin/env python |
| michael@0 | 2 | |
| michael@0 | 3 | # Copyright (c) 2009, Giampaolo Rodola'. All rights reserved. |
| michael@0 | 4 | # Use of this source code is governed by a BSD-style license that can be |
| michael@0 | 5 | # found in the LICENSE file. |
| michael@0 | 6 | |
| michael@0 | 7 | """BSD specific tests. These are implicitly run by test_psutil.py.""" |
| michael@0 | 8 | |
| michael@0 | 9 | import unittest |
| michael@0 | 10 | import subprocess |
| michael@0 | 11 | import time |
| michael@0 | 12 | import re |
| michael@0 | 13 | import sys |
| michael@0 | 14 | import os |
| michael@0 | 15 | |
| michael@0 | 16 | import psutil |
| michael@0 | 17 | |
| michael@0 | 18 | from psutil._compat import PY3 |
| michael@0 | 19 | from test_psutil import * |
| michael@0 | 20 | |
| michael@0 | 21 | |
| michael@0 | 22 | PAGESIZE = os.sysconf("SC_PAGE_SIZE") |
| michael@0 | 23 | MUSE_AVAILABLE = which('muse') |
| michael@0 | 24 | |
| michael@0 | 25 | |
| michael@0 | 26 | def sysctl(cmdline): |
| michael@0 | 27 | """Expects a sysctl command with an argument and parse the result |
| michael@0 | 28 | returning only the value of interest. |
| michael@0 | 29 | """ |
| michael@0 | 30 | result = sh("sysctl " + cmdline) |
| michael@0 | 31 | result = result[result.find(": ") + 2:] |
| michael@0 | 32 | try: |
| michael@0 | 33 | return int(result) |
| michael@0 | 34 | except ValueError: |
| michael@0 | 35 | return result |
| michael@0 | 36 | |
| michael@0 | 37 | def muse(field): |
| michael@0 | 38 | """Thin wrapper around 'muse' cmdline utility.""" |
| michael@0 | 39 | out = sh('muse', stderr=DEVNULL) |
| michael@0 | 40 | for line in out.split('\n'): |
| michael@0 | 41 | if line.startswith(field): |
| michael@0 | 42 | break |
| michael@0 | 43 | else: |
| michael@0 | 44 | raise ValueError("line not found") |
| michael@0 | 45 | return int(line.split()[1]) |
| michael@0 | 46 | |
| michael@0 | 47 | |
| michael@0 | 48 | class BSDSpecificTestCase(unittest.TestCase): |
| michael@0 | 49 | |
| michael@0 | 50 | def setUp(self): |
| michael@0 | 51 | self.pid = get_test_subprocess().pid |
| michael@0 | 52 | |
| michael@0 | 53 | def tearDown(self): |
| michael@0 | 54 | reap_children() |
| michael@0 | 55 | |
| michael@0 | 56 | def test_BOOT_TIME(self): |
| michael@0 | 57 | s = sysctl('sysctl kern.boottime') |
| michael@0 | 58 | s = s[s.find(" sec = ") + 7:] |
| michael@0 | 59 | s = s[:s.find(',')] |
| michael@0 | 60 | btime = int(s) |
| michael@0 | 61 | self.assertEqual(btime, psutil.BOOT_TIME) |
| michael@0 | 62 | |
| michael@0 | 63 | def test_process_create_time(self): |
| michael@0 | 64 | cmdline = "ps -o lstart -p %s" %self.pid |
| michael@0 | 65 | p = subprocess.Popen(cmdline, shell=1, stdout=subprocess.PIPE) |
| michael@0 | 66 | output = p.communicate()[0] |
| michael@0 | 67 | if PY3: |
| michael@0 | 68 | output = str(output, sys.stdout.encoding) |
| michael@0 | 69 | start_ps = output.replace('STARTED', '').strip() |
| michael@0 | 70 | start_psutil = psutil.Process(self.pid).create_time |
| michael@0 | 71 | start_psutil = time.strftime("%a %b %e %H:%M:%S %Y", |
| michael@0 | 72 | time.localtime(start_psutil)) |
| michael@0 | 73 | self.assertEqual(start_ps, start_psutil) |
| michael@0 | 74 | |
| michael@0 | 75 | def test_disks(self): |
| michael@0 | 76 | # test psutil.disk_usage() and psutil.disk_partitions() |
| michael@0 | 77 | # against "df -a" |
| michael@0 | 78 | def df(path): |
| michael@0 | 79 | out = sh('df -k "%s"' % path).strip() |
| michael@0 | 80 | lines = out.split('\n') |
| michael@0 | 81 | lines.pop(0) |
| michael@0 | 82 | line = lines.pop(0) |
| michael@0 | 83 | dev, total, used, free = line.split()[:4] |
| michael@0 | 84 | if dev == 'none': |
| michael@0 | 85 | dev = '' |
| michael@0 | 86 | total = int(total) * 1024 |
| michael@0 | 87 | used = int(used) * 1024 |
| michael@0 | 88 | free = int(free) * 1024 |
| michael@0 | 89 | return dev, total, used, free |
| michael@0 | 90 | |
| michael@0 | 91 | for part in psutil.disk_partitions(all=False): |
| michael@0 | 92 | usage = psutil.disk_usage(part.mountpoint) |
| michael@0 | 93 | dev, total, used, free = df(part.mountpoint) |
| michael@0 | 94 | self.assertEqual(part.device, dev) |
| michael@0 | 95 | self.assertEqual(usage.total, total) |
| michael@0 | 96 | # 10 MB tollerance |
| michael@0 | 97 | if abs(usage.free - free) > 10 * 1024 * 1024: |
| michael@0 | 98 | self.fail("psutil=%s, df=%s" % (usage.free, free)) |
| michael@0 | 99 | if abs(usage.used - used) > 10 * 1024 * 1024: |
| michael@0 | 100 | self.fail("psutil=%s, df=%s" % (usage.used, used)) |
| michael@0 | 101 | |
| michael@0 | 102 | def test_memory_maps(self): |
| michael@0 | 103 | out = sh('procstat -v %s' % self.pid) |
| michael@0 | 104 | maps = psutil.Process(self.pid).get_memory_maps(grouped=False) |
| michael@0 | 105 | lines = out.split('\n')[1:] |
| michael@0 | 106 | while lines: |
| michael@0 | 107 | line = lines.pop() |
| michael@0 | 108 | fields = line.split() |
| michael@0 | 109 | _, start, stop, perms, res = fields[:5] |
| michael@0 | 110 | map = maps.pop() |
| michael@0 | 111 | self.assertEqual("%s-%s" % (start, stop), map.addr) |
| michael@0 | 112 | self.assertEqual(int(res), map.rss) |
| michael@0 | 113 | if not map.path.startswith('['): |
| michael@0 | 114 | self.assertEqual(fields[10], map.path) |
| michael@0 | 115 | |
| michael@0 | 116 | # --- virtual_memory(); tests against sysctl |
| michael@0 | 117 | |
| michael@0 | 118 | def test_vmem_total(self): |
| michael@0 | 119 | syst = sysctl("sysctl vm.stats.vm.v_page_count") * PAGESIZE |
| michael@0 | 120 | self.assertEqual(psutil.virtual_memory().total, syst) |
| michael@0 | 121 | |
| michael@0 | 122 | @retry_before_failing() |
| michael@0 | 123 | def test_vmem_active(self): |
| michael@0 | 124 | syst = sysctl("vm.stats.vm.v_active_count") * PAGESIZE |
| michael@0 | 125 | self.assertAlmostEqual(psutil.virtual_memory().active, syst, |
| michael@0 | 126 | delta=TOLERANCE) |
| michael@0 | 127 | |
| michael@0 | 128 | @retry_before_failing() |
| michael@0 | 129 | def test_vmem_inactive(self): |
| michael@0 | 130 | syst = sysctl("vm.stats.vm.v_inactive_count") * PAGESIZE |
| michael@0 | 131 | self.assertAlmostEqual(psutil.virtual_memory().inactive, syst, |
| michael@0 | 132 | delta=TOLERANCE) |
| michael@0 | 133 | |
| michael@0 | 134 | @retry_before_failing() |
| michael@0 | 135 | def test_vmem_wired(self): |
| michael@0 | 136 | syst = sysctl("vm.stats.vm.v_wire_count") * PAGESIZE |
| michael@0 | 137 | self.assertAlmostEqual(psutil.virtual_memory().wired, syst, |
| michael@0 | 138 | delta=TOLERANCE) |
| michael@0 | 139 | |
| michael@0 | 140 | @retry_before_failing() |
| michael@0 | 141 | def test_vmem_cached(self): |
| michael@0 | 142 | syst = sysctl("vm.stats.vm.v_cache_count") * PAGESIZE |
| michael@0 | 143 | self.assertAlmostEqual(psutil.virtual_memory().cached, syst, |
| michael@0 | 144 | delta=TOLERANCE) |
| michael@0 | 145 | |
| michael@0 | 146 | @retry_before_failing() |
| michael@0 | 147 | def test_vmem_free(self): |
| michael@0 | 148 | syst = sysctl("vm.stats.vm.v_free_count") * PAGESIZE |
| michael@0 | 149 | self.assertAlmostEqual(psutil.virtual_memory().free, syst, |
| michael@0 | 150 | delta=TOLERANCE) |
| michael@0 | 151 | |
| michael@0 | 152 | @retry_before_failing() |
| michael@0 | 153 | def test_vmem_buffers(self): |
| michael@0 | 154 | syst = sysctl("vfs.bufspace") |
| michael@0 | 155 | self.assertAlmostEqual(psutil.virtual_memory().buffers, syst, |
| michael@0 | 156 | delta=TOLERANCE) |
| michael@0 | 157 | |
| michael@0 | 158 | # --- virtual_memory(); tests against muse |
| michael@0 | 159 | |
| michael@0 | 160 | @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available") |
| michael@0 | 161 | def test_total(self): |
| michael@0 | 162 | num = muse('Total') |
| michael@0 | 163 | self.assertEqual(psutil.virtual_memory().total, num) |
| michael@0 | 164 | |
| michael@0 | 165 | @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available") |
| michael@0 | 166 | @retry_before_failing() |
| michael@0 | 167 | def test_active(self): |
| michael@0 | 168 | num = muse('Active') |
| michael@0 | 169 | self.assertAlmostEqual(psutil.virtual_memory().active, num, |
| michael@0 | 170 | delta=TOLERANCE) |
| michael@0 | 171 | |
| michael@0 | 172 | @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available") |
| michael@0 | 173 | @retry_before_failing() |
| michael@0 | 174 | def test_inactive(self): |
| michael@0 | 175 | num = muse('Inactive') |
| michael@0 | 176 | self.assertAlmostEqual(psutil.virtual_memory().inactive, num, |
| michael@0 | 177 | delta=TOLERANCE) |
| michael@0 | 178 | |
| michael@0 | 179 | @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available") |
| michael@0 | 180 | @retry_before_failing() |
| michael@0 | 181 | def test_wired(self): |
| michael@0 | 182 | num = muse('Wired') |
| michael@0 | 183 | self.assertAlmostEqual(psutil.virtual_memory().wired, num, |
| michael@0 | 184 | delta=TOLERANCE) |
| michael@0 | 185 | |
| michael@0 | 186 | @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available") |
| michael@0 | 187 | @retry_before_failing() |
| michael@0 | 188 | def test_cached(self): |
| michael@0 | 189 | num = muse('Cache') |
| michael@0 | 190 | self.assertAlmostEqual(psutil.virtual_memory().cached, num, |
| michael@0 | 191 | delta=TOLERANCE) |
| michael@0 | 192 | |
| michael@0 | 193 | @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available") |
| michael@0 | 194 | @retry_before_failing() |
| michael@0 | 195 | def test_free(self): |
| michael@0 | 196 | num = muse('Free') |
| michael@0 | 197 | self.assertAlmostEqual(psutil.virtual_memory().free, num, |
| michael@0 | 198 | delta=TOLERANCE) |
| michael@0 | 199 | |
| michael@0 | 200 | @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available") |
| michael@0 | 201 | @retry_before_failing() |
| michael@0 | 202 | def test_buffers(self): |
| michael@0 | 203 | num = muse('Buffer') |
| michael@0 | 204 | self.assertAlmostEqual(psutil.virtual_memory().buffers, num, |
| michael@0 | 205 | delta=TOLERANCE) |
| michael@0 | 206 | |
| michael@0 | 207 | |
| michael@0 | 208 | def test_main(): |
| michael@0 | 209 | test_suite = unittest.TestSuite() |
| michael@0 | 210 | test_suite.addTest(unittest.makeSuite(BSDSpecificTestCase)) |
| michael@0 | 211 | result = unittest.TextTestRunner(verbosity=2).run(test_suite) |
| michael@0 | 212 | return result.wasSuccessful() |
| michael@0 | 213 | |
| michael@0 | 214 | if __name__ == '__main__': |
| michael@0 | 215 | if not test_main(): |
| michael@0 | 216 | sys.exit(1) |
