diff options
Diffstat (limited to 'lib/psutil/tests/test_windows.py')
| -rw-r--r-- | lib/psutil/tests/test_windows.py | 898 |
1 files changed, 898 insertions, 0 deletions
diff --git a/lib/psutil/tests/test_windows.py b/lib/psutil/tests/test_windows.py new file mode 100644 index 0000000..55e6731 --- /dev/null +++ b/lib/psutil/tests/test_windows.py @@ -0,0 +1,898 @@ +#!/usr/bin/env python3 +# -*- coding: UTF-8 -* + +# 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.""" + +import datetime +import errno +import glob +import os +import platform +import re +import signal +import subprocess +import sys +import time +import unittest +import warnings + +import psutil +from psutil import WINDOWS +from psutil._compat import FileNotFoundError +from psutil._compat import which +from psutil._compat import super +from psutil.tests import APPVEYOR +from psutil.tests import GITHUB_ACTIONS +from psutil.tests import HAS_BATTERY +from psutil.tests import IS_64BIT +from psutil.tests import PY3 +from psutil.tests import PYPY +from psutil.tests import TOLERANCE_DISK_USAGE +from psutil.tests import TOLERANCE_SYS_MEM +from psutil.tests import PsutilTestCase +from psutil.tests import mock +from psutil.tests import retry_on_failure +from psutil.tests import sh +from psutil.tests import spawn_testproc +from psutil.tests import terminate + + +if WINDOWS and not PYPY: + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + import win32api # requires "pip install pywin32" + import win32con + import win32process + import wmi # requires "pip install wmi" / "make setup-dev-env" + +if WINDOWS: + from psutil._pswindows import convert_oserror + + +cext = psutil._psplatform.cext + + +@unittest.skipIf(not WINDOWS, "WINDOWS only") +@unittest.skipIf(PYPY, "pywin32 not available on PYPY") +# https://github.com/giampaolo/psutil/pull/1762#issuecomment-632892692 +@unittest.skipIf(GITHUB_ACTIONS and not PY3, "pywin32 broken on GITHUB + PY2") +class WindowsTestCase(PsutilTestCase): + pass + + +def powershell(cmd): + """Currently not used, but avalable just in case. Usage: + + >>> powershell( + "Get-CIMInstance Win32_PageFileUsage | Select AllocatedBaseSize") + """ + if not which("powershell.exe"): + raise unittest.SkipTest("powershell.exe not available") + cmdline = \ + 'powershell.exe -ExecutionPolicy Bypass -NoLogo -NonInteractive ' + \ + '-NoProfile -WindowStyle Hidden -Command "%s"' % cmd + return sh(cmdline) + + +def wmic(path, what, converter=int): + """Currently not used, but avalable just in case. Usage: + + >>> wmic("Win32_OperatingSystem", "FreePhysicalMemory") + 2134124534 + """ + out = sh("wmic path %s get %s" % (path, what)).strip() + data = "".join(out.splitlines()[1:]).strip() # get rid of the header + if converter is not None: + if "," in what: + return tuple([converter(x) for x in data.split()]) + else: + return converter(data) + else: + return data + + +# =================================================================== +# System APIs +# =================================================================== + + +class TestCpuAPIs(WindowsTestCase): + + @unittest.skipIf('NUMBER_OF_PROCESSORS' not in os.environ, + 'NUMBER_OF_PROCESSORS env var is not available') + def test_cpu_count_vs_NUMBER_OF_PROCESSORS(self): + # Will likely fail on many-cores systems: + # https://stackoverflow.com/questions/31209256 + num_cpus = int(os.environ['NUMBER_OF_PROCESSORS']) + self.assertEqual(num_cpus, psutil.cpu_count()) + + def test_cpu_count_vs_GetSystemInfo(self): + # Will likely fail on many-cores systems: + # https://stackoverflow.com/questions/31209256 + sys_value = win32api.GetSystemInfo()[5] + psutil_value = psutil.cpu_count() + self.assertEqual(sys_value, psutil_value) + + def test_cpu_count_logical_vs_wmi(self): + w = wmi.WMI() + procs = sum(proc.NumberOfLogicalProcessors + for proc in w.Win32_Processor()) + self.assertEqual(psutil.cpu_count(), procs) + + def test_cpu_count_cores_vs_wmi(self): + w = wmi.WMI() + cores = sum(proc.NumberOfCores for proc in w.Win32_Processor()) + self.assertEqual(psutil.cpu_count(logical=False), cores) + + def test_cpu_count_vs_cpu_times(self): + self.assertEqual(psutil.cpu_count(), + len(psutil.cpu_times(percpu=True))) + + def test_cpu_freq(self): + w = wmi.WMI() + proc = w.Win32_Processor()[0] + self.assertEqual(proc.CurrentClockSpeed, psutil.cpu_freq().current) + self.assertEqual(proc.MaxClockSpeed, psutil.cpu_freq().max) + + +class TestSystemAPIs(WindowsTestCase): + + def test_nic_names(self): + out = sh('ipconfig /all') + 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: + raise self.fail( + "%r nic wasn't found in 'ipconfig /all' output" % nic) + + def test_total_phymem(self): + w = wmi.WMI().Win32_ComputerSystem()[0] + self.assertEqual(int(w.TotalPhysicalMemory), + psutil.virtual_memory().total) + + def test_free_phymem(self): + w = wmi.WMI().Win32_PerfRawData_PerfOS_Memory()[0] + self.assertAlmostEqual( + int(w.AvailableBytes), psutil.virtual_memory().free, + delta=TOLERANCE_SYS_MEM) + + def test_total_swapmem(self): + w = wmi.WMI().Win32_PerfRawData_PerfOS_Memory()[0] + self.assertEqual(int(w.CommitLimit) - psutil.virtual_memory().total, + psutil.swap_memory().total) + if (psutil.swap_memory().total == 0): + self.assertEqual(0, psutil.swap_memory().free) + self.assertEqual(0, psutil.swap_memory().used) + + def test_percent_swapmem(self): + if (psutil.swap_memory().total > 0): + w = wmi.WMI().Win32_PerfRawData_PerfOS_PagingFile( + Name="_Total")[0] + # calculate swap usage to percent + percentSwap = int(w.PercentUsage) * 100 / int(w.PercentUsage_Base) + # exact percent may change but should be reasonable + # assert within +/- 5% and between 0 and 100% + self.assertGreaterEqual(psutil.swap_memory().percent, 0) + self.assertAlmostEqual(psutil.swap_memory().percent, percentSwap, + delta=5) + self.assertLessEqual(psutil.swap_memory().percent, 100) + + # @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(APPVEYOR, "test not relieable on appveyor") + @retry_on_failure() + 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) + + @retry_on_failure() + 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 + if 'cdrom' in ps_part.opts: + break + if ps_part.mountpoint.startswith('A:'): + break # floppy + try: + usage = psutil.disk_usage(ps_part.mountpoint) + except FileNotFoundError: + # usually this is the floppy + break + self.assertEqual(usage.total, int(wmi_part.Size)) + wmi_free = int(wmi_part.FreeSpace) + self.assertEqual(usage.free, wmi_free) + # 10 MB tolerance + if abs(usage.free - wmi_free) > 10 * 1024 * 1024: + raise self.fail("psutil=%s, wmi=%s" % ( + usage.free, wmi_free)) + break + else: + raise self.fail("can't find partition %s" % repr(ps_part)) + + @retry_on_failure() + def test_disk_usage(self): + for disk in psutil.disk_partitions(): + if 'cdrom' in disk.opts: + continue + sys_value = win32api.GetDiskFreeSpaceEx(disk.mountpoint) + psutil_value = psutil.disk_usage(disk.mountpoint) + self.assertAlmostEqual(sys_value[0], psutil_value.free, + delta=TOLERANCE_DISK_USAGE) + self.assertAlmostEqual(sys_value[1], psutil_value.total, + delta=TOLERANCE_DISK_USAGE) + self.assertEqual(psutil_value.used, + psutil_value.total - psutil_value.free) + + def test_disk_partitions(self): + sys_value = [ + x + '\\' for x in win32api.GetLogicalDriveStrings().split("\\\x00") + if x and not x.startswith('A:')] + psutil_value = [x.mountpoint for x in psutil.disk_partitions(all=True) + if not x.mountpoint.startswith('A:')] + self.assertEqual(sys_value, psutil_value) + + def test_net_if_stats(self): + ps_names = set(cext.net_if_stats()) + wmi_adapters = wmi.WMI().Win32_NetworkAdapter() + wmi_names = set() + for wmi_adapter in wmi_adapters: + wmi_names.add(wmi_adapter.Name) + wmi_names.add(wmi_adapter.NetConnectionID) + self.assertTrue(ps_names & wmi_names, + "no common entries in %s, %s" % (ps_names, wmi_names)) + + def test_boot_time(self): + wmi_os = wmi.WMI().Win32_OperatingSystem() + wmi_btime_str = wmi_os[0].LastBootUpTime.split('.')[0] + wmi_btime_dt = datetime.datetime.strptime( + wmi_btime_str, "%Y%m%d%H%M%S") + psutil_dt = datetime.datetime.fromtimestamp(psutil.boot_time()) + diff = abs((wmi_btime_dt - psutil_dt).total_seconds()) + self.assertLessEqual(diff, 5) + + def test_boot_time_fluctuation(self): + # https://github.com/giampaolo/psutil/issues/1007 + with mock.patch('psutil._pswindows.cext.boot_time', return_value=5): + self.assertEqual(psutil.boot_time(), 5) + with mock.patch('psutil._pswindows.cext.boot_time', return_value=4): + self.assertEqual(psutil.boot_time(), 5) + with mock.patch('psutil._pswindows.cext.boot_time', return_value=6): + self.assertEqual(psutil.boot_time(), 5) + with mock.patch('psutil._pswindows.cext.boot_time', return_value=333): + self.assertEqual(psutil.boot_time(), 333) + + +# =================================================================== +# sensors_battery() +# =================================================================== + + +class TestSensorsBattery(WindowsTestCase): + + def test_has_battery(self): + if win32api.GetPwrCapabilities()['SystemBatteriesPresent']: + self.assertIsNotNone(psutil.sensors_battery()) + else: + self.assertIsNone(psutil.sensors_battery()) + + @unittest.skipIf(not HAS_BATTERY, "no battery") + def test_percent(self): + w = wmi.WMI() + battery_wmi = w.query('select * from Win32_Battery')[0] + battery_psutil = psutil.sensors_battery() + self.assertAlmostEqual( + battery_psutil.percent, battery_wmi.EstimatedChargeRemaining, + delta=1) + + @unittest.skipIf(not HAS_BATTERY, "no battery") + def test_power_plugged(self): + w = wmi.WMI() + battery_wmi = w.query('select * from Win32_Battery')[0] + battery_psutil = psutil.sensors_battery() + # Status codes: + # https://msdn.microsoft.com/en-us/library/aa394074(v=vs.85).aspx + self.assertEqual(battery_psutil.power_plugged, + battery_wmi.BatteryStatus == 2) + + def test_emulate_no_battery(self): + with mock.patch("psutil._pswindows.cext.sensors_battery", + return_value=(0, 128, 0, 0)) as m: + self.assertIsNone(psutil.sensors_battery()) + assert m.called + + def test_emulate_power_connected(self): + with mock.patch("psutil._pswindows.cext.sensors_battery", + return_value=(1, 0, 0, 0)) as m: + self.assertEqual(psutil.sensors_battery().secsleft, + psutil.POWER_TIME_UNLIMITED) + assert m.called + + def test_emulate_power_charging(self): + with mock.patch("psutil._pswindows.cext.sensors_battery", + return_value=(0, 8, 0, 0)) as m: + self.assertEqual(psutil.sensors_battery().secsleft, + psutil.POWER_TIME_UNLIMITED) + assert m.called + + def test_emulate_secs_left_unknown(self): + with mock.patch("psutil._pswindows.cext.sensors_battery", + return_value=(0, 0, 0, -1)) as m: + self.assertEqual(psutil.sensors_battery().secsleft, + psutil.POWER_TIME_UNKNOWN) + assert m.called + + +# =================================================================== +# Process APIs +# =================================================================== + + +class TestProcess(WindowsTestCase): + + @classmethod + def setUpClass(cls): + cls.pid = spawn_testproc().pid + + @classmethod + def tearDownClass(cls): + terminate(cls.pid) + + 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()[:2] + 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_num_handles_increment(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) + + def test_ctrl_signals(self): + p = psutil.Process(self.spawn_testproc().pid) + p.send_signal(signal.CTRL_C_EVENT) + p.send_signal(signal.CTRL_BREAK_EVENT) + p.kill() + p.wait() + self.assertRaises(psutil.NoSuchProcess, + p.send_signal, signal.CTRL_C_EVENT) + self.assertRaises(psutil.NoSuchProcess, + p.send_signal, signal.CTRL_BREAK_EVENT) + + def test_username(self): + name = win32api.GetUserNameEx(win32con.NameSamCompatible) + if name.endswith('$'): + # When running as a service account (most likely to be + # NetworkService), these user name calculations don't produce the + # same result, causing the test to fail. + raise unittest.SkipTest('running as service account') + self.assertEqual(psutil.Process().username(), name) + + def test_cmdline(self): + sys_value = re.sub('[ ]+', ' ', win32api.GetCommandLine()).strip() + psutil_value = ' '.join(psutil.Process().cmdline()) + if sys_value[0] == '"' != psutil_value[0]: + # The PyWin32 command line may retain quotes around argv[0] if they + # were used unnecessarily, while psutil will omit them. So remove + # the first 2 quotes from sys_value if not in psutil_value. + # A path to an executable will not contain quotes, so this is safe. + sys_value = sys_value.replace('"', '', 2) + self.assertEqual(sys_value, psutil_value) + + # XXX - occasional failures + + # def test_cpu_times(self): + # handle = win32api.OpenProcess(win32con.PROCESS_QUERY_INFORMATION, + # win32con.FALSE, os.getpid()) + # self.addCleanup(win32api.CloseHandle, handle) + # sys_value = win32process.GetProcessTimes(handle) + # psutil_value = psutil.Process().cpu_times() + # self.assertAlmostEqual( + # psutil_value.user, sys_value['UserTime'] / 10000000.0, + # delta=0.2) + # self.assertAlmostEqual( + # psutil_value.user, sys_value['KernelTime'] / 10000000.0, + # delta=0.2) + + def test_nice(self): + handle = win32api.OpenProcess(win32con.PROCESS_QUERY_INFORMATION, + win32con.FALSE, os.getpid()) + self.addCleanup(win32api.CloseHandle, handle) + sys_value = win32process.GetPriorityClass(handle) + psutil_value = psutil.Process().nice() + self.assertEqual(psutil_value, sys_value) + + def test_memory_info(self): + handle = win32api.OpenProcess(win32con.PROCESS_QUERY_INFORMATION, + win32con.FALSE, self.pid) + self.addCleanup(win32api.CloseHandle, handle) + sys_value = win32process.GetProcessMemoryInfo(handle) + psutil_value = psutil.Process(self.pid).memory_info() + self.assertEqual( + sys_value['PeakWorkingSetSize'], psutil_value.peak_wset) + self.assertEqual( + sys_value['WorkingSetSize'], psutil_value.wset) + self.assertEqual( + sys_value['QuotaPeakPagedPoolUsage'], psutil_value.peak_paged_pool) + self.assertEqual( + sys_value['QuotaPagedPoolUsage'], psutil_value.paged_pool) + self.assertEqual( + sys_value['QuotaPeakNonPagedPoolUsage'], + psutil_value.peak_nonpaged_pool) + self.assertEqual( + sys_value['QuotaNonPagedPoolUsage'], psutil_value.nonpaged_pool) + self.assertEqual( + sys_value['PagefileUsage'], psutil_value.pagefile) + self.assertEqual( + sys_value['PeakPagefileUsage'], psutil_value.peak_pagefile) + + self.assertEqual(psutil_value.rss, psutil_value.wset) + self.assertEqual(psutil_value.vms, psutil_value.pagefile) + + def test_wait(self): + handle = win32api.OpenProcess(win32con.PROCESS_QUERY_INFORMATION, + win32con.FALSE, self.pid) + self.addCleanup(win32api.CloseHandle, handle) + p = psutil.Process(self.pid) + p.terminate() + psutil_value = p.wait() + sys_value = win32process.GetExitCodeProcess(handle) + self.assertEqual(psutil_value, sys_value) + + def test_cpu_affinity(self): + def from_bitmask(x): + return [i for i in range(64) if (1 << i) & x] + + handle = win32api.OpenProcess(win32con.PROCESS_QUERY_INFORMATION, + win32con.FALSE, self.pid) + self.addCleanup(win32api.CloseHandle, handle) + sys_value = from_bitmask( + win32process.GetProcessAffinityMask(handle)[0]) + psutil_value = psutil.Process(self.pid).cpu_affinity() + self.assertEqual(psutil_value, sys_value) + + def test_io_counters(self): + handle = win32api.OpenProcess(win32con.PROCESS_QUERY_INFORMATION, + win32con.FALSE, os.getpid()) + self.addCleanup(win32api.CloseHandle, handle) + sys_value = win32process.GetProcessIoCounters(handle) + psutil_value = psutil.Process().io_counters() + self.assertEqual( + psutil_value.read_count, sys_value['ReadOperationCount']) + self.assertEqual( + psutil_value.write_count, sys_value['WriteOperationCount']) + self.assertEqual( + psutil_value.read_bytes, sys_value['ReadTransferCount']) + self.assertEqual( + psutil_value.write_bytes, sys_value['WriteTransferCount']) + self.assertEqual( + psutil_value.other_count, sys_value['OtherOperationCount']) + self.assertEqual( + psutil_value.other_bytes, sys_value['OtherTransferCount']) + + def test_num_handles(self): + import ctypes + import ctypes.wintypes + PROCESS_QUERY_INFORMATION = 0x400 + handle = ctypes.windll.kernel32.OpenProcess( + PROCESS_QUERY_INFORMATION, 0, self.pid) + self.addCleanup(ctypes.windll.kernel32.CloseHandle, handle) + + hndcnt = ctypes.wintypes.DWORD() + ctypes.windll.kernel32.GetProcessHandleCount( + handle, ctypes.byref(hndcnt)) + sys_value = hndcnt.value + psutil_value = psutil.Process(self.pid).num_handles() + self.assertEqual(psutil_value, sys_value) + + def test_error_partial_copy(self): + # https://github.com/giampaolo/psutil/issues/875 + exc = WindowsError() + exc.winerror = 299 + with mock.patch("psutil._psplatform.cext.proc_cwd", side_effect=exc): + with mock.patch("time.sleep") as m: + p = psutil.Process() + self.assertRaises(psutil.AccessDenied, p.cwd) + self.assertGreaterEqual(m.call_count, 5) + + def test_exe(self): + # NtQuerySystemInformation succeeds if process is gone. Make sure + # it raises NSP for a non existent pid. + pid = psutil.pids()[-1] + 99999 + proc = psutil._psplatform.Process(pid) + self.assertRaises(psutil.NoSuchProcess, proc.exe) + + +class TestProcessWMI(WindowsTestCase): + """Compare Process API results with WMI.""" + + @classmethod + def setUpClass(cls): + cls.pid = spawn_testproc().pid + + @classmethod + def tearDownClass(cls): + terminate(cls.pid) + + def test_name(self): + w = wmi.WMI().Win32_Process(ProcessId=self.pid)[0] + p = psutil.Process(self.pid) + self.assertEqual(p.name(), w.Caption) + + # This fail on github because using virtualenv for test environment + @unittest.skipIf(GITHUB_ACTIONS, "unreliable path on GITHUB_ACTIONS") + def test_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()) + + def test_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_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) + + @retry_on_failure() + def test_memory_rss(self): + 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)) + + @retry_on_failure() + def test_memory_vms(self): + 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): + raise self.fail("wmi=%s, psutil=%s" % (wmi_usage, vms)) + + def test_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) + + +# --- + + +@unittest.skipIf(not WINDOWS, "WINDOWS only") +class TestDualProcessImplementation(PsutilTestCase): + """ + 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 + """ + + @classmethod + def setUpClass(cls): + cls.pid = spawn_testproc().pid + + @classmethod + def tearDownClass(cls): + terminate(cls.pid) + + def test_memory_info(self): + mem_1 = psutil.Process(self.pid).memory_info() + with mock.patch("psutil._psplatform.cext.proc_memory_info", + side_effect=OSError(errno.EPERM, "msg")) as fun: + mem_2 = psutil.Process(self.pid).memory_info() + self.assertEqual(len(mem_1), len(mem_2)) + for i in range(len(mem_1)): + self.assertGreaterEqual(mem_1[i], 0) + self.assertGreaterEqual(mem_2[i], 0) + self.assertAlmostEqual(mem_1[i], mem_2[i], delta=512) + assert fun.called + + def test_create_time(self): + ctime = psutil.Process(self.pid).create_time() + with mock.patch("psutil._psplatform.cext.proc_times", + side_effect=OSError(errno.EPERM, "msg")) as fun: + self.assertEqual(psutil.Process(self.pid).create_time(), ctime) + assert fun.called + + def test_cpu_times(self): + cpu_times_1 = psutil.Process(self.pid).cpu_times() + with mock.patch("psutil._psplatform.cext.proc_times", + side_effect=OSError(errno.EPERM, "msg")) as fun: + cpu_times_2 = psutil.Process(self.pid).cpu_times() + assert fun.called + self.assertAlmostEqual( + cpu_times_1.user, cpu_times_2.user, delta=0.01) + self.assertAlmostEqual( + cpu_times_1.system, cpu_times_2.system, delta=0.01) + + def test_io_counters(self): + io_counters_1 = psutil.Process(self.pid).io_counters() + with mock.patch("psutil._psplatform.cext.proc_io_counters", + side_effect=OSError(errno.EPERM, "msg")) as fun: + io_counters_2 = psutil.Process(self.pid).io_counters() + for i in range(len(io_counters_1)): + self.assertAlmostEqual( + io_counters_1[i], io_counters_2[i], delta=5) + assert fun.called + + def test_num_handles(self): + num_handles = psutil.Process(self.pid).num_handles() + with mock.patch("psutil._psplatform.cext.proc_num_handles", + side_effect=OSError(errno.EPERM, "msg")) as fun: + self.assertEqual(psutil.Process(self.pid).num_handles(), + num_handles) + assert fun.called + + def test_cmdline(self): + for pid in psutil.pids(): + try: + a = cext.proc_cmdline(pid, use_peb=True) + b = cext.proc_cmdline(pid, use_peb=False) + except OSError as err: + err = convert_oserror(err) + if not isinstance(err, (psutil.AccessDenied, + psutil.NoSuchProcess)): + raise + else: + self.assertEqual(a, b) + + +@unittest.skipIf(not WINDOWS, "WINDOWS only") +class RemoteProcessTestCase(PsutilTestCase): + """Certain functions require calling ReadProcessMemory. + This trivially works when called on the current process. + Check that this works on other processes, especially when they + have a different bitness. + """ + + @staticmethod + def find_other_interpreter(): + # find a python interpreter that is of the opposite bitness from us + code = "import sys; sys.stdout.write(str(sys.maxsize > 2**32))" + + # XXX: a different and probably more stable approach might be to access + # the registry but accessing 64 bit paths from a 32 bit process + for filename in glob.glob(r"C:\Python*\python.exe"): + proc = subprocess.Popen(args=[filename, "-c", code], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT) + output, _ = proc.communicate() + proc.wait() + if output == str(not IS_64BIT): + return filename + + test_args = ["-c", "import sys; sys.stdin.read()"] + + def setUp(self): + super().setUp() + + other_python = self.find_other_interpreter() + if other_python is None: + raise unittest.SkipTest( + "could not find interpreter with opposite bitness") + if IS_64BIT: + self.python64 = sys.executable + self.python32 = other_python + else: + self.python64 = other_python + self.python32 = sys.executable + + env = os.environ.copy() + env["THINK_OF_A_NUMBER"] = str(os.getpid()) + self.proc32 = self.spawn_testproc( + [self.python32] + self.test_args, + env=env, + stdin=subprocess.PIPE) + self.proc64 = self.spawn_testproc( + [self.python64] + self.test_args, + env=env, + stdin=subprocess.PIPE) + + def tearDown(self): + super().tearDown() + self.proc32.communicate() + self.proc64.communicate() + + def test_cmdline_32(self): + p = psutil.Process(self.proc32.pid) + self.assertEqual(len(p.cmdline()), 3) + self.assertEqual(p.cmdline()[1:], self.test_args) + + def test_cmdline_64(self): + p = psutil.Process(self.proc64.pid) + self.assertEqual(len(p.cmdline()), 3) + self.assertEqual(p.cmdline()[1:], self.test_args) + + def test_cwd_32(self): + p = psutil.Process(self.proc32.pid) + self.assertEqual(p.cwd(), os.getcwd()) + + def test_cwd_64(self): + p = psutil.Process(self.proc64.pid) + self.assertEqual(p.cwd(), os.getcwd()) + + def test_environ_32(self): + p = psutil.Process(self.proc32.pid) + e = p.environ() + self.assertIn("THINK_OF_A_NUMBER", e) + self.assertEqual(e["THINK_OF_A_NUMBER"], str(os.getpid())) + + def test_environ_64(self): + p = psutil.Process(self.proc64.pid) + try: + p.environ() + except psutil.AccessDenied: + pass + + +# =================================================================== +# Windows services +# =================================================================== + + +@unittest.skipIf(not WINDOWS, "WINDOWS only") +class TestServices(PsutilTestCase): + + def test_win_service_iter(self): + valid_statuses = set([ + "running", + "paused", + "start", + "pause", + "continue", + "stop", + "stopped", + ]) + valid_start_types = set([ + "automatic", + "manual", + "disabled", + ]) + valid_statuses = set([ + "running", + "paused", + "start_pending", + "pause_pending", + "continue_pending", + "stop_pending", + "stopped" + ]) + for serv in psutil.win_service_iter(): + data = serv.as_dict() + self.assertIsInstance(data['name'], str) + self.assertNotEqual(data['name'].strip(), "") + self.assertIsInstance(data['display_name'], str) + self.assertIsInstance(data['username'], str) + self.assertIn(data['status'], valid_statuses) + if data['pid'] is not None: + psutil.Process(data['pid']) + self.assertIsInstance(data['binpath'], str) + self.assertIsInstance(data['username'], str) + self.assertIsInstance(data['start_type'], str) + self.assertIn(data['start_type'], valid_start_types) + self.assertIn(data['status'], valid_statuses) + self.assertIsInstance(data['description'], str) + pid = serv.pid() + if pid is not None: + p = psutil.Process(pid) + self.assertTrue(p.is_running()) + # win_service_get + s = psutil.win_service_get(serv.name()) + # test __eq__ + self.assertEqual(serv, s) + + def test_win_service_get(self): + ERROR_SERVICE_DOES_NOT_EXIST = \ + psutil._psplatform.cext.ERROR_SERVICE_DOES_NOT_EXIST + ERROR_ACCESS_DENIED = psutil._psplatform.cext.ERROR_ACCESS_DENIED + + name = next(psutil.win_service_iter()).name() + with self.assertRaises(psutil.NoSuchProcess) as cm: + psutil.win_service_get(name + '???') + self.assertEqual(cm.exception.name, name + '???') + + # test NoSuchProcess + service = psutil.win_service_get(name) + if PY3: + args = (0, "msg", 0, ERROR_SERVICE_DOES_NOT_EXIST) + else: + args = (ERROR_SERVICE_DOES_NOT_EXIST, "msg") + exc = WindowsError(*args) + with mock.patch("psutil._psplatform.cext.winservice_query_status", + side_effect=exc): + self.assertRaises(psutil.NoSuchProcess, service.status) + with mock.patch("psutil._psplatform.cext.winservice_query_config", + side_effect=exc): + self.assertRaises(psutil.NoSuchProcess, service.username) + + # test AccessDenied + if PY3: + args = (0, "msg", 0, ERROR_ACCESS_DENIED) + else: + args = (ERROR_ACCESS_DENIED, "msg") + exc = WindowsError(*args) + with mock.patch("psutil._psplatform.cext.winservice_query_status", + side_effect=exc): + self.assertRaises(psutil.AccessDenied, service.status) + with mock.patch("psutil._psplatform.cext.winservice_query_config", + side_effect=exc): + self.assertRaises(psutil.AccessDenied, service.username) + + # test __str__ and __repr__ + self.assertIn(service.name(), str(service)) + self.assertIn(service.display_name(), str(service)) + self.assertIn(service.name(), repr(service)) + self.assertIn(service.display_name(), repr(service)) + + +if __name__ == '__main__': + from psutil.tests.runner import run_from_name + run_from_name(__file__) |