From 7ad1b56d2ca0b59e09c4fb40a2b844f77712e07c Mon Sep 17 00:00:00 2001 From: Padraic Fanning Date: Fri, 22 Apr 2022 20:39:07 -0400 Subject: [PATCH 1/3] Add test_faulthandler.py from CPython 3.10 --- Lib/test/test_faulthandler.py | 889 ++++++++++++++++++++++++++++++++++ 1 file changed, 889 insertions(+) create mode 100644 Lib/test/test_faulthandler.py diff --git a/Lib/test/test_faulthandler.py b/Lib/test/test_faulthandler.py new file mode 100644 index 000000000..e0f09e821 --- /dev/null +++ b/Lib/test/test_faulthandler.py @@ -0,0 +1,889 @@ +from contextlib import contextmanager +import datetime +import faulthandler +import os +import re +import signal +import subprocess +import sys +from test import support +from test.support import os_helper +from test.support import script_helper, is_android +from test.support import skip_if_sanitizer +import tempfile +import unittest +from textwrap import dedent + +try: + import _testcapi +except ImportError: + _testcapi = None + +TIMEOUT = 0.5 +MS_WINDOWS = (os.name == 'nt') + + +def expected_traceback(lineno1, lineno2, header, min_count=1): + regex = header + regex += ' File "", line %s in func\n' % lineno1 + regex += ' File "", line %s in ' % lineno2 + if 1 < min_count: + return '^' + (regex + '\n') * (min_count - 1) + regex + else: + return '^' + regex + '$' + +def skip_segfault_on_android(test): + # Issue #32138: Raising SIGSEGV on Android may not cause a crash. + return unittest.skipIf(is_android, + 'raising SIGSEGV on Android is unreliable')(test) + +@contextmanager +def temporary_filename(): + filename = tempfile.mktemp() + try: + yield filename + finally: + os_helper.unlink(filename) + +class FaultHandlerTests(unittest.TestCase): + def get_output(self, code, filename=None, fd=None): + """ + Run the specified code in Python (in a new child process) and read the + output from the standard error or from a file (if filename is set). + Return the output lines as a list. + + Strip the reference count from the standard error for Python debug + build, and replace "Current thread 0x00007f8d8fbd9700" by "Current + thread XXX". + """ + code = dedent(code).strip() + pass_fds = [] + if fd is not None: + pass_fds.append(fd) + with support.SuppressCrashReport(): + process = script_helper.spawn_python('-c', code, pass_fds=pass_fds) + with process: + output, stderr = process.communicate() + exitcode = process.wait() + output = output.decode('ascii', 'backslashreplace') + if filename: + self.assertEqual(output, '') + with open(filename, "rb") as fp: + output = fp.read() + output = output.decode('ascii', 'backslashreplace') + elif fd is not None: + self.assertEqual(output, '') + os.lseek(fd, os.SEEK_SET, 0) + with open(fd, "rb", closefd=False) as fp: + output = fp.read() + output = output.decode('ascii', 'backslashreplace') + return output.splitlines(), exitcode + + def check_error(self, code, lineno, fatal_error, *, + filename=None, all_threads=True, other_regex=None, + fd=None, know_current_thread=True, + py_fatal_error=False, + garbage_collecting=False, + function=''): + """ + Check that the fault handler for fatal errors is enabled and check the + traceback from the child process output. + + Raise an error if the output doesn't match the expected format. + """ + if all_threads: + if know_current_thread: + header = 'Current thread 0x[0-9a-f]+' + else: + header = 'Thread 0x[0-9a-f]+' + else: + header = 'Stack' + regex = [f'^{fatal_error}'] + if py_fatal_error: + regex.append("Python runtime state: initialized") + regex.append('') + regex.append(fr'{header} \(most recent call first\):') + if garbage_collecting: + regex.append(' Garbage-collecting') + regex.append(fr' File "", line {lineno} in {function}') + regex = '\n'.join(regex) + + if other_regex: + regex = f'(?:{regex}|{other_regex})' + + # Enable MULTILINE flag + regex = f'(?m){regex}' + output, exitcode = self.get_output(code, filename=filename, fd=fd) + output = '\n'.join(output) + self.assertRegex(output, regex) + self.assertNotEqual(exitcode, 0) + + def check_fatal_error(self, code, line_number, name_regex, func=None, **kw): + if func: + name_regex = '%s: %s' % (func, name_regex) + fatal_error = 'Fatal Python error: %s' % name_regex + self.check_error(code, line_number, fatal_error, **kw) + + def check_windows_exception(self, code, line_number, name_regex, **kw): + fatal_error = 'Windows fatal exception: %s' % name_regex + self.check_error(code, line_number, fatal_error, **kw) + + @unittest.skipIf(sys.platform.startswith('aix'), + "the first page of memory is a mapped read-only on AIX") + def test_read_null(self): + if not MS_WINDOWS: + self.check_fatal_error(""" + import faulthandler + faulthandler.enable() + faulthandler._read_null() + """, + 3, + # Issue #12700: Read NULL raises SIGILL on Mac OS X Lion + '(?:Segmentation fault' + '|Bus error' + '|Illegal instruction)') + else: + self.check_windows_exception(""" + import faulthandler + faulthandler.enable() + faulthandler._read_null() + """, + 3, + 'access violation') + + @skip_segfault_on_android + def test_sigsegv(self): + self.check_fatal_error(""" + import faulthandler + faulthandler.enable() + faulthandler._sigsegv() + """, + 3, + 'Segmentation fault') + + @skip_segfault_on_android + def test_gc(self): + # bpo-44466: Detect if the GC is running + self.check_fatal_error(""" + import faulthandler + import gc + import sys + + faulthandler.enable() + + class RefCycle: + def __del__(self): + faulthandler._sigsegv() + + # create a reference cycle which triggers a fatal + # error in a destructor + a = RefCycle() + b = RefCycle() + a.b = b + b.a = a + + # Delete the objects, not the cycle + a = None + b = None + + # Break the reference cycle: call __del__() + gc.collect() + + # Should not reach this line + print("exit", file=sys.stderr) + """, + 9, + 'Segmentation fault', + function='__del__', + garbage_collecting=True) + + def test_fatal_error_c_thread(self): + self.check_fatal_error(""" + import faulthandler + faulthandler.enable() + faulthandler._fatal_error_c_thread() + """, + 3, + 'in new thread', + know_current_thread=False, + func='faulthandler_fatal_error_thread', + py_fatal_error=True) + + def test_sigabrt(self): + self.check_fatal_error(""" + import faulthandler + faulthandler.enable() + faulthandler._sigabrt() + """, + 3, + 'Aborted') + + @unittest.skipIf(sys.platform == 'win32', + "SIGFPE cannot be caught on Windows") + def test_sigfpe(self): + self.check_fatal_error(""" + import faulthandler + faulthandler.enable() + faulthandler._sigfpe() + """, + 3, + 'Floating point exception') + + @unittest.skipIf(_testcapi is None, 'need _testcapi') + @unittest.skipUnless(hasattr(signal, 'SIGBUS'), 'need signal.SIGBUS') + @skip_segfault_on_android + def test_sigbus(self): + self.check_fatal_error(""" + import faulthandler + import signal + + faulthandler.enable() + signal.raise_signal(signal.SIGBUS) + """, + 5, + 'Bus error') + + @unittest.skipIf(_testcapi is None, 'need _testcapi') + @unittest.skipUnless(hasattr(signal, 'SIGILL'), 'need signal.SIGILL') + @skip_segfault_on_android + def test_sigill(self): + self.check_fatal_error(""" + import faulthandler + import signal + + faulthandler.enable() + signal.raise_signal(signal.SIGILL) + """, + 5, + 'Illegal instruction') + + def check_fatal_error_func(self, release_gil): + # Test that Py_FatalError() dumps a traceback + with support.SuppressCrashReport(): + self.check_fatal_error(f""" + import _testcapi + _testcapi.fatal_error(b'xyz', {release_gil}) + """, + 2, + 'xyz', + func='test_fatal_error', + py_fatal_error=True) + + def test_fatal_error(self): + self.check_fatal_error_func(False) + + def test_fatal_error_without_gil(self): + self.check_fatal_error_func(True) + + @unittest.skipIf(sys.platform.startswith('openbsd'), + "Issue #12868: sigaltstack() doesn't work on " + "OpenBSD if Python is compiled with pthread") + @unittest.skipIf(not hasattr(faulthandler, '_stack_overflow'), + 'need faulthandler._stack_overflow()') + def test_stack_overflow(self): + self.check_fatal_error(""" + import faulthandler + faulthandler.enable() + faulthandler._stack_overflow() + """, + 3, + '(?:Segmentation fault|Bus error)', + other_regex='unable to raise a stack overflow') + + @skip_segfault_on_android + def test_gil_released(self): + self.check_fatal_error(""" + import faulthandler + faulthandler.enable() + faulthandler._sigsegv(True) + """, + 3, + 'Segmentation fault') + + @skip_if_sanitizer(memory=True, ub=True, reason="sanitizer " + "builds change crashing process output.") + @skip_segfault_on_android + def test_enable_file(self): + with temporary_filename() as filename: + self.check_fatal_error(""" + import faulthandler + output = open({filename}, 'wb') + faulthandler.enable(output) + faulthandler._sigsegv() + """.format(filename=repr(filename)), + 4, + 'Segmentation fault', + filename=filename) + + @unittest.skipIf(sys.platform == "win32", + "subprocess doesn't support pass_fds on Windows") + @skip_if_sanitizer(memory=True, ub=True, reason="sanitizer " + "builds change crashing process output.") + @skip_segfault_on_android + def test_enable_fd(self): + with tempfile.TemporaryFile('wb+') as fp: + fd = fp.fileno() + self.check_fatal_error(""" + import faulthandler + import sys + faulthandler.enable(%s) + faulthandler._sigsegv() + """ % fd, + 4, + 'Segmentation fault', + fd=fd) + + @skip_segfault_on_android + def test_enable_single_thread(self): + self.check_fatal_error(""" + import faulthandler + faulthandler.enable(all_threads=False) + faulthandler._sigsegv() + """, + 3, + 'Segmentation fault', + all_threads=False) + + @skip_segfault_on_android + def test_disable(self): + code = """ + import faulthandler + faulthandler.enable() + faulthandler.disable() + faulthandler._sigsegv() + """ + not_expected = 'Fatal Python error' + stderr, exitcode = self.get_output(code) + stderr = '\n'.join(stderr) + self.assertTrue(not_expected not in stderr, + "%r is present in %r" % (not_expected, stderr)) + self.assertNotEqual(exitcode, 0) + + @skip_segfault_on_android + def test_dump_ext_modules(self): + code = """ + import faulthandler + import sys + # Don't filter stdlib module names + sys.stdlib_module_names = frozenset() + faulthandler.enable() + faulthandler._sigsegv() + """ + stderr, exitcode = self.get_output(code) + stderr = '\n'.join(stderr) + match = re.search(r'^Extension modules:(.*) \(total: [0-9]+\)$', + stderr, re.MULTILINE) + if not match: + self.fail(f"Cannot find 'Extension modules:' in {stderr!r}") + modules = set(match.group(1).strip().split(', ')) + for name in ('sys', 'faulthandler'): + self.assertIn(name, modules) + + def test_is_enabled(self): + orig_stderr = sys.stderr + try: + # regrtest may replace sys.stderr by io.StringIO object, but + # faulthandler.enable() requires that sys.stderr has a fileno() + # method + sys.stderr = sys.__stderr__ + + was_enabled = faulthandler.is_enabled() + try: + faulthandler.enable() + self.assertTrue(faulthandler.is_enabled()) + faulthandler.disable() + self.assertFalse(faulthandler.is_enabled()) + finally: + if was_enabled: + faulthandler.enable() + else: + faulthandler.disable() + finally: + sys.stderr = orig_stderr + + def test_disabled_by_default(self): + # By default, the module should be disabled + code = "import faulthandler; print(faulthandler.is_enabled())" + args = (sys.executable, "-E", "-c", code) + # don't use assert_python_ok() because it always enables faulthandler + output = subprocess.check_output(args) + self.assertEqual(output.rstrip(), b"False") + + def test_sys_xoptions(self): + # Test python -X faulthandler + code = "import faulthandler; print(faulthandler.is_enabled())" + args = filter(None, (sys.executable, + "-E" if sys.flags.ignore_environment else "", + "-X", "faulthandler", "-c", code)) + env = os.environ.copy() + env.pop("PYTHONFAULTHANDLER", None) + # don't use assert_python_ok() because it always enables faulthandler + output = subprocess.check_output(args, env=env) + self.assertEqual(output.rstrip(), b"True") + + def test_env_var(self): + # empty env var + code = "import faulthandler; print(faulthandler.is_enabled())" + args = (sys.executable, "-c", code) + env = dict(os.environ) + env['PYTHONFAULTHANDLER'] = '' + env['PYTHONDEVMODE'] = '' + # don't use assert_python_ok() because it always enables faulthandler + output = subprocess.check_output(args, env=env) + self.assertEqual(output.rstrip(), b"False") + + # non-empty env var + env = dict(os.environ) + env['PYTHONFAULTHANDLER'] = '1' + env['PYTHONDEVMODE'] = '' + output = subprocess.check_output(args, env=env) + self.assertEqual(output.rstrip(), b"True") + + def check_dump_traceback(self, *, filename=None, fd=None): + """ + Explicitly call dump_traceback() function and check its output. + Raise an error if the output doesn't match the expected format. + """ + code = """ + import faulthandler + + filename = {filename!r} + fd = {fd} + + def funcB(): + if filename: + with open(filename, "wb") as fp: + faulthandler.dump_traceback(fp, all_threads=False) + elif fd is not None: + faulthandler.dump_traceback(fd, + all_threads=False) + else: + faulthandler.dump_traceback(all_threads=False) + + def funcA(): + funcB() + + funcA() + """ + code = code.format( + filename=filename, + fd=fd, + ) + if filename: + lineno = 9 + elif fd is not None: + lineno = 11 + else: + lineno = 14 + expected = [ + 'Stack (most recent call first):', + ' File "", line %s in funcB' % lineno, + ' File "", line 17 in funcA', + ' File "", line 19 in ' + ] + trace, exitcode = self.get_output(code, filename, fd) + self.assertEqual(trace, expected) + self.assertEqual(exitcode, 0) + + def test_dump_traceback(self): + self.check_dump_traceback() + + def test_dump_traceback_file(self): + with temporary_filename() as filename: + self.check_dump_traceback(filename=filename) + + @unittest.skipIf(sys.platform == "win32", + "subprocess doesn't support pass_fds on Windows") + def test_dump_traceback_fd(self): + with tempfile.TemporaryFile('wb+') as fp: + self.check_dump_traceback(fd=fp.fileno()) + + def test_truncate(self): + maxlen = 500 + func_name = 'x' * (maxlen + 50) + truncated = 'x' * maxlen + '...' + code = """ + import faulthandler + + def {func_name}(): + faulthandler.dump_traceback(all_threads=False) + + {func_name}() + """ + code = code.format( + func_name=func_name, + ) + expected = [ + 'Stack (most recent call first):', + ' File "", line 4 in %s' % truncated, + ' File "", line 6 in ' + ] + trace, exitcode = self.get_output(code) + self.assertEqual(trace, expected) + self.assertEqual(exitcode, 0) + + def check_dump_traceback_threads(self, filename): + """ + Call explicitly dump_traceback(all_threads=True) and check the output. + Raise an error if the output doesn't match the expected format. + """ + code = """ + import faulthandler + from threading import Thread, Event + import time + + def dump(): + if {filename}: + with open({filename}, "wb") as fp: + faulthandler.dump_traceback(fp, all_threads=True) + else: + faulthandler.dump_traceback(all_threads=True) + + class Waiter(Thread): + # avoid blocking if the main thread raises an exception. + daemon = True + + def __init__(self): + Thread.__init__(self) + self.running = Event() + self.stop = Event() + + def run(self): + self.running.set() + self.stop.wait() + + waiter = Waiter() + waiter.start() + waiter.running.wait() + dump() + waiter.stop.set() + waiter.join() + """ + code = code.format(filename=repr(filename)) + output, exitcode = self.get_output(code, filename) + output = '\n'.join(output) + if filename: + lineno = 8 + else: + lineno = 10 + regex = r""" + ^Thread 0x[0-9a-f]+ \(most recent call first\): + (?: File ".*threading.py", line [0-9]+ in [_a-z]+ + ){{1,3}} File "", line 23 in run + File ".*threading.py", line [0-9]+ in _bootstrap_inner + File ".*threading.py", line [0-9]+ in _bootstrap + + Current thread 0x[0-9a-f]+ \(most recent call first\): + File "", line {lineno} in dump + File "", line 28 in $ + """ + regex = dedent(regex.format(lineno=lineno)).strip() + self.assertRegex(output, regex) + self.assertEqual(exitcode, 0) + + def test_dump_traceback_threads(self): + self.check_dump_traceback_threads(None) + + def test_dump_traceback_threads_file(self): + with temporary_filename() as filename: + self.check_dump_traceback_threads(filename) + + def check_dump_traceback_later(self, repeat=False, cancel=False, loops=1, + *, filename=None, fd=None): + """ + Check how many times the traceback is written in timeout x 2.5 seconds, + or timeout x 3.5 seconds if cancel is True: 1, 2 or 3 times depending + on repeat and cancel options. + + Raise an error if the output doesn't match the expect format. + """ + timeout_str = str(datetime.timedelta(seconds=TIMEOUT)) + code = """ + import faulthandler + import time + import sys + + timeout = {timeout} + repeat = {repeat} + cancel = {cancel} + loops = {loops} + filename = {filename!r} + fd = {fd} + + def func(timeout, repeat, cancel, file, loops): + for loop in range(loops): + faulthandler.dump_traceback_later(timeout, repeat=repeat, file=file) + if cancel: + faulthandler.cancel_dump_traceback_later() + time.sleep(timeout * 5) + faulthandler.cancel_dump_traceback_later() + + if filename: + file = open(filename, "wb") + elif fd is not None: + file = sys.stderr.fileno() + else: + file = None + func(timeout, repeat, cancel, file, loops) + if filename: + file.close() + """ + code = code.format( + timeout=TIMEOUT, + repeat=repeat, + cancel=cancel, + loops=loops, + filename=filename, + fd=fd, + ) + trace, exitcode = self.get_output(code, filename) + trace = '\n'.join(trace) + + if not cancel: + count = loops + if repeat: + count *= 2 + header = r'Timeout \(%s\)!\nThread 0x[0-9a-f]+ \(most recent call first\):\n' % timeout_str + regex = expected_traceback(17, 26, header, min_count=count) + self.assertRegex(trace, regex) + else: + self.assertEqual(trace, '') + self.assertEqual(exitcode, 0) + + def test_dump_traceback_later(self): + self.check_dump_traceback_later() + + def test_dump_traceback_later_repeat(self): + self.check_dump_traceback_later(repeat=True) + + def test_dump_traceback_later_cancel(self): + self.check_dump_traceback_later(cancel=True) + + def test_dump_traceback_later_file(self): + with temporary_filename() as filename: + self.check_dump_traceback_later(filename=filename) + + @unittest.skipIf(sys.platform == "win32", + "subprocess doesn't support pass_fds on Windows") + def test_dump_traceback_later_fd(self): + with tempfile.TemporaryFile('wb+') as fp: + self.check_dump_traceback_later(fd=fp.fileno()) + + def test_dump_traceback_later_twice(self): + self.check_dump_traceback_later(loops=2) + + @unittest.skipIf(not hasattr(faulthandler, "register"), + "need faulthandler.register") + def check_register(self, filename=False, all_threads=False, + unregister=False, chain=False, fd=None): + """ + Register a handler displaying the traceback on a user signal. Raise the + signal and check the written traceback. + + If chain is True, check that the previous signal handler is called. + + Raise an error if the output doesn't match the expected format. + """ + signum = signal.SIGUSR1 + code = """ + import faulthandler + import os + import signal + import sys + + all_threads = {all_threads} + signum = {signum:d} + unregister = {unregister} + chain = {chain} + filename = {filename!r} + fd = {fd} + + def func(signum): + os.kill(os.getpid(), signum) + + def handler(signum, frame): + handler.called = True + handler.called = False + + if filename: + file = open(filename, "wb") + elif fd is not None: + file = sys.stderr.fileno() + else: + file = None + if chain: + signal.signal(signum, handler) + faulthandler.register(signum, file=file, + all_threads=all_threads, chain={chain}) + if unregister: + faulthandler.unregister(signum) + func(signum) + if chain and not handler.called: + if file is not None: + output = file + else: + output = sys.stderr + print("Error: signal handler not called!", file=output) + exitcode = 1 + else: + exitcode = 0 + if filename: + file.close() + sys.exit(exitcode) + """ + code = code.format( + all_threads=all_threads, + signum=signum, + unregister=unregister, + chain=chain, + filename=filename, + fd=fd, + ) + trace, exitcode = self.get_output(code, filename) + trace = '\n'.join(trace) + if not unregister: + if all_threads: + regex = r'Current thread 0x[0-9a-f]+ \(most recent call first\):\n' + else: + regex = r'Stack \(most recent call first\):\n' + regex = expected_traceback(14, 32, regex) + self.assertRegex(trace, regex) + else: + self.assertEqual(trace, '') + if unregister: + self.assertNotEqual(exitcode, 0) + else: + self.assertEqual(exitcode, 0) + + def test_register(self): + self.check_register() + + def test_unregister(self): + self.check_register(unregister=True) + + def test_register_file(self): + with temporary_filename() as filename: + self.check_register(filename=filename) + + @unittest.skipIf(sys.platform == "win32", + "subprocess doesn't support pass_fds on Windows") + def test_register_fd(self): + with tempfile.TemporaryFile('wb+') as fp: + self.check_register(fd=fp.fileno()) + + def test_register_threads(self): + self.check_register(all_threads=True) + + def test_register_chain(self): + self.check_register(chain=True) + + @contextmanager + def check_stderr_none(self): + stderr = sys.stderr + try: + sys.stderr = None + with self.assertRaises(RuntimeError) as cm: + yield + self.assertEqual(str(cm.exception), "sys.stderr is None") + finally: + sys.stderr = stderr + + def test_stderr_None(self): + # Issue #21497: provide a helpful error if sys.stderr is None, + # instead of just an attribute error: "None has no attribute fileno". + with self.check_stderr_none(): + faulthandler.enable() + with self.check_stderr_none(): + faulthandler.dump_traceback() + with self.check_stderr_none(): + faulthandler.dump_traceback_later(1e-3) + if hasattr(faulthandler, "register"): + with self.check_stderr_none(): + faulthandler.register(signal.SIGUSR1) + + @unittest.skipUnless(MS_WINDOWS, 'specific to Windows') + def test_raise_exception(self): + for exc, name in ( + ('EXCEPTION_ACCESS_VIOLATION', 'access violation'), + ('EXCEPTION_INT_DIVIDE_BY_ZERO', 'int divide by zero'), + ('EXCEPTION_STACK_OVERFLOW', 'stack overflow'), + ): + self.check_windows_exception(f""" + import faulthandler + faulthandler.enable() + faulthandler._raise_exception(faulthandler._{exc}) + """, + 3, + name) + + @unittest.skipUnless(MS_WINDOWS, 'specific to Windows') + def test_ignore_exception(self): + for exc_code in ( + 0xE06D7363, # MSC exception ("Emsc") + 0xE0434352, # COM Callable Runtime exception ("ECCR") + ): + code = f""" + import faulthandler + faulthandler.enable() + faulthandler._raise_exception({exc_code}) + """ + code = dedent(code) + output, exitcode = self.get_output(code) + self.assertEqual(output, []) + self.assertEqual(exitcode, exc_code) + + @unittest.skipUnless(MS_WINDOWS, 'specific to Windows') + def test_raise_nonfatal_exception(self): + # These exceptions are not strictly errors. Letting + # faulthandler display the traceback when they are + # raised is likely to result in noise. However, they + # may still terminate the process if there is no + # handler installed for them (which there typically + # is, e.g. for debug messages). + for exc in ( + 0x00000000, + 0x34567890, + 0x40000000, + 0x40001000, + 0x70000000, + 0x7FFFFFFF, + ): + output, exitcode = self.get_output(f""" + import faulthandler + faulthandler.enable() + faulthandler._raise_exception(0x{exc:x}) + """ + ) + self.assertEqual(output, []) + # On Windows older than 7 SP1, the actual exception code has + # bit 29 cleared. + self.assertIn(exitcode, + (exc, exc & ~0x10000000)) + + @unittest.skipUnless(MS_WINDOWS, 'specific to Windows') + def test_disable_windows_exc_handler(self): + code = dedent(""" + import faulthandler + faulthandler.enable() + faulthandler.disable() + code = faulthandler._EXCEPTION_ACCESS_VIOLATION + faulthandler._raise_exception(code) + """) + output, exitcode = self.get_output(code) + self.assertEqual(output, []) + self.assertEqual(exitcode, 0xC0000005) + + def test_cancel_later_without_dump_traceback_later(self): + # bpo-37933: Calling cancel_dump_traceback_later() + # without dump_traceback_later() must not segfault. + code = dedent(""" + import faulthandler + faulthandler.cancel_dump_traceback_later() + """) + output, exitcode = self.get_output(code) + self.assertEqual(output, []) + self.assertEqual(exitcode, 0) + + +if __name__ == "__main__": + unittest.main() From 62673fc10c2127689d9f58c94e59a8b5df9a1df0 Mon Sep 17 00:00:00 2001 From: Padraic Fanning Date: Fri, 22 Apr 2022 21:01:42 -0400 Subject: [PATCH 2/3] Mark erroring/failing tests --- Lib/test/test_faulthandler.py | 74 +++++++++++++++++++++++++++++++++++ 1 file changed, 74 insertions(+) diff --git a/Lib/test/test_faulthandler.py b/Lib/test/test_faulthandler.py index e0f09e821..f2bc27262 100644 --- a/Lib/test/test_faulthandler.py +++ b/Lib/test/test_faulthandler.py @@ -128,6 +128,8 @@ class FaultHandlerTests(unittest.TestCase): fatal_error = 'Windows fatal exception: %s' % name_regex self.check_error(code, line_number, fatal_error, **kw) + # TODO: RUSTPYTHON + @unittest.expectedFailure @unittest.skipIf(sys.platform.startswith('aix'), "the first page of memory is a mapped read-only on AIX") def test_read_null(self): @@ -151,6 +153,8 @@ class FaultHandlerTests(unittest.TestCase): 3, 'access violation') + # TODO: RUSTPYTHON, AssertionError: Regex didn't match + @unittest.expectedFailure @skip_segfault_on_android def test_sigsegv(self): self.check_fatal_error(""" @@ -161,6 +165,8 @@ class FaultHandlerTests(unittest.TestCase): 3, 'Segmentation fault') + # TODO: RUSTPYTHON + @unittest.expectedFailure @skip_segfault_on_android def test_gc(self): # bpo-44466: Detect if the GC is running @@ -197,6 +203,8 @@ class FaultHandlerTests(unittest.TestCase): function='__del__', garbage_collecting=True) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_fatal_error_c_thread(self): self.check_fatal_error(""" import faulthandler @@ -209,6 +217,8 @@ class FaultHandlerTests(unittest.TestCase): func='faulthandler_fatal_error_thread', py_fatal_error=True) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_sigabrt(self): self.check_fatal_error(""" import faulthandler @@ -218,6 +228,8 @@ class FaultHandlerTests(unittest.TestCase): 3, 'Aborted') + # TODO: RUSTPYTHON + @unittest.expectedFailure @unittest.skipIf(sys.platform == 'win32', "SIGFPE cannot be caught on Windows") def test_sigfpe(self): @@ -269,9 +281,13 @@ class FaultHandlerTests(unittest.TestCase): func='test_fatal_error', py_fatal_error=True) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_fatal_error(self): self.check_fatal_error_func(False) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_fatal_error_without_gil(self): self.check_fatal_error_func(True) @@ -290,6 +306,8 @@ class FaultHandlerTests(unittest.TestCase): '(?:Segmentation fault|Bus error)', other_regex='unable to raise a stack overflow') + # TODO: RUSTPYTHON + @unittest.expectedFailure @skip_segfault_on_android def test_gil_released(self): self.check_fatal_error(""" @@ -300,6 +318,8 @@ class FaultHandlerTests(unittest.TestCase): 3, 'Segmentation fault') + # TODO: RUSTPYTHON + @unittest.expectedFailure @skip_if_sanitizer(memory=True, ub=True, reason="sanitizer " "builds change crashing process output.") @skip_segfault_on_android @@ -315,6 +335,8 @@ class FaultHandlerTests(unittest.TestCase): 'Segmentation fault', filename=filename) + # TODO: RUSTPYTHON + @unittest.expectedFailure @unittest.skipIf(sys.platform == "win32", "subprocess doesn't support pass_fds on Windows") @skip_if_sanitizer(memory=True, ub=True, reason="sanitizer " @@ -333,6 +355,8 @@ class FaultHandlerTests(unittest.TestCase): 'Segmentation fault', fd=fd) + # TODO: RUSTPYTHON + @unittest.expectedFailure @skip_segfault_on_android def test_enable_single_thread(self): self.check_fatal_error(""" @@ -359,6 +383,8 @@ class FaultHandlerTests(unittest.TestCase): "%r is present in %r" % (not_expected, stderr)) self.assertNotEqual(exitcode, 0) + # TODO: RUSTPYTHON + @unittest.expectedFailure @skip_segfault_on_android def test_dump_ext_modules(self): code = """ @@ -379,6 +405,8 @@ class FaultHandlerTests(unittest.TestCase): for name in ('sys', 'faulthandler'): self.assertIn(name, modules) + # TODO: RUSTPYTHON, AttributeError: module 'faulthandler' has no attribute 'is_enabled' + @unittest.expectedFailure def test_is_enabled(self): orig_stderr = sys.stderr try: @@ -401,6 +429,8 @@ class FaultHandlerTests(unittest.TestCase): finally: sys.stderr = orig_stderr + # TODO: RUSTPYTHON, subprocess.CalledProcessError: Command ... returned non-zero exit status 1. + @unittest.expectedFailure def test_disabled_by_default(self): # By default, the module should be disabled code = "import faulthandler; print(faulthandler.is_enabled())" @@ -409,6 +439,8 @@ class FaultHandlerTests(unittest.TestCase): output = subprocess.check_output(args) self.assertEqual(output.rstrip(), b"False") + # TODO: RUSTPYTHON, subprocess.CalledProcessError: Command '' returned non-zero exit status 1. + @unittest.expectedFailure def test_sys_xoptions(self): # Test python -X faulthandler code = "import faulthandler; print(faulthandler.is_enabled())" @@ -421,6 +453,8 @@ class FaultHandlerTests(unittest.TestCase): output = subprocess.check_output(args, env=env) self.assertEqual(output.rstrip(), b"True") + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_env_var(self): # empty env var code = "import faulthandler; print(faulthandler.is_enabled())" @@ -485,19 +519,27 @@ class FaultHandlerTests(unittest.TestCase): self.assertEqual(trace, expected) self.assertEqual(exitcode, 0) + # TODO: RUSTPYTHON, AssertionError: Lists differ + @unittest.expectedFailure def test_dump_traceback(self): self.check_dump_traceback() + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_dump_traceback_file(self): with temporary_filename() as filename: self.check_dump_traceback(filename=filename) + # TODO: RUSTPYTHON + @unittest.expectedFailure @unittest.skipIf(sys.platform == "win32", "subprocess doesn't support pass_fds on Windows") def test_dump_traceback_fd(self): with tempfile.TemporaryFile('wb+') as fp: self.check_dump_traceback(fd=fp.fileno()) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_truncate(self): maxlen = 500 func_name = 'x' * (maxlen + 50) @@ -581,9 +623,13 @@ class FaultHandlerTests(unittest.TestCase): self.assertRegex(output, regex) self.assertEqual(exitcode, 0) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_dump_traceback_threads(self): self.check_dump_traceback_threads(None) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_dump_traceback_threads_file(self): with temporary_filename() as filename: self.check_dump_traceback_threads(filename) @@ -650,25 +696,37 @@ class FaultHandlerTests(unittest.TestCase): self.assertEqual(trace, '') self.assertEqual(exitcode, 0) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_dump_traceback_later(self): self.check_dump_traceback_later() + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_dump_traceback_later_repeat(self): self.check_dump_traceback_later(repeat=True) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_dump_traceback_later_cancel(self): self.check_dump_traceback_later(cancel=True) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_dump_traceback_later_file(self): with temporary_filename() as filename: self.check_dump_traceback_later(filename=filename) + # TODO: RUSTPYTHON + @unittest.expectedFailure @unittest.skipIf(sys.platform == "win32", "subprocess doesn't support pass_fds on Windows") def test_dump_traceback_later_fd(self): with tempfile.TemporaryFile('wb+') as fp: self.check_dump_traceback_later(fd=fp.fileno()) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_dump_traceback_later_twice(self): self.check_dump_traceback_later(loops=2) @@ -755,25 +813,37 @@ class FaultHandlerTests(unittest.TestCase): else: self.assertEqual(exitcode, 0) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_register(self): self.check_register() + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_unregister(self): self.check_register(unregister=True) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_register_file(self): with temporary_filename() as filename: self.check_register(filename=filename) + # TODO: RUSTPYTHON + @unittest.expectedFailure @unittest.skipIf(sys.platform == "win32", "subprocess doesn't support pass_fds on Windows") def test_register_fd(self): with tempfile.TemporaryFile('wb+') as fp: self.check_register(fd=fp.fileno()) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_register_threads(self): self.check_register(all_threads=True) + # TODO: RUSTPYTHON + @unittest.expectedFailure def test_register_chain(self): self.check_register(chain=True) @@ -788,6 +858,8 @@ class FaultHandlerTests(unittest.TestCase): finally: sys.stderr = stderr + # TODO: RUSTPYTHON, AssertionError: RuntimeError not raised + @unittest.expectedFailure def test_stderr_None(self): # Issue #21497: provide a helpful error if sys.stderr is None, # instead of just an attribute error: "None has no attribute fileno". @@ -873,6 +945,8 @@ class FaultHandlerTests(unittest.TestCase): self.assertEqual(output, []) self.assertEqual(exitcode, 0xC0000005) + # TODO: RUSTPYTHON, AssertionError: Lists differ + @unittest.expectedFailure def test_cancel_later_without_dump_traceback_later(self): # bpo-37933: Calling cancel_dump_traceback_later() # without dump_traceback_later() must not segfault. From 5266e93e2a769b1dbf3c70658f4e8de556084377 Mon Sep 17 00:00:00 2001 From: Padraic Fanning Date: Fri, 22 Apr 2022 22:45:55 -0400 Subject: [PATCH 3/3] Mark erroring tests on Windows --- Lib/test/test_faulthandler.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/Lib/test/test_faulthandler.py b/Lib/test/test_faulthandler.py index f2bc27262..5e73f7381 100644 --- a/Lib/test/test_faulthandler.py +++ b/Lib/test/test_faulthandler.py @@ -383,6 +383,9 @@ class FaultHandlerTests(unittest.TestCase): "%r is present in %r" % (not_expected, stderr)) self.assertNotEqual(exitcode, 0) + if sys.platform == "win32": # TODO: RUSTPYTHON, AttributeError: module 'msvcrt' has no attribute 'GetErrorMode' + test_disable = unittest.expectedFailure(test_disable) + # TODO: RUSTPYTHON @unittest.expectedFailure @skip_segfault_on_android @@ -873,6 +876,8 @@ class FaultHandlerTests(unittest.TestCase): with self.check_stderr_none(): faulthandler.register(signal.SIGUSR1) + # TODO: RUSTPYTHON, AttributeError: module 'msvcrt' has no attribute 'GetErrorMode' + @unittest.expectedFailure @unittest.skipUnless(MS_WINDOWS, 'specific to Windows') def test_raise_exception(self): for exc, name in ( @@ -888,6 +893,8 @@ class FaultHandlerTests(unittest.TestCase): 3, name) + # TODO: RUSTPYTHON, AttributeError: module 'msvcrt' has no attribute 'GetErrorMode' + @unittest.expectedFailure @unittest.skipUnless(MS_WINDOWS, 'specific to Windows') def test_ignore_exception(self): for exc_code in ( @@ -904,6 +911,8 @@ class FaultHandlerTests(unittest.TestCase): self.assertEqual(output, []) self.assertEqual(exitcode, exc_code) + # TODO: RUSTPYTHON, AttributeError: module 'msvcrt' has no attribute 'GetErrorMode' + @unittest.expectedFailure @unittest.skipUnless(MS_WINDOWS, 'specific to Windows') def test_raise_nonfatal_exception(self): # These exceptions are not strictly errors. Letting @@ -932,6 +941,8 @@ class FaultHandlerTests(unittest.TestCase): self.assertIn(exitcode, (exc, exc & ~0x10000000)) + # TODO: RUSTPYTHON, AttributeError: module 'msvcrt' has no attribute 'GetErrorMode' + @unittest.expectedFailure @unittest.skipUnless(MS_WINDOWS, 'specific to Windows') def test_disable_windows_exc_handler(self): code = dedent("""