diff options
Diffstat (limited to 'psutil/tests/test_process.py')
| -rwxr-xr-x | psutil/tests/test_process.py | 149 |
1 files changed, 95 insertions, 54 deletions
diff --git a/psutil/tests/test_process.py b/psutil/tests/test_process.py index 8b88f766..3b60d38a 100755 --- a/psutil/tests/test_process.py +++ b/psutil/tests/test_process.py @@ -9,6 +9,7 @@ import collections import errno import getpass +import itertools import os import signal import socket @@ -22,6 +23,7 @@ import types import psutil +from psutil import AIX from psutil import BSD from psutil import LINUX from psutil import NETBSD @@ -49,9 +51,10 @@ from psutil.tests import HAS_MEMORY_MAPS from psutil.tests import HAS_PROC_CPU_NUM from psutil.tests import HAS_PROC_IO_COUNTERS from psutil.tests import HAS_RLIMIT +from psutil.tests import HAS_THREADS from psutil.tests import mock from psutil.tests import PYPY -from psutil.tests import PYTHON +from psutil.tests import PYTHON_EXE from psutil.tests import reap_children from psutil.tests import retry_before_failing from psutil.tests import run_test_module_by_name @@ -62,7 +65,6 @@ from psutil.tests import skip_on_not_implemented from psutil.tests import TESTFILE_PREFIX from psutil.tests import TESTFN from psutil.tests import ThreadTask -from psutil.tests import TOX from psutil.tests import TRAVIS from psutil.tests import unittest from psutil.tests import wait_for_pid @@ -151,7 +153,7 @@ class TestProcess(unittest.TestCase): if POSIX: self.assertEqual(code, -signal.SIGKILL) else: - self.assertEqual(code, 0) + self.assertEqual(code, signal.SIGTERM) self.assertFalse(p.is_running()) sproc = get_test_subprocess() @@ -161,12 +163,12 @@ class TestProcess(unittest.TestCase): if POSIX: self.assertEqual(code, -signal.SIGTERM) else: - self.assertEqual(code, 0) + self.assertEqual(code, signal.SIGTERM) 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]) + sproc = get_test_subprocess([PYTHON_EXE, "-c", code]) p = psutil.Process(sproc.pid) self.assertEqual(p.wait(), 5) self.assertFalse(p.is_running()) @@ -175,7 +177,7 @@ class TestProcess(unittest.TestCase): # 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]) + sproc = get_test_subprocess([PYTHON_EXE, "-c", code]) p = psutil.Process(sproc.pid) self.assertEqual(p.wait(), 5) self.assertIn(p.wait(), (5, None)) @@ -207,8 +209,8 @@ class TestProcess(unittest.TestCase): # to get None. self.assertEqual(ret2, None) else: - self.assertEqual(ret1, 0) - self.assertEqual(ret1, 0) + self.assertEqual(ret1, signal.SIGTERM) + self.assertEqual(ret1, signal.SIGTERM) def test_wait_timeout_0(self): sproc = get_test_subprocess() @@ -227,7 +229,7 @@ class TestProcess(unittest.TestCase): if POSIX: self.assertEqual(code, -signal.SIGKILL) else: - self.assertEqual(code, 0) + self.assertEqual(code, signal.SIGTERM) self.assertFalse(p.is_running()) def test_cpu_percent(self): @@ -316,10 +318,10 @@ class TestProcess(unittest.TestCase): # test reads io1 = p.io_counters() - with open(PYTHON, 'rb') as f: + with open(PYTHON_EXE, 'rb') as f: f.read() io2 = p.io_counters() - if not BSD: + if not BSD and not AIX: self.assertGreater(io2.read_count, io1.read_count) self.assertEqual(io2.write_count, io1.write_count) if LINUX: @@ -528,6 +530,7 @@ class TestProcess(unittest.TestCase): p = psutil.Process() self.assertGreater(p.num_handles(), 0) + @unittest.skipIf(not HAS_THREADS, 'not supported') def test_threads(self): p = psutil.Process() if OPENBSD: @@ -552,6 +555,7 @@ class TestProcess(unittest.TestCase): @retry_before_failing() @skip_on_access_denied(only_if=OSX) + @unittest.skipIf(not HAS_THREADS, 'not supported') def test_threads_2(self): sproc = get_test_subprocess() p = psutil.Process(sproc.pid) @@ -693,12 +697,12 @@ class TestProcess(unittest.TestCase): sproc = get_test_subprocess() exe = psutil.Process(sproc.pid).exe() try: - self.assertEqual(exe, PYTHON) + self.assertEqual(exe, PYTHON_EXE) except AssertionError: - if WINDOWS and len(exe) == len(PYTHON): + if WINDOWS and len(exe) == len(PYTHON_EXE): # on Windows we don't care about case sensitivity normcase = os.path.normcase - self.assertEqual(normcase(exe), normcase(PYTHON)) + self.assertEqual(normcase(exe), normcase(PYTHON_EXE)) else: # certain platforms such as BSD are more accurate returning: # "/usr/local/bin/python2.7" @@ -709,7 +713,7 @@ class TestProcess(unittest.TestCase): ver = "%s.%s" % (sys.version_info[0], sys.version_info[1]) try: self.assertEqual(exe.replace(ver, ''), - PYTHON.replace(ver, '')) + PYTHON_EXE.replace(ver, '')) except AssertionError: # Tipically OSX. Really not sure what to do here. pass @@ -718,7 +722,7 @@ class TestProcess(unittest.TestCase): self.assertEqual(out, 'hey') def test_cmdline(self): - cmdline = [PYTHON, "-c", "import time; time.sleep(60)"] + cmdline = [PYTHON_EXE, "-c", "import time; time.sleep(60)"] sproc = get_test_subprocess(cmdline) try: self.assertEqual(' '.join(psutil.Process(sproc.pid).cmdline()), @@ -728,27 +732,39 @@ class TestProcess(unittest.TestCase): # and Open BSD returns a truncated string. # Also /proc/pid/cmdline behaves the same so it looks # like this is a kernel bug. - if NETBSD or OPENBSD: + # XXX - AIX truncates long arguments in /proc/pid/cmdline + if NETBSD or OPENBSD or AIX: self.assertEqual( - psutil.Process(sproc.pid).cmdline()[0], PYTHON) + psutil.Process(sproc.pid).cmdline()[0], PYTHON_EXE) else: raise def test_name(self): - sproc = get_test_subprocess(PYTHON) + sproc = get_test_subprocess(PYTHON_EXE) name = psutil.Process(sproc.pid).name().lower() pyexe = os.path.basename(os.path.realpath(sys.executable)).lower() assert pyexe.startswith(name), (pyexe, name) # XXX @unittest.skipIf(SUNOS, "broken on SUNOS") + @unittest.skipIf(AIX, "broken on AIX") 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 + + def rm(): + # Try to limit occasional failures on Appveyor: + # https://ci.appveyor.com/project/giampaolo/psutil/build/1350/ + # job/lbo3bkju55le850n + try: + safe_rmpath(funky_path) + except OSError: + pass + funky_path = TESTFN + 'foo bar )' create_exe(funky_path) - self.addCleanup(safe_rmpath, funky_path) + self.addCleanup(rm) cmdline = [funky_path, "-c", "import time; [time.sleep(0.01) for x in range(3000)];" "arg1", "arg2", "", "arg3", ""] @@ -853,7 +869,8 @@ class TestProcess(unittest.TestCase): self.assertEqual(p.cwd(), os.getcwd()) def test_cwd_2(self): - cmd = [PYTHON, "-c", "import os, time; os.chdir('..'); time.sleep(60)"] + cmd = [PYTHON_EXE, "-c", + "import os, time; os.chdir('..'); time.sleep(60)"] sproc = get_test_subprocess(cmd) p = psutil.Process(sproc.pid) call_until(p.cwd, "ret == os.path.dirname(os.getcwd())") @@ -870,10 +887,9 @@ class TestProcess(unittest.TestCase): 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: + # Work around travis failure: + # https://travis-ci.org/giampaolo/psutil/builds/284173194 + for n in all_cpus if not TRAVIS else initial: p.cpu_affinity([n]) self.assertEqual(p.cpu_affinity(), [n]) if hasattr(os, "sched_getaffinity"): @@ -911,6 +927,24 @@ class TestProcess(unittest.TestCase): self.assertRaises(TypeError, p.cpu_affinity, [0, "1"]) self.assertRaises(ValueError, p.cpu_affinity, [0, -1]) + @unittest.skipIf(not HAS_CPU_AFFINITY, 'not supported') + def test_cpu_affinity_all_combinations(self): + p = psutil.Process() + initial = p.cpu_affinity() + assert initial, initial + self.addCleanup(p.cpu_affinity, initial) + + # All possible CPU set combinations. + combos = [] + for l in range(0, len(initial) + 1): + for subset in itertools.combinations(initial, l): + if subset: + combos.append(list(subset)) + + for combo in combos: + p.cpu_affinity(combo) + self.assertEqual(p.cpu_affinity(), combo) + # TODO: #595 @unittest.skipIf(BSD, "broken on BSD") # can't find any process file on Appveyor @@ -937,7 +971,7 @@ class TestProcess(unittest.TestCase): # another process cmdline = "import time; f = open(r'%s', 'r'); time.sleep(60);" % TESTFN - sproc = get_test_subprocess([PYTHON, "-c", cmdline]) + sproc = get_test_subprocess([PYTHON_EXE, "-c", cmdline]) p = psutil.Process(sproc.pid) for x in range(100): @@ -1268,7 +1302,15 @@ class TestProcess(unittest.TestCase): # set methods succeed_or_zombie_p_exc(zproc.parent) if hasattr(zproc, 'cpu_affinity'): - succeed_or_zombie_p_exc(zproc.cpu_affinity, [0]) + try: + succeed_or_zombie_p_exc(zproc.cpu_affinity, [0]) + except ValueError as err: + if TRAVIS and LINUX and "not eligible" in str(err): + # https://travis-ci.org/giampaolo/psutil/jobs/279890461 + pass + else: + raise + succeed_or_zombie_p_exc(zproc.nice, 0) if hasattr(zproc, 'ionice'): if LINUX: @@ -1294,10 +1336,14 @@ class TestProcess(unittest.TestCase): # 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()]) + if not TRAVIS and OSX: + # For some reason this started failing all of the sudden. + # Maybe they upgraded OSX version? + # https://travis-ci.org/giampaolo/psutil/jobs/310896404 + 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()]) @unittest.skipIf(not POSIX, 'POSIX only') def test_zombie_process_is_running_w_exc(self): @@ -1360,28 +1406,23 @@ class TestProcess(unittest.TestCase): @unittest.skipIf(not HAS_ENVIRON, "not supported") def test_environ(self): + def clean_dict(d): + # Most of these are problematic on Travis. + d.pop("PSUTIL_TESTING", None) + d.pop("PLAT", None) + d.pop("HOME", None) + if OSX: + d.pop("__CF_USER_TEXT_ENCODING", None) + d.pop("VERSIONER_PYTHON_PREFER_32_BIT", None) + d.pop("VERSIONER_PYTHON_VERSION", None) + return dict( + [(k.rstrip("\r\n"), v.rstrip("\r\n")) for k, v in d.items()]) + self.maxDiff = None p = psutil.Process() - d = p.environ() - d2 = os.environ.copy() - - removes = [] - if 'PSUTIL_TESTING' in os.environ: - removes.append('PSUTIL_TESTING') - if OSX: - removes.extend([ - "__CF_USER_TEXT_ENCODING", - "VERSIONER_PYTHON_PREFER_32_BIT", - "VERSIONER_PYTHON_VERSION"]) - if LINUX or OSX: - removes.extend(['PLAT']) - if TOX: - removes.extend(['HOME']) - for key in removes: - d.pop(key, None) - d2.pop(key, None) - - self.assertEqual(d, d2) + d1 = clean_dict(p.environ()) + d2 = clean_dict(os.environ.copy()) + self.assertEqual(d1, d2) @unittest.skipIf(not HAS_ENVIRON, "not supported") @unittest.skipIf(not POSIX, "POSIX only") @@ -1485,7 +1526,7 @@ class TestPopen(unittest.TestCase): # 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);"] + cmd = [PYTHON_EXE, "-c", "import time; time.sleep(60);"] with psutil.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as proc: proc.name() @@ -1496,7 +1537,7 @@ class TestPopen(unittest.TestCase): proc.terminate() def test_ctx_manager(self): - with psutil.Popen([PYTHON, "-V"], + with psutil.Popen([PYTHON_EXE, "-V"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE) as proc: @@ -1510,7 +1551,7 @@ class TestPopen(unittest.TestCase): # subprocess.Popen()'s terminate(), kill() and send_signal() do # not raise exception after the process is gone. psutil.Popen # diverges from that. - cmd = [PYTHON, "-c", "import time; time.sleep(60);"] + cmd = [PYTHON_EXE, "-c", "import time; time.sleep(60);"] with psutil.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as proc: proc.terminate() |