import FIREFOX_52_6_0esr_RELEASE from mozilla-esr52 hg repo

python/psutil/test/_windows.py

@@ -0,0 +1,464 @@
+#!/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 errno
+import os
+import platform
+import signal
+import subprocess
+import sys
+import time
+import traceback
+
+from test_psutil import APPVEYOR, WINDOWS
+from test_psutil import get_test_subprocess, reap_children, unittest
+
+import mock
+try:
+    import wmi
+except ImportError:
+    wmi = None
+try:
+    import win32api
+    import win32con
+except ImportError:
+    win32api = win32con = None
+
+from psutil._compat import PY3, callable, long
+import psutil
+
+
+cext = psutil._psplatform.cext
+
+
+def wrap_exceptions(fun):
+    def wrapper(self, *args, **kwargs):
+        try:
+            return fun(self, *args, **kwargs)
+        except OSError as err:
+            from psutil._pswindows import ACCESS_DENIED_SET
+            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
+
+
+@unittest.skipUnless(WINDOWS, "not a Windows system")
+class WindowsSpecificTestCase(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        cls.pid = get_test_subprocess().pid
+
+    @classmethod
+    def tearDownClass(cls):
+        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.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_send_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
+
+    # --- 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)
+
+    @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)
+        # Note: wmi reports the exe as a lower case string.
+        # Being Windows paths case-insensitive we ignore that.
+        self.assertEqual(p.exe().lower(), w.ExecutablePath.lower())
+
+    @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('"', ''))
+
+    @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)
+
+    @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))
+
+    @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))
+
+    @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)
+
+    # --- psutil namespace functions and constants tests
+
+    @unittest.skipUnless('NUMBER_OF_PROCESSORS' in os.environ,
+                         '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())
+
+    @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.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()))
+    #
+
+    # Note: this test is not very reliable
+    @unittest.skipIf(wmi is None, "wmi module is not installed")
+    @unittest.skipIf(APPVEYOR, "test not relieable on appveyor")
+    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 = set([x.ProcessId for x in w])
+        psutil_pids = set(psutil.pids())
+        self.assertEqual(wmi_pids, psutil_pids)
+
+    @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
+                    try:
+                        usage = psutil.disk_usage(ps_part.mountpoint)
+                    except OSError as err:
+                        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)
+
+    @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 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 test_name_always_available(self):
+        # On Windows name() is never supposed to raise AccessDenied,
+        # see https://github.com/giampaolo/psutil/issues/627
+        for p in psutil.process_iter():
+            try:
+                p.name()
+            except psutil.NoSuchProcess():
+                pass
+
+
+@unittest.skipUnless(WINDOWS, "not a Windows system")
+class TestDualProcessImplementation(unittest.TestCase):
+    """
+    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:
+    https://github.com/giampaolo/psutil/issues/304
+    """
+
+    fun_names = [
+        # 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_memory_info', 1024),  # KB
+        ('proc_io_counters', 0),
+    ]
+
+    def test_compare_values(self):
+        def assert_ge_0(obj):
+            if isinstance(obj, tuple):
+                for value in obj:
+                    self.assertGreaterEqual(value, 0, msg=obj)
+            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)
+
+        from psutil._pswindows import ntpinfo
+        failures = []
+        for p in psutil.process_iter():
+            try:
+                nt = ntpinfo(*cext.proc_info(p.pid))
+            except psutil.NoSuchProcess:
+                continue
+            assert_ge_0(nt)
+
+            for name, tolerance in self.fun_names:
+                if name == 'proc_memory_info' and p.pid == os.getpid():
+                    continue
+                if name == 'proc_create_time' and p.pid in (0, 4):
+                    continue
+                meth = wrap_exceptions(getattr(cext, name))
+                try:
+                    ret = meth(p.pid)
+                except (psutil.NoSuchProcess, psutil.AccessDenied):
+                    continue
+                # compare values
+                try:
+                    if name == 'proc_cpu_times':
+                        compare_with_tolerance(ret[0], nt.user_time, tolerance)
+                        compare_with_tolerance(ret[1],
+                                               nt.kernel_time, tolerance)
+                    elif name == 'proc_create_time':
+                        compare_with_tolerance(ret, nt.create_time, tolerance)
+                    elif name == 'proc_num_handles':
+                        compare_with_tolerance(ret, nt.num_handles, tolerance)
+                    elif name == 'proc_io_counters':
+                        compare_with_tolerance(ret[0], nt.io_rcount, tolerance)
+                        compare_with_tolerance(ret[1], nt.io_wcount, tolerance)
+                        compare_with_tolerance(ret[2], nt.io_rbytes, tolerance)
+                        compare_with_tolerance(ret[3], nt.io_wbytes, tolerance)
+                    elif name == 'proc_memory_info':
+                        try:
+                            rawtupl = cext.proc_memory_info_2(p.pid)
+                        except psutil.NoSuchProcess:
+                            continue
+                        compare_with_tolerance(ret, rawtupl, tolerance)
+                except AssertionError:
+                    trace = traceback.format_exc()
+                    msg = '%s\npid=%s, method=%r, ret_1=%r, ret_2=%r' % (
+                        trace, p.pid, name, ret, nt)
+                    failures.append(msg)
+                    break
+
+        if failures:
+            self.fail('\n\n'.join(failures))
+
+    # ---
+    # same tests as above but mimicks the AccessDenied failure of
+    # the first (fast) method failing with AD.
+    # TODO: currently does not take tolerance into account.
+
+    def test_name(self):
+        name = psutil.Process().name()
+        with mock.patch("psutil._psplatform.cext.proc_exe",
+                        side_effect=psutil.AccessDenied(os.getpid())) as fun:
+            psutil.Process().name() == name
+            assert fun.called
+
+    def test_memory_info(self):
+        mem = psutil.Process().memory_info()
+        with mock.patch("psutil._psplatform.cext.proc_memory_info",
+                        side_effect=OSError(errno.EPERM, "msg")) as fun:
+            psutil.Process().memory_info() == mem
+            assert fun.called
+
+    def test_create_time(self):
+        ctime = psutil.Process().create_time()
+        with mock.patch("psutil._psplatform.cext.proc_create_time",
+                        side_effect=OSError(errno.EPERM, "msg")) as fun:
+            psutil.Process().create_time() == ctime
+            assert fun.called
+
+    def test_cpu_times(self):
+        cpu_times = psutil.Process().cpu_times()
+        with mock.patch("psutil._psplatform.cext.proc_cpu_times",
+                        side_effect=OSError(errno.EPERM, "msg")) as fun:
+            psutil.Process().cpu_times() == cpu_times
+            assert fun.called
+
+    def test_io_counters(self):
+        io_counters = psutil.Process().io_counters()
+        with mock.patch("psutil._psplatform.cext.proc_io_counters",
+                        side_effect=OSError(errno.EPERM, "msg")) as fun:
+            psutil.Process().io_counters() == io_counters
+            assert fun.called
+
+    def test_num_handles(self):
+        io_counters = psutil.Process().io_counters()
+        with mock.patch("psutil._psplatform.cext.proc_io_counters",
+                        side_effect=OSError(errno.EPERM, "msg")) as fun:
+            psutil.Process().io_counters() == io_counters
+            assert fun.called
+
+    # --- other tests
+
+    def test_compare_name_exe(self):
+        for p in psutil.process_iter():
+            try:
+                a = os.path.basename(p.exe())
+                b = p.name()
+            except (psutil.NoSuchProcess, psutil.AccessDenied):
+                pass
+            else:
+                self.assertEqual(a, b)
+
+    def test_zombies(self):
+        # test that NPS is raised by the 2nd implementation in case a
+        # process no longer exists
+        ZOMBIE_PID = max(psutil.pids()) + 5000
+        for name, _ in self.fun_names:
+            meth = wrap_exceptions(getattr(cext, name))
+            self.assertRaises(psutil.NoSuchProcess, meth, ZOMBIE_PID)
+
+
+def 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 main():
+        sys.exit(1)