gecko/python/psutil/test/_windows.py

369 lines
14 KiB
Python

#!/usr/bin/env python
#
# $Id: _windows.py 1453 2012-07-13 19:55:11Z g.rodola $
#
# Copyright (c) 2009, Jay Loden, 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 os
import unittest
import platform
import signal
import time
import warnings
import atexit
import sys
import subprocess
import errno
import traceback
import psutil
import _psutil_mswindows
from psutil._compat import PY3, callable, long
from test_psutil import reap_children, get_test_subprocess, wait_for_pid, warn
try:
import wmi
except ImportError:
err = sys.exc_info()[1]
atexit.register(warn, "Couldn't run wmi tests: %s" % str(err))
wmi = None
try:
import win32api
import win32con
except ImportError:
err = sys.exc_info()[1]
atexit.register(warn, "Couldn't run pywin32 tests: %s" % str(err))
win32api = None
class WindowsSpecificTestCase(unittest.TestCase):
def setUp(self):
sproc = get_test_subprocess()
wait_for_pid(sproc.pid)
self.pid = sproc.pid
def tearDown(self):
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.get_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_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.network_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
if wmi is not None:
# --- Process class tests
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_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_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_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_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))
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))
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
def test_NUM_CPUS(self):
num_cpus = int(os.environ['NUMBER_OF_PROCESSORS'])
self.assertEqual(num_cpus, psutil.NUM_CPUS)
def test_TOTAL_PHYMEM(self):
w = wmi.WMI().Win32_ComputerSystem()[0]
self.assertEqual(int(w.TotalPhysicalMemory), psutil.TOTAL_PHYMEM)
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)
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))
import _psutil_mswindows
from psutil._psmswindows import ACCESS_DENIED_SET
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
]
def test_compare_values(self):
# 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:
# http://code.google.com/p/psutil/issues/detail?id=304
def assert_ge_0(obj):
if isinstance(obj, tuple):
for value in obj:
assert value >= 0, value
elif isinstance(obj, (int, long, float)):
assert obj >= 0, obj
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)
assert diff <= tolerance, diff
elif isinstance(ret2, tuple):
for a, b in zip(ret1, ret2):
diff = abs(a - b)
assert diff <= tolerance, diff
failures = []
for name, tolerance in self.fun_names:
meth1 = wrap_exceptions(getattr(_psutil_mswindows, name))
meth2 = wrap_exceptions(getattr(_psutil_mswindows, name + '_2'))
for p in psutil.process_iter():
#
try:
ret1 = meth1(p.pid)
except psutil.NoSuchProcess:
continue
except psutil.AccessDenied:
ret1 = None
#
try:
ret2 = meth2(p.pid)
except psutil.NoSuchProcess:
# this is supposed to fail only in case of zombie process
# never for permission error
continue
# compare values
try:
if ret1 is None:
assert_ge_0(ret2)
else:
compare_with_tolerance(ret1, ret2, tolerance)
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)
failures.append(msg)
break
if failures:
self.fail('\n\n'.join(failures))
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
for name, _ in self.fun_names:
meth = wrap_exceptions(getattr(_psutil_mswindows, name))
self.assertRaises(psutil.NoSuchProcess, meth, ZOMBIE_PID)
if __name__ == '__main__':
test_suite = unittest.TestSuite()
test_suite.addTest(unittest.makeSuite(WindowsSpecificTestCase))
test_suite.addTest(unittest.makeSuite(TestDualProcessImplementation))
unittest.TextTestRunner(verbosity=2).run(test_suite)