The Tor Browser: python/psutil/test/_windows.py@b8a032363ba2 (annotated)

python/psutil/test/_windows.py@b8a032363ba2 (annotated)

python/psutil/test/_windows.py

Thu, 22 Jan 2015 13:21:57 +0100

author: Michael Schloh von Bennewitz <michael@schloh.com>
date: Thu, 22 Jan 2015 13:21:57 +0100
branch: TOR_BUG_9701
changeset 15: b8a032363ba2
permissions: -rw-r--r--

Incorporate requested changes from Mozilla in review:
https://bugzilla.mozilla.org/show_bug.cgi?id=1123480#c6

 #!/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.
 """Windows specific tests.  These are implicitly run by test_psutil.py."""
 import os
 import unittest
 import platform
 import signal
 import time
 import sys
 import subprocess
 import errno
 import traceback
 import psutil
 import _psutil_mswindows
 from psutil._compat import PY3, callable, long
 from test_psutil import *
 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
 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.net_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
         @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)
         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:
                     self.assertGreaterEqual(value, 0)
             elif isinstance(obj, (int, long, float)):
                 self.assertGreaterEqual(obj, 0)
             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)
                     self.assertLessEqual(diff, tolerance)
                 elif isinstance(ret2, tuple):
                     for a, b in zip(ret1, ret2):
                         diff = abs(a - b)
                         self.assertLessEqual(diff, tolerance)
         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():
                 if name == 'get_process_memory_info' and p.pid == os.getpid():
                     continue
                 #
                 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)
 def test_main():
     test_suite = unittest.TestSuite()
     test_suite.addTest(unittest.makeSuite(WindowsSpecificTestCase))
     test_suite.addTest(unittest.makeSuite(TestDualProcessImplementation))
     result = unittest.TextTestRunner(verbosity=2).run(test_suite)
     return result.wasSuccessful()
 if __name__ == '__main__':
     if not test_main():
         sys.exit(1)

The Tor Browser / annotate

python/psutil/test/_windows.py@b8a032363ba2 (annotated)

python/psutil/test/_windows.py