Python code coverage for Lib/test/test_faulthandler.py

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