diff options
Diffstat (limited to 'python/psutil/test')
| -rw-r--r-- | python/psutil/test/README.rst | 21 | ||||
| -rw-r--r-- | python/psutil/test/_bsd.py | 252 | ||||
| -rw-r--r-- | python/psutil/test/_linux.py | 473 | ||||
| -rw-r--r-- | python/psutil/test/_osx.py | 160 | ||||
| -rw-r--r-- | python/psutil/test/_posix.py | 258 | ||||
| -rw-r--r-- | python/psutil/test/_sunos.py | 48 | ||||
| -rw-r--r-- | python/psutil/test/_windows.py | 464 | ||||
| -rw-r--r-- | python/psutil/test/test_memory_leaks.py | 445 | ||||
| -rw-r--r-- | python/psutil/test/test_psutil.py | 3013 |
9 files changed, 5134 insertions, 0 deletions
diff --git a/python/psutil/test/README.rst b/python/psutil/test/README.rst new file mode 100644 index 0000000000..3f2a468efd --- /dev/null +++ b/python/psutil/test/README.rst @@ -0,0 +1,21 @@ +- The recommended way to run tests (also on Windows) is to cd into parent + directory and run ``make test`` + +- Dependencies for running tests: + - python 2.6: ipaddress, mock, unittest2 + - python 2.7: ipaddress, mock + - python 3.2: ipaddress, mock + - python 3.3: ipaddress + - python >= 3.4: no deps required + +- The main test script is ``test_psutil.py``, which also imports platform-specific + ``_*.py`` scripts (which should be ignored). + +- ``test_memory_leaks.py`` looks for memory leaks into C extension modules and must + be run separately with ``make test-memleaks``. + +- To run tests on all supported Python version install tox (pip install tox) + then run ``tox``. + +- Every time a commit is pushed tests are automatically run on Travis: + https://travis-ci.org/giampaolo/psutil/ diff --git a/python/psutil/test/_bsd.py b/python/psutil/test/_bsd.py new file mode 100644 index 0000000000..e4a3225d20 --- /dev/null +++ b/python/psutil/test/_bsd.py @@ -0,0 +1,252 @@ +#!/usr/bin/env python + +# Copyright (c) 2009, Giampaolo Rodola'. All rights reserved. +# Use of this source code is governed by a BSD-style license that can be +# found in the LICENSE file. + +# TODO: add test for comparing connections with 'sockstat' cmd + +"""BSD specific tests. These are implicitly run by test_psutil.py.""" + +import os +import subprocess +import sys +import time + +import psutil + +from psutil._compat import PY3 +from test_psutil import (TOLERANCE, BSD, sh, get_test_subprocess, which, + retry_before_failing, reap_children, unittest) + + +PAGESIZE = os.sysconf("SC_PAGE_SIZE") +if os.getuid() == 0: # muse requires root privileges + MUSE_AVAILABLE = which('muse') +else: + MUSE_AVAILABLE = False + + +def sysctl(cmdline): + """Expects a sysctl command with an argument and parse the result + returning only the value of interest. + """ + result = sh("sysctl " + cmdline) + result = result[result.find(": ") + 2:] + try: + return int(result) + except ValueError: + return result + + +def muse(field): + """Thin wrapper around 'muse' cmdline utility.""" + out = sh('muse') + for line in out.split('\n'): + if line.startswith(field): + break + else: + raise ValueError("line not found") + return int(line.split()[1]) + + +@unittest.skipUnless(BSD, "not a BSD system") +class BSDSpecificTestCase(unittest.TestCase): + + @classmethod + def setUpClass(cls): + cls.pid = get_test_subprocess().pid + + @classmethod + def tearDownClass(cls): + reap_children() + + def test_boot_time(self): + s = sysctl('sysctl kern.boottime') + s = s[s.find(" sec = ") + 7:] + s = s[:s.find(',')] + btime = int(s) + self.assertEqual(btime, psutil.boot_time()) + + def test_process_create_time(self): + cmdline = "ps -o lstart -p %s" % self.pid + p = subprocess.Popen(cmdline, shell=1, stdout=subprocess.PIPE) + output = p.communicate()[0] + if PY3: + output = str(output, sys.stdout.encoding) + start_ps = output.replace('STARTED', '').strip() + start_psutil = psutil.Process(self.pid).create_time() + start_psutil = time.strftime("%a %b %e %H:%M:%S %Y", + time.localtime(start_psutil)) + self.assertEqual(start_ps, start_psutil) + + def test_disks(self): + # test psutil.disk_usage() and psutil.disk_partitions() + # against "df -a" + def df(path): + out = sh('df -k "%s"' % path).strip() + lines = out.split('\n') + lines.pop(0) + line = lines.pop(0) + dev, total, used, free = line.split()[:4] + if dev == 'none': + dev = '' + total = int(total) * 1024 + used = int(used) * 1024 + free = int(free) * 1024 + return dev, total, used, free + + for part in psutil.disk_partitions(all=False): + usage = psutil.disk_usage(part.mountpoint) + dev, total, used, free = df(part.mountpoint) + self.assertEqual(part.device, dev) + self.assertEqual(usage.total, total) + # 10 MB tollerance + if abs(usage.free - free) > 10 * 1024 * 1024: + self.fail("psutil=%s, df=%s" % (usage.free, free)) + if abs(usage.used - used) > 10 * 1024 * 1024: + self.fail("psutil=%s, df=%s" % (usage.used, used)) + + @retry_before_failing() + def test_memory_maps(self): + out = sh('procstat -v %s' % self.pid) + maps = psutil.Process(self.pid).memory_maps(grouped=False) + lines = out.split('\n')[1:] + while lines: + line = lines.pop() + fields = line.split() + _, start, stop, perms, res = fields[:5] + map = maps.pop() + self.assertEqual("%s-%s" % (start, stop), map.addr) + self.assertEqual(int(res), map.rss) + if not map.path.startswith('['): + self.assertEqual(fields[10], map.path) + + def test_exe(self): + out = sh('procstat -b %s' % self.pid) + self.assertEqual(psutil.Process(self.pid).exe(), + out.split('\n')[1].split()[-1]) + + def test_cmdline(self): + out = sh('procstat -c %s' % self.pid) + self.assertEqual(' '.join(psutil.Process(self.pid).cmdline()), + ' '.join(out.split('\n')[1].split()[2:])) + + def test_uids_gids(self): + out = sh('procstat -s %s' % self.pid) + euid, ruid, suid, egid, rgid, sgid = out.split('\n')[1].split()[2:8] + p = psutil.Process(self.pid) + uids = p.uids() + gids = p.gids() + self.assertEqual(uids.real, int(ruid)) + self.assertEqual(uids.effective, int(euid)) + self.assertEqual(uids.saved, int(suid)) + self.assertEqual(gids.real, int(rgid)) + self.assertEqual(gids.effective, int(egid)) + self.assertEqual(gids.saved, int(sgid)) + + # --- virtual_memory(); tests against sysctl + + def test_vmem_total(self): + syst = sysctl("sysctl vm.stats.vm.v_page_count") * PAGESIZE + self.assertEqual(psutil.virtual_memory().total, syst) + + @retry_before_failing() + def test_vmem_active(self): + syst = sysctl("vm.stats.vm.v_active_count") * PAGESIZE + self.assertAlmostEqual(psutil.virtual_memory().active, syst, + delta=TOLERANCE) + + @retry_before_failing() + def test_vmem_inactive(self): + syst = sysctl("vm.stats.vm.v_inactive_count") * PAGESIZE + self.assertAlmostEqual(psutil.virtual_memory().inactive, syst, + delta=TOLERANCE) + + @retry_before_failing() + def test_vmem_wired(self): + syst = sysctl("vm.stats.vm.v_wire_count") * PAGESIZE + self.assertAlmostEqual(psutil.virtual_memory().wired, syst, + delta=TOLERANCE) + + @retry_before_failing() + def test_vmem_cached(self): + syst = sysctl("vm.stats.vm.v_cache_count") * PAGESIZE + self.assertAlmostEqual(psutil.virtual_memory().cached, syst, + delta=TOLERANCE) + + @retry_before_failing() + def test_vmem_free(self): + syst = sysctl("vm.stats.vm.v_free_count") * PAGESIZE + self.assertAlmostEqual(psutil.virtual_memory().free, syst, + delta=TOLERANCE) + + @retry_before_failing() + def test_vmem_buffers(self): + syst = sysctl("vfs.bufspace") + self.assertAlmostEqual(psutil.virtual_memory().buffers, syst, + delta=TOLERANCE) + + def test_cpu_count_logical(self): + syst = sysctl("hw.ncpu") + self.assertEqual(psutil.cpu_count(logical=True), syst) + + # --- virtual_memory(); tests against muse + + @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available") + def test_total(self): + num = muse('Total') + self.assertEqual(psutil.virtual_memory().total, num) + + @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available") + @retry_before_failing() + def test_active(self): + num = muse('Active') + self.assertAlmostEqual(psutil.virtual_memory().active, num, + delta=TOLERANCE) + + @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available") + @retry_before_failing() + def test_inactive(self): + num = muse('Inactive') + self.assertAlmostEqual(psutil.virtual_memory().inactive, num, + delta=TOLERANCE) + + @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available") + @retry_before_failing() + def test_wired(self): + num = muse('Wired') + self.assertAlmostEqual(psutil.virtual_memory().wired, num, + delta=TOLERANCE) + + @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available") + @retry_before_failing() + def test_cached(self): + num = muse('Cache') + self.assertAlmostEqual(psutil.virtual_memory().cached, num, + delta=TOLERANCE) + + @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available") + @retry_before_failing() + def test_free(self): + num = muse('Free') + self.assertAlmostEqual(psutil.virtual_memory().free, num, + delta=TOLERANCE) + + @unittest.skipUnless(MUSE_AVAILABLE, "muse cmdline tool is not available") + @retry_before_failing() + def test_buffers(self): + num = muse('Buffer') + self.assertAlmostEqual(psutil.virtual_memory().buffers, num, + delta=TOLERANCE) + + +def main(): + test_suite = unittest.TestSuite() + test_suite.addTest(unittest.makeSuite(BSDSpecificTestCase)) + result = unittest.TextTestRunner(verbosity=2).run(test_suite) + return result.wasSuccessful() + +if __name__ == '__main__': + if not main(): + sys.exit(1) diff --git a/python/psutil/test/_linux.py b/python/psutil/test/_linux.py new file mode 100644 index 0000000000..c1927ea8b2 --- /dev/null +++ b/python/psutil/test/_linux.py @@ -0,0 +1,473 @@ +#!/usr/bin/env python + +# Copyright (c) 2009, Giampaolo Rodola'. All rights reserved. +# Use of this source code is governed by a BSD-style license that can be +# found in the LICENSE file. + +"""Linux specific tests. These are implicitly run by test_psutil.py.""" + +from __future__ import division +import contextlib +import errno +import fcntl +import io +import os +import pprint +import re +import socket +import struct +import sys +import tempfile +import time +import warnings + +try: + from unittest import mock # py3 +except ImportError: + import mock # requires "pip install mock" + +from test_psutil import POSIX, TOLERANCE, TRAVIS, LINUX +from test_psutil import (skip_on_not_implemented, sh, get_test_subprocess, + retry_before_failing, get_kernel_version, unittest, + which, call_until) + +import psutil +import psutil._pslinux +from psutil._compat import PY3, u + + +SIOCGIFADDR = 0x8915 +SIOCGIFCONF = 0x8912 +SIOCGIFHWADDR = 0x8927 + + +def get_ipv4_address(ifname): + ifname = ifname[:15] + if PY3: + ifname = bytes(ifname, 'ascii') + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + with contextlib.closing(s): + return socket.inet_ntoa( + fcntl.ioctl(s.fileno(), + SIOCGIFADDR, + struct.pack('256s', ifname))[20:24]) + + +def get_mac_address(ifname): + ifname = ifname[:15] + if PY3: + ifname = bytes(ifname, 'ascii') + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + with contextlib.closing(s): + info = fcntl.ioctl( + s.fileno(), SIOCGIFHWADDR, struct.pack('256s', ifname)) + if PY3: + def ord(x): + return x + else: + import __builtin__ + ord = __builtin__.ord + return ''.join(['%02x:' % ord(char) for char in info[18:24]])[:-1] + + +@unittest.skipUnless(LINUX, "not a Linux system") +class LinuxSpecificTestCase(unittest.TestCase): + + @unittest.skipIf( + POSIX and not hasattr(os, 'statvfs'), + reason="os.statvfs() function not available on this platform") + @skip_on_not_implemented() + def test_disks(self): + # test psutil.disk_usage() and psutil.disk_partitions() + # against "df -a" + def df(path): + out = sh('df -P -B 1 "%s"' % path).strip() + lines = out.split('\n') + lines.pop(0) + line = lines.pop(0) + dev, total, used, free = line.split()[:4] + if dev == 'none': + dev = '' + total, used, free = int(total), int(used), int(free) + return dev, total, used, free + + for part in psutil.disk_partitions(all=False): + usage = psutil.disk_usage(part.mountpoint) + dev, total, used, free = df(part.mountpoint) + self.assertEqual(part.device, dev) + self.assertEqual(usage.total, total) + # 10 MB tollerance + if abs(usage.free - free) > 10 * 1024 * 1024: + self.fail("psutil=%s, df=%s" % (usage.free, free)) + if abs(usage.used - used) > 10 * 1024 * 1024: + self.fail("psutil=%s, df=%s" % (usage.used, used)) + + def test_memory_maps(self): + sproc = get_test_subprocess() + time.sleep(1) + p = psutil.Process(sproc.pid) + maps = p.memory_maps(grouped=False) + pmap = sh('pmap -x %s' % p.pid).split('\n') + # get rid of header + del pmap[0] + del pmap[0] + while maps and pmap: + this = maps.pop(0) + other = pmap.pop(0) + addr, _, rss, dirty, mode, path = other.split(None, 5) + if not path.startswith('[') and not path.endswith(']'): + self.assertEqual(path, os.path.basename(this.path)) + self.assertEqual(int(rss) * 1024, this.rss) + # test only rwx chars, ignore 's' and 'p' + self.assertEqual(mode[:3], this.perms[:3]) + + def test_vmem_total(self): + lines = sh('free').split('\n')[1:] + total = int(lines[0].split()[1]) * 1024 + self.assertEqual(total, psutil.virtual_memory().total) + + @retry_before_failing() + def test_vmem_used(self): + lines = sh('free').split('\n')[1:] + used = int(lines[0].split()[2]) * 1024 + self.assertAlmostEqual(used, psutil.virtual_memory().used, + delta=TOLERANCE) + + @retry_before_failing() + def test_vmem_free(self): + lines = sh('free').split('\n')[1:] + free = int(lines[0].split()[3]) * 1024 + self.assertAlmostEqual(free, psutil.virtual_memory().free, + delta=TOLERANCE) + + @retry_before_failing() + def test_vmem_buffers(self): + lines = sh('free').split('\n')[1:] + buffers = int(lines[0].split()[5]) * 1024 + self.assertAlmostEqual(buffers, psutil.virtual_memory().buffers, + delta=TOLERANCE) + + @retry_before_failing() + def test_vmem_cached(self): + lines = sh('free').split('\n')[1:] + cached = int(lines[0].split()[6]) * 1024 + self.assertAlmostEqual(cached, psutil.virtual_memory().cached, + delta=TOLERANCE) + + def test_swapmem_total(self): + lines = sh('free').split('\n')[1:] + total = int(lines[2].split()[1]) * 1024 + self.assertEqual(total, psutil.swap_memory().total) + + @retry_before_failing() + def test_swapmem_used(self): + lines = sh('free').split('\n')[1:] + used = int(lines[2].split()[2]) * 1024 + self.assertAlmostEqual(used, psutil.swap_memory().used, + delta=TOLERANCE) + + @retry_before_failing() + def test_swapmem_free(self): + lines = sh('free').split('\n')[1:] + free = int(lines[2].split()[3]) * 1024 + self.assertAlmostEqual(free, psutil.swap_memory().free, + delta=TOLERANCE) + + @unittest.skipIf(TRAVIS, "unknown failure on travis") + def test_cpu_times(self): + fields = psutil.cpu_times()._fields + kernel_ver = re.findall('\d+\.\d+\.\d+', os.uname()[2])[0] + kernel_ver_info = tuple(map(int, kernel_ver.split('.'))) + if kernel_ver_info >= (2, 6, 11): + self.assertIn('steal', fields) + else: + self.assertNotIn('steal', fields) + if kernel_ver_info >= (2, 6, 24): + self.assertIn('guest', fields) + else: + self.assertNotIn('guest', fields) + if kernel_ver_info >= (3, 2, 0): + self.assertIn('guest_nice', fields) + else: + self.assertNotIn('guest_nice', fields) + + def test_net_if_addrs_ips(self): + for name, addrs in psutil.net_if_addrs().items(): + for addr in addrs: + if addr.family == psutil.AF_LINK: + self.assertEqual(addr.address, get_mac_address(name)) + elif addr.family == socket.AF_INET: + self.assertEqual(addr.address, get_ipv4_address(name)) + # TODO: test for AF_INET6 family + + @unittest.skipUnless(which('ip'), "'ip' utility not available") + @unittest.skipIf(TRAVIS, "skipped on Travis") + def test_net_if_names(self): + out = sh("ip addr").strip() + nics = psutil.net_if_addrs() + found = 0 + for line in out.split('\n'): + line = line.strip() + if re.search("^\d+:", line): + found += 1 + name = line.split(':')[1].strip() + self.assertIn(name, nics.keys()) + self.assertEqual(len(nics), found, msg="%s\n---\n%s" % ( + pprint.pformat(nics), out)) + + @unittest.skipUnless(which("nproc"), "nproc utility not available") + def test_cpu_count_logical_w_nproc(self): + num = int(sh("nproc --all")) + self.assertEqual(psutil.cpu_count(logical=True), num) + + @unittest.skipUnless(which("lscpu"), "lscpu utility not available") + def test_cpu_count_logical_w_lscpu(self): + out = sh("lscpu -p") + num = len([x for x in out.split('\n') if not x.startswith('#')]) + self.assertEqual(psutil.cpu_count(logical=True), num) + + # --- mocked tests + + def test_virtual_memory_mocked_warnings(self): + with mock.patch('psutil._pslinux.open', create=True) as m: + with warnings.catch_warnings(record=True) as ws: + warnings.simplefilter("always") + ret = psutil._pslinux.virtual_memory() + assert m.called + self.assertEqual(len(ws), 1) + w = ws[0] + self.assertTrue(w.filename.endswith('psutil/_pslinux.py')) + self.assertIn( + "'cached', 'active' and 'inactive' memory stats couldn't " + "be determined", str(w.message)) + self.assertEqual(ret.cached, 0) + self.assertEqual(ret.active, 0) + self.assertEqual(ret.inactive, 0) + + def test_swap_memory_mocked_warnings(self): + with mock.patch('psutil._pslinux.open', create=True) as m: + with warnings.catch_warnings(record=True) as ws: + warnings.simplefilter("always") + ret = psutil._pslinux.swap_memory() + assert m.called + self.assertEqual(len(ws), 1) + w = ws[0] + self.assertTrue(w.filename.endswith('psutil/_pslinux.py')) + self.assertIn( + "'sin' and 'sout' swap memory stats couldn't " + "be determined", str(w.message)) + self.assertEqual(ret.sin, 0) + self.assertEqual(ret.sout, 0) + + def test_cpu_count_logical_mocked(self): + import psutil._pslinux + original = psutil._pslinux.cpu_count_logical() + # Here we want to mock os.sysconf("SC_NPROCESSORS_ONLN") in + # order to cause the parsing of /proc/cpuinfo and /proc/stat. + with mock.patch( + 'psutil._pslinux.os.sysconf', side_effect=ValueError) as m: + self.assertEqual(psutil._pslinux.cpu_count_logical(), original) + assert m.called + + # Let's have open() return emtpy data and make sure None is + # returned ('cause we mimick os.cpu_count()). + with mock.patch('psutil._pslinux.open', create=True) as m: + self.assertIsNone(psutil._pslinux.cpu_count_logical()) + self.assertEqual(m.call_count, 2) + # /proc/stat should be the last one + self.assertEqual(m.call_args[0][0], '/proc/stat') + + # Let's push this a bit further and make sure /proc/cpuinfo + # parsing works as expected. + with open('/proc/cpuinfo', 'rb') as f: + cpuinfo_data = f.read() + fake_file = io.BytesIO(cpuinfo_data) + with mock.patch('psutil._pslinux.open', + return_value=fake_file, create=True) as m: + self.assertEqual(psutil._pslinux.cpu_count_logical(), original) + + def test_cpu_count_physical_mocked(self): + # Have open() return emtpy data and make sure None is returned + # ('cause we want to mimick os.cpu_count()) + with mock.patch('psutil._pslinux.open', create=True) as m: + self.assertIsNone(psutil._pslinux.cpu_count_physical()) + assert m.called + + def test_proc_open_files_file_gone(self): + # simulates a file which gets deleted during open_files() + # execution + p = psutil.Process() + files = p.open_files() + with tempfile.NamedTemporaryFile(): + # give the kernel some time to see the new file + call_until(p.open_files, "len(ret) != %i" % len(files)) + with mock.patch('psutil._pslinux.os.readlink', + side_effect=OSError(errno.ENOENT, "")) as m: + files = p.open_files() + assert not files + assert m.called + # also simulate the case where os.readlink() returns EINVAL + # in which case psutil is supposed to 'continue' + with mock.patch('psutil._pslinux.os.readlink', + side_effect=OSError(errno.EINVAL, "")) as m: + self.assertEqual(p.open_files(), []) + assert m.called + + def test_proc_terminal_mocked(self): + with mock.patch('psutil._pslinux._psposix._get_terminal_map', + return_value={}) as m: + self.assertIsNone(psutil._pslinux.Process(os.getpid()).terminal()) + assert m.called + + def test_proc_num_ctx_switches_mocked(self): + with mock.patch('psutil._pslinux.open', create=True) as m: + self.assertRaises( + NotImplementedError, + psutil._pslinux.Process(os.getpid()).num_ctx_switches) + assert m.called + + def test_proc_num_threads_mocked(self): + with mock.patch('psutil._pslinux.open', create=True) as m: + self.assertRaises( + NotImplementedError, + psutil._pslinux.Process(os.getpid()).num_threads) + assert m.called + + def test_proc_ppid_mocked(self): + with mock.patch('psutil._pslinux.open', create=True) as m: + self.assertRaises( + NotImplementedError, + psutil._pslinux.Process(os.getpid()).ppid) + assert m.called + + def test_proc_uids_mocked(self): + with mock.patch('psutil._pslinux.open', create=True) as m: + self.assertRaises( + NotImplementedError, + psutil._pslinux.Process(os.getpid()).uids) + assert m.called + + def test_proc_gids_mocked(self): + with mock.patch('psutil._pslinux.open', create=True) as m: + self.assertRaises( + NotImplementedError, + psutil._pslinux.Process(os.getpid()).gids) + assert m.called + + def test_proc_cmdline_mocked(self): + # see: https://github.com/giampaolo/psutil/issues/639 + p = psutil.Process() + fake_file = io.StringIO(u('foo\x00bar\x00')) + with mock.patch('psutil._pslinux.open', + return_value=fake_file, create=True) as m: + p.cmdline() == ['foo', 'bar'] + assert m.called + fake_file = io.StringIO(u('foo\x00bar\x00\x00')) + with mock.patch('psutil._pslinux.open', + return_value=fake_file, create=True) as m: + p.cmdline() == ['foo', 'bar', ''] + assert m.called + + def test_proc_io_counters_mocked(self): + with mock.patch('psutil._pslinux.open', create=True) as m: + self.assertRaises( + NotImplementedError, + psutil._pslinux.Process(os.getpid()).io_counters) + assert m.called + + def test_boot_time_mocked(self): + with mock.patch('psutil._pslinux.open', create=True) as m: + self.assertRaises( + RuntimeError, + psutil._pslinux.boot_time) + assert m.called + + def test_users_mocked(self): + # Make sure ':0' and ':0.0' (returned by C ext) are converted + # to 'localhost'. + with mock.patch('psutil._pslinux.cext.users', + return_value=[('giampaolo', 'pts/2', ':0', + 1436573184.0, True)]) as m: + self.assertEqual(psutil.users()[0].host, 'localhost') + assert m.called + with mock.patch('psutil._pslinux.cext.users', + return_value=[('giampaolo', 'pts/2', ':0.0', + 1436573184.0, True)]) as m: + self.assertEqual(psutil.users()[0].host, 'localhost') + assert m.called + # ...otherwise it should be returned as-is + with mock.patch('psutil._pslinux.cext.users', + return_value=[('giampaolo', 'pts/2', 'foo', + 1436573184.0, True)]) as m: + self.assertEqual(psutil.users()[0].host, 'foo') + assert m.called + + def test_disk_partitions_mocked(self): + # Test that ZFS partitions are returned. + with open("/proc/filesystems", "r") as f: + data = f.read() + if 'zfs' in data: + for part in psutil.disk_partitions(): + if part.fstype == 'zfs': + break + else: + self.fail("couldn't find any ZFS partition") + else: + # No ZFS partitions on this system. Let's fake one. + fake_file = io.StringIO(u("nodev\tzfs\n")) + with mock.patch('psutil._pslinux.open', + return_value=fake_file, create=True) as m1: + with mock.patch( + 'psutil._pslinux.cext.disk_partitions', + return_value=[('/dev/sdb3', '/', 'zfs', 'rw')]) as m2: + ret = psutil.disk_partitions() + assert m1.called + assert m2.called + assert ret + self.assertEqual(ret[0].fstype, 'zfs') + + # --- tests for specific kernel versions + + @unittest.skipUnless( + get_kernel_version() >= (2, 6, 36), + "prlimit() not available on this Linux kernel version") + def test_prlimit_availability(self): + # prlimit() should be available starting from kernel 2.6.36 + p = psutil.Process(os.getpid()) + p.rlimit(psutil.RLIMIT_NOFILE) + # if prlimit() is supported *at least* these constants should + # be available + self.assertTrue(hasattr(psutil, "RLIM_INFINITY")) + self.assertTrue(hasattr(psutil, "RLIMIT_AS")) + self.assertTrue(hasattr(psutil, "RLIMIT_CORE")) + self.assertTrue(hasattr(psutil, "RLIMIT_CPU")) + self.assertTrue(hasattr(psutil, "RLIMIT_DATA")) + self.assertTrue(hasattr(psutil, "RLIMIT_FSIZE")) + self.assertTrue(hasattr(psutil, "RLIMIT_LOCKS")) + self.assertTrue(hasattr(psutil, "RLIMIT_MEMLOCK")) + self.assertTrue(hasattr(psutil, "RLIMIT_NOFILE")) + self.assertTrue(hasattr(psutil, "RLIMIT_NPROC")) + self.assertTrue(hasattr(psutil, "RLIMIT_RSS")) + self.assertTrue(hasattr(psutil, "RLIMIT_STACK")) + + @unittest.skipUnless( + get_kernel_version() >= (3, 0), + "prlimit constants not available on this Linux kernel version") + def test_resource_consts_kernel_v(self): + # more recent constants + self.assertTrue(hasattr(psutil, "RLIMIT_MSGQUEUE")) + self.assertTrue(hasattr(psutil, "RLIMIT_NICE")) + self.assertTrue(hasattr(psutil, "RLIMIT_RTPRIO")) + self.assertTrue(hasattr(psutil, "RLIMIT_RTTIME")) + self.assertTrue(hasattr(psutil, "RLIMIT_SIGPENDING")) + + +def main(): + test_suite = unittest.TestSuite() + test_suite.addTest(unittest.makeSuite(LinuxSpecificTestCase)) + result = unittest.TextTestRunner(verbosity=2).run(test_suite) + return result.wasSuccessful() + +if __name__ == '__main__': + if not main(): + sys.exit(1) diff --git a/python/psutil/test/_osx.py b/python/psutil/test/_osx.py new file mode 100644 index 0000000000..6e6e4380ef --- /dev/null +++ b/python/psutil/test/_osx.py @@ -0,0 +1,160 @@ +#!/usr/bin/env python + +# Copyright (c) 2009, Giampaolo Rodola'. All rights reserved. +# Use of this source code is governed by a BSD-style license that can be +# found in the LICENSE file. + +"""OSX specific tests. These are implicitly run by test_psutil.py.""" + +import os +import re +import subprocess +import sys +import time + +import psutil + +from psutil._compat import PY3 +from test_psutil import (TOLERANCE, OSX, sh, get_test_subprocess, + reap_children, retry_before_failing, unittest) + + +PAGESIZE = os.sysconf("SC_PAGE_SIZE") + + +def sysctl(cmdline): + """Expects a sysctl command with an argument and parse the result + returning only the value of interest. + """ + p = subprocess.Popen(cmdline, shell=1, stdout=subprocess.PIPE) + result = p.communicate()[0].strip().split()[1] + if PY3: + result = str(result, sys.stdout.encoding) + try: + return int(result) + except ValueError: + return result + + +def vm_stat(field): + """Wrapper around 'vm_stat' cmdline utility.""" + out = sh('vm_stat') + for line in out.split('\n'): + if field in line: + break + else: + raise ValueError("line not found") + return int(re.search('\d+', line).group(0)) * PAGESIZE + + +@unittest.skipUnless(OSX, "not an OSX system") +class OSXSpecificTestCase(unittest.TestCase): + + @classmethod + def setUpClass(cls): + cls.pid = get_test_subprocess().pid + + @classmethod + def tearDownClass(cls): + reap_children() + + def test_process_create_time(self): + cmdline = "ps -o lstart -p %s" % self.pid + p = subprocess.Popen(cmdline, shell=1, stdout=subprocess.PIPE) + output = p.communicate()[0] + if PY3: + output = str(output, sys.stdout.encoding) + start_ps = output.replace('STARTED', '').strip() + start_psutil = psutil.Process(self.pid).create_time() + start_psutil = time.strftime("%a %b %e %H:%M:%S %Y", + time.localtime(start_psutil)) + self.assertEqual(start_ps, start_psutil) + + def test_disks(self): + # test psutil.disk_usage() and psutil.disk_partitions() + # against "df -a" + def df(path): + out = sh('df -k "%s"' % path).strip() + lines = out.split('\n') + lines.pop(0) + line = lines.pop(0) + dev, total, used, free = line.split()[:4] + if dev == 'none': + dev = '' + total = int(total) * 1024 + used = int(used) * 1024 + free = int(free) * 1024 + return dev, total, used, free + + for part in psutil.disk_partitions(all=False): + usage = psutil.disk_usage(part.mountpoint) + dev, total, used, free = df(part.mountpoint) + self.assertEqual(part.device, dev) + self.assertEqual(usage.total, total) + # 10 MB tollerance + if abs(usage.free - free) > 10 * 1024 * 1024: + self.fail("psutil=%s, df=%s" % usage.free, free) + if abs(usage.used - used) > 10 * 1024 * 1024: + self.fail("psutil=%s, df=%s" % usage.used, used) + + # --- virtual mem + + def test_vmem_total(self): + sysctl_hwphymem = sysctl('sysctl hw.memsize') + self.assertEqual(sysctl_hwphymem, psutil.virtual_memory().total) + + @retry_before_failing() + def test_vmem_free(self): + num = vm_stat("free") + self.assertAlmostEqual(psutil.virtual_memory().free, num, + delta=TOLERANCE) + + @retry_before_failing() + def test_vmem_active(self): + num = vm_stat("active") + self.assertAlmostEqual(psutil.virtual_memory().active, num, + delta=TOLERANCE) + + @retry_before_failing() + def test_vmem_inactive(self): + num = vm_stat("inactive") + self.assertAlmostEqual(psutil.virtual_memory().inactive, num, + delta=TOLERANCE) + + @retry_before_failing() + def test_vmem_wired(self): + num = vm_stat("wired") + self.assertAlmostEqual(psutil.virtual_memory().wired, num, + delta=TOLERANCE) + + # --- swap mem + + def test_swapmem_sin(self): + num = vm_stat("Pageins") + self.assertEqual(psutil.swap_memory().sin, num) + + def test_swapmem_sout(self): + num = vm_stat("Pageouts") + self.assertEqual(psutil.swap_memory().sout, num) + + def test_swapmem_total(self): + tot1 = psutil.swap_memory().total + tot2 = 0 + # OSX uses multiple cache files: + # http://en.wikipedia.org/wiki/Paging#OS_X + for name in os.listdir("/var/vm/"): + file = os.path.join("/var/vm", name) + if os.path.isfile(file): + tot2 += os.path.getsize(file) + self.assertEqual(tot1, tot2) + + +def main(): + test_suite = unittest.TestSuite() + test_suite.addTest(unittest.makeSuite(OSXSpecificTestCase)) + result = unittest.TextTestRunner(verbosity=2).run(test_suite) + return result.wasSuccessful() + +if __name__ == '__main__': + if not main(): + sys.exit(1) diff --git a/python/psutil/test/_posix.py b/python/psutil/test/_posix.py new file mode 100644 index 0000000000..e6c56aac3e --- /dev/null +++ b/python/psutil/test/_posix.py @@ -0,0 +1,258 @@ +#!/usr/bin/env python + +# Copyright (c) 2009, Giampaolo Rodola'. All rights reserved. +# Use of this source code is governed by a BSD-style license that can be +# found in the LICENSE file. + +"""POSIX specific tests. These are implicitly run by test_psutil.py.""" + +import datetime +import os +import subprocess +import sys +import time + +import psutil + +from psutil._compat import PY3, callable +from test_psutil import LINUX, SUNOS, OSX, BSD, PYTHON, POSIX, TRAVIS +from test_psutil import (get_test_subprocess, skip_on_access_denied, + retry_before_failing, reap_children, sh, unittest, + get_kernel_version, wait_for_pid) + + +def ps(cmd): + """Expects a ps command with a -o argument and parse the result + returning only the value of interest. + """ + if not LINUX: + cmd = cmd.replace(" --no-headers ", " ") + if SUNOS: + cmd = cmd.replace("-o command", "-o comm") + cmd = cmd.replace("-o start", "-o stime") + p = subprocess.Popen(cmd, shell=1, stdout=subprocess.PIPE) + output = p.communicate()[0].strip() + if PY3: + output = str(output, sys.stdout.encoding) + if not LINUX: + output = output.split('\n')[1].strip() + try: + return int(output) + except ValueError: + return output + + +@unittest.skipUnless(POSIX, "not a POSIX system") +class PosixSpecificTestCase(unittest.TestCase): + """Compare psutil results against 'ps' command line utility.""" + + @classmethod + def setUpClass(cls): + cls.pid = get_test_subprocess([PYTHON, "-E", "-O"], + stdin=subprocess.PIPE).pid + wait_for_pid(cls.pid) + + @classmethod + def tearDownClass(cls): + reap_children() + + # for ps -o arguments see: http://unixhelp.ed.ac.uk/CGI/man-cgi?ps + + def test_process_parent_pid(self): + ppid_ps = ps("ps --no-headers -o ppid -p %s" % self.pid) + ppid_psutil = psutil.Process(self.pid).ppid() + self.assertEqual(ppid_ps, ppid_psutil) + + def test_process_uid(self): + uid_ps = ps("ps --no-headers -o uid -p %s" % self.pid) + uid_psutil = psutil.Process(self.pid).uids().real + self.assertEqual(uid_ps, uid_psutil) + + def test_process_gid(self): + gid_ps = ps("ps --no-headers -o rgid -p %s" % self.pid) + gid_psutil = psutil.Process(self.pid).gids().real + self.assertEqual(gid_ps, gid_psutil) + + def test_process_username(self): + username_ps = ps("ps --no-headers -o user -p %s" % self.pid) + username_psutil = psutil.Process(self.pid).username() + self.assertEqual(username_ps, username_psutil) + + @skip_on_access_denied() + @retry_before_failing() + def test_process_rss_memory(self): + # give python interpreter some time to properly initialize + # so that the results are the same + time.sleep(0.1) + rss_ps = ps("ps --no-headers -o rss -p %s" % self.pid) + rss_psutil = psutil.Process(self.pid).memory_info()[0] / 1024 + self.assertEqual(rss_ps, rss_psutil) + + @skip_on_access_denied() + @retry_before_failing() + def test_process_vsz_memory(self): + # give python interpreter some time to properly initialize + # so that the results are the same + time.sleep(0.1) + vsz_ps = ps("ps --no-headers -o vsz -p %s" % self.pid) + vsz_psutil = psutil.Process(self.pid).memory_info()[1] / 1024 + self.assertEqual(vsz_ps, vsz_psutil) + + def test_process_name(self): + # use command + arg since "comm" keyword not supported on all platforms + name_ps = ps("ps --no-headers -o command -p %s" % ( + self.pid)).split(' ')[0] + # remove path if there is any, from the command + name_ps = os.path.basename(name_ps).lower() + name_psutil = psutil.Process(self.pid).name().lower() + self.assertEqual(name_ps, name_psutil) + + @unittest.skipIf(OSX or BSD, + 'ps -o start not available') + def test_process_create_time(self): + time_ps = ps("ps --no-headers -o start -p %s" % self.pid).split(' ')[0] + time_psutil = psutil.Process(self.pid).create_time() + time_psutil_tstamp = datetime.datetime.fromtimestamp( + time_psutil).strftime("%H:%M:%S") + # sometimes ps shows the time rounded up instead of down, so we check + # for both possible values + round_time_psutil = round(time_psutil) + round_time_psutil_tstamp = datetime.datetime.fromtimestamp( + round_time_psutil).strftime("%H:%M:%S") + self.assertIn(time_ps, [time_psutil_tstamp, round_time_psutil_tstamp]) + + def test_process_exe(self): + ps_pathname = ps("ps --no-headers -o command -p %s" % + self.pid).split(' ')[0] + psutil_pathname = psutil.Process(self.pid).exe() + try: + self.assertEqual(ps_pathname, psutil_pathname) + except AssertionError: + # certain platforms such as BSD are more accurate returning: + # "/usr/local/bin/python2.7" + # ...instead of: + # "/usr/local/bin/python" + # We do not want to consider this difference in accuracy + # an error. + adjusted_ps_pathname = ps_pathname[:len(ps_pathname)] + self.assertEqual(ps_pathname, adjusted_ps_pathname) + + def test_process_cmdline(self): + ps_cmdline = ps("ps --no-headers -o command -p %s" % self.pid) + psutil_cmdline = " ".join(psutil.Process(self.pid).cmdline()) + if SUNOS: + # ps on Solaris only shows the first part of the cmdline + psutil_cmdline = psutil_cmdline.split(" ")[0] + self.assertEqual(ps_cmdline, psutil_cmdline) + + @retry_before_failing() + def test_pids(self): + # Note: this test might fail if the OS is starting/killing + # other processes in the meantime + if SUNOS: + cmd = ["ps", "ax"] + else: + cmd = ["ps", "ax", "-o", "pid"] + p = get_test_subprocess(cmd, stdout=subprocess.PIPE) + output = p.communicate()[0].strip() + if PY3: + output = str(output, sys.stdout.encoding) + pids_ps = [] + for line in output.split('\n')[1:]: + if line: + pid = int(line.split()[0].strip()) + pids_ps.append(pid) + # remove ps subprocess pid which is supposed to be dead in meantime + pids_ps.remove(p.pid) + pids_psutil = psutil.pids() + pids_ps.sort() + pids_psutil.sort() + + # on OSX ps doesn't show pid 0 + if OSX and 0 not in pids_ps: + pids_ps.insert(0, 0) + + if pids_ps != pids_psutil: + difference = [x for x in pids_psutil if x not in pids_ps] + \ + [x for x in pids_ps if x not in pids_psutil] + self.fail("difference: " + str(difference)) + + # for some reason ifconfig -a does not report all interfaces + # returned by psutil + @unittest.skipIf(SUNOS, "test not reliable on SUNOS") + @unittest.skipIf(TRAVIS, "test not reliable on Travis") + def test_nic_names(self): + p = subprocess.Popen("ifconfig -a", shell=1, stdout=subprocess.PIPE) + output = p.communicate()[0].strip() + if PY3: + output = str(output, sys.stdout.encoding) + for nic in psutil.net_io_counters(pernic=True).keys(): + for line in output.split(): + if line.startswith(nic): + break + else: + self.fail( + "couldn't find %s nic in 'ifconfig -a' output\n%s" % ( + nic, output)) + + @retry_before_failing() + def test_users(self): + out = sh("who") + lines = out.split('\n') + users = [x.split()[0] for x in lines] + self.assertEqual(len(users), len(psutil.users())) + terminals = [x.split()[1] for x in lines] + for u in psutil.users(): + self.assertTrue(u.name in users, u.name) + self.assertTrue(u.terminal in terminals, u.terminal) + + def test_fds_open(self): + # Note: this fails from time to time; I'm keen on thinking + # it doesn't mean something is broken + def call(p, attr): + args = () + attr = getattr(p, name, None) + if attr is not None and callable(attr): + if name == 'rlimit': + args = (psutil.RLIMIT_NOFILE,) + attr(*args) + else: + attr + + p = psutil.Process(os.getpid()) + failures = [] + ignored_names = ['terminate', 'kill', 'suspend', 'resume', 'nice', + 'send_signal', 'wait', 'children', 'as_dict'] + if LINUX and get_kernel_version() < (2, 6, 36): + ignored_names.append('rlimit') + if LINUX and get_kernel_version() < (2, 6, 23): + ignored_names.append('num_ctx_switches') + for name in dir(psutil.Process): + if (name.startswith('_') or name in ignored_names): + continue + else: + try: + num1 = p.num_fds() + for x in range(2): + call(p, name) + num2 = p.num_fds() + except psutil.AccessDenied: + pass + else: + if abs(num2 - num1) > 1: + fail = "failure while processing Process.%s method " \ + "(before=%s, after=%s)" % (name, num1, num2) + failures.append(fail) + if failures: + self.fail('\n' + '\n'.join(failures)) + + +def main(): + test_suite = unittest.TestSuite() + test_suite.addTest(unittest.makeSuite(PosixSpecificTestCase)) + result = unittest.TextTestRunner(verbosity=2).run(test_suite) + return result.wasSuccessful() + +if __name__ == '__main__': + if not main(): + sys.exit(1) diff --git a/python/psutil/test/_sunos.py b/python/psutil/test/_sunos.py new file mode 100644 index 0000000000..3d54ccd8cf --- /dev/null +++ b/python/psutil/test/_sunos.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python + +# Copyright (c) 2009, Giampaolo Rodola'. All rights reserved. +# Use of this source code is governed by a BSD-style license that can be +# found in the LICENSE file. + +"""Sun OS specific tests. These are implicitly run by test_psutil.py.""" + +import sys +import os + +from test_psutil import SUNOS, sh, unittest +import psutil + + +@unittest.skipUnless(SUNOS, "not a SunOS system") +class SunOSSpecificTestCase(unittest.TestCase): + + def test_swap_memory(self): + out = sh('env PATH=/usr/sbin:/sbin:%s swap -l -k' % os.environ['PATH']) + lines = out.strip().split('\n')[1:] + if not lines: + raise ValueError('no swap device(s) configured') + total = free = 0 + for line in lines: + line = line.split() + t, f = line[-2:] + t = t.replace('K', '') + f = f.replace('K', '') + total += int(int(t) * 1024) + free += int(int(f) * 1024) + used = total - free + + psutil_swap = psutil.swap_memory() + self.assertEqual(psutil_swap.total, total) + self.assertEqual(psutil_swap.used, used) + self.assertEqual(psutil_swap.free, free) + + +def main(): + test_suite = unittest.TestSuite() + test_suite.addTest(unittest.makeSuite(SunOSSpecificTestCase)) + result = unittest.TextTestRunner(verbosity=2).run(test_suite) + return result.wasSuccessful() + +if __name__ == '__main__': + if not main(): + sys.exit(1) diff --git a/python/psutil/test/_windows.py b/python/psutil/test/_windows.py new file mode 100644 index 0000000000..b7477bfeb4 --- /dev/null +++ b/python/psutil/test/_windows.py @@ -0,0 +1,464 @@ +#!/usr/bin/env python + +# Copyright (c) 2009, Giampaolo Rodola'. All rights reserved. +# Use of this source code is governed by a BSD-style license that can be +# found in the LICENSE file. + +"""Windows specific tests. These are implicitly run by test_psutil.py.""" + +import errno +import os +import platform +import signal +import subprocess +import sys +import time +import traceback + +from test_psutil import APPVEYOR, WINDOWS +from test_psutil import get_test_subprocess, reap_children, unittest + +import mock +try: + import wmi +except ImportError: + wmi = None +try: + import win32api + import win32con +except ImportError: + win32api = win32con = None + +from psutil._compat import PY3, callable, long +import psutil + + +cext = psutil._psplatform.cext + + +def wrap_exceptions(fun): + def wrapper(self, *args, **kwargs): + try: + return fun(self, *args, **kwargs) + except OSError as err: + from psutil._pswindows import ACCESS_DENIED_SET + if err.errno in ACCESS_DENIED_SET: + raise psutil.AccessDenied(None, None) + if err.errno == errno.ESRCH: + raise psutil.NoSuchProcess(None, None) + raise + return wrapper + + +@unittest.skipUnless(WINDOWS, "not a Windows system") +class WindowsSpecificTestCase(unittest.TestCase): + + @classmethod + def setUpClass(cls): + cls.pid = get_test_subprocess().pid + + @classmethod + def tearDownClass(cls): + reap_children() + + def test_issue_24(self): + p = psutil.Process(0) + self.assertRaises(psutil.AccessDenied, p.kill) + + def test_special_pid(self): + p = psutil.Process(4) + self.assertEqual(p.name(), 'System') + # use __str__ to access all common Process properties to check + # that nothing strange happens + str(p) + p.username() + self.assertTrue(p.create_time() >= 0.0) + try: + rss, vms = p.memory_info() + except psutil.AccessDenied: + # expected on Windows Vista and Windows 7 + if not platform.uname()[1] in ('vista', 'win-7', 'win7'): + raise + else: + self.assertTrue(rss > 0) + + def test_send_signal(self): + p = psutil.Process(self.pid) + self.assertRaises(ValueError, p.send_signal, signal.SIGINT) + + def test_nic_names(self): + p = subprocess.Popen(['ipconfig', '/all'], stdout=subprocess.PIPE) + out = p.communicate()[0] + if PY3: + out = str(out, sys.stdout.encoding) + nics = psutil.net_io_counters(pernic=True).keys() + for nic in nics: + if "pseudo-interface" in nic.replace(' ', '-').lower(): + continue + if nic not in out: + self.fail( + "%r nic wasn't found in 'ipconfig /all' output" % nic) + + def test_exe(self): + for p in psutil.process_iter(): + try: + self.assertEqual(os.path.basename(p.exe()), p.name()) + except psutil.Error: + pass + + # --- Process class tests + + @unittest.skipIf(wmi is None, "wmi module is not installed") + def test_process_name(self): + w = wmi.WMI().Win32_Process(ProcessId=self.pid)[0] + p = psutil.Process(self.pid) + self.assertEqual(p.name(), w.Caption) + + @unittest.skipIf(wmi is None, "wmi module is not installed") + def test_process_exe(self): + w = wmi.WMI().Win32_Process(ProcessId=self.pid)[0] + p = psutil.Process(self.pid) + # Note: wmi reports the exe as a lower case string. + # Being Windows paths case-insensitive we ignore that. + self.assertEqual(p.exe().lower(), w.ExecutablePath.lower()) + + @unittest.skipIf(wmi is None, "wmi module is not installed") + def test_process_cmdline(self): + w = wmi.WMI().Win32_Process(ProcessId=self.pid)[0] + p = psutil.Process(self.pid) + self.assertEqual(' '.join(p.cmdline()), + w.CommandLine.replace('"', '')) + + @unittest.skipIf(wmi is None, "wmi module is not installed") + def test_process_username(self): + w = wmi.WMI().Win32_Process(ProcessId=self.pid)[0] + p = psutil.Process(self.pid) + domain, _, username = w.GetOwner() + username = "%s\\%s" % (domain, username) + self.assertEqual(p.username(), username) + + @unittest.skipIf(wmi is None, "wmi module is not installed") + def test_process_rss_memory(self): + time.sleep(0.1) + w = wmi.WMI().Win32_Process(ProcessId=self.pid)[0] + p = psutil.Process(self.pid) + rss = p.memory_info().rss + self.assertEqual(rss, int(w.WorkingSetSize)) + + @unittest.skipIf(wmi is None, "wmi module is not installed") + def test_process_vms_memory(self): + time.sleep(0.1) + w = wmi.WMI().Win32_Process(ProcessId=self.pid)[0] + p = psutil.Process(self.pid) + vms = p.memory_info().vms + # http://msdn.microsoft.com/en-us/library/aa394372(VS.85).aspx + # ...claims that PageFileUsage is represented in Kilo + # bytes but funnily enough on certain platforms bytes are + # returned instead. + wmi_usage = int(w.PageFileUsage) + if (vms != wmi_usage) and (vms != wmi_usage * 1024): + self.fail("wmi=%s, psutil=%s" % (wmi_usage, vms)) + + @unittest.skipIf(wmi is None, "wmi module is not installed") + def test_process_create_time(self): + w = wmi.WMI().Win32_Process(ProcessId=self.pid)[0] + p = psutil.Process(self.pid) + wmic_create = str(w.CreationDate.split('.')[0]) + psutil_create = time.strftime("%Y%m%d%H%M%S", + time.localtime(p.create_time())) + self.assertEqual(wmic_create, psutil_create) + + # --- psutil namespace functions and constants tests + + @unittest.skipUnless('NUMBER_OF_PROCESSORS' in os.environ, + 'NUMBER_OF_PROCESSORS env var is not available') + def test_cpu_count(self): + num_cpus = int(os.environ['NUMBER_OF_PROCESSORS']) + self.assertEqual(num_cpus, psutil.cpu_count()) + + @unittest.skipIf(wmi is None, "wmi module is not installed") + def test_total_phymem(self): + w = wmi.WMI().Win32_ComputerSystem()[0] + self.assertEqual(int(w.TotalPhysicalMemory), + psutil.virtual_memory().total) + + # @unittest.skipIf(wmi is None, "wmi module is not installed") + # def test__UPTIME(self): + # # _UPTIME constant is not public but it is used internally + # # as value to return for pid 0 creation time. + # # WMI behaves the same. + # w = wmi.WMI().Win32_Process(ProcessId=self.pid)[0] + # p = psutil.Process(0) + # wmic_create = str(w.CreationDate.split('.')[0]) + # psutil_create = time.strftime("%Y%m%d%H%M%S", + # time.localtime(p.create_time())) + # + + # Note: this test is not very reliable + @unittest.skipIf(wmi is None, "wmi module is not installed") + @unittest.skipIf(APPVEYOR, "test not relieable on appveyor") + def test_pids(self): + # Note: this test might fail if the OS is starting/killing + # other processes in the meantime + w = wmi.WMI().Win32_Process() + wmi_pids = set([x.ProcessId for x in w]) + psutil_pids = set(psutil.pids()) + self.assertEqual(wmi_pids, psutil_pids) + + @unittest.skipIf(wmi is None, "wmi module is not installed") + def test_disks(self): + ps_parts = psutil.disk_partitions(all=True) + wmi_parts = wmi.WMI().Win32_LogicalDisk() + for ps_part in ps_parts: + for wmi_part in wmi_parts: + if ps_part.device.replace('\\', '') == wmi_part.DeviceID: + if not ps_part.mountpoint: + # this is usually a CD-ROM with no disk inserted + break + try: + usage = psutil.disk_usage(ps_part.mountpoint) + except OSError as err: + if err.errno == errno.ENOENT: + # usually this is the floppy + break + else: + raise + self.assertEqual(usage.total, int(wmi_part.Size)) + wmi_free = int(wmi_part.FreeSpace) + self.assertEqual(usage.free, wmi_free) + # 10 MB tollerance + if abs(usage.free - wmi_free) > 10 * 1024 * 1024: + self.fail("psutil=%s, wmi=%s" % ( + usage.free, wmi_free)) + break + else: + self.fail("can't find partition %s" % repr(ps_part)) + + @unittest.skipIf(win32api is None, "pywin32 module is not installed") + def test_num_handles(self): + p = psutil.Process(os.getpid()) + before = p.num_handles() + handle = win32api.OpenProcess(win32con.PROCESS_QUERY_INFORMATION, + win32con.FALSE, os.getpid()) + after = p.num_handles() + self.assertEqual(after, before + 1) + win32api.CloseHandle(handle) + self.assertEqual(p.num_handles(), before) + + @unittest.skipIf(win32api is None, "pywin32 module is not installed") + def test_num_handles_2(self): + # Note: this fails from time to time; I'm keen on thinking + # it doesn't mean something is broken + def call(p, attr): + attr = getattr(p, name, None) + if attr is not None and callable(attr): + attr() + else: + attr + + p = psutil.Process(self.pid) + failures = [] + for name in dir(psutil.Process): + if name.startswith('_') \ + or name in ('terminate', 'kill', 'suspend', 'resume', + 'nice', 'send_signal', 'wait', 'children', + 'as_dict'): + continue + else: + try: + call(p, name) + num1 = p.num_handles() + call(p, name) + num2 = p.num_handles() + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + else: + if num2 > num1: + fail = \ + "failure while processing Process.%s method " \ + "(before=%s, after=%s)" % (name, num1, num2) + failures.append(fail) + if failures: + self.fail('\n' + '\n'.join(failures)) + + def test_name_always_available(self): + # On Windows name() is never supposed to raise AccessDenied, + # see https://github.com/giampaolo/psutil/issues/627 + for p in psutil.process_iter(): + try: + p.name() + except psutil.NoSuchProcess(): + pass + + +@unittest.skipUnless(WINDOWS, "not a Windows system") +class TestDualProcessImplementation(unittest.TestCase): + """ + Certain APIs on Windows have 2 internal implementations, one + based on documented Windows APIs, another one based + NtQuerySystemInformation() which gets called as fallback in + case the first fails because of limited permission error. + Here we test that the two methods return the exact same value, + see: + https://github.com/giampaolo/psutil/issues/304 + """ + + fun_names = [ + # function name, tolerance + ('proc_cpu_times', 0.2), + ('proc_create_time', 0.5), + ('proc_num_handles', 1), # 1 because impl #1 opens a handle + ('proc_memory_info', 1024), # KB + ('proc_io_counters', 0), + ] + + def test_compare_values(self): + def assert_ge_0(obj): + if isinstance(obj, tuple): + for value in obj: + self.assertGreaterEqual(value, 0, msg=obj) + elif isinstance(obj, (int, long, float)): + self.assertGreaterEqual(obj, 0) + else: + assert 0 # case not handled which needs to be fixed + + def compare_with_tolerance(ret1, ret2, tolerance): + if ret1 == ret2: + return + else: + if isinstance(ret2, (int, long, float)): + diff = abs(ret1 - ret2) + self.assertLessEqual(diff, tolerance) + elif isinstance(ret2, tuple): + for a, b in zip(ret1, ret2): + diff = abs(a - b) + self.assertLessEqual(diff, tolerance) + + from psutil._pswindows import ntpinfo + failures = [] + for p in psutil.process_iter(): + try: + nt = ntpinfo(*cext.proc_info(p.pid)) + except psutil.NoSuchProcess: + continue + assert_ge_0(nt) + + for name, tolerance in self.fun_names: + if name == 'proc_memory_info' and p.pid == os.getpid(): + continue + if name == 'proc_create_time' and p.pid in (0, 4): + continue + meth = wrap_exceptions(getattr(cext, name)) + try: + ret = meth(p.pid) + except (psutil.NoSuchProcess, psutil.AccessDenied): + continue + # compare values + try: + if name == 'proc_cpu_times': + compare_with_tolerance(ret[0], nt.user_time, tolerance) + compare_with_tolerance(ret[1], + nt.kernel_time, tolerance) + elif name == 'proc_create_time': + compare_with_tolerance(ret, nt.create_time, tolerance) + elif name == 'proc_num_handles': + compare_with_tolerance(ret, nt.num_handles, tolerance) + elif name == 'proc_io_counters': + compare_with_tolerance(ret[0], nt.io_rcount, tolerance) + compare_with_tolerance(ret[1], nt.io_wcount, tolerance) + compare_with_tolerance(ret[2], nt.io_rbytes, tolerance) + compare_with_tolerance(ret[3], nt.io_wbytes, tolerance) + elif name == 'proc_memory_info': + try: + rawtupl = cext.proc_memory_info_2(p.pid) + except psutil.NoSuchProcess: + continue + compare_with_tolerance(ret, rawtupl, tolerance) + except AssertionError: + trace = traceback.format_exc() + msg = '%s\npid=%s, method=%r, ret_1=%r, ret_2=%r' % ( + trace, p.pid, name, ret, nt) + failures.append(msg) + break + + if failures: + self.fail('\n\n'.join(failures)) + + # --- + # same tests as above but mimicks the AccessDenied failure of + # the first (fast) method failing with AD. + # TODO: currently does not take tolerance into account. + + def test_name(self): + name = psutil.Process().name() + with mock.patch("psutil._psplatform.cext.proc_exe", + side_effect=psutil.AccessDenied(os.getpid())) as fun: + psutil.Process().name() == name + assert fun.called + + def test_memory_info(self): + mem = psutil.Process().memory_info() + with mock.patch("psutil._psplatform.cext.proc_memory_info", + side_effect=OSError(errno.EPERM, "msg")) as fun: + psutil.Process().memory_info() == mem + assert fun.called + + def test_create_time(self): + ctime = psutil.Process().create_time() + with mock.patch("psutil._psplatform.cext.proc_create_time", + side_effect=OSError(errno.EPERM, "msg")) as fun: + psutil.Process().create_time() == ctime + assert fun.called + + def test_cpu_times(self): + cpu_times = psutil.Process().cpu_times() + with mock.patch("psutil._psplatform.cext.proc_cpu_times", + side_effect=OSError(errno.EPERM, "msg")) as fun: + psutil.Process().cpu_times() == cpu_times + assert fun.called + + def test_io_counters(self): + io_counters = psutil.Process().io_counters() + with mock.patch("psutil._psplatform.cext.proc_io_counters", + side_effect=OSError(errno.EPERM, "msg")) as fun: + psutil.Process().io_counters() == io_counters + assert fun.called + + def test_num_handles(self): + io_counters = psutil.Process().io_counters() + with mock.patch("psutil._psplatform.cext.proc_io_counters", + side_effect=OSError(errno.EPERM, "msg")) as fun: + psutil.Process().io_counters() == io_counters + assert fun.called + + # --- other tests + + def test_compare_name_exe(self): + for p in psutil.process_iter(): + try: + a = os.path.basename(p.exe()) + b = p.name() + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + else: + self.assertEqual(a, b) + + def test_zombies(self): + # test that NPS is raised by the 2nd implementation in case a + # process no longer exists + ZOMBIE_PID = max(psutil.pids()) + 5000 + for name, _ in self.fun_names: + meth = wrap_exceptions(getattr(cext, name)) + self.assertRaises(psutil.NoSuchProcess, meth, ZOMBIE_PID) + + +def main(): + test_suite = unittest.TestSuite() + test_suite.addTest(unittest.makeSuite(WindowsSpecificTestCase)) + test_suite.addTest(unittest.makeSuite(TestDualProcessImplementation)) + result = unittest.TextTestRunner(verbosity=2).run(test_suite) + return result.wasSuccessful() + +if __name__ == '__main__': + if not main(): + sys.exit(1) diff --git a/python/psutil/test/test_memory_leaks.py b/python/psutil/test/test_memory_leaks.py new file mode 100644 index 0000000000..6f02dc0acf --- /dev/null +++ b/python/psutil/test/test_memory_leaks.py @@ -0,0 +1,445 @@ +#!/usr/bin/env python + +# Copyright (c) 2009, Giampaolo Rodola'. All rights reserved. +# Use of this source code is governed by a BSD-style license that can be +# found in the LICENSE file. + +""" +A test script which attempts to detect memory leaks by calling C +functions many times and compare process memory usage before and +after the calls. It might produce false positives. +""" + +import functools +import gc +import os +import socket +import sys +import threading +import time + +import psutil +import psutil._common + +from psutil._compat import xrange, callable +from test_psutil import (WINDOWS, POSIX, OSX, LINUX, SUNOS, BSD, TESTFN, + RLIMIT_SUPPORT, TRAVIS) +from test_psutil import (reap_children, supports_ipv6, safe_remove, + get_test_subprocess) + +if sys.version_info < (2, 7): + import unittest2 as unittest # https://pypi.python.org/pypi/unittest2 +else: + import unittest + + +LOOPS = 1000 +TOLERANCE = 4096 +SKIP_PYTHON_IMPL = True + + +def skip_if_linux(): + return unittest.skipIf(LINUX and SKIP_PYTHON_IMPL, + "not worth being tested on LINUX (pure python)") + + +class Base(unittest.TestCase): + proc = psutil.Process() + + def execute(self, function, *args, **kwargs): + def call_many_times(): + for x in xrange(LOOPS - 1): + self.call(function, *args, **kwargs) + del x + gc.collect() + return self.get_mem() + + self.call(function, *args, **kwargs) + self.assertEqual(gc.garbage, []) + self.assertEqual(threading.active_count(), 1) + + # RSS comparison + # step 1 + rss1 = call_many_times() + # step 2 + rss2 = call_many_times() + + difference = rss2 - rss1 + if difference > TOLERANCE: + # This doesn't necessarily mean we have a leak yet. + # At this point we assume that after having called the + # function so many times the memory usage is stabilized + # and if there are no leaks it should not increase any + # more. + # Let's keep calling fun for 3 more seconds and fail if + # we notice any difference. + stop_at = time.time() + 3 + while True: + self.call(function, *args, **kwargs) + if time.time() >= stop_at: + break + del stop_at + gc.collect() + rss3 = self.get_mem() + difference = rss3 - rss2 + if rss3 > rss2: + self.fail("rss2=%s, rss3=%s, difference=%s" + % (rss2, rss3, difference)) + + def execute_w_exc(self, exc, function, *args, **kwargs): + kwargs['_exc'] = exc + self.execute(function, *args, **kwargs) + + def get_mem(self): + return psutil.Process().memory_info()[0] + + def call(self, function, *args, **kwargs): + raise NotImplementedError("must be implemented in subclass") + + +class TestProcessObjectLeaks(Base): + """Test leaks of Process class methods and properties""" + + def setUp(self): + gc.collect() + + def tearDown(self): + reap_children() + + def call(self, function, *args, **kwargs): + if callable(function): + if '_exc' in kwargs: + exc = kwargs.pop('_exc') + self.assertRaises(exc, function, *args, **kwargs) + else: + try: + function(*args, **kwargs) + except psutil.Error: + pass + else: + meth = getattr(self.proc, function) + if '_exc' in kwargs: + exc = kwargs.pop('_exc') + self.assertRaises(exc, meth, *args, **kwargs) + else: + try: + meth(*args, **kwargs) + except psutil.Error: + pass + + @skip_if_linux() + def test_name(self): + self.execute('name') + + @skip_if_linux() + def test_cmdline(self): + self.execute('cmdline') + + @skip_if_linux() + def test_exe(self): + self.execute('exe') + + @skip_if_linux() + def test_ppid(self): + self.execute('ppid') + + @unittest.skipUnless(POSIX, "POSIX only") + @skip_if_linux() + def test_uids(self): + self.execute('uids') + + @unittest.skipUnless(POSIX, "POSIX only") + @skip_if_linux() + def test_gids(self): + self.execute('gids') + + @skip_if_linux() + def test_status(self): + self.execute('status') + + def test_nice_get(self): + self.execute('nice') + + def test_nice_set(self): + niceness = psutil.Process().nice() + self.execute('nice', niceness) + + @unittest.skipUnless(hasattr(psutil.Process, 'ionice'), + "Linux and Windows Vista only") + def test_ionice_get(self): + self.execute('ionice') + + @unittest.skipUnless(hasattr(psutil.Process, 'ionice'), + "Linux and Windows Vista only") + def test_ionice_set(self): + if WINDOWS: + value = psutil.Process().ionice() + self.execute('ionice', value) + else: + from psutil._pslinux import cext + self.execute('ionice', psutil.IOPRIO_CLASS_NONE) + fun = functools.partial(cext.proc_ioprio_set, os.getpid(), -1, 0) + self.execute_w_exc(OSError, fun) + + @unittest.skipIf(OSX or SUNOS, "feature not supported on this platform") + @skip_if_linux() + def test_io_counters(self): + self.execute('io_counters') + + @unittest.skipUnless(WINDOWS, "not worth being tested on posix") + def test_username(self): + self.execute('username') + + @skip_if_linux() + def test_create_time(self): + self.execute('create_time') + + @skip_if_linux() + def test_num_threads(self): + self.execute('num_threads') + + @unittest.skipUnless(WINDOWS, "Windows only") + def test_num_handles(self): + self.execute('num_handles') + + @unittest.skipUnless(POSIX, "POSIX only") + @skip_if_linux() + def test_num_fds(self): + self.execute('num_fds') + + @skip_if_linux() + def test_threads(self): + self.execute('threads') + + @skip_if_linux() + def test_cpu_times(self): + self.execute('cpu_times') + + @skip_if_linux() + def test_memory_info(self): + self.execute('memory_info') + + @skip_if_linux() + def test_memory_info_ex(self): + self.execute('memory_info_ex') + + @unittest.skipUnless(POSIX, "POSIX only") + @skip_if_linux() + def test_terminal(self): + self.execute('terminal') + + @unittest.skipIf(POSIX and SKIP_PYTHON_IMPL, + "not worth being tested on POSIX (pure python)") + def test_resume(self): + self.execute('resume') + + @skip_if_linux() + def test_cwd(self): + self.execute('cwd') + + @unittest.skipUnless(WINDOWS or LINUX or BSD, + "Windows or Linux or BSD only") + def test_cpu_affinity_get(self): + self.execute('cpu_affinity') + + @unittest.skipUnless(WINDOWS or LINUX or BSD, + "Windows or Linux or BSD only") + def test_cpu_affinity_set(self): + affinity = psutil.Process().cpu_affinity() + self.execute('cpu_affinity', affinity) + if not TRAVIS: + self.execute_w_exc(ValueError, 'cpu_affinity', [-1]) + + @skip_if_linux() + def test_open_files(self): + safe_remove(TESTFN) # needed after UNIX socket test has run + with open(TESTFN, 'w'): + self.execute('open_files') + + # OSX implementation is unbelievably slow + @unittest.skipIf(OSX, "OSX implementation is too slow") + @skip_if_linux() + def test_memory_maps(self): + self.execute('memory_maps') + + @unittest.skipUnless(LINUX, "Linux only") + @unittest.skipUnless(LINUX and RLIMIT_SUPPORT, + "only available on Linux >= 2.6.36") + def test_rlimit_get(self): + self.execute('rlimit', psutil.RLIMIT_NOFILE) + + @unittest.skipUnless(LINUX, "Linux only") + @unittest.skipUnless(LINUX and RLIMIT_SUPPORT, + "only available on Linux >= 2.6.36") + def test_rlimit_set(self): + limit = psutil.Process().rlimit(psutil.RLIMIT_NOFILE) + self.execute('rlimit', psutil.RLIMIT_NOFILE, limit) + self.execute_w_exc(OSError, 'rlimit', -1) + + @skip_if_linux() + # Windows implementation is based on a single system-wide function + @unittest.skipIf(WINDOWS, "tested later") + def test_connections(self): + def create_socket(family, type): + sock = socket.socket(family, type) + sock.bind(('', 0)) + if type == socket.SOCK_STREAM: + sock.listen(1) + return sock + + socks = [] + socks.append(create_socket(socket.AF_INET, socket.SOCK_STREAM)) + socks.append(create_socket(socket.AF_INET, socket.SOCK_DGRAM)) + if supports_ipv6(): + socks.append(create_socket(socket.AF_INET6, socket.SOCK_STREAM)) + socks.append(create_socket(socket.AF_INET6, socket.SOCK_DGRAM)) + if hasattr(socket, 'AF_UNIX'): + safe_remove(TESTFN) + s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + s.bind(TESTFN) + s.listen(1) + socks.append(s) + kind = 'all' + # TODO: UNIX sockets are temporarily implemented by parsing + # 'pfiles' cmd output; we don't want that part of the code to + # be executed. + if SUNOS: + kind = 'inet' + try: + self.execute('connections', kind=kind) + finally: + for s in socks: + s.close() + + +p = get_test_subprocess() +DEAD_PROC = psutil.Process(p.pid) +DEAD_PROC.kill() +DEAD_PROC.wait() +del p + + +class TestProcessObjectLeaksZombie(TestProcessObjectLeaks): + """Same as above but looks for leaks occurring when dealing with + zombie processes raising NoSuchProcess exception. + """ + proc = DEAD_PROC + + def call(self, *args, **kwargs): + try: + TestProcessObjectLeaks.call(self, *args, **kwargs) + except psutil.NoSuchProcess: + pass + + if not POSIX: + def test_kill(self): + self.execute('kill') + + def test_terminate(self): + self.execute('terminate') + + def test_suspend(self): + self.execute('suspend') + + def test_resume(self): + self.execute('resume') + + def test_wait(self): + self.execute('wait') + + +class TestModuleFunctionsLeaks(Base): + """Test leaks of psutil module functions.""" + + def setUp(self): + gc.collect() + + def call(self, function, *args, **kwargs): + fun = getattr(psutil, function) + fun(*args, **kwargs) + + @skip_if_linux() + def test_cpu_count_logical(self): + psutil.cpu_count = psutil._psplatform.cpu_count_logical + self.execute('cpu_count') + + @skip_if_linux() + def test_cpu_count_physical(self): + psutil.cpu_count = psutil._psplatform.cpu_count_physical + self.execute('cpu_count') + + @skip_if_linux() + def test_boot_time(self): + self.execute('boot_time') + + @unittest.skipIf(POSIX and SKIP_PYTHON_IMPL, + "not worth being tested on POSIX (pure python)") + def test_pid_exists(self): + self.execute('pid_exists', os.getpid()) + + def test_virtual_memory(self): + self.execute('virtual_memory') + + # TODO: remove this skip when this gets fixed + @unittest.skipIf(SUNOS, + "not worth being tested on SUNOS (uses a subprocess)") + def test_swap_memory(self): + self.execute('swap_memory') + + @skip_if_linux() + def test_cpu_times(self): + self.execute('cpu_times') + + @skip_if_linux() + def test_per_cpu_times(self): + self.execute('cpu_times', percpu=True) + + @unittest.skipIf(POSIX and SKIP_PYTHON_IMPL, + "not worth being tested on POSIX (pure python)") + def test_disk_usage(self): + self.execute('disk_usage', '.') + + def test_disk_partitions(self): + self.execute('disk_partitions') + + @skip_if_linux() + def test_net_io_counters(self): + self.execute('net_io_counters') + + @unittest.skipIf(LINUX and not os.path.exists('/proc/diskstats'), + '/proc/diskstats not available on this Linux version') + @skip_if_linux() + def test_disk_io_counters(self): + self.execute('disk_io_counters') + + # XXX - on Windows this produces a false positive + @unittest.skipIf(WINDOWS, "XXX produces a false positive on Windows") + def test_users(self): + self.execute('users') + + @unittest.skipIf(LINUX, + "not worth being tested on Linux (pure python)") + def test_net_connections(self): + self.execute('net_connections') + + def test_net_if_addrs(self): + self.execute('net_if_addrs') + + @unittest.skipIf(TRAVIS, "EPERM on travis") + def test_net_if_stats(self): + self.execute('net_if_stats') + + +def main(): + test_suite = unittest.TestSuite() + tests = [TestProcessObjectLeaksZombie, + TestProcessObjectLeaks, + TestModuleFunctionsLeaks] + for test in tests: + test_suite.addTest(unittest.makeSuite(test)) + result = unittest.TextTestRunner(verbosity=2).run(test_suite) + return result.wasSuccessful() + +if __name__ == '__main__': + if not main(): + sys.exit(1) diff --git a/python/psutil/test/test_psutil.py b/python/psutil/test/test_psutil.py new file mode 100644 index 0000000000..3b2e3587ae --- /dev/null +++ b/python/psutil/test/test_psutil.py @@ -0,0 +1,3013 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +# Copyright (c) 2009, Giampaolo Rodola'. All rights reserved. +# Use of this source code is governed by a BSD-style license that can be +# found in the LICENSE file. + +""" +psutil test suite. Run it with: +$ make test + +If you're on Python < 2.7 unittest2 module must be installed first: +https://pypi.python.org/pypi/unittest2 +""" + +from __future__ import division + +import ast +import atexit +import collections +import contextlib +import datetime +import errno +import functools +import imp +import json +import os +import pickle +import pprint +import re +import select +import shutil +import signal +import socket +import stat +import subprocess +import sys +import tempfile +import textwrap +import threading +import time +import traceback +import types +import warnings +from socket import AF_INET, SOCK_STREAM, SOCK_DGRAM +try: + import ipaddress # python >= 3.3 +except ImportError: + ipaddress = None +try: + from unittest import mock # py3 +except ImportError: + import mock # requires "pip install mock" + +import psutil +from psutil._compat import PY3, callable, long, unicode + +if sys.version_info < (2, 7): + import unittest2 as unittest # https://pypi.python.org/pypi/unittest2 +else: + import unittest +if sys.version_info >= (3, 4): + import enum +else: + enum = None + + +# =================================================================== +# --- Constants +# =================================================================== + +# conf for retry_before_failing() decorator +NO_RETRIES = 10 +# bytes tolerance for OS memory related tests +TOLERANCE = 500 * 1024 # 500KB +# the timeout used in functions which have to wait +GLOBAL_TIMEOUT = 3 + +AF_INET6 = getattr(socket, "AF_INET6") +AF_UNIX = getattr(socket, "AF_UNIX", None) +PYTHON = os.path.realpath(sys.executable) +DEVNULL = open(os.devnull, 'r+') +TESTFN = os.path.join(os.getcwd(), "$testfile") +TESTFN_UNICODE = TESTFN + "ƒőő" +TESTFILE_PREFIX = 'psutil-test-suite-' +if not PY3: + try: + TESTFN_UNICODE = unicode(TESTFN_UNICODE, sys.getfilesystemencoding()) + except UnicodeDecodeError: + TESTFN_UNICODE = TESTFN + "???" + +EXAMPLES_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), + '..', 'examples')) + +POSIX = os.name == 'posix' +WINDOWS = os.name == 'nt' +if WINDOWS: + WIN_VISTA = (6, 0, 0) +LINUX = sys.platform.startswith("linux") +OSX = sys.platform.startswith("darwin") +BSD = sys.platform.startswith("freebsd") +SUNOS = sys.platform.startswith("sunos") +VALID_PROC_STATUSES = [getattr(psutil, x) for x in dir(psutil) + if x.startswith('STATUS_')] +# whether we're running this test suite on Travis (https://travis-ci.org/) +TRAVIS = bool(os.environ.get('TRAVIS')) +# whether we're running this test suite on Appveyor for Windows +# (http://www.appveyor.com/) +APPVEYOR = bool(os.environ.get('APPVEYOR')) + +if TRAVIS or 'tox' in sys.argv[0]: + import ipaddress +if TRAVIS or APPVEYOR: + GLOBAL_TIMEOUT = GLOBAL_TIMEOUT * 4 + + +# =================================================================== +# --- Utility functions +# =================================================================== + +def cleanup(): + reap_children(search_all=True) + safe_remove(TESTFN) + try: + safe_rmdir(TESTFN_UNICODE) + except UnicodeEncodeError: + pass + for path in _testfiles: + safe_remove(path) + +atexit.register(cleanup) +atexit.register(lambda: DEVNULL.close()) + + +_subprocesses_started = set() + + +def get_test_subprocess(cmd=None, stdout=DEVNULL, stderr=DEVNULL, + stdin=DEVNULL, wait=False): + """Return a subprocess.Popen object to use in tests. + By default stdout and stderr are redirected to /dev/null and the + python interpreter is used as test process. + If 'wait' is True attemps to make sure the process is in a + reasonably initialized state. + """ + if cmd is None: + pyline = "" + if wait: + pyline += "open(r'%s', 'w'); " % TESTFN + pyline += "import time; time.sleep(60);" + cmd_ = [PYTHON, "-c", pyline] + else: + cmd_ = cmd + sproc = subprocess.Popen(cmd_, stdout=stdout, stderr=stderr, stdin=stdin) + if wait: + if cmd is None: + stop_at = time.time() + 3 + while stop_at > time.time(): + if os.path.exists(TESTFN): + break + time.sleep(0.001) + else: + warn("couldn't make sure test file was actually created") + else: + wait_for_pid(sproc.pid) + _subprocesses_started.add(psutil.Process(sproc.pid)) + return sproc + + +_testfiles = [] + + +def pyrun(src): + """Run python code 'src' in a separate interpreter. + Return interpreter subprocess. + """ + if PY3: + src = bytes(src, 'ascii') + with tempfile.NamedTemporaryFile( + prefix=TESTFILE_PREFIX, delete=False) as f: + _testfiles.append(f.name) + f.write(src) + f.flush() + subp = get_test_subprocess([PYTHON, f.name], stdout=None, + stderr=None) + wait_for_pid(subp.pid) + return subp + + +def warn(msg): + """Raise a warning msg.""" + warnings.warn(msg, UserWarning) + + +def sh(cmdline, stdout=subprocess.PIPE, stderr=subprocess.PIPE): + """run cmd in a subprocess and return its output. + raises RuntimeError on error. + """ + p = subprocess.Popen(cmdline, shell=True, stdout=stdout, stderr=stderr) + stdout, stderr = p.communicate() + if p.returncode != 0: + raise RuntimeError(stderr) + if stderr: + warn(stderr) + if PY3: + stdout = str(stdout, sys.stdout.encoding) + return stdout.strip() + + +def which(program): + """Same as UNIX which command. Return None on command not found.""" + def is_exe(fpath): + return os.path.isfile(fpath) and os.access(fpath, os.X_OK) + + fpath, fname = os.path.split(program) + if fpath: + if is_exe(program): + return program + else: + for path in os.environ["PATH"].split(os.pathsep): + exe_file = os.path.join(path, program) + if is_exe(exe_file): + return exe_file + return None + + +if POSIX: + def get_kernel_version(): + """Return a tuple such as (2, 6, 36).""" + s = "" + uname = os.uname()[2] + for c in uname: + if c.isdigit() or c == '.': + s += c + else: + break + if not s: + raise ValueError("can't parse %r" % uname) + minor = 0 + micro = 0 + nums = s.split('.') + major = int(nums[0]) + if len(nums) >= 2: + minor = int(nums[1]) + if len(nums) >= 3: + micro = int(nums[2]) + return (major, minor, micro) + + +if LINUX: + RLIMIT_SUPPORT = get_kernel_version() >= (2, 6, 36) +else: + RLIMIT_SUPPORT = False + + +def wait_for_pid(pid, timeout=GLOBAL_TIMEOUT): + """Wait for pid to show up in the process list then return. + Used in the test suite to give time the sub process to initialize. + """ + raise_at = time.time() + timeout + while True: + if pid in psutil.pids(): + # give it one more iteration to allow full initialization + time.sleep(0.01) + return + time.sleep(0.0001) + if time.time() >= raise_at: + raise RuntimeError("Timed out") + + +def wait_for_file(fname, timeout=GLOBAL_TIMEOUT, delete_file=True): + """Wait for a file to be written on disk.""" + stop_at = time.time() + 3 + while time.time() < stop_at: + try: + with open(fname, "r") as f: + data = f.read() + if not data: + continue + if delete_file: + os.remove(fname) + return data + except IOError: + time.sleep(0.001) + raise RuntimeError("timed out (couldn't read file)") + + +def reap_children(search_all=False): + """Kill any subprocess started by this test suite and ensure that + no zombies stick around to hog resources and create problems when + looking for refleaks. + """ + global _subprocesses_started + procs = _subprocesses_started.copy() + if search_all: + this_process = psutil.Process() + for p in this_process.children(recursive=True): + procs.add(p) + for p in procs: + try: + p.terminate() + except psutil.NoSuchProcess: + pass + gone, alive = psutil.wait_procs(procs, timeout=GLOBAL_TIMEOUT) + for p in alive: + warn("couldn't terminate process %s" % p) + try: + p.kill() + except psutil.NoSuchProcess: + pass + _, alive = psutil.wait_procs(alive, timeout=GLOBAL_TIMEOUT) + if alive: + warn("couldn't not kill processes %s" % str(alive)) + _subprocesses_started = set(alive) + + +def check_ip_address(addr, family): + """Attempts to check IP address's validity.""" + if enum and PY3: + assert isinstance(family, enum.IntEnum), family + if family == AF_INET: + octs = [int(x) for x in addr.split('.')] + assert len(octs) == 4, addr + for num in octs: + assert 0 <= num <= 255, addr + if ipaddress: + if not PY3: + addr = unicode(addr) + ipaddress.IPv4Address(addr) + elif family == AF_INET6: + assert isinstance(addr, str), addr + if ipaddress: + if not PY3: + addr = unicode(addr) + ipaddress.IPv6Address(addr) + elif family == psutil.AF_LINK: + assert re.match('([a-fA-F0-9]{2}[:|\-]?){6}', addr) is not None, addr + else: + raise ValueError("unknown family %r", family) + + +def check_connection_ntuple(conn): + """Check validity of a connection namedtuple.""" + valid_conn_states = [getattr(psutil, x) for x in dir(psutil) if + x.startswith('CONN_')] + assert conn[0] == conn.fd + assert conn[1] == conn.family + assert conn[2] == conn.type + assert conn[3] == conn.laddr + assert conn[4] == conn.raddr + assert conn[5] == conn.status + assert conn.type in (SOCK_STREAM, SOCK_DGRAM), repr(conn.type) + assert conn.family in (AF_INET, AF_INET6, AF_UNIX), repr(conn.family) + assert conn.status in valid_conn_states, conn.status + + # check IP address and port sanity + for addr in (conn.laddr, conn.raddr): + if not addr: + continue + if conn.family in (AF_INET, AF_INET6): + assert isinstance(addr, tuple), addr + ip, port = addr + assert isinstance(port, int), port + assert 0 <= port <= 65535, port + check_ip_address(ip, conn.family) + elif conn.family == AF_UNIX: + assert isinstance(addr, (str, None)), addr + else: + raise ValueError("unknown family %r", conn.family) + + if conn.family in (AF_INET, AF_INET6): + # actually try to bind the local socket; ignore IPv6 + # sockets as their address might be represented as + # an IPv4-mapped-address (e.g. "::127.0.0.1") + # and that's rejected by bind() + if conn.family == AF_INET: + s = socket.socket(conn.family, conn.type) + with contextlib.closing(s): + try: + s.bind((conn.laddr[0], 0)) + except socket.error as err: + if err.errno != errno.EADDRNOTAVAIL: + raise + elif conn.family == AF_UNIX: + assert not conn.raddr, repr(conn.raddr) + assert conn.status == psutil.CONN_NONE, conn.status + + if getattr(conn, 'fd', -1) != -1: + assert conn.fd > 0, conn + if hasattr(socket, 'fromfd') and not WINDOWS: + try: + dupsock = socket.fromfd(conn.fd, conn.family, conn.type) + except (socket.error, OSError) as err: + if err.args[0] != errno.EBADF: + raise + else: + with contextlib.closing(dupsock): + assert dupsock.family == conn.family + assert dupsock.type == conn.type + + +def safe_remove(file): + "Convenience function for removing temporary test files" + try: + os.remove(file) + except OSError as err: + if err.errno != errno.ENOENT: + # file is being used by another process + if WINDOWS and isinstance(err, WindowsError) and err.errno == 13: + return + raise + + +def safe_rmdir(dir): + "Convenience function for removing temporary test directories" + try: + os.rmdir(dir) + except OSError as err: + if err.errno != errno.ENOENT: + raise + + +def call_until(fun, expr, timeout=GLOBAL_TIMEOUT): + """Keep calling function for timeout secs and exit if eval() + expression is True. + """ + stop_at = time.time() + timeout + while time.time() < stop_at: + ret = fun() + if eval(expr): + return ret + time.sleep(0.001) + raise RuntimeError('timed out (ret=%r)' % ret) + + +def retry_before_failing(ntimes=None): + """Decorator which runs a test function and retries N times before + actually failing. + """ + def decorator(fun): + @functools.wraps(fun) + def wrapper(*args, **kwargs): + for x in range(ntimes or NO_RETRIES): + try: + return fun(*args, **kwargs) + except AssertionError: + pass + raise + return wrapper + return decorator + + +def skip_on_access_denied(only_if=None): + """Decorator to Ignore AccessDenied exceptions.""" + def decorator(fun): + @functools.wraps(fun) + def wrapper(*args, **kwargs): + try: + return fun(*args, **kwargs) + except psutil.AccessDenied: + if only_if is not None: + if not only_if: + raise + msg = "%r was skipped because it raised AccessDenied" \ + % fun.__name__ + raise unittest.SkipTest(msg) + return wrapper + return decorator + + +def skip_on_not_implemented(only_if=None): + """Decorator to Ignore NotImplementedError exceptions.""" + def decorator(fun): + @functools.wraps(fun) + def wrapper(*args, **kwargs): + try: + return fun(*args, **kwargs) + except NotImplementedError: + if only_if is not None: + if not only_if: + raise + msg = "%r was skipped because it raised NotImplementedError" \ + % fun.__name__ + raise unittest.SkipTest(msg) + return wrapper + return decorator + + +def supports_ipv6(): + """Return True if IPv6 is supported on this platform.""" + if not socket.has_ipv6 or not hasattr(socket, "AF_INET6"): + return False + sock = None + try: + sock = socket.socket(AF_INET6, SOCK_STREAM) + sock.bind(("::1", 0)) + except (socket.error, socket.gaierror): + return False + else: + return True + finally: + if sock is not None: + sock.close() + + +if WINDOWS: + def get_winver(): + wv = sys.getwindowsversion() + if hasattr(wv, 'service_pack_major'): # python >= 2.7 + sp = wv.service_pack_major or 0 + else: + r = re.search("\s\d$", wv[4]) + if r: + sp = int(r.group(0)) + else: + sp = 0 + return (wv[0], wv[1], sp) + + +class ThreadTask(threading.Thread): + """A thread object used for running process thread tests.""" + + def __init__(self): + threading.Thread.__init__(self) + self._running = False + self._interval = None + self._flag = threading.Event() + + def __repr__(self): + name = self.__class__.__name__ + return '<%s running=%s at %#x>' % (name, self._running, id(self)) + + def start(self, interval=0.001): + """Start thread and keep it running until an explicit + stop() request. Polls for shutdown every 'timeout' seconds. + """ + if self._running: + raise ValueError("already started") + self._interval = interval + threading.Thread.start(self) + self._flag.wait() + + def run(self): + self._running = True + self._flag.set() + while self._running: + time.sleep(self._interval) + + def stop(self): + """Stop thread execution and and waits until it is stopped.""" + if not self._running: + raise ValueError("already stopped") + self._running = False + self.join() + + +# =================================================================== +# --- System-related API tests +# =================================================================== + +class TestSystemAPIs(unittest.TestCase): + """Tests for system-related APIs.""" + + def setUp(self): + safe_remove(TESTFN) + + def tearDown(self): + reap_children() + + def test_process_iter(self): + self.assertIn(os.getpid(), [x.pid for x in psutil.process_iter()]) + sproc = get_test_subprocess() + self.assertIn(sproc.pid, [x.pid for x in psutil.process_iter()]) + p = psutil.Process(sproc.pid) + p.kill() + p.wait() + self.assertNotIn(sproc.pid, [x.pid for x in psutil.process_iter()]) + + def test_wait_procs(self): + def callback(p): + l.append(p.pid) + + l = [] + sproc1 = get_test_subprocess() + sproc2 = get_test_subprocess() + sproc3 = get_test_subprocess() + procs = [psutil.Process(x.pid) for x in (sproc1, sproc2, sproc3)] + self.assertRaises(ValueError, psutil.wait_procs, procs, timeout=-1) + self.assertRaises(TypeError, psutil.wait_procs, procs, callback=1) + t = time.time() + gone, alive = psutil.wait_procs(procs, timeout=0.01, callback=callback) + + self.assertLess(time.time() - t, 0.5) + self.assertEqual(gone, []) + self.assertEqual(len(alive), 3) + self.assertEqual(l, []) + for p in alive: + self.assertFalse(hasattr(p, 'returncode')) + + @retry_before_failing(30) + def test(procs, callback): + gone, alive = psutil.wait_procs(procs, timeout=0.03, + callback=callback) + self.assertEqual(len(gone), 1) + self.assertEqual(len(alive), 2) + return gone, alive + + sproc3.terminate() + gone, alive = test(procs, callback) + self.assertIn(sproc3.pid, [x.pid for x in gone]) + if POSIX: + self.assertEqual(gone.pop().returncode, signal.SIGTERM) + else: + self.assertEqual(gone.pop().returncode, 1) + self.assertEqual(l, [sproc3.pid]) + for p in alive: + self.assertFalse(hasattr(p, 'returncode')) + + @retry_before_failing(30) + def test(procs, callback): + gone, alive = psutil.wait_procs(procs, timeout=0.03, + callback=callback) + self.assertEqual(len(gone), 3) + self.assertEqual(len(alive), 0) + return gone, alive + + sproc1.terminate() + sproc2.terminate() + gone, alive = test(procs, callback) + self.assertEqual(set(l), set([sproc1.pid, sproc2.pid, sproc3.pid])) + for p in gone: + self.assertTrue(hasattr(p, 'returncode')) + + def test_wait_procs_no_timeout(self): + sproc1 = get_test_subprocess() + sproc2 = get_test_subprocess() + sproc3 = get_test_subprocess() + procs = [psutil.Process(x.pid) for x in (sproc1, sproc2, sproc3)] + for p in procs: + p.terminate() + gone, alive = psutil.wait_procs(procs) + + def test_boot_time(self): + bt = psutil.boot_time() + self.assertIsInstance(bt, float) + self.assertGreater(bt, 0) + self.assertLess(bt, time.time()) + + @unittest.skipUnless(POSIX, 'posix only') + def test_PAGESIZE(self): + # pagesize is used internally to perform different calculations + # and it's determined by using SC_PAGE_SIZE; make sure + # getpagesize() returns the same value. + import resource + self.assertEqual(os.sysconf("SC_PAGE_SIZE"), resource.getpagesize()) + + def test_virtual_memory(self): + mem = psutil.virtual_memory() + assert mem.total > 0, mem + assert mem.available > 0, mem + assert 0 <= mem.percent <= 100, mem + assert mem.used > 0, mem + assert mem.free >= 0, mem + for name in mem._fields: + value = getattr(mem, name) + if name != 'percent': + self.assertIsInstance(value, (int, long)) + if name != 'total': + if not value >= 0: + self.fail("%r < 0 (%s)" % (name, value)) + if value > mem.total: + self.fail("%r > total (total=%s, %s=%s)" + % (name, mem.total, name, value)) + + def test_swap_memory(self): + mem = psutil.swap_memory() + assert mem.total >= 0, mem + assert mem.used >= 0, mem + if mem.total > 0: + # likely a system with no swap partition + assert mem.free > 0, mem + else: + assert mem.free == 0, mem + assert 0 <= mem.percent <= 100, mem + assert mem.sin >= 0, mem + assert mem.sout >= 0, mem + + def test_pid_exists(self): + sproc = get_test_subprocess(wait=True) + self.assertTrue(psutil.pid_exists(sproc.pid)) + p = psutil.Process(sproc.pid) + p.kill() + p.wait() + self.assertFalse(psutil.pid_exists(sproc.pid)) + self.assertFalse(psutil.pid_exists(-1)) + self.assertEqual(psutil.pid_exists(0), 0 in psutil.pids()) + # pid 0 + psutil.pid_exists(0) == 0 in psutil.pids() + + def test_pid_exists_2(self): + reap_children() + pids = psutil.pids() + for pid in pids: + try: + assert psutil.pid_exists(pid) + except AssertionError: + # in case the process disappeared in meantime fail only + # if it is no longer in psutil.pids() + time.sleep(.1) + if pid in psutil.pids(): + self.fail(pid) + pids = range(max(pids) + 5000, max(pids) + 6000) + for pid in pids: + self.assertFalse(psutil.pid_exists(pid), msg=pid) + + def test_pids(self): + plist = [x.pid for x in psutil.process_iter()] + pidlist = psutil.pids() + self.assertEqual(plist.sort(), pidlist.sort()) + # make sure every pid is unique + self.assertEqual(len(pidlist), len(set(pidlist))) + + def test_test(self): + # test for psutil.test() function + stdout = sys.stdout + sys.stdout = DEVNULL + try: + psutil.test() + finally: + sys.stdout = stdout + + def test_cpu_count(self): + logical = psutil.cpu_count() + self.assertEqual(logical, len(psutil.cpu_times(percpu=True))) + self.assertGreaterEqual(logical, 1) + # + if LINUX: + with open("/proc/cpuinfo") as fd: + cpuinfo_data = fd.read() + if "physical id" not in cpuinfo_data: + raise unittest.SkipTest("cpuinfo doesn't include physical id") + physical = psutil.cpu_count(logical=False) + self.assertGreaterEqual(physical, 1) + self.assertGreaterEqual(logical, physical) + + def test_sys_cpu_times(self): + total = 0 + times = psutil.cpu_times() + sum(times) + for cp_time in times: + self.assertIsInstance(cp_time, float) + self.assertGreaterEqual(cp_time, 0.0) + total += cp_time + self.assertEqual(total, sum(times)) + str(times) + if not WINDOWS: + # CPU times are always supposed to increase over time or + # remain the same but never go backwards, see: + # https://github.com/giampaolo/psutil/issues/392 + last = psutil.cpu_times() + for x in range(100): + new = psutil.cpu_times() + for field in new._fields: + new_t = getattr(new, field) + last_t = getattr(last, field) + self.assertGreaterEqual(new_t, last_t, + msg="%s %s" % (new_t, last_t)) + last = new + + def test_sys_cpu_times2(self): + t1 = sum(psutil.cpu_times()) + time.sleep(0.1) + t2 = sum(psutil.cpu_times()) + difference = t2 - t1 + if not difference >= 0.05: + self.fail("difference %s" % difference) + + def test_sys_per_cpu_times(self): + for times in psutil.cpu_times(percpu=True): + total = 0 + sum(times) + for cp_time in times: + self.assertIsInstance(cp_time, float) + self.assertGreaterEqual(cp_time, 0.0) + total += cp_time + self.assertEqual(total, sum(times)) + str(times) + self.assertEqual(len(psutil.cpu_times(percpu=True)[0]), + len(psutil.cpu_times(percpu=False))) + + # Note: in theory CPU times are always supposed to increase over + # time or remain the same but never go backwards. In practice + # sometimes this is not the case. + # This issue seemd to be afflict Windows: + # https://github.com/giampaolo/psutil/issues/392 + # ...but it turns out also Linux (rarely) behaves the same. + # last = psutil.cpu_times(percpu=True) + # for x in range(100): + # new = psutil.cpu_times(percpu=True) + # for index in range(len(new)): + # newcpu = new[index] + # lastcpu = last[index] + # for field in newcpu._fields: + # new_t = getattr(newcpu, field) + # last_t = getattr(lastcpu, field) + # self.assertGreaterEqual( + # new_t, last_t, msg="%s %s" % (lastcpu, newcpu)) + # last = new + + def test_sys_per_cpu_times_2(self): + tot1 = psutil.cpu_times(percpu=True) + stop_at = time.time() + 0.1 + while True: + if time.time() >= stop_at: + break + tot2 = psutil.cpu_times(percpu=True) + for t1, t2 in zip(tot1, tot2): + t1, t2 = sum(t1), sum(t2) + difference = t2 - t1 + if difference >= 0.05: + return + self.fail() + + def _test_cpu_percent(self, percent, last_ret, new_ret): + try: + self.assertIsInstance(percent, float) + self.assertGreaterEqual(percent, 0.0) + self.assertIsNot(percent, -0.0) + self.assertLessEqual(percent, 100.0 * psutil.cpu_count()) + except AssertionError as err: + raise AssertionError("\n%s\nlast=%s\nnew=%s" % ( + err, pprint.pformat(last_ret), pprint.pformat(new_ret))) + + def test_sys_cpu_percent(self): + last = psutil.cpu_percent(interval=0.001) + for x in range(100): + new = psutil.cpu_percent(interval=None) + self._test_cpu_percent(new, last, new) + last = new + + def test_sys_per_cpu_percent(self): + last = psutil.cpu_percent(interval=0.001, percpu=True) + self.assertEqual(len(last), psutil.cpu_count()) + for x in range(100): + new = psutil.cpu_percent(interval=None, percpu=True) + for percent in new: + self._test_cpu_percent(percent, last, new) + last = new + + def test_sys_cpu_times_percent(self): + last = psutil.cpu_times_percent(interval=0.001) + for x in range(100): + new = psutil.cpu_times_percent(interval=None) + for percent in new: + self._test_cpu_percent(percent, last, new) + self._test_cpu_percent(sum(new), last, new) + last = new + + def test_sys_per_cpu_times_percent(self): + last = psutil.cpu_times_percent(interval=0.001, percpu=True) + self.assertEqual(len(last), psutil.cpu_count()) + for x in range(100): + new = psutil.cpu_times_percent(interval=None, percpu=True) + for cpu in new: + for percent in cpu: + self._test_cpu_percent(percent, last, new) + self._test_cpu_percent(sum(cpu), last, new) + last = new + + def test_sys_per_cpu_times_percent_negative(self): + # see: https://github.com/giampaolo/psutil/issues/645 + psutil.cpu_times_percent(percpu=True) + zero_times = [x._make([0 for x in range(len(x._fields))]) + for x in psutil.cpu_times(percpu=True)] + with mock.patch('psutil.cpu_times', return_value=zero_times): + for cpu in psutil.cpu_times_percent(percpu=True): + for percent in cpu: + self._test_cpu_percent(percent, None, None) + + @unittest.skipIf(POSIX and not hasattr(os, 'statvfs'), + "os.statvfs() function not available on this platform") + def test_disk_usage(self): + usage = psutil.disk_usage(os.getcwd()) + assert usage.total > 0, usage + assert usage.used > 0, usage + assert usage.free > 0, usage + assert usage.total > usage.used, usage + assert usage.total > usage.free, usage + assert 0 <= usage.percent <= 100, usage.percent + if hasattr(shutil, 'disk_usage'): + # py >= 3.3, see: http://bugs.python.org/issue12442 + shutil_usage = shutil.disk_usage(os.getcwd()) + tolerance = 5 * 1024 * 1024 # 5MB + self.assertEqual(usage.total, shutil_usage.total) + self.assertAlmostEqual(usage.free, shutil_usage.free, + delta=tolerance) + self.assertAlmostEqual(usage.used, shutil_usage.used, + delta=tolerance) + + # if path does not exist OSError ENOENT is expected across + # all platforms + fname = tempfile.mktemp() + try: + psutil.disk_usage(fname) + except OSError as err: + if err.args[0] != errno.ENOENT: + raise + else: + self.fail("OSError not raised") + + @unittest.skipIf(POSIX and not hasattr(os, 'statvfs'), + "os.statvfs() function not available on this platform") + def test_disk_usage_unicode(self): + # see: https://github.com/giampaolo/psutil/issues/416 + # XXX this test is not really reliable as it always fails on + # Python 3.X (2.X is fine) + try: + safe_rmdir(TESTFN_UNICODE) + os.mkdir(TESTFN_UNICODE) + psutil.disk_usage(TESTFN_UNICODE) + safe_rmdir(TESTFN_UNICODE) + except UnicodeEncodeError: + pass + + @unittest.skipIf(POSIX and not hasattr(os, 'statvfs'), + "os.statvfs() function not available on this platform") + @unittest.skipIf(LINUX and TRAVIS, "unknown failure on travis") + def test_disk_partitions(self): + # all = False + ls = psutil.disk_partitions(all=False) + # on travis we get: + # self.assertEqual(p.cpu_affinity(), [n]) + # AssertionError: Lists differ: [0, 1, 2, 3, 4, 5, 6, 7,... != [0] + self.assertTrue(ls, msg=ls) + for disk in ls: + if WINDOWS and 'cdrom' in disk.opts: + continue + if not POSIX: + assert os.path.exists(disk.device), disk + else: + # we cannot make any assumption about this, see: + # http://goo.gl/p9c43 + disk.device + if SUNOS: + # on solaris apparently mount points can also be files + assert os.path.exists(disk.mountpoint), disk + else: + assert os.path.isdir(disk.mountpoint), disk + assert disk.fstype, disk + self.assertIsInstance(disk.opts, str) + + # all = True + ls = psutil.disk_partitions(all=True) + self.assertTrue(ls, msg=ls) + for disk in psutil.disk_partitions(all=True): + if not WINDOWS: + try: + os.stat(disk.mountpoint) + except OSError as err: + # http://mail.python.org/pipermail/python-dev/ + # 2012-June/120787.html + if err.errno not in (errno.EPERM, errno.EACCES): + raise + else: + if SUNOS: + # on solaris apparently mount points can also be files + assert os.path.exists(disk.mountpoint), disk + else: + assert os.path.isdir(disk.mountpoint), disk + self.assertIsInstance(disk.fstype, str) + self.assertIsInstance(disk.opts, str) + + def find_mount_point(path): + path = os.path.abspath(path) + while not os.path.ismount(path): + path = os.path.dirname(path) + return path + + mount = find_mount_point(__file__) + mounts = [x.mountpoint for x in psutil.disk_partitions(all=True)] + self.assertIn(mount, mounts) + psutil.disk_usage(mount) + + @skip_on_access_denied() + def test_net_connections(self): + def check(cons, families, types_): + for conn in cons: + self.assertIn(conn.family, families, msg=conn) + if conn.family != getattr(socket, 'AF_UNIX', object()): + self.assertIn(conn.type, types_, msg=conn) + + from psutil._common import conn_tmap + for kind, groups in conn_tmap.items(): + if SUNOS and kind == 'unix': + continue + families, types_ = groups + cons = psutil.net_connections(kind) + self.assertEqual(len(cons), len(set(cons))) + check(cons, families, types_) + + def test_net_io_counters(self): + def check_ntuple(nt): + self.assertEqual(nt[0], nt.bytes_sent) + self.assertEqual(nt[1], nt.bytes_recv) + self.assertEqual(nt[2], nt.packets_sent) + self.assertEqual(nt[3], nt.packets_recv) + self.assertEqual(nt[4], nt.errin) + self.assertEqual(nt[5], nt.errout) + self.assertEqual(nt[6], nt.dropin) + self.assertEqual(nt[7], nt.dropout) + assert nt.bytes_sent >= 0, nt + assert nt.bytes_recv >= 0, nt + assert nt.packets_sent >= 0, nt + assert nt.packets_recv >= 0, nt + assert nt.errin >= 0, nt + assert nt.errout >= 0, nt + assert nt.dropin >= 0, nt + assert nt.dropout >= 0, nt + + ret = psutil.net_io_counters(pernic=False) + check_ntuple(ret) + ret = psutil.net_io_counters(pernic=True) + self.assertNotEqual(ret, []) + for key in ret: + self.assertTrue(key) + check_ntuple(ret[key]) + + def test_net_if_addrs(self): + nics = psutil.net_if_addrs() + assert nics, nics + + # Not reliable on all platforms (net_if_addrs() reports more + # interfaces). + # self.assertEqual(sorted(nics.keys()), + # sorted(psutil.net_io_counters(pernic=True).keys())) + + families = set([socket.AF_INET, AF_INET6, psutil.AF_LINK]) + for nic, addrs in nics.items(): + self.assertEqual(len(set(addrs)), len(addrs)) + for addr in addrs: + self.assertIsInstance(addr.family, int) + self.assertIsInstance(addr.address, str) + self.assertIsInstance(addr.netmask, (str, type(None))) + self.assertIsInstance(addr.broadcast, (str, type(None))) + self.assertIn(addr.family, families) + if sys.version_info >= (3, 4): + self.assertIsInstance(addr.family, enum.IntEnum) + if addr.family == socket.AF_INET: + s = socket.socket(addr.family) + with contextlib.closing(s): + s.bind((addr.address, 0)) + elif addr.family == socket.AF_INET6: + info = socket.getaddrinfo( + addr.address, 0, socket.AF_INET6, socket.SOCK_STREAM, + 0, socket.AI_PASSIVE)[0] + af, socktype, proto, canonname, sa = info + s = socket.socket(af, socktype, proto) + with contextlib.closing(s): + s.bind(sa) + for ip in (addr.address, addr.netmask, addr.broadcast): + if ip is not None: + # TODO: skip AF_INET6 for now because I get: + # AddressValueError: Only hex digits permitted in + # u'c6f3%lxcbr0' in u'fe80::c8e0:fff:fe54:c6f3%lxcbr0' + if addr.family != AF_INET6: + check_ip_address(ip, addr.family) + + if BSD or OSX or SUNOS: + if hasattr(socket, "AF_LINK"): + self.assertEqual(psutil.AF_LINK, socket.AF_LINK) + elif LINUX: + self.assertEqual(psutil.AF_LINK, socket.AF_PACKET) + elif WINDOWS: + self.assertEqual(psutil.AF_LINK, -1) + + @unittest.skipIf(TRAVIS, "EPERM on travis") + def test_net_if_stats(self): + nics = psutil.net_if_stats() + assert nics, nics + all_duplexes = (psutil.NIC_DUPLEX_FULL, + psutil.NIC_DUPLEX_HALF, + psutil.NIC_DUPLEX_UNKNOWN) + for nic, stats in nics.items(): + isup, duplex, speed, mtu = stats + self.assertIsInstance(isup, bool) + self.assertIn(duplex, all_duplexes) + self.assertIn(duplex, all_duplexes) + self.assertGreaterEqual(speed, 0) + self.assertGreaterEqual(mtu, 0) + + @unittest.skipIf(LINUX and not os.path.exists('/proc/diskstats'), + '/proc/diskstats not available on this linux version') + @unittest.skipIf(APPVEYOR, + "can't find any physical disk on Appveyor") + def test_disk_io_counters(self): + def check_ntuple(nt): + self.assertEqual(nt[0], nt.read_count) + self.assertEqual(nt[1], nt.write_count) + self.assertEqual(nt[2], nt.read_bytes) + self.assertEqual(nt[3], nt.write_bytes) + self.assertEqual(nt[4], nt.read_time) + self.assertEqual(nt[5], nt.write_time) + assert nt.read_count >= 0, nt + assert nt.write_count >= 0, nt + assert nt.read_bytes >= 0, nt + assert nt.write_bytes >= 0, nt + assert nt.read_time >= 0, nt + assert nt.write_time >= 0, nt + + ret = psutil.disk_io_counters(perdisk=False) + check_ntuple(ret) + ret = psutil.disk_io_counters(perdisk=True) + # make sure there are no duplicates + self.assertEqual(len(ret), len(set(ret))) + for key in ret: + assert key, key + check_ntuple(ret[key]) + if LINUX and key[-1].isdigit(): + # if 'sda1' is listed 'sda' shouldn't, see: + # https://github.com/giampaolo/psutil/issues/338 + while key[-1].isdigit(): + key = key[:-1] + self.assertNotIn(key, ret.keys()) + + def test_users(self): + users = psutil.users() + if not APPVEYOR: + self.assertNotEqual(users, []) + for user in users: + assert user.name, user + user.terminal + user.host + assert user.started > 0.0, user + datetime.datetime.fromtimestamp(user.started) + + +# =================================================================== +# --- psutil.Process class tests +# =================================================================== + +class TestProcess(unittest.TestCase): + """Tests for psutil.Process class.""" + + def setUp(self): + safe_remove(TESTFN) + + def tearDown(self): + reap_children() + + def test_pid(self): + self.assertEqual(psutil.Process().pid, os.getpid()) + sproc = get_test_subprocess() + self.assertEqual(psutil.Process(sproc.pid).pid, sproc.pid) + + def test_kill(self): + sproc = get_test_subprocess(wait=True) + test_pid = sproc.pid + p = psutil.Process(test_pid) + p.kill() + sig = p.wait() + self.assertFalse(psutil.pid_exists(test_pid)) + if POSIX: + self.assertEqual(sig, signal.SIGKILL) + + def test_terminate(self): + sproc = get_test_subprocess(wait=True) + test_pid = sproc.pid + p = psutil.Process(test_pid) + p.terminate() + sig = p.wait() + self.assertFalse(psutil.pid_exists(test_pid)) + if POSIX: + self.assertEqual(sig, signal.SIGTERM) + + def test_send_signal(self): + sig = signal.SIGKILL if POSIX else signal.SIGTERM + sproc = get_test_subprocess() + p = psutil.Process(sproc.pid) + p.send_signal(sig) + exit_sig = p.wait() + self.assertFalse(psutil.pid_exists(p.pid)) + if POSIX: + self.assertEqual(exit_sig, sig) + # + sproc = get_test_subprocess() + p = psutil.Process(sproc.pid) + p.send_signal(sig) + with mock.patch('psutil.os.kill', + side_effect=OSError(errno.ESRCH, "")) as fun: + with self.assertRaises(psutil.NoSuchProcess): + p.send_signal(sig) + assert fun.called + # + sproc = get_test_subprocess() + p = psutil.Process(sproc.pid) + p.send_signal(sig) + with mock.patch('psutil.os.kill', + side_effect=OSError(errno.EPERM, "")) as fun: + with self.assertRaises(psutil.AccessDenied): + p.send_signal(sig) + assert fun.called + + def test_wait(self): + # check exit code signal + sproc = get_test_subprocess() + p = psutil.Process(sproc.pid) + p.kill() + code = p.wait() + if POSIX: + self.assertEqual(code, signal.SIGKILL) + else: + self.assertEqual(code, 0) + self.assertFalse(p.is_running()) + + sproc = get_test_subprocess() + p = psutil.Process(sproc.pid) + p.terminate() + code = p.wait() + if POSIX: + self.assertEqual(code, signal.SIGTERM) + else: + self.assertEqual(code, 0) + self.assertFalse(p.is_running()) + + # check sys.exit() code + code = "import time, sys; time.sleep(0.01); sys.exit(5);" + sproc = get_test_subprocess([PYTHON, "-c", code]) + p = psutil.Process(sproc.pid) + self.assertEqual(p.wait(), 5) + self.assertFalse(p.is_running()) + + # Test wait() issued twice. + # It is not supposed to raise NSP when the process is gone. + # On UNIX this should return None, on Windows it should keep + # returning the exit code. + sproc = get_test_subprocess([PYTHON, "-c", code]) + p = psutil.Process(sproc.pid) + self.assertEqual(p.wait(), 5) + self.assertIn(p.wait(), (5, None)) + + # test timeout + sproc = get_test_subprocess() + p = psutil.Process(sproc.pid) + p.name() + self.assertRaises(psutil.TimeoutExpired, p.wait, 0.01) + + # timeout < 0 not allowed + self.assertRaises(ValueError, p.wait, -1) + + # XXX why is this skipped on Windows? + @unittest.skipUnless(POSIX, 'skipped on Windows') + def test_wait_non_children(self): + # test wait() against processes which are not our children + code = "import sys;" + code += "from subprocess import Popen, PIPE;" + code += "cmd = ['%s', '-c', 'import time; time.sleep(60)'];" % PYTHON + code += "sp = Popen(cmd, stdout=PIPE);" + code += "sys.stdout.write(str(sp.pid));" + sproc = get_test_subprocess([PYTHON, "-c", code], + stdout=subprocess.PIPE) + grandson_pid = int(sproc.stdout.read()) + grandson_proc = psutil.Process(grandson_pid) + try: + self.assertRaises(psutil.TimeoutExpired, grandson_proc.wait, 0.01) + grandson_proc.kill() + ret = grandson_proc.wait() + self.assertEqual(ret, None) + finally: + if grandson_proc.is_running(): + grandson_proc.kill() + grandson_proc.wait() + + def test_wait_timeout_0(self): + sproc = get_test_subprocess() + p = psutil.Process(sproc.pid) + self.assertRaises(psutil.TimeoutExpired, p.wait, 0) + p.kill() + stop_at = time.time() + 2 + while True: + try: + code = p.wait(0) + except psutil.TimeoutExpired: + if time.time() >= stop_at: + raise + else: + break + if POSIX: + self.assertEqual(code, signal.SIGKILL) + else: + self.assertEqual(code, 0) + self.assertFalse(p.is_running()) + + def test_cpu_percent(self): + p = psutil.Process() + p.cpu_percent(interval=0.001) + p.cpu_percent(interval=0.001) + for x in range(100): + percent = p.cpu_percent(interval=None) + self.assertIsInstance(percent, float) + self.assertGreaterEqual(percent, 0.0) + if not POSIX: + self.assertLessEqual(percent, 100.0) + else: + self.assertGreaterEqual(percent, 0.0) + + def test_cpu_times(self): + times = psutil.Process().cpu_times() + assert (times.user > 0.0) or (times.system > 0.0), times + # make sure returned values can be pretty printed with strftime + time.strftime("%H:%M:%S", time.localtime(times.user)) + time.strftime("%H:%M:%S", time.localtime(times.system)) + + # Test Process.cpu_times() against os.times() + # os.times() is broken on Python 2.6 + # http://bugs.python.org/issue1040026 + # XXX fails on OSX: not sure if it's for os.times(). We should + # try this with Python 2.7 and re-enable the test. + + @unittest.skipUnless(sys.version_info > (2, 6, 1) and not OSX, + 'os.times() is not reliable on this Python version') + def test_cpu_times2(self): + user_time, kernel_time = psutil.Process().cpu_times() + utime, ktime = os.times()[:2] + + # Use os.times()[:2] as base values to compare our results + # using a tolerance of +/- 0.1 seconds. + # It will fail if the difference between the values is > 0.1s. + if (max([user_time, utime]) - min([user_time, utime])) > 0.1: + self.fail("expected: %s, found: %s" % (utime, user_time)) + + if (max([kernel_time, ktime]) - min([kernel_time, ktime])) > 0.1: + self.fail("expected: %s, found: %s" % (ktime, kernel_time)) + + def test_create_time(self): + sproc = get_test_subprocess(wait=True) + now = time.time() + p = psutil.Process(sproc.pid) + create_time = p.create_time() + + # Use time.time() as base value to compare our result using a + # tolerance of +/- 1 second. + # It will fail if the difference between the values is > 2s. + difference = abs(create_time - now) + if difference > 2: + self.fail("expected: %s, found: %s, difference: %s" + % (now, create_time, difference)) + + # make sure returned value can be pretty printed with strftime + time.strftime("%Y %m %d %H:%M:%S", time.localtime(p.create_time())) + + @unittest.skipIf(WINDOWS, 'Windows only') + def test_terminal(self): + terminal = psutil.Process().terminal() + if sys.stdin.isatty(): + self.assertEqual(terminal, sh('tty')) + else: + assert terminal, repr(terminal) + + @unittest.skipUnless(LINUX or BSD or WINDOWS, + 'not available on this platform') + @skip_on_not_implemented(only_if=LINUX) + def test_io_counters(self): + p = psutil.Process() + # test reads + io1 = p.io_counters() + with open(PYTHON, 'rb') as f: + f.read() + io2 = p.io_counters() + if not BSD: + assert io2.read_count > io1.read_count, (io1, io2) + self.assertEqual(io2.write_count, io1.write_count) + assert io2.read_bytes >= io1.read_bytes, (io1, io2) + assert io2.write_bytes >= io1.write_bytes, (io1, io2) + # test writes + io1 = p.io_counters() + with tempfile.TemporaryFile(prefix=TESTFILE_PREFIX) as f: + if PY3: + f.write(bytes("x" * 1000000, 'ascii')) + else: + f.write("x" * 1000000) + io2 = p.io_counters() + assert io2.write_count >= io1.write_count, (io1, io2) + assert io2.write_bytes >= io1.write_bytes, (io1, io2) + assert io2.read_count >= io1.read_count, (io1, io2) + assert io2.read_bytes >= io1.read_bytes, (io1, io2) + + @unittest.skipUnless(LINUX or (WINDOWS and get_winver() >= WIN_VISTA), + 'Linux and Windows Vista only') + @unittest.skipIf(LINUX and TRAVIS, "unknown failure on travis") + def test_ionice(self): + if LINUX: + from psutil import (IOPRIO_CLASS_NONE, IOPRIO_CLASS_RT, + IOPRIO_CLASS_BE, IOPRIO_CLASS_IDLE) + self.assertEqual(IOPRIO_CLASS_NONE, 0) + self.assertEqual(IOPRIO_CLASS_RT, 1) + self.assertEqual(IOPRIO_CLASS_BE, 2) + self.assertEqual(IOPRIO_CLASS_IDLE, 3) + p = psutil.Process() + try: + p.ionice(2) + ioclass, value = p.ionice() + if enum is not None: + self.assertIsInstance(ioclass, enum.IntEnum) + self.assertEqual(ioclass, 2) + self.assertEqual(value, 4) + # + p.ionice(3) + ioclass, value = p.ionice() + self.assertEqual(ioclass, 3) + self.assertEqual(value, 0) + # + p.ionice(2, 0) + ioclass, value = p.ionice() + self.assertEqual(ioclass, 2) + self.assertEqual(value, 0) + p.ionice(2, 7) + ioclass, value = p.ionice() + self.assertEqual(ioclass, 2) + self.assertEqual(value, 7) + # + self.assertRaises(ValueError, p.ionice, 2, 10) + self.assertRaises(ValueError, p.ionice, 2, -1) + self.assertRaises(ValueError, p.ionice, 4) + self.assertRaises(TypeError, p.ionice, 2, "foo") + self.assertRaisesRegexp( + ValueError, "can't specify value with IOPRIO_CLASS_NONE", + p.ionice, psutil.IOPRIO_CLASS_NONE, 1) + self.assertRaisesRegexp( + ValueError, "can't specify value with IOPRIO_CLASS_IDLE", + p.ionice, psutil.IOPRIO_CLASS_IDLE, 1) + self.assertRaisesRegexp( + ValueError, "'ioclass' argument must be specified", + p.ionice, value=1) + finally: + p.ionice(IOPRIO_CLASS_NONE) + else: + p = psutil.Process() + original = p.ionice() + self.assertIsInstance(original, int) + try: + value = 0 # very low + if original == value: + value = 1 # low + p.ionice(value) + self.assertEqual(p.ionice(), value) + finally: + p.ionice(original) + # + self.assertRaises(ValueError, p.ionice, 3) + self.assertRaises(TypeError, p.ionice, 2, 1) + + @unittest.skipUnless(LINUX and RLIMIT_SUPPORT, + "only available on Linux >= 2.6.36") + def test_rlimit_get(self): + import resource + p = psutil.Process(os.getpid()) + names = [x for x in dir(psutil) if x.startswith('RLIMIT')] + assert names, names + for name in names: + value = getattr(psutil, name) + self.assertGreaterEqual(value, 0) + if name in dir(resource): + self.assertEqual(value, getattr(resource, name)) + self.assertEqual(p.rlimit(value), resource.getrlimit(value)) + else: + ret = p.rlimit(value) + self.assertEqual(len(ret), 2) + self.assertGreaterEqual(ret[0], -1) + self.assertGreaterEqual(ret[1], -1) + + @unittest.skipUnless(LINUX and RLIMIT_SUPPORT, + "only available on Linux >= 2.6.36") + def test_rlimit_set(self): + sproc = get_test_subprocess() + p = psutil.Process(sproc.pid) + p.rlimit(psutil.RLIMIT_NOFILE, (5, 5)) + self.assertEqual(p.rlimit(psutil.RLIMIT_NOFILE), (5, 5)) + # If pid is 0 prlimit() applies to the calling process and + # we don't want that. + with self.assertRaises(ValueError): + psutil._psplatform.Process(0).rlimit(0) + with self.assertRaises(ValueError): + p.rlimit(psutil.RLIMIT_NOFILE, (5, 5, 5)) + + def test_num_threads(self): + # on certain platforms such as Linux we might test for exact + # thread number, since we always have with 1 thread per process, + # but this does not apply across all platforms (OSX, Windows) + p = psutil.Process() + step1 = p.num_threads() + + thread = ThreadTask() + thread.start() + try: + step2 = p.num_threads() + self.assertEqual(step2, step1 + 1) + thread.stop() + finally: + if thread._running: + thread.stop() + + @unittest.skipUnless(WINDOWS, 'Windows only') + def test_num_handles(self): + # a better test is done later into test/_windows.py + p = psutil.Process() + self.assertGreater(p.num_handles(), 0) + + def test_threads(self): + p = psutil.Process() + step1 = p.threads() + + thread = ThreadTask() + thread.start() + + try: + step2 = p.threads() + self.assertEqual(len(step2), len(step1) + 1) + # on Linux, first thread id is supposed to be this process + if LINUX: + self.assertEqual(step2[0].id, os.getpid()) + athread = step2[0] + # test named tuple + self.assertEqual(athread.id, athread[0]) + self.assertEqual(athread.user_time, athread[1]) + self.assertEqual(athread.system_time, athread[2]) + # test num threads + thread.stop() + finally: + if thread._running: + thread.stop() + + def test_memory_info(self): + p = psutil.Process() + + # step 1 - get a base value to compare our results + rss1, vms1 = p.memory_info() + percent1 = p.memory_percent() + self.assertGreater(rss1, 0) + self.assertGreater(vms1, 0) + + # step 2 - allocate some memory + memarr = [None] * 1500000 + + rss2, vms2 = p.memory_info() + percent2 = p.memory_percent() + # make sure that the memory usage bumped up + self.assertGreater(rss2, rss1) + self.assertGreaterEqual(vms2, vms1) # vms might be equal + self.assertGreater(percent2, percent1) + del memarr + + # def test_memory_info_ex(self): + # # tested later in fetch all test suite + + def test_memory_maps(self): + p = psutil.Process() + maps = p.memory_maps() + paths = [x for x in maps] + self.assertEqual(len(paths), len(set(paths))) + ext_maps = p.memory_maps(grouped=False) + + for nt in maps: + if not nt.path.startswith('['): + assert os.path.isabs(nt.path), nt.path + if POSIX: + assert os.path.exists(nt.path), nt.path + else: + # XXX - On Windows we have this strange behavior with + # 64 bit dlls: they are visible via explorer but cannot + # be accessed via os.stat() (wtf?). + if '64' not in os.path.basename(nt.path): + assert os.path.exists(nt.path), nt.path + for nt in ext_maps: + for fname in nt._fields: + value = getattr(nt, fname) + if fname == 'path': + continue + elif fname in ('addr', 'perms'): + assert value, value + else: + self.assertIsInstance(value, (int, long)) + assert value >= 0, value + + def test_memory_percent(self): + p = psutil.Process() + self.assertGreater(p.memory_percent(), 0.0) + + def test_is_running(self): + sproc = get_test_subprocess(wait=True) + p = psutil.Process(sproc.pid) + assert p.is_running() + assert p.is_running() + p.kill() + p.wait() + assert not p.is_running() + assert not p.is_running() + + def test_exe(self): + sproc = get_test_subprocess(wait=True) + exe = psutil.Process(sproc.pid).exe() + try: + self.assertEqual(exe, PYTHON) + except AssertionError: + if WINDOWS and len(exe) == len(PYTHON): + # on Windows we don't care about case sensitivity + self.assertEqual(exe.lower(), PYTHON.lower()) + else: + # certain platforms such as BSD are more accurate returning: + # "/usr/local/bin/python2.7" + # ...instead of: + # "/usr/local/bin/python" + # We do not want to consider this difference in accuracy + # an error. + ver = "%s.%s" % (sys.version_info[0], sys.version_info[1]) + self.assertEqual(exe.replace(ver, ''), PYTHON.replace(ver, '')) + + def test_cmdline(self): + cmdline = [PYTHON, "-c", "import time; time.sleep(60)"] + sproc = get_test_subprocess(cmdline, wait=True) + self.assertEqual(' '.join(psutil.Process(sproc.pid).cmdline()), + ' '.join(cmdline)) + + def test_name(self): + sproc = get_test_subprocess(PYTHON, wait=True) + name = psutil.Process(sproc.pid).name().lower() + pyexe = os.path.basename(os.path.realpath(sys.executable)).lower() + assert pyexe.startswith(name), (pyexe, name) + + @unittest.skipUnless(POSIX, "posix only") + # TODO: add support for other compilers + @unittest.skipUnless(which("gcc"), "gcc not available") + def test_prog_w_funky_name(self): + # Test that name(), exe() and cmdline() correctly handle programs + # with funky chars such as spaces and ")", see: + # https://github.com/giampaolo/psutil/issues/628 + funky_name = "/tmp/foo bar )" + _, c_file = tempfile.mkstemp(prefix='psutil-', suffix='.c', dir="/tmp") + self.addCleanup(lambda: safe_remove(c_file)) + self.addCleanup(lambda: safe_remove(funky_name)) + with open(c_file, "w") as f: + f.write("void main() { pause(); }") + subprocess.check_call(["gcc", c_file, "-o", funky_name]) + sproc = get_test_subprocess( + [funky_name, "arg1", "arg2", "", "arg3", ""]) + p = psutil.Process(sproc.pid) + # ...in order to try to prevent occasional failures on travis + wait_for_pid(p.pid) + self.assertEqual(p.name(), "foo bar )") + self.assertEqual(p.exe(), "/tmp/foo bar )") + self.assertEqual( + p.cmdline(), ["/tmp/foo bar )", "arg1", "arg2", "", "arg3", ""]) + + @unittest.skipUnless(POSIX, 'posix only') + def test_uids(self): + p = psutil.Process() + real, effective, saved = p.uids() + # os.getuid() refers to "real" uid + self.assertEqual(real, os.getuid()) + # os.geteuid() refers to "effective" uid + self.assertEqual(effective, os.geteuid()) + # no such thing as os.getsuid() ("saved" uid), but starting + # from python 2.7 we have os.getresuid()[2] + if hasattr(os, "getresuid"): + self.assertEqual(saved, os.getresuid()[2]) + + @unittest.skipUnless(POSIX, 'posix only') + def test_gids(self): + p = psutil.Process() + real, effective, saved = p.gids() + # os.getuid() refers to "real" uid + self.assertEqual(real, os.getgid()) + # os.geteuid() refers to "effective" uid + self.assertEqual(effective, os.getegid()) + # no such thing as os.getsuid() ("saved" uid), but starting + # from python 2.7 we have os.getresgid()[2] + if hasattr(os, "getresuid"): + self.assertEqual(saved, os.getresgid()[2]) + + def test_nice(self): + p = psutil.Process() + self.assertRaises(TypeError, p.nice, "str") + if WINDOWS: + try: + init = p.nice() + if sys.version_info > (3, 4): + self.assertIsInstance(init, enum.IntEnum) + else: + self.assertIsInstance(init, int) + self.assertEqual(init, psutil.NORMAL_PRIORITY_CLASS) + p.nice(psutil.HIGH_PRIORITY_CLASS) + self.assertEqual(p.nice(), psutil.HIGH_PRIORITY_CLASS) + p.nice(psutil.NORMAL_PRIORITY_CLASS) + self.assertEqual(p.nice(), psutil.NORMAL_PRIORITY_CLASS) + finally: + p.nice(psutil.NORMAL_PRIORITY_CLASS) + else: + try: + first_nice = p.nice() + p.nice(1) + self.assertEqual(p.nice(), 1) + # going back to previous nice value raises + # AccessDenied on OSX + if not OSX: + p.nice(0) + self.assertEqual(p.nice(), 0) + except psutil.AccessDenied: + pass + finally: + try: + p.nice(first_nice) + except psutil.AccessDenied: + pass + + def test_status(self): + p = psutil.Process() + self.assertEqual(p.status(), psutil.STATUS_RUNNING) + + def test_username(self): + sproc = get_test_subprocess() + p = psutil.Process(sproc.pid) + if POSIX: + import pwd + self.assertEqual(p.username(), pwd.getpwuid(os.getuid()).pw_name) + with mock.patch("psutil.pwd.getpwuid", + side_effect=KeyError) as fun: + p.username() == str(p.uids().real) + assert fun.called + + elif WINDOWS and 'USERNAME' in os.environ: + expected_username = os.environ['USERNAME'] + expected_domain = os.environ['USERDOMAIN'] + domain, username = p.username().split('\\') + self.assertEqual(domain, expected_domain) + self.assertEqual(username, expected_username) + else: + p.username() + + def test_cwd(self): + sproc = get_test_subprocess(wait=True) + p = psutil.Process(sproc.pid) + self.assertEqual(p.cwd(), os.getcwd()) + + def test_cwd_2(self): + cmd = [PYTHON, "-c", "import os, time; os.chdir('..'); time.sleep(60)"] + sproc = get_test_subprocess(cmd, wait=True) + p = psutil.Process(sproc.pid) + call_until(p.cwd, "ret == os.path.dirname(os.getcwd())") + + @unittest.skipUnless(WINDOWS or LINUX or BSD, + 'not available on this platform') + @unittest.skipIf(LINUX and TRAVIS, "unknown failure on travis") + def test_cpu_affinity(self): + p = psutil.Process() + initial = p.cpu_affinity() + if hasattr(os, "sched_getaffinity"): + self.assertEqual(initial, list(os.sched_getaffinity(p.pid))) + self.assertEqual(len(initial), len(set(initial))) + all_cpus = list(range(len(psutil.cpu_percent(percpu=True)))) + # setting on travis doesn't seem to work (always return all + # CPUs on get): + # AssertionError: Lists differ: [0, 1, 2, 3, 4, 5, 6, ... != [0] + for n in all_cpus: + p.cpu_affinity([n]) + self.assertEqual(p.cpu_affinity(), [n]) + if hasattr(os, "sched_getaffinity"): + self.assertEqual(p.cpu_affinity(), + list(os.sched_getaffinity(p.pid))) + # + p.cpu_affinity(all_cpus) + self.assertEqual(p.cpu_affinity(), all_cpus) + if hasattr(os, "sched_getaffinity"): + self.assertEqual(p.cpu_affinity(), + list(os.sched_getaffinity(p.pid))) + # + self.assertRaises(TypeError, p.cpu_affinity, 1) + p.cpu_affinity(initial) + # it should work with all iterables, not only lists + p.cpu_affinity(set(all_cpus)) + p.cpu_affinity(tuple(all_cpus)) + invalid_cpu = [len(psutil.cpu_times(percpu=True)) + 10] + self.assertRaises(ValueError, p.cpu_affinity, invalid_cpu) + self.assertRaises(ValueError, p.cpu_affinity, range(10000, 11000)) + self.assertRaises(TypeError, p.cpu_affinity, [0, "1"]) + + # TODO + @unittest.skipIf(BSD, "broken on BSD, see #595") + @unittest.skipIf(APPVEYOR, + "can't find any process file on Appveyor") + def test_open_files(self): + # current process + p = psutil.Process() + files = p.open_files() + self.assertFalse(TESTFN in files) + with open(TESTFN, 'w'): + # give the kernel some time to see the new file + call_until(p.open_files, "len(ret) != %i" % len(files)) + filenames = [x.path for x in p.open_files()] + self.assertIn(TESTFN, filenames) + for file in filenames: + assert os.path.isfile(file), file + + # another process + cmdline = "import time; f = open(r'%s', 'r'); time.sleep(60);" % TESTFN + sproc = get_test_subprocess([PYTHON, "-c", cmdline], wait=True) + p = psutil.Process(sproc.pid) + + for x in range(100): + filenames = [x.path for x in p.open_files()] + if TESTFN in filenames: + break + time.sleep(.01) + else: + self.assertIn(TESTFN, filenames) + for file in filenames: + assert os.path.isfile(file), file + + # TODO + @unittest.skipIf(BSD, "broken on BSD, see #595") + @unittest.skipIf(APPVEYOR, + "can't find any process file on Appveyor") + def test_open_files2(self): + # test fd and path fields + with open(TESTFN, 'w') as fileobj: + p = psutil.Process() + for path, fd in p.open_files(): + if path == fileobj.name or fd == fileobj.fileno(): + break + else: + self.fail("no file found; files=%s" % repr(p.open_files())) + self.assertEqual(path, fileobj.name) + if WINDOWS: + self.assertEqual(fd, -1) + else: + self.assertEqual(fd, fileobj.fileno()) + # test positions + ntuple = p.open_files()[0] + self.assertEqual(ntuple[0], ntuple.path) + self.assertEqual(ntuple[1], ntuple.fd) + # test file is gone + self.assertTrue(fileobj.name not in p.open_files()) + + def compare_proc_sys_cons(self, pid, proc_cons): + from psutil._common import pconn + sys_cons = [] + for c in psutil.net_connections(kind='all'): + if c.pid == pid: + sys_cons.append(pconn(*c[:-1])) + if BSD: + # on BSD all fds are set to -1 + proc_cons = [pconn(*[-1] + list(x[1:])) for x in proc_cons] + self.assertEqual(sorted(proc_cons), sorted(sys_cons)) + + @skip_on_access_denied(only_if=OSX) + def test_connections(self): + def check_conn(proc, conn, family, type, laddr, raddr, status, kinds): + all_kinds = ("all", "inet", "inet4", "inet6", "tcp", "tcp4", + "tcp6", "udp", "udp4", "udp6") + check_connection_ntuple(conn) + self.assertEqual(conn.family, family) + self.assertEqual(conn.type, type) + self.assertEqual(conn.laddr, laddr) + self.assertEqual(conn.raddr, raddr) + self.assertEqual(conn.status, status) + for kind in all_kinds: + cons = proc.connections(kind=kind) + if kind in kinds: + self.assertNotEqual(cons, []) + else: + self.assertEqual(cons, []) + # compare against system-wide connections + # XXX Solaris can't retrieve system-wide UNIX + # sockets. + if not SUNOS: + self.compare_proc_sys_cons(proc.pid, [conn]) + + tcp_template = textwrap.dedent(""" + import socket, time + s = socket.socket($family, socket.SOCK_STREAM) + s.bind(('$addr', 0)) + s.listen(1) + with open('$testfn', 'w') as f: + f.write(str(s.getsockname()[:2])) + time.sleep(60) + """) + + udp_template = textwrap.dedent(""" + import socket, time + s = socket.socket($family, socket.SOCK_DGRAM) + s.bind(('$addr', 0)) + with open('$testfn', 'w') as f: + f.write(str(s.getsockname()[:2])) + time.sleep(60) + """) + + from string import Template + testfile = os.path.basename(TESTFN) + tcp4_template = Template(tcp_template).substitute( + family=int(AF_INET), addr="127.0.0.1", testfn=testfile) + udp4_template = Template(udp_template).substitute( + family=int(AF_INET), addr="127.0.0.1", testfn=testfile) + tcp6_template = Template(tcp_template).substitute( + family=int(AF_INET6), addr="::1", testfn=testfile) + udp6_template = Template(udp_template).substitute( + family=int(AF_INET6), addr="::1", testfn=testfile) + + # launch various subprocess instantiating a socket of various + # families and types to enrich psutil results + tcp4_proc = pyrun(tcp4_template) + tcp4_addr = eval(wait_for_file(testfile)) + udp4_proc = pyrun(udp4_template) + udp4_addr = eval(wait_for_file(testfile)) + if supports_ipv6(): + tcp6_proc = pyrun(tcp6_template) + tcp6_addr = eval(wait_for_file(testfile)) + udp6_proc = pyrun(udp6_template) + udp6_addr = eval(wait_for_file(testfile)) + else: + tcp6_proc = None + udp6_proc = None + tcp6_addr = None + udp6_addr = None + + for p in psutil.Process().children(): + cons = p.connections() + self.assertEqual(len(cons), 1) + for conn in cons: + # TCP v4 + if p.pid == tcp4_proc.pid: + check_conn(p, conn, AF_INET, SOCK_STREAM, tcp4_addr, (), + psutil.CONN_LISTEN, + ("all", "inet", "inet4", "tcp", "tcp4")) + # UDP v4 + elif p.pid == udp4_proc.pid: + check_conn(p, conn, AF_INET, SOCK_DGRAM, udp4_addr, (), + psutil.CONN_NONE, + ("all", "inet", "inet4", "udp", "udp4")) + # TCP v6 + elif p.pid == getattr(tcp6_proc, "pid", None): + check_conn(p, conn, AF_INET6, SOCK_STREAM, tcp6_addr, (), + psutil.CONN_LISTEN, + ("all", "inet", "inet6", "tcp", "tcp6")) + # UDP v6 + elif p.pid == getattr(udp6_proc, "pid", None): + check_conn(p, conn, AF_INET6, SOCK_DGRAM, udp6_addr, (), + psutil.CONN_NONE, + ("all", "inet", "inet6", "udp", "udp6")) + + @unittest.skipUnless(hasattr(socket, 'AF_UNIX'), + 'AF_UNIX is not supported') + @skip_on_access_denied(only_if=OSX) + def test_connections_unix(self): + def check(type): + safe_remove(TESTFN) + sock = socket.socket(AF_UNIX, type) + with contextlib.closing(sock): + sock.bind(TESTFN) + cons = psutil.Process().connections(kind='unix') + conn = cons[0] + check_connection_ntuple(conn) + if conn.fd != -1: # != sunos and windows + self.assertEqual(conn.fd, sock.fileno()) + self.assertEqual(conn.family, AF_UNIX) + self.assertEqual(conn.type, type) + self.assertEqual(conn.laddr, TESTFN) + if not SUNOS: + # XXX Solaris can't retrieve system-wide UNIX + # sockets. + self.compare_proc_sys_cons(os.getpid(), cons) + + check(SOCK_STREAM) + check(SOCK_DGRAM) + + @unittest.skipUnless(hasattr(socket, "fromfd"), + 'socket.fromfd() is not availble') + @unittest.skipIf(WINDOWS or SUNOS, + 'connection fd not available on this platform') + def test_connection_fromfd(self): + with contextlib.closing(socket.socket()) as sock: + sock.bind(('localhost', 0)) + sock.listen(1) + p = psutil.Process() + for conn in p.connections(): + if conn.fd == sock.fileno(): + break + else: + self.fail("couldn't find socket fd") + dupsock = socket.fromfd(conn.fd, conn.family, conn.type) + with contextlib.closing(dupsock): + self.assertEqual(dupsock.getsockname(), conn.laddr) + self.assertNotEqual(sock.fileno(), dupsock.fileno()) + + def test_connection_constants(self): + ints = [] + strs = [] + for name in dir(psutil): + if name.startswith('CONN_'): + num = getattr(psutil, name) + str_ = str(num) + assert str_.isupper(), str_ + assert str_ not in strs, str_ + assert num not in ints, num + ints.append(num) + strs.append(str_) + if SUNOS: + psutil.CONN_IDLE + psutil.CONN_BOUND + if WINDOWS: + psutil.CONN_DELETE_TCB + + @unittest.skipUnless(POSIX, 'posix only') + def test_num_fds(self): + p = psutil.Process() + start = p.num_fds() + file = open(TESTFN, 'w') + self.addCleanup(file.close) + self.assertEqual(p.num_fds(), start + 1) + sock = socket.socket() + self.addCleanup(sock.close) + self.assertEqual(p.num_fds(), start + 2) + file.close() + sock.close() + self.assertEqual(p.num_fds(), start) + + @skip_on_not_implemented(only_if=LINUX) + def test_num_ctx_switches(self): + p = psutil.Process() + before = sum(p.num_ctx_switches()) + for x in range(500000): + after = sum(p.num_ctx_switches()) + if after > before: + return + self.fail("num ctx switches still the same after 50.000 iterations") + + def test_parent_ppid(self): + this_parent = os.getpid() + sproc = get_test_subprocess() + p = psutil.Process(sproc.pid) + self.assertEqual(p.ppid(), this_parent) + self.assertEqual(p.parent().pid, this_parent) + # no other process is supposed to have us as parent + for p in psutil.process_iter(): + if p.pid == sproc.pid: + continue + self.assertTrue(p.ppid() != this_parent) + + def test_children(self): + p = psutil.Process() + self.assertEqual(p.children(), []) + self.assertEqual(p.children(recursive=True), []) + sproc = get_test_subprocess() + children1 = p.children() + children2 = p.children(recursive=True) + for children in (children1, children2): + self.assertEqual(len(children), 1) + self.assertEqual(children[0].pid, sproc.pid) + self.assertEqual(children[0].ppid(), os.getpid()) + + def test_children_recursive(self): + # here we create a subprocess which creates another one as in: + # A (parent) -> B (child) -> C (grandchild) + s = "import subprocess, os, sys, time;" + s += "PYTHON = os.path.realpath(sys.executable);" + s += "cmd = [PYTHON, '-c', 'import time; time.sleep(60);'];" + s += "subprocess.Popen(cmd);" + s += "time.sleep(60);" + get_test_subprocess(cmd=[PYTHON, "-c", s]) + p = psutil.Process() + self.assertEqual(len(p.children(recursive=False)), 1) + # give the grandchild some time to start + stop_at = time.time() + GLOBAL_TIMEOUT + while time.time() < stop_at: + children = p.children(recursive=True) + if len(children) > 1: + break + self.assertEqual(len(children), 2) + self.assertEqual(children[0].ppid(), os.getpid()) + self.assertEqual(children[1].ppid(), children[0].pid) + + def test_children_duplicates(self): + # find the process which has the highest number of children + table = collections.defaultdict(int) + for p in psutil.process_iter(): + try: + table[p.ppid()] += 1 + except psutil.Error: + pass + # this is the one, now let's make sure there are no duplicates + pid = sorted(table.items(), key=lambda x: x[1])[-1][0] + p = psutil.Process(pid) + try: + c = p.children(recursive=True) + except psutil.AccessDenied: # windows + pass + else: + self.assertEqual(len(c), len(set(c))) + + def test_suspend_resume(self): + sproc = get_test_subprocess(wait=True) + p = psutil.Process(sproc.pid) + p.suspend() + for x in range(100): + if p.status() == psutil.STATUS_STOPPED: + break + time.sleep(0.01) + p.resume() + self.assertNotEqual(p.status(), psutil.STATUS_STOPPED) + + def test_invalid_pid(self): + self.assertRaises(TypeError, psutil.Process, "1") + self.assertRaises(ValueError, psutil.Process, -1) + + def test_as_dict(self): + p = psutil.Process() + d = p.as_dict(attrs=['exe', 'name']) + self.assertEqual(sorted(d.keys()), ['exe', 'name']) + + p = psutil.Process(min(psutil.pids())) + d = p.as_dict(attrs=['connections'], ad_value='foo') + if not isinstance(d['connections'], list): + self.assertEqual(d['connections'], 'foo') + + def test_halfway_terminated_process(self): + # Test that NoSuchProcess exception gets raised in case the + # process dies after we create the Process object. + # Example: + # >>> proc = Process(1234) + # >>> time.sleep(2) # time-consuming task, process dies in meantime + # >>> proc.name() + # Refers to Issue #15 + sproc = get_test_subprocess() + p = psutil.Process(sproc.pid) + p.terminate() + p.wait() + if WINDOWS: + wait_for_pid(p.pid) + self.assertFalse(p.is_running()) + self.assertFalse(p.pid in psutil.pids()) + + excluded_names = ['pid', 'is_running', 'wait', 'create_time'] + if LINUX and not RLIMIT_SUPPORT: + excluded_names.append('rlimit') + for name in dir(p): + if (name.startswith('_') or + name in excluded_names): + continue + try: + meth = getattr(p, name) + # get/set methods + if name == 'nice': + if POSIX: + ret = meth(1) + else: + ret = meth(psutil.NORMAL_PRIORITY_CLASS) + elif name == 'ionice': + ret = meth() + ret = meth(2) + elif name == 'rlimit': + ret = meth(psutil.RLIMIT_NOFILE) + ret = meth(psutil.RLIMIT_NOFILE, (5, 5)) + elif name == 'cpu_affinity': + ret = meth() + ret = meth([0]) + elif name == 'send_signal': + ret = meth(signal.SIGTERM) + else: + ret = meth() + except psutil.ZombieProcess: + self.fail("ZombieProcess for %r was not supposed to happen" % + name) + except psutil.NoSuchProcess: + pass + except NotImplementedError: + pass + else: + self.fail( + "NoSuchProcess exception not raised for %r, retval=%s" % ( + name, ret)) + + @unittest.skipUnless(POSIX, 'posix only') + def test_zombie_process(self): + def succeed_or_zombie_p_exc(fun, *args, **kwargs): + try: + fun(*args, **kwargs) + except (psutil.ZombieProcess, psutil.AccessDenied): + pass + + # Note: in this test we'll be creating two sub processes. + # Both of them are supposed to be freed / killed by + # reap_children() as they are attributable to 'us' + # (os.getpid()) via children(recursive=True). + src = textwrap.dedent("""\ + import os, sys, time, socket, contextlib + child_pid = os.fork() + if child_pid > 0: + time.sleep(3000) + else: + # this is the zombie process + s = socket.socket(socket.AF_UNIX) + with contextlib.closing(s): + s.connect('%s') + if sys.version_info < (3, ): + pid = str(os.getpid()) + else: + pid = bytes(str(os.getpid()), 'ascii') + s.sendall(pid) + """ % TESTFN) + with contextlib.closing(socket.socket(socket.AF_UNIX)) as sock: + try: + sock.settimeout(GLOBAL_TIMEOUT) + sock.bind(TESTFN) + sock.listen(1) + pyrun(src) + conn, _ = sock.accept() + select.select([conn.fileno()], [], [], GLOBAL_TIMEOUT) + zpid = int(conn.recv(1024)) + zproc = psutil.Process(zpid) + call_until(lambda: zproc.status(), + "ret == psutil.STATUS_ZOMBIE") + # A zombie process should always be instantiable + zproc = psutil.Process(zpid) + # ...and at least its status always be querable + self.assertEqual(zproc.status(), psutil.STATUS_ZOMBIE) + # ...and it should be considered 'running' + self.assertTrue(zproc.is_running()) + # ...and as_dict() shouldn't crash + zproc.as_dict() + if hasattr(zproc, "rlimit"): + succeed_or_zombie_p_exc(zproc.rlimit, psutil.RLIMIT_NOFILE) + succeed_or_zombie_p_exc(zproc.rlimit, psutil.RLIMIT_NOFILE, + (5, 5)) + # set methods + succeed_or_zombie_p_exc(zproc.parent) + if hasattr(zproc, 'cpu_affinity'): + succeed_or_zombie_p_exc(zproc.cpu_affinity, [0]) + succeed_or_zombie_p_exc(zproc.nice, 0) + if hasattr(zproc, 'ionice'): + if LINUX: + succeed_or_zombie_p_exc(zproc.ionice, 2, 0) + else: + succeed_or_zombie_p_exc(zproc.ionice, 0) # Windows + if hasattr(zproc, 'rlimit'): + succeed_or_zombie_p_exc(zproc.rlimit, + psutil.RLIMIT_NOFILE, (5, 5)) + succeed_or_zombie_p_exc(zproc.suspend) + succeed_or_zombie_p_exc(zproc.resume) + succeed_or_zombie_p_exc(zproc.terminate) + succeed_or_zombie_p_exc(zproc.kill) + + # ...its parent should 'see' it + # edit: not true on BSD and OSX + # descendants = [x.pid for x in psutil.Process().children( + # recursive=True)] + # self.assertIn(zpid, descendants) + # XXX should we also assume ppid be usable? Note: this + # would be an important use case as the only way to get + # rid of a zombie is to kill its parent. + # self.assertEqual(zpid.ppid(), os.getpid()) + # ...and all other APIs should be able to deal with it + self.assertTrue(psutil.pid_exists(zpid)) + self.assertIn(zpid, psutil.pids()) + self.assertIn(zpid, [x.pid for x in psutil.process_iter()]) + psutil._pmap = {} + self.assertIn(zpid, [x.pid for x in psutil.process_iter()]) + finally: + reap_children(search_all=True) + + def test_pid_0(self): + # Process(0) is supposed to work on all platforms except Linux + if 0 not in psutil.pids(): + self.assertRaises(psutil.NoSuchProcess, psutil.Process, 0) + return + + p = psutil.Process(0) + self.assertTrue(p.name()) + + if POSIX: + try: + self.assertEqual(p.uids().real, 0) + self.assertEqual(p.gids().real, 0) + except psutil.AccessDenied: + pass + + self.assertRaisesRegexp( + ValueError, "preventing sending signal to process with PID 0", + p.send_signal, signal.SIGTERM) + + self.assertIn(p.ppid(), (0, 1)) + # self.assertEqual(p.exe(), "") + p.cmdline() + try: + p.num_threads() + except psutil.AccessDenied: + pass + + try: + p.memory_info() + except psutil.AccessDenied: + pass + + try: + if POSIX: + self.assertEqual(p.username(), 'root') + elif WINDOWS: + self.assertEqual(p.username(), 'NT AUTHORITY\\SYSTEM') + else: + p.username() + except psutil.AccessDenied: + pass + + self.assertIn(0, psutil.pids()) + self.assertTrue(psutil.pid_exists(0)) + + def test_Popen(self): + # Popen class test + # XXX this test causes a ResourceWarning on Python 3 because + # psutil.__subproc instance doesn't get propertly freed. + # Not sure what to do though. + cmd = [PYTHON, "-c", "import time; time.sleep(60);"] + proc = psutil.Popen(cmd, stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + try: + proc.name() + proc.stdin + self.assertTrue(hasattr(proc, 'name')) + self.assertTrue(hasattr(proc, 'stdin')) + self.assertTrue(dir(proc)) + self.assertRaises(AttributeError, getattr, proc, 'foo') + finally: + proc.kill() + proc.wait() + self.assertIsNotNone(proc.returncode) + + +# =================================================================== +# --- Featch all processes test +# =================================================================== + +class TestFetchAllProcesses(unittest.TestCase): + """Test which iterates over all running processes and performs + some sanity checks against Process API's returned values. + """ + + def setUp(self): + if POSIX: + import pwd + pall = pwd.getpwall() + self._uids = set([x.pw_uid for x in pall]) + self._usernames = set([x.pw_name for x in pall]) + + def test_fetch_all(self): + valid_procs = 0 + excluded_names = set([ + 'send_signal', 'suspend', 'resume', 'terminate', 'kill', 'wait', + 'as_dict', 'cpu_percent', 'parent', 'children', 'pid']) + if LINUX and not RLIMIT_SUPPORT: + excluded_names.add('rlimit') + attrs = [] + for name in dir(psutil.Process): + if name.startswith("_"): + continue + if name in excluded_names: + continue + attrs.append(name) + + default = object() + failures = [] + for name in attrs: + for p in psutil.process_iter(): + ret = default + try: + try: + args = () + attr = getattr(p, name, None) + if attr is not None and callable(attr): + if name == 'rlimit': + args = (psutil.RLIMIT_NOFILE,) + ret = attr(*args) + else: + ret = attr + valid_procs += 1 + except NotImplementedError: + msg = "%r was skipped because not implemented" % ( + self.__class__.__name__ + '.test_' + name) + warn(msg) + except (psutil.NoSuchProcess, psutil.AccessDenied) as err: + self.assertEqual(err.pid, p.pid) + if err.name: + # make sure exception's name attr is set + # with the actual process name + self.assertEqual(err.name, p.name()) + self.assertTrue(str(err)) + self.assertTrue(err.msg) + else: + if ret not in (0, 0.0, [], None, ''): + assert ret, ret + meth = getattr(self, name) + meth(ret) + except Exception as err: + s = '\n' + '=' * 70 + '\n' + s += "FAIL: test_%s (proc=%s" % (name, p) + if ret != default: + s += ", ret=%s)" % repr(ret) + s += ')\n' + s += '-' * 70 + s += "\n%s" % traceback.format_exc() + s = "\n".join((" " * 4) + i for i in s.splitlines()) + failures.append(s) + break + + if failures: + self.fail(''.join(failures)) + + # we should always have a non-empty list, not including PID 0 etc. + # special cases. + self.assertTrue(valid_procs > 0) + + def cmdline(self, ret): + pass + + def exe(self, ret): + if not ret: + self.assertEqual(ret, '') + else: + assert os.path.isabs(ret), ret + # Note: os.stat() may return False even if the file is there + # hence we skip the test, see: + # http://stackoverflow.com/questions/3112546/os-path-exists-lies + if POSIX and os.path.isfile(ret): + if hasattr(os, 'access') and hasattr(os, "X_OK"): + # XXX may fail on OSX + self.assertTrue(os.access(ret, os.X_OK)) + + def ppid(self, ret): + self.assertTrue(ret >= 0) + + def name(self, ret): + self.assertIsInstance(ret, (str, unicode)) + self.assertTrue(ret) + + def create_time(self, ret): + self.assertTrue(ret > 0) + # this can't be taken for granted on all platforms + # self.assertGreaterEqual(ret, psutil.boot_time()) + # make sure returned value can be pretty printed + # with strftime + time.strftime("%Y %m %d %H:%M:%S", time.localtime(ret)) + + def uids(self, ret): + for uid in ret: + self.assertTrue(uid >= 0) + self.assertIn(uid, self._uids) + + def gids(self, ret): + # note: testing all gids as above seems not to be reliable for + # gid == 30 (nodoby); not sure why. + for gid in ret: + self.assertTrue(gid >= 0) + # self.assertIn(uid, self.gids + + def username(self, ret): + self.assertTrue(ret) + if POSIX: + self.assertIn(ret, self._usernames) + + def status(self, ret): + self.assertTrue(ret != "") + self.assertTrue(ret != '?') + self.assertIn(ret, VALID_PROC_STATUSES) + + def io_counters(self, ret): + for field in ret: + if field != -1: + self.assertTrue(field >= 0) + + def ionice(self, ret): + if LINUX: + self.assertTrue(ret.ioclass >= 0) + self.assertTrue(ret.value >= 0) + else: + self.assertTrue(ret >= 0) + self.assertIn(ret, (0, 1, 2)) + + def num_threads(self, ret): + self.assertTrue(ret >= 1) + + def threads(self, ret): + for t in ret: + self.assertTrue(t.id >= 0) + self.assertTrue(t.user_time >= 0) + self.assertTrue(t.system_time >= 0) + + def cpu_times(self, ret): + self.assertTrue(ret.user >= 0) + self.assertTrue(ret.system >= 0) + + def memory_info(self, ret): + self.assertTrue(ret.rss >= 0) + self.assertTrue(ret.vms >= 0) + + def memory_info_ex(self, ret): + for name in ret._fields: + self.assertTrue(getattr(ret, name) >= 0) + if POSIX and ret.vms != 0: + # VMS is always supposed to be the highest + for name in ret._fields: + if name != 'vms': + value = getattr(ret, name) + assert ret.vms > value, ret + elif WINDOWS: + assert ret.peak_wset >= ret.wset, ret + assert ret.peak_paged_pool >= ret.paged_pool, ret + assert ret.peak_nonpaged_pool >= ret.nonpaged_pool, ret + assert ret.peak_pagefile >= ret.pagefile, ret + + def open_files(self, ret): + for f in ret: + if WINDOWS: + assert f.fd == -1, f + else: + self.assertIsInstance(f.fd, int) + assert os.path.isabs(f.path), f + assert os.path.isfile(f.path), f + + def num_fds(self, ret): + self.assertTrue(ret >= 0) + + def connections(self, ret): + self.assertEqual(len(ret), len(set(ret))) + for conn in ret: + check_connection_ntuple(conn) + + def cwd(self, ret): + if ret is not None: # BSD may return None + assert os.path.isabs(ret), ret + try: + st = os.stat(ret) + except OSError as err: + # directory has been removed in mean time + if err.errno != errno.ENOENT: + raise + else: + self.assertTrue(stat.S_ISDIR(st.st_mode)) + + def memory_percent(self, ret): + assert 0 <= ret <= 100, ret + + def is_running(self, ret): + self.assertTrue(ret) + + def cpu_affinity(self, ret): + assert ret != [], ret + + def terminal(self, ret): + if ret is not None: + assert os.path.isabs(ret), ret + assert os.path.exists(ret), ret + + def memory_maps(self, ret): + for nt in ret: + for fname in nt._fields: + value = getattr(nt, fname) + if fname == 'path': + if not value.startswith('['): + assert os.path.isabs(nt.path), nt.path + # commented as on Linux we might get + # '/foo/bar (deleted)' + # assert os.path.exists(nt.path), nt.path + elif fname in ('addr', 'perms'): + self.assertTrue(value) + else: + self.assertIsInstance(value, (int, long)) + assert value >= 0, value + + def num_handles(self, ret): + if WINDOWS: + self.assertGreaterEqual(ret, 0) + else: + self.assertGreaterEqual(ret, 0) + + def nice(self, ret): + if POSIX: + assert -20 <= ret <= 20, ret + else: + priorities = [getattr(psutil, x) for x in dir(psutil) + if x.endswith('_PRIORITY_CLASS')] + self.assertIn(ret, priorities) + + def num_ctx_switches(self, ret): + self.assertTrue(ret.voluntary >= 0) + self.assertTrue(ret.involuntary >= 0) + + def rlimit(self, ret): + self.assertEqual(len(ret), 2) + self.assertGreaterEqual(ret[0], -1) + self.assertGreaterEqual(ret[1], -1) + + +# =================================================================== +# --- Limited user tests +# =================================================================== + +@unittest.skipUnless(POSIX, "UNIX only") +@unittest.skipUnless(hasattr(os, 'getuid') and os.getuid() == 0, + "super user privileges are required") +class LimitedUserTestCase(TestProcess): + """Repeat the previous tests by using a limited user. + Executed only on UNIX and only if the user who run the test script + is root. + """ + # the uid/gid the test suite runs under + if hasattr(os, 'getuid'): + PROCESS_UID = os.getuid() + PROCESS_GID = os.getgid() + + def __init__(self, *args, **kwargs): + TestProcess.__init__(self, *args, **kwargs) + # re-define all existent test methods in order to + # ignore AccessDenied exceptions + for attr in [x for x in dir(self) if x.startswith('test')]: + meth = getattr(self, attr) + + def test_(self): + try: + meth() + except psutil.AccessDenied: + pass + setattr(self, attr, types.MethodType(test_, self)) + + def setUp(self): + safe_remove(TESTFN) + TestProcess.setUp(self) + os.setegid(1000) + os.seteuid(1000) + + def tearDown(self): + os.setegid(self.PROCESS_UID) + os.seteuid(self.PROCESS_GID) + TestProcess.tearDown(self) + + def test_nice(self): + try: + psutil.Process().nice(-1) + except psutil.AccessDenied: + pass + else: + self.fail("exception not raised") + + def test_zombie_process(self): + # causes problems if test test suite is run as root + pass + + +# =================================================================== +# --- Misc tests +# =================================================================== + +class TestMisc(unittest.TestCase): + """Misc / generic tests.""" + + def test_process__repr__(self, func=repr): + p = psutil.Process() + r = func(p) + self.assertIn("psutil.Process", r) + self.assertIn("pid=%s" % p.pid, r) + self.assertIn("name=", r) + self.assertIn(p.name(), r) + with mock.patch.object(psutil.Process, "name", + side_effect=psutil.ZombieProcess(os.getpid())): + p = psutil.Process() + r = func(p) + self.assertIn("pid=%s" % p.pid, r) + self.assertIn("zombie", r) + self.assertNotIn("name=", r) + with mock.patch.object(psutil.Process, "name", + side_effect=psutil.NoSuchProcess(os.getpid())): + p = psutil.Process() + r = func(p) + self.assertIn("pid=%s" % p.pid, r) + self.assertIn("terminated", r) + self.assertNotIn("name=", r) + + def test_process__str__(self): + self.test_process__repr__(func=str) + + def test_no_such_process__repr__(self, func=repr): + self.assertEqual( + repr(psutil.NoSuchProcess(321)), + "psutil.NoSuchProcess process no longer exists (pid=321)") + self.assertEqual( + repr(psutil.NoSuchProcess(321, name='foo')), + "psutil.NoSuchProcess process no longer exists (pid=321, " + "name='foo')") + self.assertEqual( + repr(psutil.NoSuchProcess(321, msg='foo')), + "psutil.NoSuchProcess foo") + + def test_zombie_process__repr__(self, func=repr): + self.assertEqual( + repr(psutil.ZombieProcess(321)), + "psutil.ZombieProcess process still exists but it's a zombie " + "(pid=321)") + self.assertEqual( + repr(psutil.ZombieProcess(321, name='foo')), + "psutil.ZombieProcess process still exists but it's a zombie " + "(pid=321, name='foo')") + self.assertEqual( + repr(psutil.ZombieProcess(321, name='foo', ppid=1)), + "psutil.ZombieProcess process still exists but it's a zombie " + "(pid=321, name='foo', ppid=1)") + self.assertEqual( + repr(psutil.ZombieProcess(321, msg='foo')), + "psutil.ZombieProcess foo") + + def test_access_denied__repr__(self, func=repr): + self.assertEqual( + repr(psutil.AccessDenied(321)), + "psutil.AccessDenied (pid=321)") + self.assertEqual( + repr(psutil.AccessDenied(321, name='foo')), + "psutil.AccessDenied (pid=321, name='foo')") + self.assertEqual( + repr(psutil.AccessDenied(321, msg='foo')), + "psutil.AccessDenied foo") + + def test_timeout_expired__repr__(self, func=repr): + self.assertEqual( + repr(psutil.TimeoutExpired(321)), + "psutil.TimeoutExpired timeout after 321 seconds") + self.assertEqual( + repr(psutil.TimeoutExpired(321, pid=111)), + "psutil.TimeoutExpired timeout after 321 seconds (pid=111)") + self.assertEqual( + repr(psutil.TimeoutExpired(321, pid=111, name='foo')), + "psutil.TimeoutExpired timeout after 321 seconds " + "(pid=111, name='foo')") + + def test_process__eq__(self): + p1 = psutil.Process() + p2 = psutil.Process() + self.assertEqual(p1, p2) + p2._ident = (0, 0) + self.assertNotEqual(p1, p2) + self.assertNotEqual(p1, 'foo') + + def test_process__hash__(self): + s = set([psutil.Process(), psutil.Process()]) + self.assertEqual(len(s), 1) + + def test__all__(self): + dir_psutil = dir(psutil) + for name in dir_psutil: + if name in ('callable', 'error', 'namedtuple', + 'long', 'test', 'NUM_CPUS', 'BOOT_TIME', + 'TOTAL_PHYMEM'): + continue + if not name.startswith('_'): + try: + __import__(name) + except ImportError: + if name not in psutil.__all__: + fun = getattr(psutil, name) + if fun is None: + continue + if (fun.__doc__ is not None and + 'deprecated' not in fun.__doc__.lower()): + self.fail('%r not in psutil.__all__' % name) + + # Import 'star' will break if __all__ is inconsistent, see: + # https://github.com/giampaolo/psutil/issues/656 + # Can't do `from psutil import *` as it won't work on python 3 + # so we simply iterate over __all__. + for name in psutil.__all__: + self.assertIn(name, dir_psutil) + + def test_version(self): + self.assertEqual('.'.join([str(x) for x in psutil.version_info]), + psutil.__version__) + + def test_memoize(self): + from psutil._common import memoize + + @memoize + def foo(*args, **kwargs): + "foo docstring" + calls.append(None) + return (args, kwargs) + + calls = [] + # no args + for x in range(2): + ret = foo() + expected = ((), {}) + self.assertEqual(ret, expected) + self.assertEqual(len(calls), 1) + # with args + for x in range(2): + ret = foo(1) + expected = ((1, ), {}) + self.assertEqual(ret, expected) + self.assertEqual(len(calls), 2) + # with args + kwargs + for x in range(2): + ret = foo(1, bar=2) + expected = ((1, ), {'bar': 2}) + self.assertEqual(ret, expected) + self.assertEqual(len(calls), 3) + # clear cache + foo.cache_clear() + ret = foo() + expected = ((), {}) + self.assertEqual(ret, expected) + self.assertEqual(len(calls), 4) + # docstring + self.assertEqual(foo.__doc__, "foo docstring") + + def test_isfile_strict(self): + from psutil._common import isfile_strict + this_file = os.path.abspath(__file__) + assert isfile_strict(this_file) + assert not isfile_strict(os.path.dirname(this_file)) + with mock.patch('psutil._common.os.stat', + side_effect=OSError(errno.EPERM, "foo")): + self.assertRaises(OSError, isfile_strict, this_file) + with mock.patch('psutil._common.os.stat', + side_effect=OSError(errno.EACCES, "foo")): + self.assertRaises(OSError, isfile_strict, this_file) + with mock.patch('psutil._common.os.stat', + side_effect=OSError(errno.EINVAL, "foo")): + assert not isfile_strict(this_file) + with mock.patch('psutil._common.stat.S_ISREG', return_value=False): + assert not isfile_strict(this_file) + + def test_serialization(self): + def check(ret): + if json is not None: + json.loads(json.dumps(ret)) + a = pickle.dumps(ret) + b = pickle.loads(a) + self.assertEqual(ret, b) + + check(psutil.Process().as_dict()) + check(psutil.virtual_memory()) + check(psutil.swap_memory()) + check(psutil.cpu_times()) + check(psutil.cpu_times_percent(interval=0)) + check(psutil.net_io_counters()) + if LINUX and not os.path.exists('/proc/diskstats'): + pass + else: + if not APPVEYOR: + check(psutil.disk_io_counters()) + check(psutil.disk_partitions()) + check(psutil.disk_usage(os.getcwd())) + check(psutil.users()) + + def test_setup_script(self): + here = os.path.abspath(os.path.dirname(__file__)) + setup_py = os.path.realpath(os.path.join(here, '..', 'setup.py')) + module = imp.load_source('setup', setup_py) + self.assertRaises(SystemExit, module.setup) + self.assertEqual(module.get_version(), psutil.__version__) + + def test_ad_on_process_creation(self): + # We are supposed to be able to instantiate Process also in case + # of zombie processes or access denied. + with mock.patch.object(psutil.Process, 'create_time', + side_effect=psutil.AccessDenied) as meth: + psutil.Process() + assert meth.called + with mock.patch.object(psutil.Process, 'create_time', + side_effect=psutil.ZombieProcess(1)) as meth: + psutil.Process() + assert meth.called + with mock.patch.object(psutil.Process, 'create_time', + side_effect=ValueError) as meth: + with self.assertRaises(ValueError): + psutil.Process() + assert meth.called + + +# =================================================================== +# --- Example script tests +# =================================================================== + +class TestExampleScripts(unittest.TestCase): + """Tests for scripts in the examples directory.""" + + def assert_stdout(self, exe, args=None): + exe = os.path.join(EXAMPLES_DIR, exe) + if args: + exe = exe + ' ' + args + try: + out = sh(sys.executable + ' ' + exe).strip() + except RuntimeError as err: + if 'AccessDenied' in str(err): + return str(err) + else: + raise + assert out, out + return out + + def assert_syntax(self, exe, args=None): + exe = os.path.join(EXAMPLES_DIR, exe) + with open(exe, 'r') as f: + src = f.read() + ast.parse(src) + + def test_check_presence(self): + # make sure all example scripts have a test method defined + meths = dir(self) + for name in os.listdir(EXAMPLES_DIR): + if name.endswith('.py'): + if 'test_' + os.path.splitext(name)[0] not in meths: + # self.assert_stdout(name) + self.fail('no test defined for %r script' + % os.path.join(EXAMPLES_DIR, name)) + + def test_disk_usage(self): + self.assert_stdout('disk_usage.py') + + def test_free(self): + self.assert_stdout('free.py') + + def test_meminfo(self): + self.assert_stdout('meminfo.py') + + def test_process_detail(self): + self.assert_stdout('process_detail.py') + + @unittest.skipIf(APPVEYOR, "can't find users on Appveyor") + def test_who(self): + self.assert_stdout('who.py') + + def test_ps(self): + self.assert_stdout('ps.py') + + def test_pstree(self): + self.assert_stdout('pstree.py') + + def test_netstat(self): + self.assert_stdout('netstat.py') + + @unittest.skipIf(TRAVIS, "permission denied on travis") + def test_ifconfig(self): + self.assert_stdout('ifconfig.py') + + def test_pmap(self): + self.assert_stdout('pmap.py', args=str(os.getpid())) + + @unittest.skipIf(ast is None, + 'ast module not available on this python version') + def test_killall(self): + self.assert_syntax('killall.py') + + @unittest.skipIf(ast is None, + 'ast module not available on this python version') + def test_nettop(self): + self.assert_syntax('nettop.py') + + @unittest.skipIf(ast is None, + 'ast module not available on this python version') + def test_top(self): + self.assert_syntax('top.py') + + @unittest.skipIf(ast is None, + 'ast module not available on this python version') + def test_iotop(self): + self.assert_syntax('iotop.py') + + def test_pidof(self): + output = self.assert_stdout('pidof.py %s' % psutil.Process().name()) + self.assertIn(str(os.getpid()), output) + + +def main(): + tests = [] + test_suite = unittest.TestSuite() + tests.append(TestSystemAPIs) + tests.append(TestProcess) + tests.append(TestFetchAllProcesses) + tests.append(TestMisc) + tests.append(TestExampleScripts) + tests.append(LimitedUserTestCase) + + if POSIX: + from _posix import PosixSpecificTestCase + tests.append(PosixSpecificTestCase) + + # import the specific platform test suite + stc = None + if LINUX: + from _linux import LinuxSpecificTestCase as stc + elif WINDOWS: + from _windows import WindowsSpecificTestCase as stc + from _windows import TestDualProcessImplementation + tests.append(TestDualProcessImplementation) + elif OSX: + from _osx import OSXSpecificTestCase as stc + elif BSD: + from _bsd import BSDSpecificTestCase as stc + elif SUNOS: + from _sunos import SunOSSpecificTestCase as stc + if stc is not None: + tests.append(stc) + + for test_class in tests: + test_suite.addTest(unittest.makeSuite(test_class)) + result = unittest.TextTestRunner(verbosity=2).run(test_suite) + return result.wasSuccessful() + +if __name__ == '__main__': + if not main(): + sys.exit(1) |