| """ | |
| Collect various information about Python to help debugging test failures. | |
| """ | |
| import errno | |
| import re | |
| import sys | |
| import traceback | |
| import warnings | |
| MS_WINDOWS = (sys.platform == "win32") | |
| APPLE = (sys.platform in ("darwin", "ios", "tvos", "watchos")) | |
| COMMAND_TIMEOUT = 60.0 | |
| def normalize_text(text): | |
| if text is None: | |
| return None | |
| text = str(text) | |
| text = re.sub(r'\s+', ' ', text) | |
| return text.strip() | |
| def first_line(text): | |
| # Get the first line. Return text unchanged if it's empty. | |
| lines = text.splitlines() | |
| if lines: | |
| return lines[0] | |
| else: | |
| # text is an empty string | |
| return text | |
| def read_first_line(filename): | |
| # Get the first line of a text file and strip trailing spaces | |
| try: | |
| with open(filename, encoding="utf-8") as fp: | |
| return fp.readline().rstrip() | |
| except OSError: | |
| return '' | |
| class PythonInfo: | |
| def __init__(self): | |
| self.info = {} | |
| def add(self, key, value): | |
| if key in self.info: | |
| raise ValueError("duplicate key: %r" % key) | |
| if value is None: | |
| return | |
| if not isinstance(value, int): | |
| if not isinstance(value, str): | |
| # convert other objects like sys.flags to string | |
| value = str(value) | |
| value = value.strip() | |
| if not value: | |
| return | |
| self.info[key] = value | |
| def get_infos(self): | |
| """ | |
| Get information as a key:value dictionary where values are strings. | |
| """ | |
| return {key: str(value) for key, value in self.info.items()} | |
| def copy_attributes(info_add, obj, name_fmt, attributes, *, formatter=None): | |
| for attr in attributes: | |
| value = getattr(obj, attr, None) | |
| if value is None: | |
| continue | |
| name = name_fmt % attr | |
| if formatter is not None: | |
| value = formatter(attr, value) | |
| info_add(name, value) | |
| def copy_attr(info_add, name, mod, attr_name): | |
| try: | |
| value = getattr(mod, attr_name) | |
| except AttributeError: | |
| return | |
| info_add(name, value) | |
| def call_func(info_add, name, mod, func_name, *, formatter=None): | |
| try: | |
| func = getattr(mod, func_name) | |
| except AttributeError: | |
| return | |
| value = func() | |
| if formatter is not None: | |
| value = formatter(value) | |
| info_add(name, value) | |
| def collect_sys(info_add): | |
| attributes = ( | |
| '_emscripten_info', | |
| '_framework', | |
| 'abiflags', | |
| 'api_version', | |
| 'builtin_module_names', | |
| 'byteorder', | |
| 'dont_write_bytecode', | |
| 'executable', | |
| 'flags', | |
| 'float_info', | |
| 'float_repr_style', | |
| 'hash_info', | |
| 'hexversion', | |
| 'implementation', | |
| 'int_info', | |
| 'maxsize', | |
| 'maxunicode', | |
| 'path', | |
| 'platform', | |
| 'platlibdir', | |
| 'prefix', | |
| 'thread_info', | |
| 'version', | |
| 'version_info', | |
| 'winver', | |
| ) | |
| copy_attributes(info_add, sys, 'sys.%s', attributes) | |
| for func in ( | |
| '_is_gil_enabled', | |
| 'getandroidapilevel', | |
| 'getrecursionlimit', | |
| 'getwindowsversion', | |
| ): | |
| call_func(info_add, f'sys.{func}', sys, func) | |
| encoding = sys.getfilesystemencoding() | |
| if hasattr(sys, 'getfilesystemencodeerrors'): | |
| encoding = '%s/%s' % (encoding, sys.getfilesystemencodeerrors()) | |
| info_add('sys.filesystem_encoding', encoding) | |
| for name in ('stdin', 'stdout', 'stderr'): | |
| stream = getattr(sys, name) | |
| if stream is None: | |
| continue | |
| encoding = getattr(stream, 'encoding', None) | |
| if not encoding: | |
| continue | |
| errors = getattr(stream, 'errors', None) | |
| if errors: | |
| encoding = '%s/%s' % (encoding, errors) | |
| info_add('sys.%s.encoding' % name, encoding) | |
| # Were we compiled --with-pydebug? | |
| Py_DEBUG = hasattr(sys, 'gettotalrefcount') | |
| if Py_DEBUG: | |
| text = 'Yes (sys.gettotalrefcount() present)' | |
| else: | |
| text = 'No (sys.gettotalrefcount() missing)' | |
| info_add('build.Py_DEBUG', text) | |
| # Were we compiled --with-trace-refs? | |
| Py_TRACE_REFS = hasattr(sys, 'getobjects') | |
| if Py_TRACE_REFS: | |
| text = 'Yes (sys.getobjects() present)' | |
| else: | |
| text = 'No (sys.getobjects() missing)' | |
| info_add('build.Py_TRACE_REFS', text) | |
| info_add('sys.is_remote_debug_enabled', sys.is_remote_debug_enabled()) | |
| def collect_platform(info_add): | |
| import platform | |
| arch = platform.architecture() | |
| arch = ' '.join(filter(bool, arch)) | |
| info_add('platform.architecture', arch) | |
| info_add('platform.python_implementation', | |
| platform.python_implementation()) | |
| info_add('platform.platform', | |
| platform.platform(aliased=True)) | |
| libc_ver = ('%s %s' % platform.libc_ver()).strip() | |
| if libc_ver: | |
| info_add('platform.libc_ver', libc_ver) | |
| try: | |
| os_release = platform.freedesktop_os_release() | |
| except OSError: | |
| pass | |
| else: | |
| for key in ( | |
| 'ID', | |
| 'NAME', | |
| 'PRETTY_NAME' | |
| 'VARIANT', | |
| 'VARIANT_ID', | |
| 'VERSION', | |
| 'VERSION_CODENAME', | |
| 'VERSION_ID', | |
| ): | |
| if key not in os_release: | |
| continue | |
| info_add(f'platform.freedesktop_os_release[{key}]', | |
| os_release[key]) | |
| if sys.platform == 'android': | |
| call_func(info_add, 'platform.android_ver', platform, 'android_ver') | |
| def collect_locale(info_add): | |
| import locale | |
| info_add('locale.getencoding', locale.getencoding()) | |
| def collect_builtins(info_add): | |
| info_add('builtins.float.float_format', float.__getformat__("float")) | |
| info_add('builtins.float.double_format', float.__getformat__("double")) | |
| def collect_urandom(info_add): | |
| import os | |
| if hasattr(os, 'getrandom'): | |
| # PEP 524: Check if system urandom is initialized | |
| try: | |
| try: | |
| os.getrandom(1, os.GRND_NONBLOCK) | |
| state = 'ready (initialized)' | |
| except BlockingIOError as exc: | |
| state = 'not seeded yet (%s)' % exc | |
| info_add('os.getrandom', state) | |
| except OSError as exc: | |
| # Python was compiled on a more recent Linux version | |
| # than the current Linux kernel: ignore OSError(ENOSYS) | |
| if exc.errno != errno.ENOSYS: | |
| raise | |
| def collect_os(info_add): | |
| import os | |
| plain_attributes = ( | |
| 'name', | |
| 'supports_bytes_environ', | |
| ) | |
| copy_attributes(info_add, os, 'os.%s', plain_attributes) | |
| def format_set_of_functions(attr, value): | |
| return str(sorted(func.__name__ for func in value)) | |
| supports_attributes = ( | |
| 'supports_dir_fd', | |
| 'supports_effective_ids', | |
| 'supports_fd', | |
| 'supports_follow_symlinks', | |
| ) | |
| copy_attributes(info_add, os, 'os.%s', supports_attributes, | |
| formatter=format_set_of_functions) | |
| for func in ( | |
| 'cpu_count', | |
| 'getcwd', | |
| 'getegid', | |
| 'geteuid', | |
| 'getgid', | |
| 'getloadavg', | |
| 'getresgid', | |
| 'getresuid', | |
| 'getuid', | |
| 'process_cpu_count', | |
| 'uname', | |
| ): | |
| call_func(info_add, 'os.%s' % func, os, func) | |
| def format_groups(groups): | |
| return ', '.join(map(str, groups)) | |
| call_func(info_add, 'os.getgroups', os, 'getgroups', formatter=format_groups) | |
| if hasattr(os, 'getlogin'): | |
| try: | |
| login = os.getlogin() | |
| except OSError: | |
| # getlogin() fails with "OSError: [Errno 25] Inappropriate ioctl | |
| # for device" on Travis CI | |
| pass | |
| else: | |
| info_add("os.login", login) | |
| # Environment variables used by the stdlib and tests. Don't log the full | |
| # environment: filter to list to not leak sensitive information. | |
| # | |
| # HTTP_PROXY is not logged because it can contain a password. | |
| ENV_VARS = frozenset(( | |
| "APPDATA", | |
| "AR", | |
| "ARCHFLAGS", | |
| "ARFLAGS", | |
| "AUDIODEV", | |
| "BUILDPYTHON", | |
| "CC", | |
| "CFLAGS", | |
| "CI", | |
| "COLUMNS", | |
| "COMPUTERNAME", | |
| "COMSPEC", | |
| "CONTAINER", | |
| "CPP", | |
| "CPPFLAGS", | |
| "DISPLAY", | |
| "DISTUTILS_DEBUG", | |
| "DISTUTILS_USE_SDK", | |
| "DYLD_LIBRARY_PATH", | |
| "ENSUREPIP_OPTIONS", | |
| "FORCE_COLOR", | |
| "GITHUB_ACTIONS", | |
| "HISTORY_FILE", | |
| "HOME", | |
| "HOMEDRIVE", | |
| "HOMEPATH", | |
| "IDLESTARTUP", | |
| "IMAGE_OS_VERSION", | |
| "IPHONEOS_DEPLOYMENT_TARGET", | |
| "LANG", | |
| "LDFLAGS", | |
| "LDSHARED", | |
| "LD_LIBRARY_PATH", | |
| "LINES", | |
| "MACOSX_DEPLOYMENT_TARGET", | |
| "MAILCAPS", | |
| "MAKEFLAGS", | |
| "MIXERDEV", | |
| "MSSDK", | |
| "NO_COLOR", | |
| "PATH", | |
| "PATHEXT", | |
| "PIP_CONFIG_FILE", | |
| "PLAT", | |
| "POSIXLY_CORRECT", | |
| "PYTHON_COLORS", | |
| "PY_SAX_PARSER", | |
| "ProgramFiles", | |
| "ProgramFiles(x86)", | |
| "RUNNING_ON_VALGRIND", | |
| "SDK_TOOLS_BIN", | |
| "SERVER_SOFTWARE", | |
| "SHELL", | |
| "SOURCE_DATE_EPOCH", | |
| "SYSTEMROOT", | |
| "TEMP", | |
| "TERM", | |
| "TILE_LIBRARY", | |
| "TMP", | |
| "TMPDIR", | |
| "TRAVIS", | |
| "TZ", | |
| "USERPROFILE", | |
| "VIRTUAL_ENV", | |
| "WAYLAND_DISPLAY", | |
| "WINDIR", | |
| "_PYTHON_HOSTRUNNER", | |
| "_PYTHON_HOST_PLATFORM", | |
| "_PYTHON_PROJECT_BASE", | |
| "_PYTHON_SYSCONFIGDATA_NAME", | |
| "_PYTHON_SYSCONFIGDATA_PATH", | |
| "__PYVENV_LAUNCHER__", | |
| # Sanitizer options | |
| "ASAN_OPTIONS", | |
| "LSAN_OPTIONS", | |
| "MSAN_OPTIONS", | |
| "TSAN_OPTIONS", | |
| "UBSAN_OPTIONS", | |
| )) | |
| for name, value in os.environ.items(): | |
| uname = name.upper() | |
| if (uname in ENV_VARS | |
| # Copy PYTHON* variables like PYTHONPATH | |
| # Copy LC_* variables like LC_ALL | |
| or uname.startswith(("PYTHON", "LC_")) | |
| # Visual Studio: VS140COMNTOOLS | |
| or (uname.startswith("VS") and uname.endswith("COMNTOOLS"))): | |
| info_add('os.environ[%s]' % name, value) | |
| if hasattr(os, 'umask'): | |
| mask = os.umask(0) | |
| os.umask(mask) | |
| info_add("os.umask", '0o%03o' % mask) | |
| def collect_pwd(info_add): | |
| try: | |
| import pwd | |
| except ImportError: | |
| return | |
| import os | |
| uid = os.getuid() | |
| try: | |
| entry = pwd.getpwuid(uid) | |
| except KeyError: | |
| entry = None | |
| info_add('pwd.getpwuid(%s)'% uid, | |
| entry if entry is not None else '<KeyError>') | |
| if entry is None: | |
| # there is nothing interesting to read if the current user identifier | |
| # is not the password database | |
| return | |
| if hasattr(os, 'getgrouplist'): | |
| groups = os.getgrouplist(entry.pw_name, entry.pw_gid) | |
| groups = ', '.join(map(str, groups)) | |
| info_add('os.getgrouplist', groups) | |
| def collect_readline(info_add): | |
| try: | |
| import readline | |
| except ImportError: | |
| return | |
| attributes = ( | |
| "backend", | |
| "READLINE_VERSION_INFO", | |
| "readline_version_info", | |
| "readline_version", | |
| ) | |
| copy_attributes(info_add, readline, 'readline.%s', attributes) | |
| def run_command(cmd, check=True, **kwargs): | |
| import subprocess | |
| from test.support import has_subprocess_support | |
| if not has_subprocess_support: | |
| # subprocess is not supported by the current platform | |
| return '' | |
| timeout = COMMAND_TIMEOUT | |
| cmd_str = ' '.join(cmd) | |
| try: | |
| proc = subprocess.Popen(cmd, | |
| stdout=subprocess.PIPE, | |
| stderr=subprocess.DEVNULL, | |
| text=True, | |
| **kwargs) | |
| with proc: | |
| try: | |
| stdout = proc.communicate(timeout=timeout)[0] | |
| except: | |
| proc.kill() | |
| proc.communicate() | |
| raise | |
| if check and proc.returncode: | |
| print(f"Command {cmd_str} failed with exit code {proc.returncode}") | |
| return '' | |
| # Strip trailing spaces and newlines | |
| stdout = stdout.rstrip() | |
| return stdout | |
| except FileNotFoundError: | |
| return '' | |
| except OSError as exc: | |
| print(f"Command {cmd_str} failed with: {exc!r}") | |
| return '' | |
| except subprocess.TimeoutExpired: | |
| print(f"Command {cmd_str}: timeout!") | |
| return '' | |
| def collect_gdb(info_add): | |
| version = run_command(["gdb", "-nx", "--version"]) | |
| if version: | |
| # Only keep the first line | |
| info_add('gdb_version', first_line(version)) | |
| def collect_tkinter(info_add): | |
| try: | |
| import _tkinter | |
| except ImportError: | |
| pass | |
| else: | |
| attributes = ('TK_PATCH_LEVEL', 'TCL_PATCH_LEVEL') | |
| copy_attributes(info_add, _tkinter, 'tkinter.%s', attributes) | |
| try: | |
| import tkinter | |
| except ImportError: | |
| pass | |
| else: | |
| tcl = tkinter.Tcl() | |
| patchlevel = tcl.call('info', 'patchlevel') | |
| info_add('tkinter.info_patchlevel', patchlevel) | |
| def collect_time(info_add): | |
| import time | |
| info_add('time.time', time.time()) | |
| attributes = ( | |
| 'altzone', | |
| 'daylight', | |
| 'timezone', | |
| 'tzname', | |
| ) | |
| copy_attributes(info_add, time, 'time.%s', attributes) | |
| if hasattr(time, 'get_clock_info'): | |
| for clock in ('clock', 'monotonic', 'perf_counter', | |
| 'process_time', 'thread_time', 'time'): | |
| try: | |
| # prevent DeprecatingWarning on get_clock_info('clock') | |
| with warnings.catch_warnings(record=True): | |
| clock_info = time.get_clock_info(clock) | |
| except ValueError: | |
| # missing clock like time.thread_time() | |
| pass | |
| else: | |
| info_add('time.get_clock_info(%s)' % clock, clock_info) | |
| def collect_curses(info_add): | |
| try: | |
| import curses | |
| except ImportError: | |
| return | |
| copy_attr(info_add, 'curses.ncurses_version', curses, 'ncurses_version') | |
| def collect_datetime(info_add): | |
| try: | |
| import datetime | |
| except ImportError: | |
| return | |
| info_add('datetime.datetime.now', datetime.datetime.now()) | |
| def collect_sysconfig(info_add): | |
| import sysconfig | |
| info_add('sysconfig.is_python_build', sysconfig.is_python_build()) | |
| for name in ( | |
| 'ABIFLAGS', | |
| 'ALT_SOABI', | |
| 'ANDROID_API_LEVEL', | |
| 'CC', | |
| 'CCSHARED', | |
| 'CFLAGS', | |
| 'CFLAGSFORSHARED', | |
| 'CONFIG_ARGS', | |
| 'HOSTRUNNER', | |
| 'HOST_GNU_TYPE', | |
| 'MACHDEP', | |
| 'MULTIARCH', | |
| 'OPT', | |
| 'PGO_PROF_USE_FLAG', | |
| 'PY_CFLAGS', | |
| 'PY_CFLAGS_NODIST', | |
| 'PY_CORE_LDFLAGS', | |
| 'PY_CORE_EXE_LDFLAGS', | |
| 'PY_LDFLAGS', | |
| 'PY_LDFLAGS_NODIST', | |
| 'PY_STDMODULE_CFLAGS', | |
| 'Py_DEBUG', | |
| 'Py_ENABLE_SHARED', | |
| 'Py_GIL_DISABLED', | |
| 'Py_REMOTE_DEBUG', | |
| 'SHELL', | |
| 'SOABI', | |
| 'SOABI_PLATFORM', | |
| 'TEST_MODULES', | |
| 'VAPTH', | |
| 'abs_builddir', | |
| 'abs_srcdir', | |
| 'prefix', | |
| 'srcdir', | |
| ): | |
| value = sysconfig.get_config_var(name) | |
| if name == 'ANDROID_API_LEVEL' and not value: | |
| # skip ANDROID_API_LEVEL=0 | |
| continue | |
| value = normalize_text(value) | |
| info_add('sysconfig[%s]' % name, value) | |
| for name in ( | |
| 'WITH_DOC_STRINGS', | |
| 'WITH_DTRACE', | |
| 'WITH_MIMALLOC', | |
| 'WITH_PYMALLOC', | |
| 'WITH_VALGRIND', | |
| ): | |
| value = sysconfig.get_config_var(name) | |
| if value: | |
| text = 'Yes' | |
| else: | |
| text = 'No' | |
| info_add(f'build.{name}', text) | |
| def collect_ssl(info_add): | |
| import os | |
| try: | |
| import ssl | |
| except ImportError: | |
| return | |
| try: | |
| import _ssl | |
| except ImportError: | |
| _ssl = None | |
| def format_attr(attr, value): | |
| if attr.startswith('OP_'): | |
| return '%#8x' % value | |
| else: | |
| return value | |
| attributes = ( | |
| 'OPENSSL_VERSION', | |
| 'OPENSSL_VERSION_INFO', | |
| 'OPENSSL_API_VERSION_INFO', | |
| 'HAS_SNI', | |
| 'OP_ALL', | |
| 'OP_NO_TLSv1_1', | |
| ) | |
| copy_attributes(info_add, ssl, 'ssl.%s', attributes, formatter=format_attr) | |
| for name, ctx in ( | |
| ('SSLContext', ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)), | |
| ('default_https_context', ssl._create_default_https_context()), | |
| ('stdlib_context', ssl._create_stdlib_context()), | |
| ): | |
| attributes = ( | |
| 'minimum_version', | |
| 'maximum_version', | |
| 'protocol', | |
| 'options', | |
| 'verify_mode', | |
| ) | |
| copy_attributes(info_add, ctx, f'ssl.{name}.%s', attributes) | |
| env_names = ["OPENSSL_CONF", "SSLKEYLOGFILE"] | |
| if _ssl is not None and hasattr(_ssl, 'get_default_verify_paths'): | |
| parts = _ssl.get_default_verify_paths() | |
| env_names.extend((parts[0], parts[2])) | |
| for name in env_names: | |
| try: | |
| value = os.environ[name] | |
| except KeyError: | |
| continue | |
| info_add('ssl.environ[%s]' % name, value) | |
| def collect_socket(info_add): | |
| try: | |
| import socket | |
| except ImportError: | |
| return | |
| try: | |
| hostname = socket.gethostname() | |
| except (OSError, AttributeError): | |
| # WASI SDK 16.0 does not have gethostname(2). | |
| if sys.platform != "wasi": | |
| raise | |
| else: | |
| info_add('socket.hostname', hostname) | |
| def collect_sqlite(info_add): | |
| try: | |
| import sqlite3 | |
| except ImportError: | |
| return | |
| attributes = ('SQLITE_VERSION', 'sqlite_version') | |
| copy_attributes(info_add, sqlite3, 'sqlite3.%s', attributes) | |
| def collect_zlib(info_add): | |
| try: | |
| import zlib | |
| except ImportError: | |
| return | |
| attributes = ('ZLIB_VERSION', 'zlib_version', 'ZLIBNG_VERSION') | |
| copy_attributes(info_add, zlib, 'zlib.%s', attributes) | |
| def collect_bz2(info_add): | |
| try: | |
| import _bz2 | |
| except ImportError: | |
| return | |
| attributes = ('bzlib_version',) | |
| copy_attributes(info_add, _bz2, 'bz2.%s', attributes) | |
| def collect_lzma(info_add): | |
| try: | |
| import _lzma | |
| except ImportError: | |
| return | |
| attributes = ('LZMA_VERSION', 'lzma_version') | |
| copy_attributes(info_add, _lzma, 'lzma.%s', attributes) | |
| def collect_zstd(info_add): | |
| try: | |
| import _zstd | |
| except ImportError: | |
| return | |
| attributes = ('ZSTD_VERSION', 'zstd_version') | |
| copy_attributes(info_add, _zstd, 'zstd.%s', attributes) | |
| def collect_ctypes(info_add): | |
| try: | |
| import _ctypes | |
| except ImportError: | |
| return | |
| attributes = ('LIBFFI_VERSION', 'libffi_version') | |
| copy_attributes(info_add, _ctypes, 'ctypes.%s', attributes) | |
| def collect_expat(info_add): | |
| try: | |
| from xml.parsers import expat | |
| except ImportError: | |
| return | |
| attributes = ('EXPAT_VERSION', 'VERSION_INFO', 'version_info') | |
| copy_attributes(info_add, expat, 'expat.%s', attributes) | |
| def collect_decimal(info_add): | |
| try: | |
| import _decimal | |
| except ImportError: | |
| return | |
| attributes = ('LIBMPDEC_VERSION', 'libmpdec_version') | |
| copy_attributes(info_add, _decimal, '_decimal.%s', attributes) | |
| def collect_testcapi(info_add): | |
| try: | |
| import _testcapi | |
| except ImportError: | |
| return | |
| for name in ( | |
| 'LONG_MAX', # always 32-bit on Windows, 64-bit on 64-bit Unix | |
| 'PY_SSIZE_T_MAX', | |
| 'SIZEOF_TIME_T', # 32-bit or 64-bit depending on the platform | |
| 'SIZEOF_WCHAR_T', # 16-bit or 32-bit depending on the platform | |
| ): | |
| copy_attr(info_add, f'_testcapi.{name}', _testcapi, name) | |
| def collect_testinternalcapi(info_add): | |
| try: | |
| import _testinternalcapi | |
| except ImportError: | |
| return | |
| call_func(info_add, 'pymem.allocator', _testinternalcapi, 'pymem_getallocatorsname') | |
| for name in ( | |
| 'SIZEOF_PYGC_HEAD', | |
| 'SIZEOF_PYOBJECT', | |
| ): | |
| copy_attr(info_add, f'_testinternalcapi.{name}', _testinternalcapi, name) | |
| def collect_resource(info_add): | |
| try: | |
| import resource | |
| except ImportError: | |
| return | |
| limits = [attr for attr in dir(resource) if attr.startswith('RLIMIT_')] | |
| for name in limits: | |
| key = getattr(resource, name) | |
| value = resource.getrlimit(key) | |
| info_add('resource.%s' % name, value) | |
| call_func(info_add, 'resource.pagesize', resource, 'getpagesize') | |
| def collect_test_socket(info_add): | |
| import unittest | |
| try: | |
| from test import test_socket | |
| except (ImportError, unittest.SkipTest): | |
| return | |
| # all check attributes like HAVE_SOCKET_CAN | |
| attributes = [name for name in dir(test_socket) | |
| if name.startswith('HAVE_')] | |
| copy_attributes(info_add, test_socket, 'test_socket.%s', attributes) | |
| # Get IOCTL_VM_SOCKETS_GET_LOCAL_CID of /dev/vsock | |
| cid = test_socket.get_cid() | |
| info_add('test_socket.get_cid', cid) | |
| def collect_support(info_add): | |
| try: | |
| from test import support | |
| except ImportError: | |
| return | |
| attributes = ( | |
| 'MS_WINDOWS', | |
| 'has_fork_support', | |
| 'has_socket_support', | |
| 'has_strftime_extensions', | |
| 'has_subprocess_support', | |
| 'is_android', | |
| 'is_emscripten', | |
| 'is_jython', | |
| 'is_wasi', | |
| 'is_wasm32', | |
| ) | |
| copy_attributes(info_add, support, 'support.%s', attributes) | |
| call_func(info_add, 'support._is_gui_available', support, '_is_gui_available') | |
| call_func(info_add, 'support.python_is_optimized', support, 'python_is_optimized') | |
| info_add('support.check_sanitizer(address=True)', | |
| support.check_sanitizer(address=True)) | |
| info_add('support.check_sanitizer(memory=True)', | |
| support.check_sanitizer(memory=True)) | |
| info_add('support.check_sanitizer(ub=True)', | |
| support.check_sanitizer(ub=True)) | |
| info_add('support.built_with_c_assertions', | |
| support.built_with_c_assertions()) | |
| def collect_support_os_helper(info_add): | |
| try: | |
| from test.support import os_helper | |
| except ImportError: | |
| return | |
| for name in ( | |
| 'can_symlink', | |
| 'can_xattr', | |
| 'can_chmod', | |
| 'can_dac_override', | |
| ): | |
| func = getattr(os_helper, name) | |
| info_add(f'support_os_helper.{name}', func()) | |
| def collect_support_socket_helper(info_add): | |
| try: | |
| from test.support import socket_helper | |
| except ImportError: | |
| return | |
| attributes = ( | |
| 'IPV6_ENABLED', | |
| 'has_gethostname', | |
| ) | |
| copy_attributes(info_add, socket_helper, 'support_socket_helper.%s', attributes) | |
| for name in ( | |
| 'tcp_blackhole', | |
| ): | |
| func = getattr(socket_helper, name) | |
| info_add(f'support_socket_helper.{name}', func()) | |
| def collect_support_threading_helper(info_add): | |
| try: | |
| from test.support import threading_helper | |
| except ImportError: | |
| return | |
| attributes = ( | |
| 'can_start_thread', | |
| ) | |
| copy_attributes(info_add, threading_helper, 'support_threading_helper.%s', attributes) | |
| def get_compiler_version(sysconfig_var): | |
| import sysconfig | |
| program = sysconfig.get_config_var(sysconfig_var) | |
| if not program: | |
| return | |
| try: | |
| import shlex | |
| args = shlex.split(program) | |
| except ImportError: | |
| args = program.split() | |
| args.append('--version') | |
| stdout = run_command(args) | |
| if not stdout: | |
| # Cannot run the compiler, for example when Python has been | |
| # cross-compiled and installed on the target platform where the | |
| # compiler is missing. | |
| # | |
| # CC --version failed: ignore error | |
| return | |
| text = first_line(stdout) | |
| text = normalize_text(text) | |
| if text: | |
| text = f'[{program}] {text}' | |
| return text | |
| def collect_cc(info_add): | |
| # C compiler | |
| version = get_compiler_version('CC') | |
| if version: | |
| info_add('CC.version', version) | |
| # C++ compiler | |
| version = get_compiler_version('CXX') | |
| if version: | |
| info_add('CXX.version', version) | |
| def collect_ndbm(info_add): | |
| try: | |
| import _dbm | |
| except ImportError: | |
| return | |
| attributes = ('library', 'GDBM_VERSION_INFO', 'gdbm_version', | |
| 'BDB_VERSION', 'bdb_version') | |
| copy_attributes(info_add, _dbm, 'ndbm.%s', attributes) | |
| def collect_gdbm(info_add): | |
| try: | |
| import _gdbm | |
| except ImportError: | |
| return | |
| attributes = ('GDBM_VERSION_INFO', 'gdbm_version') | |
| copy_attributes(info_add, _gdbm, 'gdbm.%s', attributes) | |
| def collect_get_config(info_add): | |
| # Get global configuration variables, _PyPreConfig and _PyCoreConfig | |
| try: | |
| from _testinternalcapi import get_configs | |
| except ImportError: | |
| return | |
| all_configs = get_configs() | |
| for config_type in sorted(all_configs): | |
| config = all_configs[config_type] | |
| for key in sorted(config): | |
| info_add('%s[%s]' % (config_type, key), repr(config[key])) | |
| def collect_subprocess(info_add): | |
| import subprocess | |
| copy_attributes(info_add, subprocess, 'subprocess.%s', ('_USE_POSIX_SPAWN',)) | |
| def winreg_query(path): | |
| try: | |
| import winreg | |
| except ImportError: | |
| return None | |
| key, path = path.split('\\', 1) | |
| sub_key, value = path.rsplit('\\', 1) | |
| if key == "HKEY_LOCAL_MACHINE": | |
| key = winreg.HKEY_LOCAL_MACHINE | |
| else: | |
| raise ValueError(f"unknown key {key!r}") | |
| try: | |
| access = winreg.KEY_READ | winreg.KEY_WOW64_64KEY | |
| with winreg.OpenKey(key, sub_key, access=access) as key_handle: | |
| result, _ = winreg.QueryValueEx(key_handle, value) | |
| return result | |
| except OSError: | |
| return None | |
| def wmi_query(query): | |
| try: | |
| import _wmi | |
| except ImportError: | |
| return {} | |
| try: | |
| data = _wmi.exec_query(query) | |
| except OSError: | |
| return {} | |
| dict_data = {} | |
| for item in data.split("\0"): | |
| key, _, value = item.partition("=") | |
| dict_data[key] = value | |
| return dict_data | |
| def collect_windows(info_add): | |
| if not MS_WINDOWS: | |
| # Code specific to Windows | |
| return | |
| # windows.RtlAreLongPathsEnabled: RtlAreLongPathsEnabled() | |
| # windows.is_admin: IsUserAnAdmin() | |
| try: | |
| import ctypes.util | |
| if not hasattr(ctypes, 'WinDLL'): | |
| raise ImportError | |
| except ImportError: | |
| pass | |
| else: | |
| ntdll = ctypes.WinDLL('ntdll') | |
| BOOLEAN = ctypes.c_ubyte | |
| try: | |
| @ctypes.util.wrap_dll_function(ntdll) | |
| def RtlAreLongPathsEnabled() -> BOOLEAN: | |
| pass | |
| except AttributeError: | |
| res = '<function not available>' | |
| else: | |
| res = bool(RtlAreLongPathsEnabled()) | |
| info_add('windows.RtlAreLongPathsEnabled', res) | |
| @ctypes.util.wrap_dll_function(ctypes.windll.shell32) | |
| def IsUserAnAdmin() -> BOOLEAN: | |
| pass | |
| info_add('windows.is_admin', bool(IsUserAnAdmin())) | |
| try: | |
| import _winapi | |
| except ImportError: | |
| pass | |
| else: | |
| try: | |
| dll_path = _winapi.GetModuleFileName(sys.dllhandle) | |
| info_add('windows.dll_path', dll_path) | |
| except AttributeError: | |
| pass | |
| call_func(info_add, 'windows.ansi_code_page', _winapi, 'GetACP') | |
| call_func(info_add, 'windows.oem_code_page', _winapi, 'GetOEMCP') | |
| # Get operating system caption and version using WMI | |
| data = wmi_query("SELECT Caption, Version FROM Win32_OperatingSystem") | |
| caption = data.get('Caption', '') | |
| if caption: | |
| info_add('windows.version_caption', caption) | |
| version = data.get('Version', '') | |
| if version: | |
| info_add('windows.version', version) | |
| # windows.ver: "ver" command | |
| output = run_command(["ver"], shell=True) | |
| # "ver" output starts with an empty line: remove it | |
| output = output.strip() | |
| if output: | |
| info_add('windows.ver', first_line(output)) | |
| # windows.developer_mode: get AllowDevelopmentWithoutDevLicense registry | |
| value = winreg_query(r"HKEY_LOCAL_MACHINE\SOFTWARE\Microsoft\Windows" | |
| r"\CurrentVersion\AppModelUnlock" | |
| r"\AllowDevelopmentWithoutDevLicense") | |
| if value is not None: | |
| info_add('windows.developer_mode', "enabled" if value else "disabled") | |
| def collect_fips(info_add): | |
| try: | |
| import _hashlib | |
| except ImportError: | |
| _hashlib = None | |
| if _hashlib is not None: | |
| call_func(info_add, 'fips.openssl_fips_mode', _hashlib, 'get_fips_mode') | |
| fips_enabled = read_first_line("/proc/sys/crypto/fips_enabled") | |
| if fips_enabled: | |
| info_add('fips.linux_crypto_fips_enabled', fips_enabled) | |
| def collect_tempfile(info_add): | |
| import tempfile | |
| info_add('tempfile.gettempdir', tempfile.gettempdir()) | |
| def collect_libregrtest_utils(info_add): | |
| try: | |
| from test.libregrtest import utils | |
| except ImportError: | |
| return | |
| info_add('libregrtests.build_info', ' '.join(utils.get_build_info())) | |
| def uptime_boottime(): | |
| # Use CLOCK_BOOTTIME | |
| import time | |
| try: | |
| return time.clock_gettime(time.CLOCK_BOOTTIME) | |
| except (AttributeError, OSError): | |
| return None | |
| def uptime_linux(): | |
| # Parse the first member of /proc/uptime | |
| line = read_first_line("/proc/uptime") | |
| if not line: | |
| return | |
| try: | |
| parts = line.split() | |
| if not parts: | |
| return | |
| return float(parts[0]) | |
| except ValueError: | |
| return | |
| def uptime_bsd(): | |
| # Get sysctlbyname("kern.boottime") | |
| try: | |
| import _testcapi | |
| except ImportError: | |
| return None | |
| try: | |
| return _testcapi.uptime_bsd() | |
| except (AttributeError, OSError): | |
| return None | |
| def uptime_windows(): | |
| try: | |
| import _winapi | |
| except ImportError: | |
| return None | |
| else: | |
| return _winapi.GetTickCount64() / 1000. | |
| def get_uptime(): | |
| for func in (uptime_boottime, uptime_linux, uptime_bsd, uptime_windows): | |
| uptime = func() | |
| if uptime is not None: | |
| return uptime | |
| return None | |
| def get_machine_id(): | |
| if MS_WINDOWS: | |
| machine_guid = winreg_query(r"HKEY_LOCAL_MACHINE\SOFTWARE\Microsoft" | |
| r"\Cryptography\MachineGuid") | |
| if machine_guid: | |
| return machine_guid | |
| for filename in ( | |
| # https://www.freedesktop.org/software/systemd/man/latest/machine-id.html | |
| "/etc/machine-id", | |
| # BSD | |
| "/etc/hostid", | |
| ): | |
| machine_id = read_first_line(filename) | |
| if machine_id: | |
| return machine_id | |
| return None | |
| def detect_virt_windows(info_add): | |
| # On Windows, use WMI to detect the virtualization. | |
| # | |
| # Microsoft Hyper-V: | |
| # - Win32_Bios.Version = 'VRTUAL - 12001807' | |
| # - Win32_Bios.Manufacturer = 'American Megatrends Inc.' | |
| # - Win32_ComputerSystem.Model = 'Virtual Machine' | |
| # - Win32_ComputerSystem.Manufacturer = 'Microsoft Corporation' | |
| # | |
| # VMware: | |
| # - Win32_ComputerSystem.Model = 'VMware' | |
| # - Win32_ComputerSystem.Manufacturer = 'VMWare' (uppercase W in Ware) | |
| # - Win32_Bios.SerialNumber starts with 'VMware-' | |
| # | |
| # QEMU: | |
| # - Win32_ComputerSystem.Manufacturer = 'QEMU' | |
| # - Win32_ComputerSystem.Model = 'Standard PC (Q35 + ICH9, 2009)' | |
| # - Win32_Bios.Version = 'BOCHS - 1' | |
| # - Win32_Bios.Manufacturer = 'EDK II' | |
| # | |
| # Parallels: | |
| # - Win32_Bios.Version = 'PARALLELS' | |
| # | |
| # VirtualBox: | |
| # - Win32_Bios.Version = 'VBOX' | |
| # - Win32_ComputerSystem.Model = 'VirtualBox' | |
| # - Win32_ComputerSystem.Manufacturer = 'innotek GmbH' | |
| # | |
| # Amazon EC2: | |
| # - Win32_Bios.Version = 'AMAZON - 1' | |
| # - Win32_Bios.Manufacturer = 'Amazon EC2' | |
| # - Win32_ComputerSystem.Model = 'm7i.4xlarge' | |
| # - Win32_ComputerSystem.Manufacturer = 'Amazon EC2' | |
| KNOWN_VIRT = ( | |
| 'Amazon EC2', | |
| 'QEMU', | |
| 'VMware', | |
| 'VirtualBox', | |
| 'Xen', | |
| 'oVirt', | |
| ) | |
| KNOWN_BIOS_VERSIONS = { | |
| 'PARALLELS': 'Parallels', | |
| 'VBOX': 'VirtualBox', | |
| } | |
| computer = wmi_query('SELECT Model, Manufacturer FROM Win32_ComputerSystem') | |
| computer_model = computer.get('Model', '') | |
| computer_manufacturer = computer.get('Manufacturer', '') | |
| if computer_manufacturer == 'Amazon EC2': | |
| # Log the VM model (ex: 'm7i.4xlarge') | |
| info_add('system.computer.model', computer_model) | |
| return computer_manufacturer | |
| if computer_model in KNOWN_VIRT: | |
| return computer_model | |
| if computer_manufacturer in KNOWN_VIRT: | |
| return computer_manufacturer | |
| bios = wmi_query('SELECT Version, Manufacturer FROM Win32_Bios') | |
| bios_version = bios.get('Version', '') | |
| if bios_version in KNOWN_VIRT: | |
| return bios_version | |
| if (bios_version.startswith('VRTUAL - ') | |
| and computer_manufacturer == 'Microsoft Corporation'): | |
| return 'Microsoft Hyper-V' | |
| try: | |
| return KNOWN_BIOS_VERSIONS[bios_version] | |
| except KeyError: | |
| pass | |
| bios_manufacturer = bios.get('Manufacturer', '') | |
| if bios_manufacturer in KNOWN_VIRT: | |
| return bios_manufacturer | |
| # Log the values to update the code if a new VM is discovered | |
| if computer_model: | |
| info_add('system.computer.model', computer_model) | |
| if computer_manufacturer: | |
| info_add('system.computer.manufacturer', computer_manufacturer) | |
| if bios_version: | |
| info_add('system.bios.version', bios_version) | |
| if bios_manufacturer: | |
| info_add('system.bios.manufacturer', bios_manufacturer) | |
| def detect_virt(info_add): | |
| if MS_WINDOWS: | |
| return detect_virt_windows(info_add) | |
| # Run systemd-detect-virt command | |
| virt = run_command(["systemd-detect-virt"], check=False) | |
| if virt and virt != "none": | |
| return virt | |
| # Check if the process in running in a container | |
| import os.path | |
| if os.path.exists('/.dockerenv'): | |
| return 'docker' | |
| if os.path.exists('/run/.containerenv'): | |
| return 'podman' | |
| container = read_first_line('/run/systemd/container') | |
| if container: | |
| return container | |
| if APPLE: | |
| hv_vmm_present = run_command(['sysctl', '-n', 'kern.hv_vmm_present']) | |
| if hv_vmm_present == '1': | |
| return 'run in a VM (kern.hv_vmm_present is 1)' | |
| # Other ways to check if running in a container: | |
| # * Parse /proc/1/mounts or /proc/1/mountinfo (check "/" filesystem). | |
| # * Parse /proc/1/cgroup. | |
| # * Parse the first line of /proc/1/sched (check process name is different | |
| # than "init" and "systemd"). | |
| # * Check / inode. | |
| # * On systems using SELinux (Fedora/CentOS/RHEL), check for "container_t" | |
| # label, for example of /proc/1/attr/current. | |
| # * Check for "container" variable in /proc/1/environ | |
| # (only root can read this file). | |
| # * Check for "container" environment variable. | |
| # * Set a specific env var when creating the container image. | |
| # * Run virt-what, need to install the script, and must be run as root. | |
| # * Check for "GITHUB_ACTIONS" environmant variable (GitHub Action). | |
| def collect_system(info_add): | |
| boot_id = read_first_line("/proc/sys/kernel/random/boot_id") | |
| if boot_id: | |
| info_add('system.boot_id', boot_id) | |
| machine_id = get_machine_id() | |
| if machine_id: | |
| info_add('system.machine_id', machine_id) | |
| uptime = get_uptime() | |
| if uptime is not None: | |
| # truncate microseconds | |
| uptime = int(uptime) | |
| try: | |
| import datetime | |
| uptime = str(datetime.timedelta(seconds=uptime)) | |
| except ImportError: | |
| uptime = f'{uptime} sec' | |
| info_add('system.uptime', uptime) | |
| virt = detect_virt(info_add) | |
| if virt: | |
| info_add('system.virt', virt) | |
| if APPLE: | |
| hardware = run_command(['sysctl', '-n', 'hw.model']) | |
| if hardware: | |
| info_add('system.hardware', hardware) | |
| def collect_importlib(info_add): | |
| import importlib.machinery | |
| info_add('importlib.extension_suffixes', | |
| importlib.machinery.EXTENSION_SUFFIXES) | |
| def collect_info(info): | |
| error = False | |
| info_add = info.add | |
| for collect_func in ( | |
| # collect_urandom() must be the first, to check the getrandom() status. | |
| # Other functions may block on os.urandom() indirectly and so change | |
| # its state. | |
| collect_urandom, | |
| collect_builtins, | |
| collect_bz2, | |
| collect_cc, | |
| collect_curses, | |
| collect_datetime, | |
| collect_decimal, | |
| collect_ctypes, | |
| collect_expat, | |
| collect_fips, | |
| collect_gdb, | |
| collect_ndbm, | |
| collect_gdbm, | |
| collect_get_config, | |
| collect_locale, | |
| collect_lzma, | |
| collect_os, | |
| collect_platform, | |
| collect_pwd, | |
| collect_readline, | |
| collect_resource, | |
| collect_socket, | |
| collect_sqlite, | |
| collect_ssl, | |
| collect_subprocess, | |
| collect_sys, | |
| collect_sysconfig, | |
| collect_testcapi, | |
| collect_testinternalcapi, | |
| collect_tempfile, | |
| collect_time, | |
| collect_tkinter, | |
| collect_windows, | |
| collect_zlib, | |
| collect_zstd, | |
| collect_libregrtest_utils, | |
| collect_system, | |
| collect_importlib, | |
| # Collecting from tests should be last as they have side effects. | |
| collect_test_socket, | |
| collect_support, | |
| collect_support_os_helper, | |
| collect_support_socket_helper, | |
| collect_support_threading_helper, | |
| ): | |
| try: | |
| collect_func(info_add) | |
| except Exception: | |
| error = True | |
| print("ERROR: %s() failed" % (collect_func.__name__), | |
| file=sys.stderr) | |
| traceback.print_exc(file=sys.stderr) | |
| print(file=sys.stderr) | |
| sys.stderr.flush() | |
| return error | |
| def dump_info(info, file=None): | |
| title = "Python build information" | |
| print(title) | |
| print("#" * len(title)) | |
| print() | |
| infos = info.get_infos() | |
| infos = sorted(infos.items()) | |
| for key, value in infos: | |
| value = value.replace("\n", " ") | |
| print("%s: %s" % (key, value)) | |
| def main(): | |
| info = PythonInfo() | |
| error = collect_info(info) | |
| dump_info(info) | |
| if error: | |
| print() | |
| print("Collection failed: exit with error", file=sys.stderr) | |
| sys.exit(1) | |
| if __name__ == "__main__": | |
| main() |