summaryrefslogtreecommitdiffstats
path: root/python/psutil/test/_posix.py
diff options
context:
space:
mode:
authorMatt A. Tobin <mattatobin@localhost.localdomain>2018-02-02 04:16:08 -0500
committerMatt A. Tobin <mattatobin@localhost.localdomain>2018-02-02 04:16:08 -0500
commit5f8de423f190bbb79a62f804151bc24824fa32d8 (patch)
tree10027f336435511475e392454359edea8e25895d /python/psutil/test/_posix.py
parent49ee0794b5d912db1f95dce6eb52d781dc210db5 (diff)
downloadUXP-5f8de423f190bbb79a62f804151bc24824fa32d8.tar
UXP-5f8de423f190bbb79a62f804151bc24824fa32d8.tar.gz
UXP-5f8de423f190bbb79a62f804151bc24824fa32d8.tar.lz
UXP-5f8de423f190bbb79a62f804151bc24824fa32d8.tar.xz
UXP-5f8de423f190bbb79a62f804151bc24824fa32d8.zip
Add m-esr52 at 52.6.0
Diffstat (limited to 'python/psutil/test/_posix.py')
-rw-r--r--python/psutil/test/_posix.py258
1 files changed, 258 insertions, 0 deletions
diff --git a/python/psutil/test/_posix.py b/python/psutil/test/_posix.py
new file mode 100644
index 000000000..e6c56aac3
--- /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)