#!/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)