diff --git a/MANIFEST.in b/MANIFEST.in
index 14ede9455..bb60aa849 100644
--- a/MANIFEST.in
+++ b/MANIFEST.in
@@ -164,6 +164,7 @@ include psutil/tests/test_misc.py
include psutil/tests/test_osx.py
include psutil/tests/test_posix.py
include psutil/tests/test_process.py
+include psutil/tests/test_process_all.py
include psutil/tests/test_sunos.py
include psutil/tests/test_system.py
include psutil/tests/test_testutils.py
diff --git a/Makefile b/Makefile
index 6b6c3f32e..34279c90a 100644
--- a/Makefile
+++ b/Makefile
@@ -140,6 +140,10 @@ test-process: ## Run process-related API tests.
${MAKE} build
$(TEST_PREFIX) $(PYTHON) $(TSCRIPT) $(ARGS) psutil/tests/test_process.py
+test-process-all: ## Run tests which iterate over all process PIDs.
+ ${MAKE} build
+ $(TEST_PREFIX) $(PYTHON) $(TSCRIPT) $(ARGS) psutil/tests/test_process_all.py
+
test-system: ## Run system-related API tests.
${MAKE} build
$(TEST_PREFIX) $(PYTHON) $(TSCRIPT) $(ARGS) psutil/tests/test_system.py
diff --git a/psutil/tests/__init__.py b/psutil/tests/__init__.py
index 05f0d63b8..df30e3069 100644
--- a/psutil/tests/__init__.py
+++ b/psutil/tests/__init__.py
@@ -47,6 +47,7 @@
from psutil import SUNOS
from psutil import WINDOWS
from psutil._common import bytes2human
+from psutil._common import debug
from psutil._common import memoize
from psutil._common import print_color
from psutil._common import supports_ipv6
@@ -105,7 +106,7 @@
# sync primitives
'call_until', 'wait_for_pid', 'wait_for_file',
# network
- 'check_net_address',
+ 'check_net_address', 'filter_proc_connections',
'get_free_port', 'bind_socket', 'bind_unix_socket', 'tcp_socketpair',
'unix_socketpair', 'create_sockets',
# compat
@@ -1871,6 +1872,20 @@ def check_status(conn):
check_status(conn)
+def filter_proc_connections(cons):
+ """Our process may start with some open UNIX sockets which are not
+ initialized by us, invalidating unit tests.
+ """
+ new = []
+ for conn in cons:
+ if POSIX and conn.family == socket.AF_UNIX:
+ if MACOS and "/syslog" in conn.raddr:
+ debug("skipping %s" % str(conn))
+ continue
+ new.append(conn)
+ return new
+
+
# ===================================================================
# --- compatibility
# ===================================================================
diff --git a/psutil/tests/test_connections.py b/psutil/tests/test_connections.py
index 082bf981a..e261fc0f9 100755
--- a/psutil/tests/test_connections.py
+++ b/psutil/tests/test_connections.py
@@ -35,6 +35,7 @@
from psutil.tests import bind_unix_socket
from psutil.tests import check_connection_ntuple
from psutil.tests import create_sockets
+from psutil.tests import filter_proc_connections
from psutil.tests import reap_children
from psutil.tests import retry_on_failure
from psutil.tests import serialrun
@@ -44,26 +45,24 @@
from psutil.tests import wait_for_file
-thisproc = psutil.Process()
SOCK_SEQPACKET = getattr(socket, "SOCK_SEQPACKET", object())
+def this_proc_connections(kind):
+ cons = psutil.Process().connections(kind=kind)
+ if kind in ("all", "unix"):
+ return filter_proc_connections(cons)
+ return cons
+
+
@serialrun
class ConnectionTestCase(PsutilTestCase):
def setUp(self):
- if NETBSD or FREEBSD or (MACOS and not PY3):
- # Process opens a UNIX socket to /var/log/run.
- return
- cons = thisproc.connections(kind='all')
- self.assertEqual(cons, [])
+ self.assertEqual(this_proc_connections(kind='all'), [])
def tearDown(self):
# Make sure we closed all resources.
- # Some BSDs open a UNIX socket to /var/log/run.
- if NETBSD or FREEBSD or (MACOS and not PY3):
- return
- cons = thisproc.connections(kind='all')
- self.assertEqual(cons, [])
+ self.assertEqual(this_proc_connections(kind='all'), [])
def compare_procsys_connections(self, pid, proc_cons, kind='all'):
"""Given a process PID and its list of connections compare
@@ -95,11 +94,11 @@ def test_system(self):
def test_process(self):
with create_sockets():
- for conn in psutil.Process().connections(kind='all'):
+ for conn in this_proc_connections(kind='all'):
check_connection_ntuple(conn)
def test_invalid_kind(self):
- self.assertRaises(ValueError, thisproc.connections, kind='???')
+ self.assertRaises(ValueError, this_proc_connections, kind='???')
self.assertRaises(ValueError, psutil.net_connections, kind='???')
@@ -108,7 +107,7 @@ class TestUnconnectedSockets(ConnectionTestCase):
"""Tests sockets which are open but not connected to anything."""
def get_conn_from_sock(self, sock):
- cons = thisproc.connections(kind='all')
+ cons = this_proc_connections(kind='all')
smap = dict([(c.fd, c) for c in cons])
if NETBSD or FREEBSD:
# NetBSD opens a UNIX socket to /var/log/run
@@ -148,7 +147,7 @@ def check_socket(self, sock):
# XXX Solaris can't retrieve system-wide UNIX sockets
if sock.family == AF_UNIX and HAS_CONNECTIONS_UNIX:
- cons = thisproc.connections(kind='all')
+ cons = this_proc_connections(kind='all')
self.compare_procsys_connections(os.getpid(), cons, kind='all')
return conn
@@ -210,17 +209,17 @@ class TestConnectedSocket(ConnectionTestCase):
@unittest.skipIf(SUNOS, "unreliable on SUONS")
def test_tcp(self):
addr = ("127.0.0.1", 0)
- self.assertEqual(thisproc.connections(kind='tcp4'), [])
+ self.assertEqual(this_proc_connections(kind='tcp4'), [])
server, client = tcp_socketpair(AF_INET, addr=addr)
try:
- cons = thisproc.connections(kind='tcp4')
+ cons = this_proc_connections(kind='tcp4')
self.assertEqual(len(cons), 2)
self.assertEqual(cons[0].status, psutil.CONN_ESTABLISHED)
self.assertEqual(cons[1].status, psutil.CONN_ESTABLISHED)
# May not be fast enough to change state so it stays
# commenteed.
# client.close()
- # cons = thisproc.connections(kind='all')
+ # cons = this_proc_connections(kind='all')
# self.assertEqual(len(cons), 1)
# self.assertEqual(cons[0].status, psutil.CONN_CLOSE_WAIT)
finally:
@@ -232,7 +231,7 @@ def test_unix(self):
testfn = self.get_testfn()
server, client = unix_socketpair(testfn)
try:
- cons = thisproc.connections(kind='unix')
+ cons = this_proc_connections(kind='unix')
assert not (cons[0].laddr and cons[0].raddr), cons
assert not (cons[1].laddr and cons[1].raddr), cons
if NETBSD or FREEBSD:
@@ -258,7 +257,7 @@ def test_unix(self):
class TestFilters(ConnectionTestCase):
def test_filters(self):
def check(kind, families, types):
- for conn in thisproc.connections(kind=kind):
+ for conn in this_proc_connections(kind=kind):
self.assertIn(conn.family, families)
self.assertIn(conn.type, types)
if not SKIP_SYSCONS:
@@ -373,7 +372,7 @@ def check_conn(proc, conn, family, type, laddr, raddr, status, kinds):
tcp6_addr = None
udp6_addr = None
- for p in thisproc.children():
+ for p in psutil.Process().children():
cons = p.connections()
self.assertEqual(len(cons), 1)
for conn in cons:
@@ -429,48 +428,48 @@ def check_conn(proc, conn, family, type, laddr, raddr, status, kinds):
def test_count(self):
with create_sockets():
# tcp
- cons = thisproc.connections(kind='tcp')
+ cons = this_proc_connections(kind='tcp')
self.assertEqual(len(cons), 2 if supports_ipv6() else 1)
for conn in cons:
self.assertIn(conn.family, (AF_INET, AF_INET6))
self.assertEqual(conn.type, SOCK_STREAM)
# tcp4
- cons = thisproc.connections(kind='tcp4')
+ cons = this_proc_connections(kind='tcp4')
self.assertEqual(len(cons), 1)
self.assertEqual(cons[0].family, AF_INET)
self.assertEqual(cons[0].type, SOCK_STREAM)
# tcp6
if supports_ipv6():
- cons = thisproc.connections(kind='tcp6')
+ cons = this_proc_connections(kind='tcp6')
self.assertEqual(len(cons), 1)
self.assertEqual(cons[0].family, AF_INET6)
self.assertEqual(cons[0].type, SOCK_STREAM)
# udp
- cons = thisproc.connections(kind='udp')
+ cons = this_proc_connections(kind='udp')
self.assertEqual(len(cons), 2 if supports_ipv6() else 1)
for conn in cons:
self.assertIn(conn.family, (AF_INET, AF_INET6))
self.assertEqual(conn.type, SOCK_DGRAM)
# udp4
- cons = thisproc.connections(kind='udp4')
+ cons = this_proc_connections(kind='udp4')
self.assertEqual(len(cons), 1)
self.assertEqual(cons[0].family, AF_INET)
self.assertEqual(cons[0].type, SOCK_DGRAM)
# udp6
if supports_ipv6():
- cons = thisproc.connections(kind='udp6')
+ cons = this_proc_connections(kind='udp6')
self.assertEqual(len(cons), 1)
self.assertEqual(cons[0].family, AF_INET6)
self.assertEqual(cons[0].type, SOCK_DGRAM)
# inet
- cons = thisproc.connections(kind='inet')
+ cons = this_proc_connections(kind='inet')
self.assertEqual(len(cons), 4 if supports_ipv6() else 2)
for conn in cons:
self.assertIn(conn.family, (AF_INET, AF_INET6))
self.assertIn(conn.type, (SOCK_STREAM, SOCK_DGRAM))
# inet6
if supports_ipv6():
- cons = thisproc.connections(kind='inet6')
+ cons = this_proc_connections(kind='inet6')
self.assertEqual(len(cons), 2)
for conn in cons:
self.assertEqual(conn.family, AF_INET6)
@@ -478,7 +477,7 @@ def test_count(self):
# Skipped on BSD becayse by default the Python process
# creates a UNIX socket to '/var/run/log'.
if HAS_CONNECTIONS_UNIX and not (FREEBSD or NETBSD):
- cons = thisproc.connections(kind='unix')
+ cons = this_proc_connections(kind='unix')
self.assertEqual(len(cons), 3)
for conn in cons:
self.assertEqual(conn.family, AF_UNIX)
diff --git a/psutil/tests/test_contracts.py b/psutil/tests/test_contracts.py
index 9dc54f864..4736f5f1a 100755
--- a/psutil/tests/test_contracts.py
+++ b/psutil/tests/test_contracts.py
@@ -9,34 +9,21 @@
Some of these are duplicates of tests test_system.py and test_process.py.
"""
-import errno
-import multiprocessing
-import os
import platform
import signal
-import stat
-import time
-import traceback
import unittest
import psutil
from psutil import AIX
-from psutil import BSD
from psutil import FREEBSD
from psutil import LINUX
from psutil import MACOS
from psutil import NETBSD
from psutil import OPENBSD
-from psutil import OSX
from psutil import POSIX
from psutil import SUNOS
from psutil import WINDOWS
-from psutil._compat import PY3
-from psutil._compat import FileNotFoundError
from psutil._compat import long
-from psutil._compat import range
-from psutil._compat import unicode
-from psutil.tests import CI_TESTING
from psutil.tests import GITHUB_ACTIONS
from psutil.tests import HAS_CPU_FREQ
from psutil.tests import HAS_NET_IO_COUNTERS
@@ -44,16 +31,11 @@
from psutil.tests import HAS_SENSORS_TEMPERATURES
from psutil.tests import PYPY
from psutil.tests import SKIP_SYSCONS
-from psutil.tests import VALID_PROC_STATUSES
from psutil.tests import PsutilTestCase
-from psutil.tests import check_connection_ntuple
from psutil.tests import create_sockets
from psutil.tests import enum
from psutil.tests import is_namedtuple
-from psutil.tests import is_win_secure_system_proc
from psutil.tests import kernel_version
-from psutil.tests import process_namespace
-from psutil.tests import serialrun
# ===================================================================
@@ -361,425 +343,6 @@ def test_negative_signal(self):
self.assertIsInstance(code, int)
-# ===================================================================
-# --- Featch all processes test
-# ===================================================================
-
-
-def proc_info(pid):
- tcase = PsutilTestCase()
-
- def check_exception(exc, proc, name, ppid):
- tcase.assertEqual(exc.pid, pid)
- if exc.name is not None:
- tcase.assertEqual(exc.name, name)
- if isinstance(exc, psutil.ZombieProcess):
- tcase.assertProcessZombie(proc)
- if exc.ppid is not None:
- tcase.assertGreaterEqual(exc.ppid, 0)
- tcase.assertEqual(exc.ppid, ppid)
- elif isinstance(exc, psutil.NoSuchProcess):
- tcase.assertProcessGone(proc)
- str(exc)
- repr(exc)
-
- def do_wait():
- if pid != 0:
- try:
- proc.wait(0)
- except psutil.Error as exc:
- check_exception(exc, proc, name, ppid)
-
- try:
- proc = psutil.Process(pid)
- except psutil.NoSuchProcess:
- tcase.assertPidGone(pid)
- return {}
- try:
- d = proc.as_dict(['ppid', 'name'])
- except psutil.NoSuchProcess:
- tcase.assertProcessGone(proc)
- else:
- name, ppid = d['name'], d['ppid']
- info = {'pid': proc.pid}
- ns = process_namespace(proc)
- # We don't use oneshot() because in order not to fool
- # check_exception() in case of NSP.
- for fun, fun_name in ns.iter(ns.getters, clear_cache=False):
- try:
- info[fun_name] = fun()
- except psutil.Error as exc:
- check_exception(exc, proc, name, ppid)
- continue
- do_wait()
- return info
-
-
-@serialrun
-class TestFetchAllProcesses(PsutilTestCase):
- """Test which iterates over all running processes and performs
- some sanity checks against Process API's returned values.
- Uses a process pool to get info about all processes.
- """
-
- use_proc_pool = 0
-
- def setUp(self):
- # Using a pool in a CI env may result in deadlock, see:
- # https://github.com/giampaolo/psutil/issues/2104
- if self.use_proc_pool:
- self.pool = multiprocessing.Pool()
-
- def tearDown(self):
- if self.use_proc_pool:
- self.pool.terminate()
- self.pool.join()
-
- def iter_proc_info(self):
- # Fixes "can't pickle <function proc_info>: it's not the
- # same object as test_contracts.proc_info".
- from psutil.tests.test_contracts import proc_info
-
- if self.use_proc_pool:
- return self.pool.imap_unordered(proc_info, psutil.pids())
- else:
- ls = []
- for pid in psutil.pids():
- ls.append(proc_info(pid))
- return ls
-
- def test_all(self):
- failures = []
- for info in self.iter_proc_info():
- for name, value in info.items():
- meth = getattr(self, name)
- try:
- meth(value, info)
- except Exception: # noqa: BLE001
- s = '\n' + '=' * 70 + '\n'
- s += "FAIL: name=test_%s, pid=%s, ret=%s\ninfo=%s\n" % (
- name,
- info['pid'],
- repr(value),
- info,
- )
- s += '-' * 70
- s += "\n%s" % traceback.format_exc()
- s = "\n".join((" " * 4) + i for i in s.splitlines()) + "\n"
- failures.append(s)
- else:
- if value not in (0, 0.0, [], None, '', {}):
- assert value, value
- if failures:
- raise self.fail(''.join(failures))
-
- def cmdline(self, ret, info):
- self.assertIsInstance(ret, list)
- for part in ret:
- self.assertIsInstance(part, str)
-
- def exe(self, ret, info):
- self.assertIsInstance(ret, (str, unicode))
- self.assertEqual(ret.strip(), ret)
- if ret:
- if WINDOWS and not ret.endswith('.exe'):
- return # May be "Registry", "MemCompression", ...
- assert os.path.isabs(ret), ret
- # Note: os.stat() may return False even if the file is there
- # hence we skip the test, see:
- # http://stackoverflow.com/questions/3112546/os-path-exists-lies
- if POSIX and os.path.isfile(ret):
- if hasattr(os, 'access') and hasattr(os, "X_OK"):
- # XXX: may fail on MACOS
- try:
- assert os.access(ret, os.X_OK)
- except AssertionError:
- if os.path.exists(ret) and not CI_TESTING:
- raise
-
- def pid(self, ret, info):
- self.assertIsInstance(ret, int)
- self.assertGreaterEqual(ret, 0)
-
- def ppid(self, ret, info):
- self.assertIsInstance(ret, (int, long))
- self.assertGreaterEqual(ret, 0)
- proc_info(ret)
-
- def name(self, ret, info):
- self.assertIsInstance(ret, (str, unicode))
- if WINDOWS and not ret and is_win_secure_system_proc(info['pid']):
- # https://github.com/giampaolo/psutil/issues/2338
- return
- # on AIX, "<exiting>" processes don't have names
- if not AIX:
- assert ret, repr(ret)
-
- def create_time(self, ret, info):
- self.assertIsInstance(ret, float)
- try:
- self.assertGreaterEqual(ret, 0)
- except AssertionError:
- # XXX
- if OPENBSD and info['status'] == psutil.STATUS_ZOMBIE:
- pass
- else:
- raise
- # this can't be taken for granted on all platforms
- # self.assertGreaterEqual(ret, psutil.boot_time())
- # make sure returned value can be pretty printed
- # with strftime
- time.strftime("%Y %m %d %H:%M:%S", time.localtime(ret))
-
- def uids(self, ret, info):
- assert is_namedtuple(ret)
- for uid in ret:
- self.assertIsInstance(uid, int)
- self.assertGreaterEqual(uid, 0)
-
- def gids(self, ret, info):
- assert is_namedtuple(ret)
- # note: testing all gids as above seems not to be reliable for
- # gid == 30 (nodoby); not sure why.
- for gid in ret:
- self.assertIsInstance(gid, int)
- if not MACOS and not NETBSD:
- self.assertGreaterEqual(gid, 0)
-
- def username(self, ret, info):
- self.assertIsInstance(ret, str)
- self.assertEqual(ret.strip(), ret)
- assert ret.strip()
-
- def status(self, ret, info):
- self.assertIsInstance(ret, str)
- assert ret, ret
- self.assertNotEqual(ret, '?') # XXX
- self.assertIn(ret, VALID_PROC_STATUSES)
-
- def io_counters(self, ret, info):
- assert is_namedtuple(ret)
- for field in ret:
- self.assertIsInstance(field, (int, long))
- if field != -1:
- self.assertGreaterEqual(field, 0)
-
- def ionice(self, ret, info):
- if LINUX:
- self.assertIsInstance(ret.ioclass, int)
- self.assertIsInstance(ret.value, int)
- self.assertGreaterEqual(ret.ioclass, 0)
- self.assertGreaterEqual(ret.value, 0)
- else: # Windows, Cygwin
- choices = [
- psutil.IOPRIO_VERYLOW,
- psutil.IOPRIO_LOW,
- psutil.IOPRIO_NORMAL,
- psutil.IOPRIO_HIGH,
- ]
- self.assertIsInstance(ret, int)
- self.assertGreaterEqual(ret, 0)
- self.assertIn(ret, choices)
-
- def num_threads(self, ret, info):
- self.assertIsInstance(ret, int)
- if WINDOWS and ret == 0 and is_win_secure_system_proc(info['pid']):
- # https://github.com/giampaolo/psutil/issues/2338
- return
- self.assertGreaterEqual(ret, 1)
-
- def threads(self, ret, info):
- self.assertIsInstance(ret, list)
- for t in ret:
- assert is_namedtuple(t)
- self.assertGreaterEqual(t.id, 0)
- self.assertGreaterEqual(t.user_time, 0)
- self.assertGreaterEqual(t.system_time, 0)
- for field in t:
- self.assertIsInstance(field, (int, float))
-
- def cpu_times(self, ret, info):
- assert is_namedtuple(ret)
- for n in ret:
- self.assertIsInstance(n, float)
- self.assertGreaterEqual(n, 0)
- # TODO: check ntuple fields
-
- def cpu_percent(self, ret, info):
- self.assertIsInstance(ret, float)
- assert 0.0 <= ret <= 100.0, ret
-
- def cpu_num(self, ret, info):
- self.assertIsInstance(ret, int)
- if FREEBSD and ret == -1:
- return
- self.assertGreaterEqual(ret, 0)
- if psutil.cpu_count() == 1:
- self.assertEqual(ret, 0)
- self.assertIn(ret, list(range(psutil.cpu_count())))
-
- def memory_info(self, ret, info):
- assert is_namedtuple(ret)
- for value in ret:
- self.assertIsInstance(value, (int, long))
- self.assertGreaterEqual(value, 0)
- if WINDOWS:
- self.assertGreaterEqual(ret.peak_wset, ret.wset)
- self.assertGreaterEqual(ret.peak_paged_pool, ret.paged_pool)
- self.assertGreaterEqual(ret.peak_nonpaged_pool, ret.nonpaged_pool)
- self.assertGreaterEqual(ret.peak_pagefile, ret.pagefile)
-
- def memory_full_info(self, ret, info):
- assert is_namedtuple(ret)
- total = psutil.virtual_memory().total
- for name in ret._fields:
- value = getattr(ret, name)
- self.assertIsInstance(value, (int, long))
- self.assertGreaterEqual(value, 0, msg=(name, value))
- if LINUX or OSX and name in ('vms', 'data'):
- # On Linux there are processes (e.g. 'goa-daemon') whose
- # VMS is incredibly high for some reason.
- continue
- self.assertLessEqual(value, total, msg=(name, value, total))
-
- if LINUX:
- self.assertGreaterEqual(ret.pss, ret.uss)
-
- def open_files(self, ret, info):
- self.assertIsInstance(ret, list)
- for f in ret:
- self.assertIsInstance(f.fd, int)
- self.assertIsInstance(f.path, str)
- self.assertEqual(f.path.strip(), f.path)
- if WINDOWS:
- self.assertEqual(f.fd, -1)
- elif LINUX:
- self.assertIsInstance(f.position, int)
- self.assertIsInstance(f.mode, str)
- self.assertIsInstance(f.flags, int)
- self.assertGreaterEqual(f.position, 0)
- self.assertIn(f.mode, ('r', 'w', 'a', 'r+', 'a+'))
- self.assertGreater(f.flags, 0)
- elif BSD and not f.path:
- # XXX see: https://github.com/giampaolo/psutil/issues/595
- continue
- assert os.path.isabs(f.path), f
- try:
- st = os.stat(f.path)
- except FileNotFoundError:
- pass
- else:
- assert stat.S_ISREG(st.st_mode), f
-
- def num_fds(self, ret, info):
- self.assertIsInstance(ret, int)
- self.assertGreaterEqual(ret, 0)
-
- def connections(self, ret, info):
- with create_sockets():
- self.assertEqual(len(ret), len(set(ret)))
- for conn in ret:
- assert is_namedtuple(conn)
- check_connection_ntuple(conn)
-
- def cwd(self, ret, info):
- self.assertIsInstance(ret, (str, unicode))
- self.assertEqual(ret.strip(), ret)
- if ret:
- assert os.path.isabs(ret), ret
- try:
- st = os.stat(ret)
- except OSError as err:
- if WINDOWS and psutil._psplatform.is_permission_err(err):
- pass
- # directory has been removed in mean time
- elif err.errno != errno.ENOENT:
- raise
- else:
- assert stat.S_ISDIR(st.st_mode)
-
- def memory_percent(self, ret, info):
- self.assertIsInstance(ret, float)
- assert 0 <= ret <= 100, ret
-
- def is_running(self, ret, info):
- self.assertIsInstance(ret, bool)
-
- def cpu_affinity(self, ret, info):
- self.assertIsInstance(ret, list)
- self.assertNotEqual(ret, [])
- cpus = list(range(psutil.cpu_count()))
- for n in ret:
- self.assertIsInstance(n, int)
- self.assertIn(n, cpus)
-
- def terminal(self, ret, info):
- self.assertIsInstance(ret, (str, type(None)))
- if ret is not None:
- assert os.path.isabs(ret), ret
- assert os.path.exists(ret), ret
-
- def memory_maps(self, ret, info):
- for nt in ret:
- self.assertIsInstance(nt.addr, str)
- self.assertIsInstance(nt.perms, str)
- self.assertIsInstance(nt.path, str)
- for fname in nt._fields:
- value = getattr(nt, fname)
- if fname == 'path':
- if not value.startswith(("[", "anon_inode:")):
- assert os.path.isabs(nt.path), nt.path
- # commented as on Linux we might get
- # '/foo/bar (deleted)'
- # assert os.path.exists(nt.path), nt.path
- elif fname == 'addr':
- assert value, repr(value)
- elif fname == 'perms':
- if not WINDOWS:
- assert value, repr(value)
- else:
- self.assertIsInstance(value, (int, long))
- self.assertGreaterEqual(value, 0)
-
- def num_handles(self, ret, info):
- self.assertIsInstance(ret, int)
- self.assertGreaterEqual(ret, 0)
-
- def nice(self, ret, info):
- self.assertIsInstance(ret, int)
- if POSIX:
- assert -20 <= ret <= 20, ret
- else:
- priorities = [
- getattr(psutil, x)
- for x in dir(psutil)
- if x.endswith('_PRIORITY_CLASS')
- ]
- self.assertIn(ret, priorities)
- if PY3:
- self.assertIsInstance(ret, enum.IntEnum)
- else:
- self.assertIsInstance(ret, int)
-
- def num_ctx_switches(self, ret, info):
- assert is_namedtuple(ret)
- for value in ret:
- self.assertIsInstance(value, (int, long))
- self.assertGreaterEqual(value, 0)
-
- def rlimit(self, ret, info):
- self.assertIsInstance(ret, tuple)
- self.assertEqual(len(ret), 2)
- self.assertGreaterEqual(ret[0], -1)
- self.assertGreaterEqual(ret[1], -1)
-
- def environ(self, ret, info):
- self.assertIsInstance(ret, dict)
- for k, v in ret.items():
- self.assertIsInstance(k, str)
- self.assertIsInstance(v, str)
-
-
if __name__ == '__main__':
from psutil.tests.runner import run_from_name
diff --git a/psutil/tests/test_process.py b/psutil/tests/test_process.py
index 0418fd54e..f03a22da0 100755
--- a/psutil/tests/test_process.py
+++ b/psutil/tests/test_process.py
@@ -746,9 +746,9 @@ def test_cmdline(self):
@unittest.skipIf(PYPY, "broken on PYPY")
def test_long_cmdline(self):
- testfn = self.get_testfn()
- create_exe(testfn)
- cmdline = [testfn] + (["0123456789"] * 20)
+ cmdline = [PYTHON_EXE]
+ cmdline.extend(["-v"] * 50)
+ cmdline.extend(["-c", "time.sleep(10)"])
p = self.spawn_psproc(cmdline)
if OPENBSD:
# XXX: for some reason the test process may turn into a
@@ -770,7 +770,7 @@ def test_name(self):
def test_long_name(self):
testfn = self.get_testfn(suffix="0123456789" * 2)
create_exe(testfn)
- cmdline = [testfn] + (["0123456789"] * 20)
+ cmdline = [testfn, "-c", "time.sleep(10)"]
p = self.spawn_psproc(cmdline)
if OPENBSD:
# XXX: for some reason the test process may turn into a
@@ -800,15 +800,7 @@ def test_prog_w_funky_name(self):
# https://github.com/giampaolo/psutil/issues/628
funky_path = self.get_testfn(suffix='foo bar )')
create_exe(funky_path)
- cmdline = [
- funky_path,
- "-c",
- "import time; [time.sleep(0.01) for x in range(3000)];arg1",
- "arg2",
- "",
- "arg3",
- "",
- ]
+ cmdline = [funky_path, "-c", "time.sleep(10)"]
p = self.spawn_psproc(cmdline)
self.assertEqual(p.cmdline(), cmdline)
self.assertEqual(p.name(), os.path.basename(funky_path))
diff --git a/psutil/tests/test_process_all.py b/psutil/tests/test_process_all.py
new file mode 100755
index 000000000..ee1a9b58d
--- /dev/null
+++ b/psutil/tests/test_process_all.py
@@ -0,0 +1,464 @@
+#!/usr/bin/env python3
+
+# 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.
+
+"""Iterate over all process PIDs and for each one of them invoke and
+test all psutil.Process() methods.
+"""
+
+import enum
+import errno
+import multiprocessing
+import os
+import stat
+import time
+import traceback
+
+import psutil
+from psutil import AIX
+from psutil import BSD
+from psutil import FREEBSD
+from psutil import LINUX
+from psutil import MACOS
+from psutil import NETBSD
+from psutil import OPENBSD
+from psutil import OSX
+from psutil import POSIX
+from psutil import WINDOWS
+from psutil._compat import PY3
+from psutil._compat import long
+from psutil._compat import unicode
+from psutil.tests import CI_TESTING
+from psutil.tests import VALID_PROC_STATUSES
+from psutil.tests import PsutilTestCase
+from psutil.tests import check_connection_ntuple
+from psutil.tests import create_sockets
+from psutil.tests import is_namedtuple
+from psutil.tests import is_win_secure_system_proc
+from psutil.tests import process_namespace
+from psutil.tests import serialrun
+
+
+# Cuts the time in half, but (e.g.) on macOS the process pool stays
+# alive after join() (multiprocessing bug?), messing up other tests.
+USE_PROC_POOL = LINUX and not CI_TESTING
+
+
+def proc_info(pid):
+ tcase = PsutilTestCase()
+
+ def check_exception(exc, proc, name, ppid):
+ tcase.assertEqual(exc.pid, pid)
+ if exc.name is not None:
+ tcase.assertEqual(exc.name, name)
+ if isinstance(exc, psutil.ZombieProcess):
+ tcase.assertProcessZombie(proc)
+ if exc.ppid is not None:
+ tcase.assertGreaterEqual(exc.ppid, 0)
+ tcase.assertEqual(exc.ppid, ppid)
+ elif isinstance(exc, psutil.NoSuchProcess):
+ tcase.assertProcessGone(proc)
+ str(exc)
+ repr(exc)
+
+ def do_wait():
+ if pid != 0:
+ try:
+ proc.wait(0)
+ except psutil.Error as exc:
+ check_exception(exc, proc, name, ppid)
+
+ try:
+ proc = psutil.Process(pid)
+ except psutil.NoSuchProcess:
+ tcase.assertPidGone(pid)
+ return {}
+ try:
+ d = proc.as_dict(['ppid', 'name'])
+ except psutil.NoSuchProcess:
+ tcase.assertProcessGone(proc)
+ else:
+ name, ppid = d['name'], d['ppid']
+ info = {'pid': proc.pid}
+ ns = process_namespace(proc)
+ # We don't use oneshot() because in order not to fool
+ # check_exception() in case of NSP.
+ for fun, fun_name in ns.iter(ns.getters, clear_cache=False):
+ try:
+ info[fun_name] = fun()
+ except psutil.Error as exc:
+ check_exception(exc, proc, name, ppid)
+ continue
+ do_wait()
+ return info
+
+
+@serialrun
+class TestFetchAllProcesses(PsutilTestCase):
+ """Test which iterates over all running processes and performs
+ some sanity checks against Process API's returned values.
+ Uses a process pool to get info about all processes.
+ """
+
+ def setUp(self):
+ # Using a pool in a CI env may result in deadlock, see:
+ # https://github.com/giampaolo/psutil/issues/2104
+ if USE_PROC_POOL:
+ self.pool = multiprocessing.Pool()
+
+ def tearDown(self):
+ if USE_PROC_POOL:
+ self.pool.terminate()
+ self.pool.join()
+
+ def iter_proc_info(self):
+ # Fixes "can't pickle <function proc_info>: it's not the
+ # same object as test_process_all.proc_info".
+ from psutil.tests.test_process_all import proc_info
+
+ if USE_PROC_POOL:
+ return self.pool.imap_unordered(proc_info, psutil.pids())
+ else:
+ ls = []
+ for pid in psutil.pids():
+ ls.append(proc_info(pid))
+ return ls
+
+ def test_all(self):
+ failures = []
+ for info in self.iter_proc_info():
+ for name, value in info.items():
+ meth = getattr(self, name)
+ try:
+ meth(value, info)
+ except Exception: # noqa: BLE001
+ s = '\n' + '=' * 70 + '\n'
+ s += "FAIL: name=test_%s, pid=%s, ret=%s\ninfo=%s\n" % (
+ name,
+ info['pid'],
+ repr(value),
+ info,
+ )
+ s += '-' * 70
+ s += "\n%s" % traceback.format_exc()
+ s = "\n".join((" " * 4) + i for i in s.splitlines()) + "\n"
+ failures.append(s)
+ else:
+ if value not in (0, 0.0, [], None, '', {}):
+ assert value, value
+ if failures:
+ raise self.fail(''.join(failures))
+
+ def cmdline(self, ret, info):
+ self.assertIsInstance(ret, list)
+ for part in ret:
+ self.assertIsInstance(part, str)
+
+ def exe(self, ret, info):
+ self.assertIsInstance(ret, (str, unicode))
+ self.assertEqual(ret.strip(), ret)
+ if ret:
+ if WINDOWS and not ret.endswith('.exe'):
+ return # May be "Registry", "MemCompression", ...
+ assert os.path.isabs(ret), ret
+ # Note: os.stat() may return False even if the file is there
+ # hence we skip the test, see:
+ # http://stackoverflow.com/questions/3112546/os-path-exists-lies
+ if POSIX and os.path.isfile(ret):
+ if hasattr(os, 'access') and hasattr(os, "X_OK"):
+ # XXX: may fail on MACOS
+ try:
+ assert os.access(ret, os.X_OK)
+ except AssertionError:
+ if os.path.exists(ret) and not CI_TESTING:
+ raise
+
+ def pid(self, ret, info):
+ self.assertIsInstance(ret, int)
+ self.assertGreaterEqual(ret, 0)
+
+ def ppid(self, ret, info):
+ self.assertIsInstance(ret, (int, long))
+ self.assertGreaterEqual(ret, 0)
+ proc_info(ret)
+
+ def name(self, ret, info):
+ self.assertIsInstance(ret, (str, unicode))
+ if WINDOWS and not ret and is_win_secure_system_proc(info['pid']):
+ # https://github.com/giampaolo/psutil/issues/2338
+ return
+ # on AIX, "<exiting>" processes don't have names
+ if not AIX:
+ assert ret, repr(ret)
+
+ def create_time(self, ret, info):
+ self.assertIsInstance(ret, float)
+ try:
+ self.assertGreaterEqual(ret, 0)
+ except AssertionError:
+ # XXX
+ if OPENBSD and info['status'] == psutil.STATUS_ZOMBIE:
+ pass
+ else:
+ raise
+ # this can't be taken for granted on all platforms
+ # self.assertGreaterEqual(ret, psutil.boot_time())
+ # make sure returned value can be pretty printed
+ # with strftime
+ time.strftime("%Y %m %d %H:%M:%S", time.localtime(ret))
+
+ def uids(self, ret, info):
+ assert is_namedtuple(ret)
+ for uid in ret:
+ self.assertIsInstance(uid, int)
+ self.assertGreaterEqual(uid, 0)
+
+ def gids(self, ret, info):
+ assert is_namedtuple(ret)
+ # note: testing all gids as above seems not to be reliable for
+ # gid == 30 (nodoby); not sure why.
+ for gid in ret:
+ self.assertIsInstance(gid, int)
+ if not MACOS and not NETBSD:
+ self.assertGreaterEqual(gid, 0)
+
+ def username(self, ret, info):
+ self.assertIsInstance(ret, str)
+ self.assertEqual(ret.strip(), ret)
+ assert ret.strip()
+
+ def status(self, ret, info):
+ self.assertIsInstance(ret, str)
+ assert ret, ret
+ self.assertNotEqual(ret, '?') # XXX
+ self.assertIn(ret, VALID_PROC_STATUSES)
+
+ def io_counters(self, ret, info):
+ assert is_namedtuple(ret)
+ for field in ret:
+ self.assertIsInstance(field, (int, long))
+ if field != -1:
+ self.assertGreaterEqual(field, 0)
+
+ def ionice(self, ret, info):
+ if LINUX:
+ self.assertIsInstance(ret.ioclass, int)
+ self.assertIsInstance(ret.value, int)
+ self.assertGreaterEqual(ret.ioclass, 0)
+ self.assertGreaterEqual(ret.value, 0)
+ else: # Windows, Cygwin
+ choices = [
+ psutil.IOPRIO_VERYLOW,
+ psutil.IOPRIO_LOW,
+ psutil.IOPRIO_NORMAL,
+ psutil.IOPRIO_HIGH,
+ ]
+ self.assertIsInstance(ret, int)
+ self.assertGreaterEqual(ret, 0)
+ self.assertIn(ret, choices)
+
+ def num_threads(self, ret, info):
+ self.assertIsInstance(ret, int)
+ if WINDOWS and ret == 0 and is_win_secure_system_proc(info['pid']):
+ # https://github.com/giampaolo/psutil/issues/2338
+ return
+ self.assertGreaterEqual(ret, 1)
+
+ def threads(self, ret, info):
+ self.assertIsInstance(ret, list)
+ for t in ret:
+ assert is_namedtuple(t)
+ self.assertGreaterEqual(t.id, 0)
+ self.assertGreaterEqual(t.user_time, 0)
+ self.assertGreaterEqual(t.system_time, 0)
+ for field in t:
+ self.assertIsInstance(field, (int, float))
+
+ def cpu_times(self, ret, info):
+ assert is_namedtuple(ret)
+ for n in ret:
+ self.assertIsInstance(n, float)
+ self.assertGreaterEqual(n, 0)
+ # TODO: check ntuple fields
+
+ def cpu_percent(self, ret, info):
+ self.assertIsInstance(ret, float)
+ assert 0.0 <= ret <= 100.0, ret
+
+ def cpu_num(self, ret, info):
+ self.assertIsInstance(ret, int)
+ if FREEBSD and ret == -1:
+ return
+ self.assertGreaterEqual(ret, 0)
+ if psutil.cpu_count() == 1:
+ self.assertEqual(ret, 0)
+ self.assertIn(ret, list(range(psutil.cpu_count())))
+
+ def memory_info(self, ret, info):
+ assert is_namedtuple(ret)
+ for value in ret:
+ self.assertIsInstance(value, (int, long))
+ self.assertGreaterEqual(value, 0)
+ if WINDOWS:
+ self.assertGreaterEqual(ret.peak_wset, ret.wset)
+ self.assertGreaterEqual(ret.peak_paged_pool, ret.paged_pool)
+ self.assertGreaterEqual(ret.peak_nonpaged_pool, ret.nonpaged_pool)
+ self.assertGreaterEqual(ret.peak_pagefile, ret.pagefile)
+
+ def memory_full_info(self, ret, info):
+ assert is_namedtuple(ret)
+ total = psutil.virtual_memory().total
+ for name in ret._fields:
+ value = getattr(ret, name)
+ self.assertIsInstance(value, (int, long))
+ self.assertGreaterEqual(value, 0, msg=(name, value))
+ if LINUX or OSX and name in ('vms', 'data'):
+ # On Linux there are processes (e.g. 'goa-daemon') whose
+ # VMS is incredibly high for some reason.
+ continue
+ self.assertLessEqual(value, total, msg=(name, value, total))
+
+ if LINUX:
+ self.assertGreaterEqual(ret.pss, ret.uss)
+
+ def open_files(self, ret, info):
+ self.assertIsInstance(ret, list)
+ for f in ret:
+ self.assertIsInstance(f.fd, int)
+ self.assertIsInstance(f.path, str)
+ self.assertEqual(f.path.strip(), f.path)
+ if WINDOWS:
+ self.assertEqual(f.fd, -1)
+ elif LINUX:
+ self.assertIsInstance(f.position, int)
+ self.assertIsInstance(f.mode, str)
+ self.assertIsInstance(f.flags, int)
+ self.assertGreaterEqual(f.position, 0)
+ self.assertIn(f.mode, ('r', 'w', 'a', 'r+', 'a+'))
+ self.assertGreater(f.flags, 0)
+ elif BSD and not f.path:
+ # XXX see: https://github.com/giampaolo/psutil/issues/595
+ continue
+ assert os.path.isabs(f.path), f
+ try:
+ st = os.stat(f.path)
+ except FileNotFoundError:
+ pass
+ else:
+ assert stat.S_ISREG(st.st_mode), f
+
+ def num_fds(self, ret, info):
+ self.assertIsInstance(ret, int)
+ self.assertGreaterEqual(ret, 0)
+
+ def connections(self, ret, info):
+ with create_sockets():
+ self.assertEqual(len(ret), len(set(ret)))
+ for conn in ret:
+ assert is_namedtuple(conn)
+ check_connection_ntuple(conn)
+
+ def cwd(self, ret, info):
+ self.assertIsInstance(ret, (str, unicode))
+ self.assertEqual(ret.strip(), ret)
+ if ret:
+ assert os.path.isabs(ret), ret
+ try:
+ st = os.stat(ret)
+ except OSError as err:
+ if WINDOWS and psutil._psplatform.is_permission_err(err):
+ pass
+ # directory has been removed in mean time
+ elif err.errno != errno.ENOENT:
+ raise
+ else:
+ assert stat.S_ISDIR(st.st_mode)
+
+ def memory_percent(self, ret, info):
+ self.assertIsInstance(ret, float)
+ assert 0 <= ret <= 100, ret
+
+ def is_running(self, ret, info):
+ self.assertIsInstance(ret, bool)
+
+ def cpu_affinity(self, ret, info):
+ self.assertIsInstance(ret, list)
+ self.assertNotEqual(ret, [])
+ cpus = list(range(psutil.cpu_count()))
+ for n in ret:
+ self.assertIsInstance(n, int)
+ self.assertIn(n, cpus)
+
+ def terminal(self, ret, info):
+ self.assertIsInstance(ret, (str, type(None)))
+ if ret is not None:
+ assert os.path.isabs(ret), ret
+ assert os.path.exists(ret), ret
+
+ def memory_maps(self, ret, info):
+ for nt in ret:
+ self.assertIsInstance(nt.addr, str)
+ self.assertIsInstance(nt.perms, str)
+ self.assertIsInstance(nt.path, str)
+ for fname in nt._fields:
+ value = getattr(nt, fname)
+ if fname == 'path':
+ if not value.startswith(("[", "anon_inode:")):
+ assert os.path.isabs(nt.path), nt.path
+ # commented as on Linux we might get
+ # '/foo/bar (deleted)'
+ # assert os.path.exists(nt.path), nt.path
+ elif fname == 'addr':
+ assert value, repr(value)
+ elif fname == 'perms':
+ if not WINDOWS:
+ assert value, repr(value)
+ else:
+ self.assertIsInstance(value, (int, long))
+ self.assertGreaterEqual(value, 0)
+
+ def num_handles(self, ret, info):
+ self.assertIsInstance(ret, int)
+ self.assertGreaterEqual(ret, 0)
+
+ def nice(self, ret, info):
+ self.assertIsInstance(ret, int)
+ if POSIX:
+ assert -20 <= ret <= 20, ret
+ else:
+ priorities = [
+ getattr(psutil, x)
+ for x in dir(psutil)
+ if x.endswith('_PRIORITY_CLASS')
+ ]
+ self.assertIn(ret, priorities)
+ if PY3:
+ self.assertIsInstance(ret, enum.IntEnum)
+ else:
+ self.assertIsInstance(ret, int)
+
+ def num_ctx_switches(self, ret, info):
+ assert is_namedtuple(ret)
+ for value in ret:
+ self.assertIsInstance(value, (int, long))
+ self.assertGreaterEqual(value, 0)
+
+ def rlimit(self, ret, info):
+ self.assertIsInstance(ret, tuple)
+ self.assertEqual(len(ret), 2)
+ self.assertGreaterEqual(ret[0], -1)
+ self.assertGreaterEqual(ret[1], -1)
+
+ def environ(self, ret, info):
+ self.assertIsInstance(ret, dict)
+ for k, v in ret.items():
+ self.assertIsInstance(k, str)
+ self.assertIsInstance(v, str)
+
+
+if __name__ == '__main__':
+ from psutil.tests.runner import run_from_name
+
+ run_from_name(__file__)
diff --git a/psutil/tests/test_testutils.py b/psutil/tests/test_testutils.py
index e92c4e11e..a93f9f09f 100755
--- a/psutil/tests/test_testutils.py
+++ b/psutil/tests/test_testutils.py
@@ -36,6 +36,7 @@
from psutil.tests import call_until
from psutil.tests import chdir
from psutil.tests import create_sockets
+from psutil.tests import filter_proc_connections
from psutil.tests import get_free_port
from psutil.tests import is_namedtuple
from psutil.tests import mock
@@ -318,14 +319,18 @@ def tcp_tcp_socketpair(self):
def test_unix_socketpair(self):
p = psutil.Process()
num_fds = p.num_fds()
- self.assertEqual(p.connections(kind='unix'), [])
+ self.assertEqual(
+ filter_proc_connections(p.connections(kind='unix')), []
+ )
name = self.get_testfn()
server, client = unix_socketpair(name)
try:
assert os.path.exists(name)
assert stat.S_ISSOCK(os.stat(name).st_mode)
self.assertEqual(p.num_fds() - num_fds, 2)
- self.assertEqual(len(p.connections(kind='unix')), 2)
+ self.assertEqual(
+ len(filter_proc_connections(p.connections(kind='unix'))), 2
+ )
self.assertEqual(server.getsockname(), name)
self.assertEqual(client.getpeername(), name)
finally:
@@ -374,10 +379,10 @@ def test_leak_mem(self):
ls = []
def fun(ls=ls):
- ls.append("x" * 24 * 1024)
+ ls.append("x" * 124 * 1024)
try:
- # will consume around 3M in total
+ # will consume around 30M in total
self.assertRaisesRegex(
AssertionError, "extra-mem", self.execute, fun, times=50
)
diff --git a/scripts/internal/winmake.py b/scripts/internal/winmake.py
index 158c46eec..43db68fd9 100755
--- a/scripts/internal/winmake.py
+++ b/scripts/internal/winmake.py
@@ -397,6 +397,12 @@ def test_process():
sh("%s psutil\\tests\\test_process.py" % PYTHON)
+def test_process_all():
+ """Run process all tests."""
+ build()
+ sh("%s psutil\\tests\\test_process_all.py" % PYTHON)
+
+
def test_system():
"""Run system tests."""
build()
@@ -517,23 +523,15 @@ def get_python(path):
# try to look for a python installation given a shortcut name
path = path.replace('.', '')
vers = (
- '26',
- '26-32',
- '26-64',
'27',
'27-32',
'27-64',
- '36',
- '36-32',
- '36-64',
- '37',
- '37-32',
- '37-64',
- '38',
- '38-32',
- '38-64',
- '39-32',
- '39-64',
+ '310-32',
+ '310-64',
+ '311-32',
+ '311-64',
+ '312-32',
+ '312-64',
)
for v in vers:
pypath = r'C:\\python%s\python.exe' % v
@@ -571,6 +569,7 @@ def parse_args():
sp.add_parser('test-misc', help="run misc tests")
sp.add_parser('test-platform', help="run windows only tests")
sp.add_parser('test-process', help="run process tests")
+ sp.add_parser('test-process-all', help="run process all tests")
sp.add_parser('test-system', help="run system tests")
sp.add_parser('test-unicode', help="run unicode tests")
sp.add_parser('test-testutils', help="run test utils tests")