B u9aïoã@s:ddlmZddlZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z m Z ddlZddlZddlZddlmZy ddlZWnek r¨dZYnXdZejdkZe d¡pÄd Ze d ¡pÒd Zd ekpâd ekZd ekpòdekZddd„Zdd„Zedd„ƒZGdd„dej ƒZ!e"dkr6e #¡dS)é)ÚcontextmanagerN)Úsupport)Ú script_helperÚ is_android)Údedentgà?ÚntZCFLAGSÚZ CONFIG_ARGSz-fsanitize=undefinedz#--with-undefined-behavior-sanitizerz-fsanitize=memoryz--with-memory-sanitizerécCsL|}|d|7}|d|7}d|kr", line %s in func z& File "", line %s in r ú^Ú ú$©)Zlineno1Zlineno2ÚheaderÚ min_countÚregexr r ú'/usr/lib/python3.7/test_faulthandler.pyÚexpected_traceback#s   rcCst td¡|ƒS)Nz(raising SIGSEGV on Android is unreliable)ÚunittestÚskipIfr)Útestr r rÚskip_segfault_on_android,src cs$t ¡}z |VWdt |¡XdS)N)ÚtempfileZmktemprÚunlink)Úfilenamer r rÚtemporary_filename1s rc@sBeZdZddd„Zddddddœdd„Zdd „Zd d „Ze e j   d ¡d ¡dd„ƒZ e dd„ƒZdd„Zdd„Ze e j dkd¡dd„ƒZe edkd¡e eedƒd¡e dd„ƒƒƒZe edkd¡e eedƒd ¡e d!d"„ƒƒƒZd#d$„Zd%d&„Ze e j   d'¡d(¡e eed)ƒ d*¡d+d,„ƒƒZe d-d.„ƒZe ep>ed/¡e d0d1„ƒƒZe e j dkd2¡e epled/¡e d3d4„ƒƒƒZ e d5d6„ƒZ!e d7d8„ƒZ"d9d:„Z#d;d<„Z$d=d>„Z%d?d@„Z&dddAœdBdC„Z'dDdE„Z(dFdG„Z)e e j dkd2¡dHdI„ƒZ*dJdK„Z+dLdM„Z,dNdO„Z-dPdQ„Z.e eedRƒ dS¡d‚dddAœdVdW„ƒZ/dXdY„Z0dZd[„Z1d\d]„Z2d^d_„Z3e e j dkd2¡d`da„ƒZ4dbdc„Z5e eeddƒ de¡dƒdfdg„ƒZ6dhdi„Z7djdk„Z8dldm„Z9e e j dkd2¡dndo„ƒZ:dpdq„Z;drds„Zdvdw„Z?e e@dx¡dydz„ƒZAe e@dx¡d{d|„ƒZBe e@dx¡d}d~„ƒZCe e@dx¡dd€„ƒZDdS)„ÚFaultHandlerTestsNc Cst|ƒ ¡}g}|dk r"| |¡t ¡:tjd||d}|| ¡\}}| ¡}WdQRXWdQRXt  |¡} |   dd¡} |r¼|  | d¡t |dƒ} |   ¡} WdQRX|   dd¡} nT|dk r|  | d¡t |tjd¡t |dd d } |   ¡} WdQRX|   dd¡} |  ¡|fS) a{ 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". Nz-c)Úpass_fdsÚasciiÚbackslashreplacerÚrbrF)Úclosefd)rÚstripÚappendrZSuppressCrashReportrZ spawn_pythonZ communicateÚwaitZstrip_python_stderrÚdecodeÚ assertEqualÚopenÚreadÚosÚlseekÚSEEK_SETÚ splitlines) ÚselfÚcoderÚfdrZprocessÚstdoutÚstderrÚexitcodeÚoutputÚfpr r rÚ get_output:s.           zFaultHandlerTests.get_outputT)rÚ all_threadsÚ other_regexr.Úknow_current_threadc Cs~|r|rd} qd} nd} d} t| j||| dƒ ¡} |rD| d|7} |j|||d\} } d | ¡} | | | ¡| | d ¡d S) zÒ 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. zCurrent thread 0x[0-9a-f]+zThread 0x[0-9a-f]+ZStackz— (?m)^{fatal_error} {header} \(most recent call first\): File "", line {lineno} in )ÚlinenoÚ fatal_errorrú|)rr.r rN)rÚformatr!r4ÚjoinÚ assertRegexÚassertNotEqual) r,r-Ú line_numberr9rr5r6r.r7rrr2r1r r rÚ check_error\s    zFaultHandlerTests.check_errorcKsd|}|j|||f|ŽdS)NzFatal Python error: %s)r@)r,r-r?Ú name_regexÚkwr9r r rÚcheck_fatal_error}sz#FaultHandlerTests.check_fatal_errorcKsd|}|j|||f|ŽdS)NzWindows fatal exception: %s)r@)r,r-r?rArBr9r r rÚcheck_windows_exceptionsz)FaultHandlerTests.check_windows_exceptionZaixz5the first page of memory is a mapped read-only on AIXcCs&ts| ddd¡n| ddd¡dS)Nz… import faulthandler faulthandler.enable() faulthandler._read_null() éz4(?:Segmentation fault|Bus error|Illegal instruction)zaccess violation)Ú MS_WINDOWSrCrD)r,r r rÚtest_read_null…sz FaultHandlerTests.test_read_nullcCs| ddd¡dS)Nzs import faulthandler faulthandler.enable() faulthandler._sigsegv() rEzSegmentation fault)rC)r,r r rÚ test_sigsegvœszFaultHandlerTests.test_sigsegvcCs|jddddddS)Nz€ import faulthandler faulthandler.enable() faulthandler._fatal_error_c_thread() rEz in new threadF)r7)rC)r,r r rÚtest_fatal_error_c_thread¦s z+FaultHandlerTests.test_fatal_error_c_threadcCs| ddd¡dS)Nzs import faulthandler faulthandler.enable() faulthandler._sigabrt() rEZAborted)rC)r,r r rÚ test_sigabrt°szFaultHandlerTests.test_sigabrtÚwin32z"SIGFPE cannot be caught on WindowscCs| ddd¡dS)Nzr import faulthandler faulthandler.enable() faulthandler._sigfpe() rEzFloating point exception)rC)r,r r rÚ test_sigfpe¹szFaultHandlerTests.test_sigfpezneed _testcapiÚSIGBUSzneed signal.SIGBUScCs| ddd¡dS)Nz¹ import _testcapi import faulthandler import signal faulthandler.enable() _testcapi.raise_signal(signal.SIGBUS) éz Bus error)rC)r,r r rÚ test_sigbusÄszFaultHandlerTests.test_sigbusÚSIGILLzneed signal.SIGILLcCs| ddd¡dS)Nz¹ import _testcapi import faulthandler import signal faulthandler.enable() _testcapi.raise_signal(signal.SIGILL) rNzIllegal instruction)rC)r,r r rÚ test_sigillÓszFaultHandlerTests.test_sigillcCs| ddd¡dS)Nz[ import faulthandler faulthandler._fatal_error(b'xyz') éÚxyz)rC)r,r r rÚtest_fatal_errorâsz"FaultHandlerTests.test_fatal_errorcCs| ddd¡dS)Nza import faulthandler faulthandler._fatal_error(b'xyz', True) rRrS)rC)r,r r rÚtest_fatal_error_without_gilêsz.FaultHandlerTests.test_fatal_error_without_gilZopenbsdzVIssue #12868: sigaltstack() doesn't work on OpenBSD if Python is compiled with pthreadZ_stack_overflowz#need faulthandler._stack_overflow()cCs|jddddddS)Nzz import faulthandler faulthandler.enable() faulthandler._stack_overflow() rEz (?:Segmentation fault|Bus error)z unable to raise a stack overflow)r6)rC)r,r r rÚtest_stack_overflowòs z%FaultHandlerTests.test_stack_overflowcCs| ddd¡dS)Nzw import faulthandler faulthandler.enable() faulthandler._sigsegv(True) rEzSegmentation fault)rC)r,r r rÚtest_gil_releasedsz#FaultHandlerTests.test_gil_releasedz0sanitizer builds change crashing process output.c Cs4tƒ$}|jdjt|ƒddd|dWdQRXdS)Nz¹ import faulthandler output = open({filename}, 'wb') faulthandler.enable(output) faulthandler._sigsegv() )rézSegmentation fault)rrCr;Úrepr)r,rr r rÚtest_enable_file s z"FaultHandlerTests.test_enable_filez.subprocess doesn't support pass_fds on Windowsc Cs8t d¡$}| ¡}|jd|dd|dWdQRXdS)Nzwb+z  import faulthandler import sys faulthandler.enable(%s) faulthandler._sigsegv() rXzSegmentation fault)r.)rÚ TemporaryFileÚfilenorC)r,r3r.r r rÚtest_enable_fds z FaultHandlerTests.test_enable_fdcCs|jddddddS)Nz„ import faulthandler faulthandler.enable(all_threads=False) faulthandler._sigsegv() rEzSegmentation faultF)r5)rC)r,r r rÚtest_enable_single_thread,s z+FaultHandlerTests.test_enable_single_threadcCsHd}d}| |¡\}}d |¡}| ||kd||f¡| |d¡dS)Nz– import faulthandler faulthandler.enable() faulthandler.disable() faulthandler._sigsegv() zFatal Python errorr z%r is present in %rr)r4r<Ú assertTruer>)r,r-Z not_expectedr0r1r r rÚ test_disable7s  zFaultHandlerTests.test_disablecCsrtj}z^tjt_t ¡}z0t ¡| t ¡¡t ¡| t ¡¡Wd|rXt ¡nt ¡XWd|t_XdS)N) Úsysr0Ú __stderr__Ú faulthandlerZ is_enabledÚenabler_ZdisableZ assertFalse)r,Z orig_stderrZ was_enabledr r rÚtest_is_enabledFs z!FaultHandlerTests.test_is_enabledcCs0d}tjdd|f}t |¡}| | ¡d¡dS)Nz5import faulthandler; print(faulthandler.is_enabled())z-Ez-csFalse)raÚ executableÚ subprocessÚ check_outputr%Úrstrip)r,r-Úargsr2r r rÚtest_disabled_by_default\s z*FaultHandlerTests.test_disabled_by_defaultcCs`d}tdtjtjjrdndddd|fƒ}tj ¡}| dd¡t j ||d}|  |  ¡d ¡dS) Nz5import faulthandler; print(faulthandler.is_enabled())z-Erz-Xrcz-cÚPYTHONFAULTHANDLER)ÚenvsTrue) ÚfilterrarfÚflagsÚignore_environmentr(ÚenvironÚcopyÚpoprgrhr%ri)r,r-rjrmr2r r rÚtest_sys_xoptionsds  z#FaultHandlerTests.test_sys_xoptionscCs„d}tjd|f}ttjƒ}d|d<d|d<tj||d}| | ¡d¡ttjƒ}d|d<d|d<tj||d}| | ¡d ¡dS) Nz5import faulthandler; print(faulthandler.is_enabled())z-crrlZ PYTHONDEVMODE)rmsFalseÚ1sTrue) rarfÚdictr(rqrgrhr%ri)r,r-rjrmr2r r rÚ test_env_varps   zFaultHandlerTests.test_env_var)rr.cCsld}|j||d}|rd}n|dk r*d}nd}dd|d d g}| |||¡\}}| ||¡| |d ¡dS) z™ Explicitly call dump_traceback() function and check its output. Raise an error if the output doesn't match the expected format. a[ 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() )rr.é Né ézStack (most recent call first):z# File "", line %s in funcBz# File "", line 17 in funcAz& File "", line 19 in r)r;r4r%)r,rr.r-r8ÚexpectedÚtracer1r r rÚcheck_dump_traceback‚s  z&FaultHandlerTests.check_dump_tracebackcCs | ¡dS)N)r})r,r r rÚtest_dump_traceback°sz%FaultHandlerTests.test_dump_tracebackc Cs"tƒ}|j|dWdQRXdS)N)r)rr})r,rr r rÚtest_dump_traceback_file³sz*FaultHandlerTests.test_dump_traceback_filec Cs*t d¡}|j| ¡dWdQRXdS)Nzwb+)r.)rr[r}r\)r,r3r r rÚtest_dump_traceback_fd·s z(FaultHandlerTests.test_dump_traceback_fdcCsdd}d|d}d|d}d}|j|d}dd|d g}| |¡\}}| ||¡| |d ¡dS) NiôÚxé2z...z§ import faulthandler def {func_name}(): faulthandler.dump_traceback(all_threads=False) {func_name}() )Ú func_namezStack (most recent call first):z File "", line 4 in %sz% File "", line 6 in r)r;r4r%)r,ÚmaxlenrƒZ truncatedr-r{r|r1r r rÚ test_truncate½s   zFaultHandlerTests.test_truncatecCspd}|jt|ƒd}| ||¡\}}d |¡}|r8d}nd}d}t|j|dƒ ¡}| ||¡| |d¡d S) z  Call explicitly dump_traceback(all_threads=True) and check the output. Raise an error if the output doesn't match the expected format. aÅ 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() )rr éé aï ^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 $ )r8rN)r;rYr4r<rr!r=r%)r,rr-r2r1r8rr r rÚcheck_dump_traceback_threadsÕs$   z.FaultHandlerTests.check_dump_traceback_threadscCs| d¡dS)N)rˆ)r,r r rÚtest_dump_traceback_threadssz-FaultHandlerTests.test_dump_traceback_threadsc Cs tƒ}| |¡WdQRXdS)N)rrˆ)r,rr r rÚ test_dump_traceback_threads_filesz2FaultHandlerTests.test_dump_traceback_threads_fileÚdump_traceback_laterz(need faulthandler.dump_traceback_later()Fr c Csšttjtdƒ}d}|jt|||||d}| ||¡\}} d |¡}|s~|} |rX| d9} d|} tdd| | d } | || ¡n |  |d ¡|  | d ¡d S) a 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. )ZsecondsaÉ 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() )ZtimeoutÚrepeatÚcancelÚloopsrr.r rRzATimeout \(%s\)!\nThread 0x[0-9a-f]+ \(most recent call first\):\néé)rrrN) ÚstrÚdatetimeZ timedeltaÚTIMEOUTr;r4r<rr=r%) r,rŒrrŽrr.Z timeout_strr-r|r1Úcountrrr r rÚcheck_dump_traceback_laters(   z,FaultHandlerTests.check_dump_traceback_latercCs | ¡dS)N)r•)r,r r rÚtest_dump_traceback_laterWsz+FaultHandlerTests.test_dump_traceback_latercCs|jdddS)NT)rŒ)r•)r,r r rÚ test_dump_traceback_later_repeatZsz2FaultHandlerTests.test_dump_traceback_later_repeatcCs|jdddS)NT)r)r•)r,r r rÚ test_dump_traceback_later_cancel]sz2FaultHandlerTests.test_dump_traceback_later_cancelc Cs"tƒ}|j|dWdQRXdS)N)r)rr•)r,rr r rÚtest_dump_traceback_later_file`sz0FaultHandlerTests.test_dump_traceback_later_filec Cs*t d¡}|j| ¡dWdQRXdS)Nzwb+)r.)rr[r•r\)r,r3r r rÚtest_dump_traceback_later_fdds z.FaultHandlerTests.test_dump_traceback_later_fdcCs|jdddS)NrR)rŽ)r•)r,r r rÚtest_dump_traceback_later_twicejsz1FaultHandlerTests.test_dump_traceback_later_twiceÚregisterzneed faulthandler.registerc Cs”tj}d}|j||||||d}| ||¡\}} d |¡}|sf|rHd} nd} tdd| ƒ} | || ¡n | |d¡|r„| | d ¡n | | d ¡d S) a 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. ax import faulthandler import os import signal import sys all_threads = {all_threads} signum = {signum} 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) )r5ÚsignumÚ unregisterÚchainrr.r z8Current thread 0x[0-9a-f]+ \(most recent call first\):\nz#Stack \(most recent call first\):\nrzé rrN) ÚsignalÚSIGUSR1r;r4r<rr=r%r>) r,rr5ržrŸr.rr-r|r1rr r rÚcheck_registerms* .   z FaultHandlerTests.check_registercCs | ¡dS)N)r£)r,r r rÚ test_registerÀszFaultHandlerTests.test_registercCs|jdddS)NT)rž)r£)r,r r rÚtest_unregisterÃsz!FaultHandlerTests.test_unregisterc Cs"tƒ}|j|dWdQRXdS)N)r)rr£)r,rr r rÚtest_register_fileÆsz$FaultHandlerTests.test_register_filec Cs*t d¡}|j| ¡dWdQRXdS)Nzwb+)r.)rr[r£r\)r,r3r r rÚtest_register_fdÊs z"FaultHandlerTests.test_register_fdcCs|jdddS)NT)r5)r£)r,r r rÚtest_register_threadsÐsz'FaultHandlerTests.test_register_threadscCs|jdddS)NT)rŸ)r£)r,r r rÚtest_register_chainÓsz%FaultHandlerTests.test_register_chainc csLtj}z8dt_| t¡ }dVWdQRX| t|jƒd¡Wd|t_XdS)Nzsys.stderr is None)rar0Z assertRaisesÚ RuntimeErrorr%r‘Z exception)r,r0Úcmr r rÚcheck_stderr_noneÖs z#FaultHandlerTests.check_stderr_nonec CsŽ| ¡t ¡WdQRX| ¡t ¡WdQRXttdƒr`| ¡t d¡WdQRXttdƒrŠ| ¡t tj¡WdQRXdS)Nr‹gü©ñÒMbP?rœ) r¬rcrdZdump_tracebackÚhasattrr‹rœr¡r¢)r,r r rÚtest_stderr_Noneás      z"FaultHandlerTests.test_stderr_Nonezspecific to WindowscCs,x&dD]\}}| d|›dd|¡qWdS)N))ZEXCEPTION_ACCESS_VIOLATIONzaccess violation)ZEXCEPTION_INT_DIVIDE_BY_ZEROzint divide by zero)ZEXCEPTION_STACK_OVERFLOWzstack overflowz‡ import faulthandler faulthandler.enable() faulthandler._raise_exception(faulthandler._z) rE)rD)r,ÚexcÚnamer r rÚtest_raise_exceptionïs   z&FaultHandlerTests.test_raise_exceptioncCsLxFdD]>}d|›d}t|ƒ}| |¡\}}| |g¡| ||¡qWdS)N)lcsÚ@lRC†@z… import faulthandler faulthandler.enable() faulthandler._raise_exception(z) )rr4r%)r,Zexc_coder-r2r1r r rÚtest_ignore_exceptionþs  z'FaultHandlerTests.test_ignore_exceptioncCsJxDdD]<}| d|d›d¡\}}| |g¡| |||d@f¡qWdS)N)rixV4i@i@ipiÿÿÿz{ import faulthandler faulthandler.enable() faulthandler._raise_exception(0xrz) iÿÿÿï)r4r%ZassertIn)r,r¯r2r1r r rÚtest_raise_nonfatal_exceptions z/FaultHandlerTests.test_raise_nonfatal_exceptioncCs2tdƒ}| |¡\}}| |g¡| |d¡dS)NzÚ import faulthandler faulthandler.enable() faulthandler.disable() code = faulthandler._EXCEPTION_ACCESS_VIOLATION faulthandler._raise_exception(code) l)rr4r%)r,r-r2r1r r rÚ test_disable_windows_exc_handler*s  z2FaultHandlerTests.test_disable_windows_exc_handler)NN)FFr )FFFFN)EÚ__name__Ú __module__Ú __qualname__r4r@rCrDrrraÚplatformÚ startswithrGrrHrIrJrLÚ _testcapiZ skipUnlessr­r¡rOrQrTrUrcrVrWÚ UB_SANITIZERÚMEMORY_SANITIZERrZr]r^r`rerkrtrwr}r~rr€r…rˆr‰rŠr•r–r—r˜r™ršr›r£r¤r¥r¦r§r¨r©rr¬r®rFr±r²r³r´r r r rr9s˜ #           . ;=  P  rÚ__main__)r )$Ú contextlibrr’rcr(r¡rgraZ sysconfigrrZ test.supportrrrZ threadingrÚtextwraprrºÚ ImportErrorr“r°rFZget_config_varZ_cflagsZ _config_argsr»r¼rrrZTestCaserrµÚmainr r r rÚsJ