• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1from contextlib import contextmanager
2import datetime
3import faulthandler
4import os
5import re
6import signal
7import subprocess
8import sys
9from test import support
10from test.support import os_helper
11from test.support import script_helper, is_android
12from test.support import skip_if_sanitizer
13import tempfile
14import unittest
15from textwrap import dedent
16
17try:
18    import _testcapi
19except ImportError:
20    _testcapi = None
21
22TIMEOUT = 0.5
23MS_WINDOWS = (os.name == 'nt')
24
25
26def expected_traceback(lineno1, lineno2, header, min_count=1):
27    regex = header
28    regex += '  File "<string>", line %s in func\n' % lineno1
29    regex += '  File "<string>", line %s in <module>' % lineno2
30    if 1 < min_count:
31        return '^' + (regex + '\n') * (min_count - 1) + regex
32    else:
33        return '^' + regex + '$'
34
35def skip_segfault_on_android(test):
36    # Issue #32138: Raising SIGSEGV on Android may not cause a crash.
37    return unittest.skipIf(is_android,
38                           'raising SIGSEGV on Android is unreliable')(test)
39
40@contextmanager
41def temporary_filename():
42    filename = tempfile.mktemp()
43    try:
44        yield filename
45    finally:
46        os_helper.unlink(filename)
47
48class FaultHandlerTests(unittest.TestCase):
49    def get_output(self, code, filename=None, fd=None):
50        """
51        Run the specified code in Python (in a new child process) and read the
52        output from the standard error or from a file (if filename is set).
53        Return the output lines as a list.
54
55        Strip the reference count from the standard error for Python debug
56        build, and replace "Current thread 0x00007f8d8fbd9700" by "Current
57        thread XXX".
58        """
59        code = dedent(code).strip()
60        pass_fds = []
61        if fd is not None:
62            pass_fds.append(fd)
63        with support.SuppressCrashReport():
64            process = script_helper.spawn_python('-c', code, pass_fds=pass_fds)
65            with process:
66                output, stderr = process.communicate()
67                exitcode = process.wait()
68        output = output.decode('ascii', 'backslashreplace')
69        if filename:
70            self.assertEqual(output, '')
71            with open(filename, "rb") as fp:
72                output = fp.read()
73            output = output.decode('ascii', 'backslashreplace')
74        elif fd is not None:
75            self.assertEqual(output, '')
76            os.lseek(fd, os.SEEK_SET, 0)
77            with open(fd, "rb", closefd=False) as fp:
78                output = fp.read()
79            output = output.decode('ascii', 'backslashreplace')
80        return output.splitlines(), exitcode
81
82    def check_error(self, code, lineno, fatal_error, *,
83                    filename=None, all_threads=True, other_regex=None,
84                    fd=None, know_current_thread=True,
85                    py_fatal_error=False,
86                    garbage_collecting=False,
87                    function='<module>'):
88        """
89        Check that the fault handler for fatal errors is enabled and check the
90        traceback from the child process output.
91
92        Raise an error if the output doesn't match the expected format.
93        """
94        if all_threads:
95            if know_current_thread:
96                header = 'Current thread 0x[0-9a-f]+'
97            else:
98                header = 'Thread 0x[0-9a-f]+'
99        else:
100            header = 'Stack'
101        regex = [f'^{fatal_error}']
102        if py_fatal_error:
103            regex.append("Python runtime state: initialized")
104        regex.append('')
105        regex.append(fr'{header} \(most recent call first\):')
106        if garbage_collecting:
107            regex.append('  Garbage-collecting')
108        regex.append(fr'  File "<string>", line {lineno} in {function}')
109        regex = '\n'.join(regex)
110
111        if other_regex:
112            regex = f'(?:{regex}|{other_regex})'
113
114        # Enable MULTILINE flag
115        regex = f'(?m){regex}'
116        output, exitcode = self.get_output(code, filename=filename, fd=fd)
117        output = '\n'.join(output)
118        self.assertRegex(output, regex)
119        self.assertNotEqual(exitcode, 0)
120
121    def check_fatal_error(self, code, line_number, name_regex, func=None, **kw):
122        if func:
123            name_regex = '%s: %s' % (func, name_regex)
124        fatal_error = 'Fatal Python error: %s' % name_regex
125        self.check_error(code, line_number, fatal_error, **kw)
126
127    def check_windows_exception(self, code, line_number, name_regex, **kw):
128        fatal_error = 'Windows fatal exception: %s' % name_regex
129        self.check_error(code, line_number, fatal_error, **kw)
130
131    @unittest.skipIf(sys.platform.startswith('aix'),
132                     "the first page of memory is a mapped read-only on AIX")
133    def test_read_null(self):
134        if not MS_WINDOWS:
135            self.check_fatal_error("""
136                import faulthandler
137                faulthandler.enable()
138                faulthandler._read_null()
139                """,
140                3,
141                # Issue #12700: Read NULL raises SIGILL on Mac OS X Lion
142                '(?:Segmentation fault'
143                    '|Bus error'
144                    '|Illegal instruction)')
145        else:
146            self.check_windows_exception("""
147                import faulthandler
148                faulthandler.enable()
149                faulthandler._read_null()
150                """,
151                3,
152                'access violation')
153
154    @skip_segfault_on_android
155    def test_sigsegv(self):
156        self.check_fatal_error("""
157            import faulthandler
158            faulthandler.enable()
159            faulthandler._sigsegv()
160            """,
161            3,
162            'Segmentation fault')
163
164    @skip_segfault_on_android
165    def test_gc(self):
166        # bpo-44466: Detect if the GC is running
167        self.check_fatal_error("""
168            import faulthandler
169            import gc
170            import sys
171
172            faulthandler.enable()
173
174            class RefCycle:
175                def __del__(self):
176                    faulthandler._sigsegv()
177
178            # create a reference cycle which triggers a fatal
179            # error in a destructor
180            a = RefCycle()
181            b = RefCycle()
182            a.b = b
183            b.a = a
184
185            # Delete the objects, not the cycle
186            a = None
187            b = None
188
189            # Break the reference cycle: call __del__()
190            gc.collect()
191
192            # Should not reach this line
193            print("exit", file=sys.stderr)
194            """,
195            9,
196            'Segmentation fault',
197            function='__del__',
198            garbage_collecting=True)
199
200    def test_fatal_error_c_thread(self):
201        self.check_fatal_error("""
202            import faulthandler
203            faulthandler.enable()
204            faulthandler._fatal_error_c_thread()
205            """,
206            3,
207            'in new thread',
208            know_current_thread=False,
209            func='faulthandler_fatal_error_thread',
210            py_fatal_error=True)
211
212    def test_sigabrt(self):
213        self.check_fatal_error("""
214            import faulthandler
215            faulthandler.enable()
216            faulthandler._sigabrt()
217            """,
218            3,
219            'Aborted')
220
221    @unittest.skipIf(sys.platform == 'win32',
222                     "SIGFPE cannot be caught on Windows")
223    def test_sigfpe(self):
224        self.check_fatal_error("""
225            import faulthandler
226            faulthandler.enable()
227            faulthandler._sigfpe()
228            """,
229            3,
230            'Floating point exception')
231
232    @unittest.skipIf(_testcapi is None, 'need _testcapi')
233    @unittest.skipUnless(hasattr(signal, 'SIGBUS'), 'need signal.SIGBUS')
234    @skip_segfault_on_android
235    def test_sigbus(self):
236        self.check_fatal_error("""
237            import faulthandler
238            import signal
239
240            faulthandler.enable()
241            signal.raise_signal(signal.SIGBUS)
242            """,
243            5,
244            'Bus error')
245
246    @unittest.skipIf(_testcapi is None, 'need _testcapi')
247    @unittest.skipUnless(hasattr(signal, 'SIGILL'), 'need signal.SIGILL')
248    @skip_segfault_on_android
249    def test_sigill(self):
250        self.check_fatal_error("""
251            import faulthandler
252            import signal
253
254            faulthandler.enable()
255            signal.raise_signal(signal.SIGILL)
256            """,
257            5,
258            'Illegal instruction')
259
260    def check_fatal_error_func(self, release_gil):
261        # Test that Py_FatalError() dumps a traceback
262        with support.SuppressCrashReport():
263            self.check_fatal_error(f"""
264                import _testcapi
265                _testcapi.fatal_error(b'xyz', {release_gil})
266                """,
267                2,
268                'xyz',
269                func='test_fatal_error',
270                py_fatal_error=True)
271
272    def test_fatal_error(self):
273        self.check_fatal_error_func(False)
274
275    def test_fatal_error_without_gil(self):
276        self.check_fatal_error_func(True)
277
278    @unittest.skipIf(sys.platform.startswith('openbsd'),
279                     "Issue #12868: sigaltstack() doesn't work on "
280                     "OpenBSD if Python is compiled with pthread")
281    @unittest.skipIf(not hasattr(faulthandler, '_stack_overflow'),
282                     'need faulthandler._stack_overflow()')
283    def test_stack_overflow(self):
284        self.check_fatal_error("""
285            import faulthandler
286            faulthandler.enable()
287            faulthandler._stack_overflow()
288            """,
289            3,
290            '(?:Segmentation fault|Bus error)',
291            other_regex='unable to raise a stack overflow')
292
293    @skip_segfault_on_android
294    def test_gil_released(self):
295        self.check_fatal_error("""
296            import faulthandler
297            faulthandler.enable()
298            faulthandler._sigsegv(True)
299            """,
300            3,
301            'Segmentation fault')
302
303    @skip_if_sanitizer(memory=True, ub=True, reason="sanitizer "
304                       "builds change crashing process output.")
305    @skip_segfault_on_android
306    def test_enable_file(self):
307        with temporary_filename() as filename:
308            self.check_fatal_error("""
309                import faulthandler
310                output = open({filename}, 'wb')
311                faulthandler.enable(output)
312                faulthandler._sigsegv()
313                """.format(filename=repr(filename)),
314                4,
315                'Segmentation fault',
316                filename=filename)
317
318    @unittest.skipIf(sys.platform == "win32",
319                     "subprocess doesn't support pass_fds on Windows")
320    @skip_if_sanitizer(memory=True, ub=True, reason="sanitizer "
321                       "builds change crashing process output.")
322    @skip_segfault_on_android
323    def test_enable_fd(self):
324        with tempfile.TemporaryFile('wb+') as fp:
325            fd = fp.fileno()
326            self.check_fatal_error("""
327                import faulthandler
328                import sys
329                faulthandler.enable(%s)
330                faulthandler._sigsegv()
331                """ % fd,
332                4,
333                'Segmentation fault',
334                fd=fd)
335
336    @skip_segfault_on_android
337    def test_enable_single_thread(self):
338        self.check_fatal_error("""
339            import faulthandler
340            faulthandler.enable(all_threads=False)
341            faulthandler._sigsegv()
342            """,
343            3,
344            'Segmentation fault',
345            all_threads=False)
346
347    @skip_segfault_on_android
348    def test_disable(self):
349        code = """
350            import faulthandler
351            faulthandler.enable()
352            faulthandler.disable()
353            faulthandler._sigsegv()
354            """
355        not_expected = 'Fatal Python error'
356        stderr, exitcode = self.get_output(code)
357        stderr = '\n'.join(stderr)
358        self.assertTrue(not_expected not in stderr,
359                     "%r is present in %r" % (not_expected, stderr))
360        self.assertNotEqual(exitcode, 0)
361
362    @skip_segfault_on_android
363    def test_dump_ext_modules(self):
364        code = """
365            import faulthandler
366            import sys
367            # Don't filter stdlib module names
368            sys.stdlib_module_names = frozenset()
369            faulthandler.enable()
370            faulthandler._sigsegv()
371            """
372        stderr, exitcode = self.get_output(code)
373        stderr = '\n'.join(stderr)
374        match = re.search(r'^Extension modules:(.*) \(total: [0-9]+\)$',
375                          stderr, re.MULTILINE)
376        if not match:
377            self.fail(f"Cannot find 'Extension modules:' in {stderr!r}")
378        modules = set(match.group(1).strip().split(', '))
379        for name in ('sys', 'faulthandler'):
380            self.assertIn(name, modules)
381
382    def test_is_enabled(self):
383        orig_stderr = sys.stderr
384        try:
385            # regrtest may replace sys.stderr by io.StringIO object, but
386            # faulthandler.enable() requires that sys.stderr has a fileno()
387            # method
388            sys.stderr = sys.__stderr__
389
390            was_enabled = faulthandler.is_enabled()
391            try:
392                faulthandler.enable()
393                self.assertTrue(faulthandler.is_enabled())
394                faulthandler.disable()
395                self.assertFalse(faulthandler.is_enabled())
396            finally:
397                if was_enabled:
398                    faulthandler.enable()
399                else:
400                    faulthandler.disable()
401        finally:
402            sys.stderr = orig_stderr
403
404    def test_disabled_by_default(self):
405        # By default, the module should be disabled
406        code = "import faulthandler; print(faulthandler.is_enabled())"
407        args = (sys.executable, "-E", "-c", code)
408        # don't use assert_python_ok() because it always enables faulthandler
409        output = subprocess.check_output(args)
410        self.assertEqual(output.rstrip(), b"False")
411
412    def test_sys_xoptions(self):
413        # Test python -X faulthandler
414        code = "import faulthandler; print(faulthandler.is_enabled())"
415        args = filter(None, (sys.executable,
416                             "-E" if sys.flags.ignore_environment else "",
417                             "-X", "faulthandler", "-c", code))
418        env = os.environ.copy()
419        env.pop("PYTHONFAULTHANDLER", None)
420        # don't use assert_python_ok() because it always enables faulthandler
421        output = subprocess.check_output(args, env=env)
422        self.assertEqual(output.rstrip(), b"True")
423
424    def test_env_var(self):
425        # empty env var
426        code = "import faulthandler; print(faulthandler.is_enabled())"
427        args = (sys.executable, "-c", code)
428        env = dict(os.environ)
429        env['PYTHONFAULTHANDLER'] = ''
430        env['PYTHONDEVMODE'] = ''
431        # don't use assert_python_ok() because it always enables faulthandler
432        output = subprocess.check_output(args, env=env)
433        self.assertEqual(output.rstrip(), b"False")
434
435        # non-empty env var
436        env = dict(os.environ)
437        env['PYTHONFAULTHANDLER'] = '1'
438        env['PYTHONDEVMODE'] = ''
439        output = subprocess.check_output(args, env=env)
440        self.assertEqual(output.rstrip(), b"True")
441
442    def check_dump_traceback(self, *, filename=None, fd=None):
443        """
444        Explicitly call dump_traceback() function and check its output.
445        Raise an error if the output doesn't match the expected format.
446        """
447        code = """
448            import faulthandler
449
450            filename = {filename!r}
451            fd = {fd}
452
453            def funcB():
454                if filename:
455                    with open(filename, "wb") as fp:
456                        faulthandler.dump_traceback(fp, all_threads=False)
457                elif fd is not None:
458                    faulthandler.dump_traceback(fd,
459                                                all_threads=False)
460                else:
461                    faulthandler.dump_traceback(all_threads=False)
462
463            def funcA():
464                funcB()
465
466            funcA()
467            """
468        code = code.format(
469            filename=filename,
470            fd=fd,
471        )
472        if filename:
473            lineno = 9
474        elif fd is not None:
475            lineno = 11
476        else:
477            lineno = 14
478        expected = [
479            'Stack (most recent call first):',
480            '  File "<string>", line %s in funcB' % lineno,
481            '  File "<string>", line 17 in funcA',
482            '  File "<string>", line 19 in <module>'
483        ]
484        trace, exitcode = self.get_output(code, filename, fd)
485        self.assertEqual(trace, expected)
486        self.assertEqual(exitcode, 0)
487
488    def test_dump_traceback(self):
489        self.check_dump_traceback()
490
491    def test_dump_traceback_file(self):
492        with temporary_filename() as filename:
493            self.check_dump_traceback(filename=filename)
494
495    @unittest.skipIf(sys.platform == "win32",
496                     "subprocess doesn't support pass_fds on Windows")
497    def test_dump_traceback_fd(self):
498        with tempfile.TemporaryFile('wb+') as fp:
499            self.check_dump_traceback(fd=fp.fileno())
500
501    def test_truncate(self):
502        maxlen = 500
503        func_name = 'x' * (maxlen + 50)
504        truncated = 'x' * maxlen + '...'
505        code = """
506            import faulthandler
507
508            def {func_name}():
509                faulthandler.dump_traceback(all_threads=False)
510
511            {func_name}()
512            """
513        code = code.format(
514            func_name=func_name,
515        )
516        expected = [
517            'Stack (most recent call first):',
518            '  File "<string>", line 4 in %s' % truncated,
519            '  File "<string>", line 6 in <module>'
520        ]
521        trace, exitcode = self.get_output(code)
522        self.assertEqual(trace, expected)
523        self.assertEqual(exitcode, 0)
524
525    def check_dump_traceback_threads(self, filename):
526        """
527        Call explicitly dump_traceback(all_threads=True) and check the output.
528        Raise an error if the output doesn't match the expected format.
529        """
530        code = """
531            import faulthandler
532            from threading import Thread, Event
533            import time
534
535            def dump():
536                if {filename}:
537                    with open({filename}, "wb") as fp:
538                        faulthandler.dump_traceback(fp, all_threads=True)
539                else:
540                    faulthandler.dump_traceback(all_threads=True)
541
542            class Waiter(Thread):
543                # avoid blocking if the main thread raises an exception.
544                daemon = True
545
546                def __init__(self):
547                    Thread.__init__(self)
548                    self.running = Event()
549                    self.stop = Event()
550
551                def run(self):
552                    self.running.set()
553                    self.stop.wait()
554
555            waiter = Waiter()
556            waiter.start()
557            waiter.running.wait()
558            dump()
559            waiter.stop.set()
560            waiter.join()
561            """
562        code = code.format(filename=repr(filename))
563        output, exitcode = self.get_output(code, filename)
564        output = '\n'.join(output)
565        if filename:
566            lineno = 8
567        else:
568            lineno = 10
569        regex = r"""
570            ^Thread 0x[0-9a-f]+ \(most recent call first\):
571            (?:  File ".*threading.py", line [0-9]+ in [_a-z]+
572            ){{1,3}}  File "<string>", line 23 in run
573              File ".*threading.py", line [0-9]+ in _bootstrap_inner
574              File ".*threading.py", line [0-9]+ in _bootstrap
575
576            Current thread 0x[0-9a-f]+ \(most recent call first\):
577              File "<string>", line {lineno} in dump
578              File "<string>", line 28 in <module>$
579            """
580        regex = dedent(regex.format(lineno=lineno)).strip()
581        self.assertRegex(output, regex)
582        self.assertEqual(exitcode, 0)
583
584    def test_dump_traceback_threads(self):
585        self.check_dump_traceback_threads(None)
586
587    def test_dump_traceback_threads_file(self):
588        with temporary_filename() as filename:
589            self.check_dump_traceback_threads(filename)
590
591    def check_dump_traceback_later(self, repeat=False, cancel=False, loops=1,
592                                   *, filename=None, fd=None):
593        """
594        Check how many times the traceback is written in timeout x 2.5 seconds,
595        or timeout x 3.5 seconds if cancel is True: 1, 2 or 3 times depending
596        on repeat and cancel options.
597
598        Raise an error if the output doesn't match the expect format.
599        """
600        timeout_str = str(datetime.timedelta(seconds=TIMEOUT))
601        code = """
602            import faulthandler
603            import time
604            import sys
605
606            timeout = {timeout}
607            repeat = {repeat}
608            cancel = {cancel}
609            loops = {loops}
610            filename = {filename!r}
611            fd = {fd}
612
613            def func(timeout, repeat, cancel, file, loops):
614                for loop in range(loops):
615                    faulthandler.dump_traceback_later(timeout, repeat=repeat, file=file)
616                    if cancel:
617                        faulthandler.cancel_dump_traceback_later()
618                    time.sleep(timeout * 5)
619                    faulthandler.cancel_dump_traceback_later()
620
621            if filename:
622                file = open(filename, "wb")
623            elif fd is not None:
624                file = sys.stderr.fileno()
625            else:
626                file = None
627            func(timeout, repeat, cancel, file, loops)
628            if filename:
629                file.close()
630            """
631        code = code.format(
632            timeout=TIMEOUT,
633            repeat=repeat,
634            cancel=cancel,
635            loops=loops,
636            filename=filename,
637            fd=fd,
638        )
639        trace, exitcode = self.get_output(code, filename)
640        trace = '\n'.join(trace)
641
642        if not cancel:
643            count = loops
644            if repeat:
645                count *= 2
646            header = r'Timeout \(%s\)!\nThread 0x[0-9a-f]+ \(most recent call first\):\n' % timeout_str
647            regex = expected_traceback(17, 26, header, min_count=count)
648            self.assertRegex(trace, regex)
649        else:
650            self.assertEqual(trace, '')
651        self.assertEqual(exitcode, 0)
652
653    def test_dump_traceback_later(self):
654        self.check_dump_traceback_later()
655
656    def test_dump_traceback_later_repeat(self):
657        self.check_dump_traceback_later(repeat=True)
658
659    def test_dump_traceback_later_cancel(self):
660        self.check_dump_traceback_later(cancel=True)
661
662    def test_dump_traceback_later_file(self):
663        with temporary_filename() as filename:
664            self.check_dump_traceback_later(filename=filename)
665
666    @unittest.skipIf(sys.platform == "win32",
667                     "subprocess doesn't support pass_fds on Windows")
668    def test_dump_traceback_later_fd(self):
669        with tempfile.TemporaryFile('wb+') as fp:
670            self.check_dump_traceback_later(fd=fp.fileno())
671
672    def test_dump_traceback_later_twice(self):
673        self.check_dump_traceback_later(loops=2)
674
675    @unittest.skipIf(not hasattr(faulthandler, "register"),
676                     "need faulthandler.register")
677    def check_register(self, filename=False, all_threads=False,
678                       unregister=False, chain=False, fd=None):
679        """
680        Register a handler displaying the traceback on a user signal. Raise the
681        signal and check the written traceback.
682
683        If chain is True, check that the previous signal handler is called.
684
685        Raise an error if the output doesn't match the expected format.
686        """
687        signum = signal.SIGUSR1
688        code = """
689            import faulthandler
690            import os
691            import signal
692            import sys
693
694            all_threads = {all_threads}
695            signum = {signum:d}
696            unregister = {unregister}
697            chain = {chain}
698            filename = {filename!r}
699            fd = {fd}
700
701            def func(signum):
702                os.kill(os.getpid(), signum)
703
704            def handler(signum, frame):
705                handler.called = True
706            handler.called = False
707
708            if filename:
709                file = open(filename, "wb")
710            elif fd is not None:
711                file = sys.stderr.fileno()
712            else:
713                file = None
714            if chain:
715                signal.signal(signum, handler)
716            faulthandler.register(signum, file=file,
717                                  all_threads=all_threads, chain={chain})
718            if unregister:
719                faulthandler.unregister(signum)
720            func(signum)
721            if chain and not handler.called:
722                if file is not None:
723                    output = file
724                else:
725                    output = sys.stderr
726                print("Error: signal handler not called!", file=output)
727                exitcode = 1
728            else:
729                exitcode = 0
730            if filename:
731                file.close()
732            sys.exit(exitcode)
733            """
734        code = code.format(
735            all_threads=all_threads,
736            signum=signum,
737            unregister=unregister,
738            chain=chain,
739            filename=filename,
740            fd=fd,
741        )
742        trace, exitcode = self.get_output(code, filename)
743        trace = '\n'.join(trace)
744        if not unregister:
745            if all_threads:
746                regex = r'Current thread 0x[0-9a-f]+ \(most recent call first\):\n'
747            else:
748                regex = r'Stack \(most recent call first\):\n'
749            regex = expected_traceback(14, 32, regex)
750            self.assertRegex(trace, regex)
751        else:
752            self.assertEqual(trace, '')
753        if unregister:
754            self.assertNotEqual(exitcode, 0)
755        else:
756            self.assertEqual(exitcode, 0)
757
758    def test_register(self):
759        self.check_register()
760
761    def test_unregister(self):
762        self.check_register(unregister=True)
763
764    def test_register_file(self):
765        with temporary_filename() as filename:
766            self.check_register(filename=filename)
767
768    @unittest.skipIf(sys.platform == "win32",
769                     "subprocess doesn't support pass_fds on Windows")
770    def test_register_fd(self):
771        with tempfile.TemporaryFile('wb+') as fp:
772            self.check_register(fd=fp.fileno())
773
774    def test_register_threads(self):
775        self.check_register(all_threads=True)
776
777    def test_register_chain(self):
778        self.check_register(chain=True)
779
780    @contextmanager
781    def check_stderr_none(self):
782        stderr = sys.stderr
783        try:
784            sys.stderr = None
785            with self.assertRaises(RuntimeError) as cm:
786                yield
787            self.assertEqual(str(cm.exception), "sys.stderr is None")
788        finally:
789            sys.stderr = stderr
790
791    def test_stderr_None(self):
792        # Issue #21497: provide a helpful error if sys.stderr is None,
793        # instead of just an attribute error: "None has no attribute fileno".
794        with self.check_stderr_none():
795            faulthandler.enable()
796        with self.check_stderr_none():
797            faulthandler.dump_traceback()
798        with self.check_stderr_none():
799            faulthandler.dump_traceback_later(1e-3)
800        if hasattr(faulthandler, "register"):
801            with self.check_stderr_none():
802                faulthandler.register(signal.SIGUSR1)
803
804    @unittest.skipUnless(MS_WINDOWS, 'specific to Windows')
805    def test_raise_exception(self):
806        for exc, name in (
807            ('EXCEPTION_ACCESS_VIOLATION', 'access violation'),
808            ('EXCEPTION_INT_DIVIDE_BY_ZERO', 'int divide by zero'),
809            ('EXCEPTION_STACK_OVERFLOW', 'stack overflow'),
810        ):
811            self.check_windows_exception(f"""
812                import faulthandler
813                faulthandler.enable()
814                faulthandler._raise_exception(faulthandler._{exc})
815                """,
816                3,
817                name)
818
819    @unittest.skipUnless(MS_WINDOWS, 'specific to Windows')
820    def test_ignore_exception(self):
821        for exc_code in (
822            0xE06D7363,   # MSC exception ("Emsc")
823            0xE0434352,   # COM Callable Runtime exception ("ECCR")
824        ):
825            code = f"""
826                    import faulthandler
827                    faulthandler.enable()
828                    faulthandler._raise_exception({exc_code})
829                    """
830            code = dedent(code)
831            output, exitcode = self.get_output(code)
832            self.assertEqual(output, [])
833            self.assertEqual(exitcode, exc_code)
834
835    @unittest.skipUnless(MS_WINDOWS, 'specific to Windows')
836    def test_raise_nonfatal_exception(self):
837        # These exceptions are not strictly errors. Letting
838        # faulthandler display the traceback when they are
839        # raised is likely to result in noise. However, they
840        # may still terminate the process if there is no
841        # handler installed for them (which there typically
842        # is, e.g. for debug messages).
843        for exc in (
844            0x00000000,
845            0x34567890,
846            0x40000000,
847            0x40001000,
848            0x70000000,
849            0x7FFFFFFF,
850        ):
851            output, exitcode = self.get_output(f"""
852                import faulthandler
853                faulthandler.enable()
854                faulthandler._raise_exception(0x{exc:x})
855                """
856            )
857            self.assertEqual(output, [])
858            # On Windows older than 7 SP1, the actual exception code has
859            # bit 29 cleared.
860            self.assertIn(exitcode,
861                          (exc, exc & ~0x10000000))
862
863    @unittest.skipUnless(MS_WINDOWS, 'specific to Windows')
864    def test_disable_windows_exc_handler(self):
865        code = dedent("""
866            import faulthandler
867            faulthandler.enable()
868            faulthandler.disable()
869            code = faulthandler._EXCEPTION_ACCESS_VIOLATION
870            faulthandler._raise_exception(code)
871        """)
872        output, exitcode = self.get_output(code)
873        self.assertEqual(output, [])
874        self.assertEqual(exitcode, 0xC0000005)
875
876    def test_cancel_later_without_dump_traceback_later(self):
877        # bpo-37933: Calling cancel_dump_traceback_later()
878        # without dump_traceback_later() must not segfault.
879        code = dedent("""
880            import faulthandler
881            faulthandler.cancel_dump_traceback_later()
882        """)
883        output, exitcode = self.get_output(code)
884        self.assertEqual(output, [])
885        self.assertEqual(exitcode, 0)
886
887
888if __name__ == "__main__":
889    unittest.main()
890