From 82a43754a7f4898e7a3a2f9721ba32bf7c00a925 Mon Sep 17 00:00:00 2001 From: Giampaolo Rodola Date: Sun, 7 Jan 2024 02:09:37 +0100 Subject: [PATCH] improve tests reliability --- MANIFEST.in | 1 + Makefile | 4 + psutil/tests/__init__.py | 17 +- psutil/tests/test_connections.py | 59 ++-- psutil/tests/test_contracts.py | 437 ----------------------------- psutil/tests/test_process.py | 18 +- psutil/tests/test_process_all.py | 464 +++++++++++++++++++++++++++++++ psutil/tests/test_testutils.py | 13 +- scripts/internal/winmake.py | 27 +- 9 files changed, 541 insertions(+), 499 deletions(-) create mode 100755 psutil/tests/test_process_all.py 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 : 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, "" 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 : 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, "" 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")