Bug 930808 - Upgrade to psutil 2.1.3; r=glandium
psutil 2.1.3 is replacing psutil 1.0.1. There are numerous bug fixes and feature enhancements in psutil worth obtaining. Source code was obtained from https://pypi.python.org/packages/source/p/psutil/psutil-2.1.3.tar.gz and uncompressed into python/psutil without modification except for the removal of the egg-info directory and the .travis.yml file.
This commit is contained in:
Gregory Szorc
@@ -6,44 +6,55 @@
|
||||
|
||||
"""Windows specific tests. These are implicitly run by test_psutil.py."""
|
||||
|
||||
import errno
|
||||
import os
|
||||
import unittest
|
||||
import platform
|
||||
import signal
|
||||
import time
|
||||
import sys
|
||||
import subprocess
|
||||
import errno
|
||||
import sys
|
||||
import time
|
||||
import traceback
|
||||
|
||||
import psutil
|
||||
import _psutil_mswindows
|
||||
from psutil._compat import PY3, callable, long
|
||||
from test_psutil import *
|
||||
from test_psutil import (get_test_subprocess, reap_children, unittest)
|
||||
|
||||
try:
|
||||
import wmi
|
||||
except ImportError:
|
||||
err = sys.exc_info()[1]
|
||||
register_warning("Couldn't run wmi tests: %s" % str(err))
|
||||
wmi = None
|
||||
try:
|
||||
import win32api
|
||||
import win32con
|
||||
except ImportError:
|
||||
err = sys.exc_info()[1]
|
||||
register_warning("Couldn't run pywin32 tests: %s" % str(err))
|
||||
win32api = None
|
||||
win32api = win32con = None
|
||||
|
||||
from psutil._compat import PY3, callable, long
|
||||
from psutil._pswindows import ACCESS_DENIED_SET
|
||||
import _psutil_windows
|
||||
import psutil
|
||||
|
||||
|
||||
def wrap_exceptions(fun):
|
||||
def wrapper(self, *args, **kwargs):
|
||||
try:
|
||||
return fun(self, *args, **kwargs)
|
||||
except OSError:
|
||||
err = sys.exc_info()[1]
|
||||
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
|
||||
|
||||
|
||||
class WindowsSpecificTestCase(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
sproc = get_test_subprocess()
|
||||
wait_for_pid(sproc.pid)
|
||||
self.pid = sproc.pid
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.pid = get_test_subprocess().pid
|
||||
|
||||
def tearDown(self):
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
reap_children()
|
||||
|
||||
def test_issue_24(self):
|
||||
@@ -52,14 +63,14 @@ class WindowsSpecificTestCase(unittest.TestCase):
|
||||
|
||||
def test_special_pid(self):
|
||||
p = psutil.Process(4)
|
||||
self.assertEqual(p.name, 'System')
|
||||
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)
|
||||
p.username()
|
||||
self.assertTrue(p.create_time() >= 0.0)
|
||||
try:
|
||||
rss, vms = p.get_memory_info()
|
||||
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'):
|
||||
@@ -67,7 +78,7 @@ class WindowsSpecificTestCase(unittest.TestCase):
|
||||
else:
|
||||
self.assertTrue(rss > 0)
|
||||
|
||||
def test_signal(self):
|
||||
def test_send_signal(self):
|
||||
p = psutil.Process(self.pid)
|
||||
self.assertRaises(ValueError, p.send_signal, signal.SIGINT)
|
||||
|
||||
@@ -81,206 +92,205 @@ class WindowsSpecificTestCase(unittest.TestCase):
|
||||
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)
|
||||
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)
|
||||
self.assertEqual(os.path.basename(p.exe()), p.name())
|
||||
except psutil.Error:
|
||||
pass
|
||||
|
||||
if wmi is not None:
|
||||
# --- Process class tests
|
||||
|
||||
# --- 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)
|
||||
|
||||
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)
|
||||
self.assertEqual(p.exe(), w.ExecutablePath)
|
||||
|
||||
def test_process_exe(self):
|
||||
w = wmi.WMI().Win32_Process(ProcessId=self.pid)[0]
|
||||
p = psutil.Process(self.pid)
|
||||
self.assertEqual(p.exe, w.ExecutablePath)
|
||||
@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('"', ''))
|
||||
|
||||
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)
|
||||
|
||||
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))
|
||||
|
||||
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.get_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))
|
||||
|
||||
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.get_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)
|
||||
|
||||
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(hasattr(os, 'NUMBER_OF_PROCESSORS'),
|
||||
'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())
|
||||
|
||||
# --- psutil namespace functions and constants tests
|
||||
@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.skipUnless(hasattr(os, 'NUMBER_OF_PROCESSORS'),
|
||||
'NUMBER_OF_PROCESSORS env var is not available')
|
||||
def test_NUM_CPUS(self):
|
||||
num_cpus = int(os.environ['NUMBER_OF_PROCESSORS'])
|
||||
self.assertEqual(num_cpus, psutil.NUM_CPUS)
|
||||
# @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()))
|
||||
#
|
||||
|
||||
def test_TOTAL_PHYMEM(self):
|
||||
w = wmi.WMI().Win32_ComputerSystem()[0]
|
||||
self.assertEqual(int(w.TotalPhysicalMemory), psutil.TOTAL_PHYMEM)
|
||||
@unittest.skipIf(wmi is None, "wmi module is not installed")
|
||||
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 = [x.ProcessId for x in w]
|
||||
wmi_pids.sort()
|
||||
psutil_pids = psutil.pids()
|
||||
psutil_pids.sort()
|
||||
if wmi_pids != psutil_pids:
|
||||
difference = \
|
||||
filter(lambda x: x not in wmi_pids, psutil_pids) + \
|
||||
filter(lambda x: x not in psutil_pids, wmi_pids)
|
||||
self.fail("difference: " + str(difference))
|
||||
|
||||
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))
|
||||
# XXX - ? no actual test here
|
||||
|
||||
def test_get_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 = [x.ProcessId for x in w]
|
||||
wmi_pids.sort()
|
||||
psutil_pids = psutil.get_pid_list()
|
||||
psutil_pids.sort()
|
||||
if wmi_pids != psutil_pids:
|
||||
difference = filter(lambda x:x not in wmi_pids, psutil_pids) + \
|
||||
filter(lambda x:x not in psutil_pids, wmi_pids)
|
||||
self.fail("difference: " + str(difference))
|
||||
|
||||
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:
|
||||
err = sys.exc_info()[1]
|
||||
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)
|
||||
@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
|
||||
else:
|
||||
self.fail("can't find partition %s" % repr(ps_part))
|
||||
|
||||
if win32api is not None:
|
||||
|
||||
def test_get_num_handles(self):
|
||||
p = psutil.Process(os.getpid())
|
||||
before = p.get_num_handles()
|
||||
handle = win32api.OpenProcess(win32con.PROCESS_QUERY_INFORMATION,
|
||||
win32con.FALSE, os.getpid())
|
||||
after = p.get_num_handles()
|
||||
self.assertEqual(after, before+1)
|
||||
win32api.CloseHandle(handle)
|
||||
self.assertEqual(p.get_num_handles(), before)
|
||||
|
||||
def test_get_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):
|
||||
ret = attr()
|
||||
else:
|
||||
ret = attr
|
||||
|
||||
p = psutil.Process(self.pid)
|
||||
attrs = []
|
||||
failures = []
|
||||
for name in dir(psutil.Process):
|
||||
if name.startswith('_') \
|
||||
or name.startswith('set_') \
|
||||
or name in ('terminate', 'kill', 'suspend', 'resume', 'nice',
|
||||
'send_signal', 'wait', 'get_children', 'as_dict'):
|
||||
continue
|
||||
else:
|
||||
try:
|
||||
call(p, name)
|
||||
num1 = p.get_num_handles()
|
||||
call(p, name)
|
||||
num2 = p.get_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))
|
||||
usage = psutil.disk_usage(ps_part.mountpoint)
|
||||
except OSError:
|
||||
err = sys.exc_info()[1]
|
||||
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)
|
||||
|
||||
import _psutil_mswindows
|
||||
from psutil._psmswindows import ACCESS_DENIED_SET
|
||||
@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.startswith('set_') \
|
||||
or name.startswith('get') \
|
||||
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 wrap_exceptions(callable):
|
||||
def wrapper(self, *args, **kwargs):
|
||||
try:
|
||||
return callable(self, *args, **kwargs)
|
||||
except OSError:
|
||||
err = sys.exc_info()[1]
|
||||
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
|
||||
|
||||
class TestDualProcessImplementation(unittest.TestCase):
|
||||
fun_names = [
|
||||
# function name tolerance
|
||||
('get_process_cpu_times', 0.2),
|
||||
('get_process_create_time', 0.5),
|
||||
('get_process_num_handles', 1), # 1 because impl #1 opens a handle
|
||||
('get_process_io_counters', 0),
|
||||
('get_process_memory_info', 1024), # KB
|
||||
# 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_io_counters', 0),
|
||||
('proc_memory_info', 1024), # KB
|
||||
]
|
||||
|
||||
def test_compare_values(self):
|
||||
@@ -290,7 +300,7 @@ class TestDualProcessImplementation(unittest.TestCase):
|
||||
# case the first fails because of limited permission error.
|
||||
# Here we test that the two methods return the exact same value,
|
||||
# see:
|
||||
# http://code.google.com/p/psutil/issues/detail?id=304
|
||||
# https://github.com/giampaolo/psutil/issues/304
|
||||
def assert_ge_0(obj):
|
||||
if isinstance(obj, tuple):
|
||||
for value in obj:
|
||||
@@ -314,10 +324,10 @@ class TestDualProcessImplementation(unittest.TestCase):
|
||||
|
||||
failures = []
|
||||
for name, tolerance in self.fun_names:
|
||||
meth1 = wrap_exceptions(getattr(_psutil_mswindows, name))
|
||||
meth2 = wrap_exceptions(getattr(_psutil_mswindows, name + '_2'))
|
||||
meth1 = wrap_exceptions(getattr(_psutil_windows, name))
|
||||
meth2 = wrap_exceptions(getattr(_psutil_windows, name + '_2'))
|
||||
for p in psutil.process_iter():
|
||||
if name == 'get_process_memory_info' and p.pid == os.getpid():
|
||||
if name == 'proc_memory_info' and p.pid == os.getpid():
|
||||
continue
|
||||
#
|
||||
try:
|
||||
@@ -343,10 +353,9 @@ class TestDualProcessImplementation(unittest.TestCase):
|
||||
assert_ge_0(ret1)
|
||||
assert_ge_0(ret2)
|
||||
except AssertionError:
|
||||
err = sys.exc_info()[1]
|
||||
trace = traceback.format_exc()
|
||||
msg = '%s\npid=%s, method=%r, ret_1=%r, ret_2=%r' \
|
||||
% (trace, p.pid, name, ret1, ret2)
|
||||
msg = '%s\npid=%s, method=%r, ret_1=%r, ret_2=%r' % (
|
||||
trace, p.pid, name, ret1, ret2)
|
||||
failures.append(msg)
|
||||
break
|
||||
if failures:
|
||||
@@ -355,9 +364,9 @@ class TestDualProcessImplementation(unittest.TestCase):
|
||||
def test_zombies(self):
|
||||
# test that NPS is raised by the 2nd implementation in case a
|
||||
# process no longer exists
|
||||
ZOMBIE_PID = max(psutil.get_pid_list()) + 5000
|
||||
ZOMBIE_PID = max(psutil.pids()) + 5000
|
||||
for name, _ in self.fun_names:
|
||||
meth = wrap_exceptions(getattr(_psutil_mswindows, name))
|
||||
meth = wrap_exceptions(getattr(_psutil_windows, name))
|
||||
self.assertRaises(psutil.NoSuchProcess, meth, ZOMBIE_PID)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user