def get_default_session():
"""
Locates the first ancestor process which is a shell. Returns
its pid, or None if not found.
"""
if psutil.POSIX:
def predicate(name):
return name.endswith("sh")
elif psutil.WINDOWS:
def predicate(name):
return name in ("cmd.exe", "powershell.exe")
else:
return None
proc = psutil.Process()
while proc.parent().pid:
proc = proc.parent()
if predicate(proc.name()):
return proc.pid
return None
python类POSIX的实例源码
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_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_net_if_addrs_mac_null_bytes(self):
# Simulate that the underlying C function returns an incomplete
# MAC address. psutil is supposed to fill it with null bytes.
# https://github.com/giampaolo/psutil/issues/786
if POSIX:
ret = [('em1', psutil.AF_LINK, '06:3d:29', None, None, None)]
else:
ret = [('em1', -1, '06-3d-29', None, None, None)]
with mock.patch('psutil._psplatform.net_if_addrs',
return_value=ret) as m:
addr = psutil.net_if_addrs()['em1'][0]
assert m.called
if POSIX:
self.assertEqual(addr.address, '06:3d:29:00:00:00')
else:
self.assertEqual(addr.address, '06-3d-29-00-00-00')
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_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_net_if_addrs_mac_null_bytes(self):
# Simulate that the underlying C function returns an incomplete
# MAC address. psutil is supposed to fill it with null bytes.
# https://github.com/giampaolo/psutil/issues/786
if POSIX:
ret = [('em1', psutil.AF_LINK, '06:3d:29', None, None, None)]
else:
ret = [('em1', -1, '06-3d-29', None, None, None)]
with mock.patch('psutil._psplatform.net_if_addrs',
return_value=ret) as m:
addr = psutil.net_if_addrs()['em1'][0]
assert m.called
if POSIX:
self.assertEqual(addr.address, '06:3d:29:00:00:00')
else:
self.assertEqual(addr.address, '06-3d-29-00-00-00')
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_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_net_if_addrs_mac_null_bytes(self):
# Simulate that the underlying C function returns an incomplete
# MAC address. psutil is supposed to fill it with null bytes.
# https://github.com/giampaolo/psutil/issues/786
if POSIX:
ret = [('em1', psutil.AF_LINK, '06:3d:29', None, None, None)]
else:
ret = [('em1', -1, '06-3d-29', None, None, None)]
with mock.patch('psutil._psplatform.net_if_addrs',
return_value=ret) as m:
addr = psutil.net_if_addrs()['em1'][0]
assert m.called
if POSIX:
self.assertEqual(addr.address, '06:3d:29:00:00:00')
else:
self.assertEqual(addr.address, '06-3d-29-00-00-00')
def test_kill(self):
sproc = get_test_subprocess()
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()
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, "")):
with self.assertRaises(psutil.NoSuchProcess):
p.send_signal(sig)
#
sproc = get_test_subprocess()
p = psutil.Process(sproc.pid)
p.send_signal(sig)
with mock.patch('psutil.os.kill',
side_effect=OSError(errno.EPERM, "")):
with self.assertRaises(psutil.AccessDenied):
psutil.Process().send_signal(sig)
# Sending a signal to process with PID 0 is not allowed as
# it would affect every process in the process group of
# the calling process (os.getpid()) instead of PID 0").
if 0 in psutil.pids():
p = psutil.Process(0)
self.assertRaises(ValueError, p.send_signal, signal.SIGTERM)
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:
try:
assert os.path.exists(nt.path) or \
os.path.islink(nt.path), nt.path
except AssertionError:
if not LINUX:
raise
else:
# https://github.com/giampaolo/psutil/issues/759
with open('/proc/self/smaps') as f:
data = f.read()
if "%s (deleted)" % nt.path not in data:
raise
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_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
# test all methods
p = psutil.Process(0)
for name in psutil._as_dict_attrnames:
if name == 'pid':
continue
meth = getattr(p, name)
try:
ret = meth()
except psutil.AccessDenied:
pass
else:
if name in ("uids", "gids"):
self.assertEqual(ret.real, 0)
elif name == "username":
if POSIX:
self.assertEqual(p.username(), 'root')
elif WINDOWS:
self.assertEqual(p.username(), 'NT AUTHORITY\\SYSTEM')
elif name == "name":
assert name, name
if hasattr(p, 'rlimit'):
try:
p.rlimit(psutil.RLIMIT_FSIZE)
except psutil.AccessDenied:
pass
p.as_dict()
if not OPENBSD:
self.assertIn(0, psutil.pids())
self.assertTrue(psutil.pid_exists(0))
def setUp(self):
if POSIX:
import pwd
import grp
users = pwd.getpwall()
groups = grp.getgrall()
self.all_uids = set([x.pw_uid for x in users])
self.all_usernames = set([x.pw_name for x in users])
self.all_gids = set([x.gr_gid for x in groups])
def exe(self, ret, proc):
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 memory_info(self, ret, proc):
for name in ret._fields:
self.assertGreaterEqual(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 nice(self, ret, proc):
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 create_exe(outpath, c_code=None):
assert not os.path.exists(outpath), outpath
if which("gcc"):
if c_code is None:
c_code = textwrap.dedent(
"""
#include <unistd.h>
int main() {
pause();
return 1;
}
""")
with tempfile.NamedTemporaryFile(
suffix='.c', delete=False, mode='wt') as f:
f.write(c_code)
try:
subprocess.check_call(["gcc", f.name, "-o", outpath])
finally:
safe_rmpath(f.name)
else:
# fallback - use python's executable
shutil.copyfile(sys.executable, outpath)
if POSIX:
st = os.stat(outpath)
os.chmod(outpath, st.st_mode | stat.S_IEXEC)
# ===================================================================
# --- testing
# ===================================================================
def test_kill(self):
sproc = get_test_subprocess()
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()
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, "")):
with self.assertRaises(psutil.NoSuchProcess):
p.send_signal(sig)
#
sproc = get_test_subprocess()
p = psutil.Process(sproc.pid)
p.send_signal(sig)
with mock.patch('psutil.os.kill',
side_effect=OSError(errno.EPERM, "")):
with self.assertRaises(psutil.AccessDenied):
psutil.Process().send_signal(sig)
# Sending a signal to process with PID 0 is not allowed as
# it would affect every process in the process group of
# the calling process (os.getpid()) instead of PID 0").
if 0 in psutil.pids():
p = psutil.Process(0)
self.assertRaises(ValueError, p.send_signal, signal.SIGTERM)
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:
try:
assert os.path.exists(nt.path) or \
os.path.islink(nt.path), nt.path
except AssertionError:
if not LINUX:
raise
else:
# https://github.com/giampaolo/psutil/issues/759
with open('/proc/self/smaps') as f:
data = f.read()
if "%s (deleted)" % nt.path not in data:
raise
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_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
# test all methods
p = psutil.Process(0)
for name in psutil._as_dict_attrnames:
if name == 'pid':
continue
meth = getattr(p, name)
try:
ret = meth()
except psutil.AccessDenied:
pass
else:
if name in ("uids", "gids"):
self.assertEqual(ret.real, 0)
elif name == "username":
if POSIX:
self.assertEqual(p.username(), 'root')
elif WINDOWS:
self.assertEqual(p.username(), 'NT AUTHORITY\\SYSTEM')
elif name == "name":
assert name, name
if hasattr(p, 'rlimit'):
try:
p.rlimit(psutil.RLIMIT_FSIZE)
except psutil.AccessDenied:
pass
p.as_dict()
if not OPENBSD:
self.assertIn(0, psutil.pids())
self.assertTrue(psutil.pid_exists(0))
def setUp(self):
if POSIX:
import pwd
import grp
users = pwd.getpwall()
groups = grp.getgrall()
self.all_uids = set([x.pw_uid for x in users])
self.all_usernames = set([x.pw_name for x in users])
self.all_gids = set([x.gr_gid for x in groups])
def exe(self, ret, proc):
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 memory_info(self, ret, proc):
for name in ret._fields:
self.assertGreaterEqual(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 nice(self, ret, proc):
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 create_exe(outpath, c_code=None):
"""Creates an executable file in the given location."""
assert not os.path.exists(outpath), outpath
if which("gcc"):
if c_code is None:
c_code = textwrap.dedent(
"""
#include <unistd.h>
int main() {
pause();
return 1;
}
""")
with tempfile.NamedTemporaryFile(
suffix='.c', delete=False, mode='wt') as f:
f.write(c_code)
try:
subprocess.check_call(["gcc", f.name, "-o", outpath])
finally:
safe_rmpath(f.name)
else:
# fallback - use python's executable
if c_code is not None:
raise ValueError(
"can't specify c_code arg as gcc is not installed")
shutil.copyfile(sys.executable, outpath)
if POSIX:
st = os.stat(outpath)
os.chmod(outpath, st.st_mode | stat.S_IEXEC)
# ===================================================================
# --- testing
# ===================================================================
