B u9aȠ@sddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m Z mZy ddlZWnek rdZYnXGddde jZe ejdkdGdd d e jZe ejdkd Gd d d e jZGd dde jZe ejdkdGddde jZe eeddGddde jZe ejdkdGddde jZe ejdkdGddde jZGddde jZGddde jZddZ e!dkre "dS) N)support)assert_python_ok spawn_pythonc@seZdZddZdS) GenericTestscCsxttD]}tt|}|dkr0||tjq |dkrH||tjq |drl|dsl||tjq |dr ||tj|t j dq WdS)N>SIG_DFLSIG_IGN> SIG_SETMASK SIG_BLOCK SIG_UNBLOCKZSIGZSIG_ZCTRL_win32) dirsignalgetattrassertIsInstanceHandlersZSigmasks startswithZSignals assertEqualsysplatform)selfnamesigr!/usr/lib/python3.7/test_signal.py test_enumss  zGenericTests.test_enumsN)__name__ __module__ __qualname__rrrrrrsrr zNot valid on Windowsc@s4eZdZddZddZddZddZd d Zd S) PosixTestscGsdS)Nr)rargsrrrtrivial_signal_handler%sz!PosixTests.trivial_signal_handlercCs(|ttjd|ttjd|jdS)Ni) assertRaises ValueErrorr getsignalr )rrrr,test_out_of_range_signal_number_raises_error(s z7PosixTests.test_out_of_range_signal_number_raises_errorcCs|ttjtjddS)N)r! TypeErrorr SIGUSR1)rrrr0test_setting_signal_handler_to_none_raises_error.s z;PosixTests.test_setting_signal_handler_to_none_raises_errorcCsZttj|j}||tj|ttj|jttj||ttj|dS)N)r SIGHUPr rrrr#)rZhuprrrtest_getsignal2s zPosixTests.test_getsignalcCs&tjt}tj|d}t|dS)Nzsignalinterproctester.py)ospathdirname__file__joinr)rr,Zscriptrrrtest_interprocess_signal;s z#PosixTests.test_interprocess_signalN)rrrr r$r'r)r/rrrrr#s  rzWindows specificc@seZdZddZdS)WindowsSignalTestsc Csdd}t}xTtjtjtjtjtjtjtjfD]0}t |dk r0t|t||| |q0W| || t td|WdQRX| t td|WdQRXdS)NcSsdS)Nr)xyrrrEz3WindowsSignalTests.test_issue9324..)setr SIGABRTZSIGBREAKSIGFPESIGILLSIGINTSIGSEGVSIGTERMr#add assertTruer!r")rhandlercheckedrrrrtest_issue9324Cs     z!WindowsSignalTests.test_issue9324N)rrrrBrrrrr0Asr0c@sNeZdZddZddZddZddZd d Ze e j d kd d dZ dS) WakeupFDTestsc CsL|ttjtjdWdQRX|tttjdWdQRXdS)N)signumF)r!r%r set_wakeup_fdr;)rrrrtest_invalid_call[s  zWakeupFDTests.test_invalid_callcCs t}|ttftj|dS)N)rZ make_bad_fdr!r"OSErrorr rE)rfdrrrtest_invalid_fdds zWakeupFDTests.test_invalid_fdcCs0t}|}||ttftj|dS)N)socketfilenocloser!r"rGr rE)rZsockrHrrrtest_invalid_socketis  z!WakeupFDTests.test_invalid_socketcCst\}}|tj||tj|t\}}|tj||tj|ttdrrt|dt|dt||t|||td||tdddS)N set_blockingFr5) r*pipe addCleanuprLhasattrrNr rEr)rZr1Zw1Zr2Zw2rrrtest_set_wakeup_fd_resultps      z'WakeupFDTests.test_set_wakeup_fd_resultcCst}||j|d|}t}||j|d|}t||t|||td||tdddS)NFr5)rJrPrLZ setblockingrKr rEr)rZsock1Zfd1Zsock2Zfd2rrr test_set_wakeup_fd_socket_results     z.WakeupFDTests.test_set_wakeup_fd_socket_resultr ztests specific to POSIXc Cst\}}|tj||tj|t|d|t}t|WdQRX| t |j d|t|dt|tddS)NTz&the fd %s must be in non-blocking modeFr5) r*rOrPrLrNr!r"r rErstrZ exception)rZrfdZwfdcmrrrtest_set_wakeup_fd_blockings       z)WakeupFDTests.test_set_wakeup_fd_blockingN) rrrrFrIrMrRrSunittestskipIfrrrVrrrrrCYs  rCc@steZdZeedkdddddZeedkdddZd d Zd d Z d dZ e e e ddddZdS)WakeupSignalTestsNzneed _testcapiT)orderedcGs&dttt|||}td|dS)Naif 1: import _testcapi import os import signal import struct signals = {!r} def handler(signum, frame): pass def check_signum(signals): data = os.read(read, len(signals)+1) raised = struct.unpack('%uB' % len(data), data) if not {!r}: raised = set(raised) signals = set(signals) if raised != signals: raise Exception("%r != %r" % (raised, signals)) {} signal.signal(signal.SIGALRM, handler) read, write = os.pipe() os.set_blocking(write, False) signal.set_wakeup_fd(write) test() check_signum(signals) os.close(read) os.close(write) z-c)formattuplemapintr)rZ test_bodyrZZsignalscoderrr check_wakeups#zWakeupSignalTests.check_wakeupc Csjd}t\}}z4yt|dWntk r6Yn X|dWdt|t|Xtd|dS)Na)if 1: import _testcapi import errno import os import signal import sys from test.support import captured_stderr def handler(signum, frame): 1/0 signal.signal(signal.SIGALRM, handler) r, w = os.pipe() os.set_blocking(r, False) # Set wakeup_fd a read-only file descriptor to trigger the error signal.set_wakeup_fd(r) try: with captured_stderr() as err: _testcapi.raise_signal(signal.SIGALRM) except ZeroDivisionError: # An ignored exception should have been printed out on stderr err = err.getvalue() if ('Exception ignored when trying to write to the signal wakeup fd' not in err): raise AssertionError(err) if ('OSError: [Errno %d]' % errno.EBADF) not in err: raise AssertionError(err) else: raise AssertionError("ZeroDivisionError not raised") os.close(r) os.close(w) xz9OS doesn't report write() error on the read end of a pipez-c)r*rOwriterGskipTestrLr)rr_rwrrrtest_wakeup_write_errors&   z)WakeupSignalTests.test_wakeup_write_errorcCs|dtjdS)Nadef test(): import select import time TIMEOUT_FULL = 10 TIMEOUT_HALF = 5 class InterruptSelect(Exception): pass def handler(signum, frame): raise InterruptSelect signal.signal(signal.SIGALRM, handler) signal.alarm(1) # We attempt to get a signal during the sleep, # before select is called try: select.select([], [], [], TIMEOUT_FULL) except InterruptSelect: pass else: raise Exception("select() was not interrupted") before_time = time.monotonic() select.select([read], [], [], TIMEOUT_FULL) after_time = time.monotonic() dt = after_time - before_time if dt >= TIMEOUT_HALF: raise Exception("%s >= %s" % (dt, TIMEOUT_HALF)) )r`r SIGALRM)rrrrtest_wakeup_fd_earlysz&WakeupSignalTests.test_wakeup_fd_earlycCs|dtjdS)Na`def test(): import select import time TIMEOUT_FULL = 10 TIMEOUT_HALF = 5 class InterruptSelect(Exception): pass def handler(signum, frame): raise InterruptSelect signal.signal(signal.SIGALRM, handler) signal.alarm(1) before_time = time.monotonic() # We attempt to get a signal during the select call try: select.select([read], [], [], TIMEOUT_FULL) except InterruptSelect: pass else: raise Exception("select() was not interrupted") after_time = time.monotonic() dt = after_time - before_time if dt >= TIMEOUT_HALF: raise Exception("%s >= %s" % (dt, TIMEOUT_HALF)) )r`r rg)rrrrtest_wakeup_fd_during%sz'WakeupSignalTests.test_wakeup_fd_duringcCs|dtjtjdS)Nzdef test(): import _testcapi signal.signal(signal.SIGUSR1, handler) _testcapi.raise_signal(signal.SIGUSR1) _testcapi.raise_signal(signal.SIGALRM) )r`r r&rg)rrrr test_signumCszWakeupSignalTests.test_signumpthread_sigmaskzneed signal.pthread_sigmask()cCs|jdtjtjdddS)Nadef test(): signum1 = signal.SIGUSR1 signum2 = signal.SIGUSR2 signal.signal(signum1, handler) signal.signal(signum2, handler) signal.pthread_sigmask(signal.SIG_BLOCK, (signum1, signum2)) _testcapi.raise_signal(signum1) _testcapi.raise_signal(signum2) # Unblocking the 2 signals calls the C signal handler twice signal.pthread_sigmask(signal.SIG_UNBLOCK, (signum1, signum2)) F)rZ)r`r r&SIGUSR2)rrrr test_pendingKs zWakeupSignalTests.test_pending)rrrrWrX _testcapir`rfrhrirj skipUnlessrQr rmrrrrrYs&4" rYZ socketpairzneed socket.socketpairc@sTeZdZeedkdddZeedkdddZeedkdddZdS) WakeupSocketSignalTestsNzneed _testcapicCsd}td|dS)Naif 1: import signal import socket import struct import _testcapi signum = signal.SIGINT signals = (signum,) def handler(signum, frame): pass signal.signal(signum, handler) read, write = socket.socketpair() write.setblocking(False) signal.set_wakeup_fd(write.fileno()) _testcapi.raise_signal(signum) data = read.recv(1) if not data: raise Exception("no signum written") raised = struct.unpack('B', data) if raised != signals: raise Exception("%r != %r" % (raised, signals)) read.close() write.close() z-c)r)rr_rrr test_socket`s z#WakeupSocketSignalTests.test_socketcCs.tjdkrd}nd}dj|d}td|dS)Nntsendrba.if 1: import errno import signal import socket import sys import time import _testcapi from test.support import captured_stderr signum = signal.SIGINT def handler(signum, frame): pass signal.signal(signum, handler) read, write = socket.socketpair() read.setblocking(False) write.setblocking(False) signal.set_wakeup_fd(write.fileno()) # Close sockets: send() will fail read.close() write.close() with captured_stderr() as err: _testcapi.raise_signal(signum) err = err.getvalue() if ('Exception ignored when trying to {action} to the signal wakeup fd' not in err): raise AssertionError(err) )actionz-c)r*rr[r)rrtr_rrrtest_send_errors  " z'WakeupSocketSignalTests.test_send_errorcCs.tjdkrd}nd}dj|d}td|dS)Nrrrsrba if 1: import errno import signal import socket import sys import time import _testcapi from test.support import captured_stderr signum = signal.SIGINT # This handler will be called, but we intentionally won't read from # the wakeup fd. def handler(signum, frame): pass signal.signal(signum, handler) read, write = socket.socketpair() # Fill the socketpair buffer if sys.platform == 'win32': # bpo-34130: On Windows, sometimes non-blocking send fails to fill # the full socketpair buffer, so use a timeout of 50 ms instead. write.settimeout(0.050) else: write.setblocking(False) # Start with large chunk size to reduce the # number of send needed to fill the buffer. written = 0 for chunk_size in (2 ** 16, 2 ** 8, 1): chunk = b"x" * chunk_size try: while True: write.send(chunk) written += chunk_size except (BlockingIOError, socket.timeout): pass print(f"%s bytes written into the socketpair" % written, flush=True) write.setblocking(False) try: write.send(b"x") except BlockingIOError: # The socketpair buffer seems full pass else: raise AssertionError("%s bytes failed to fill the socketpair " "buffer" % written) # By default, we get a warning when a signal arrives msg = ('Exception ignored when trying to {action} ' 'to the signal wakeup fd') signal.set_wakeup_fd(write.fileno()) with captured_stderr() as err: _testcapi.raise_signal(signum) err = err.getvalue() if msg not in err: raise AssertionError("first set_wakeup_fd() test failed, " "stderr: %r" % err) # And also if warn_on_full_buffer=True signal.set_wakeup_fd(write.fileno(), warn_on_full_buffer=True) with captured_stderr() as err: _testcapi.raise_signal(signum) err = err.getvalue() if msg not in err: raise AssertionError("set_wakeup_fd(warn_on_full_buffer=True) " "test failed, stderr: %r" % err) # But not if warn_on_full_buffer=False signal.set_wakeup_fd(write.fileno(), warn_on_full_buffer=False) with captured_stderr() as err: _testcapi.raise_signal(signum) err = err.getvalue() if err != "": raise AssertionError("set_wakeup_fd(warn_on_full_buffer=False) " "test failed, stderr: %r" % err) # And then check the default again, to make sure warn_on_full_buffer # settings don't leak across calls. signal.set_wakeup_fd(write.fileno()) with captured_stderr() as err: _testcapi.raise_signal(signum) err = err.getvalue() if msg not in err: raise AssertionError("second set_wakeup_fd() test failed, " "stderr: %r" % err) )rtz-c)r*rr[r)rrtr_rrrtest_warn_on_full_buffers  d z0WakeupSocketSignalTests.test_warn_on_full_buffer) rrrrWrXrnrqrurvrrrrrp]s$+rpc@s,eZdZddZddZddZddZd S) SiginterruptTestc Csd|f}td|t}y|j}|jdd\}}Wntjk rR|dSX||}|}|dkr|td||f|dkSWd QRXd S) zPerform a read during which a signal will arrive. Return True if the read is interrupted by the signal and raises an exception. Return False if it returns normally. aif 1: import errno import os import signal import sys interrupt = %r r, w = os.pipe() def handler(signum, frame): 1 / 0 signal.signal(signal.SIGALRM, handler) if interrupt is not None: signal.siginterrupt(signal.SIGALRM, interrupt) print("ready") sys.stdout.flush() # run the test twice try: for loop in range(2): # send a SIGALRM in a second (during the read) signal.alarm(1) try: # blocking call: read from a pipe without data os.read(r, 1) except ZeroDivisionError: pass else: sys.exit(2) sys.exit(3) finally: os.close(r) os.close(w) z-cg@)ZtimeoutF)zChild error (exit code %s): %rryN) rstdoutreadline communicate subprocessZTimeoutExpiredkillwait Exception)rZ interruptr_processZ first_linerzstderrexitcoderrrreadpipe_interrupted s*    z%SiginterruptTest.readpipe_interruptedcCs|d}||dS)N)rr?)r interruptedrrrtest_without_siginterrupt\s z*SiginterruptTest.test_without_siginterruptcCs|d}||dS)NT)rr?)rrrrrtest_siginterrupt_oncs z%SiginterruptTest.test_siginterrupt_oncCs|d}||dS)NF)rZ assertFalse)rrrrrtest_siginterrupt_offjs z&SiginterruptTest.test_siginterrupt_offN)rrrrrrrrrrrrws<rwc@sneZdZddZddZddZddZd d Zd d Zd dZ e e j dkdddZddZddZdS) ItimerTestcCs(d|_d|_d|_ttj|j|_dS)NFr) hndl_called hndl_countitimerr rgsig_alrm old_alarm)rrrrsetUptszItimerTest.setUpcCs,ttj|j|jdk r(t|jddS)Nr)r rgrr setitimer)rrrrtearDownzs zItimerTest.tearDowncGs d|_dS)NT)r)rrrrrrszItimerTest.sig_alrmcGsFd|_|jdkrtdn|jdkr4ttjd|jd7_dS)NTryz.setitimer didn't disable ITIMER_VIRTUAL timer.r)rrr ItimerErrorrITIMER_VIRTUAL)rrrrr sig_vtalrms    zItimerTest.sig_vtalrmcGsd|_ttjddS)NTr)rr r ITIMER_PROF)rrrrrsig_profszItimerTest.sig_profcCs|tjtjdddS)Nr5r)r!r rr)rrrrtest_itimer_excszItimerTest.test_itimer_exccCs0tj|_t|jdt||jddS)Ng?T)r ITIMER_REALrrpauserr)rrrrtest_itimer_realszItimerTest.test_itimer_real)Znetbsd5zDitimer not reliable (does not mix well with threading) on some BSDs.cCstj|_ttj|jt|jddt}xrT)r rrrrsleeprr)rrrrtest_setitimer_tinys zItimerTest.test_setitimer_tinyN)rrrrrrrrrrrWrXrrrrrrrrrrrs   rc@seZdZdZeeeddddZeeeddeeedddd Z eeed d d d Z eeeddddZ eeeddddZ eeeddddZ eeeddddZeeeddddZeeeddddZeeeddd d!Zeeeddeeeddd"d#Zeeeddd$d%Zeeeddd&d'Zeeed d d(d)Zd*S)+PendingSignalsTestsz[ Test pthread_sigmask(), pthread_kill(), sigpending() and sigwait() functions. sigpendingzneed signal.sigpending()cCs|ttdS)N)rr rr7)rrrrtest_sigpending_emptysz)PendingSignalsTests.test_sigpending_emptyrkzneed signal.pthread_sigmask()cCsd}td|dS)Na if 1: import os import signal def handler(signum, frame): 1/0 signum = signal.SIGUSR1 signal.signal(signum, handler) signal.pthread_sigmask(signal.SIG_BLOCK, [signum]) os.kill(os.getpid(), signum) pending = signal.sigpending() for sig in pending: assert isinstance(sig, signal.Signals), repr(pending) if pending != {signum}: raise Exception('%s != {%s}' % (pending, signum)) try: signal.pthread_sigmask(signal.SIG_UNBLOCK, [signum]) except ZeroDivisionError: pass else: raise Exception("ZeroDivisionError not raised") z-c)r)rr_rrrtest_sigpendingsz#PendingSignalsTests.test_sigpending pthread_killzneed signal.pthread_kill()cCsd}td|dS)Naif 1: import signal import threading import sys signum = signal.SIGUSR1 def handler(signum, frame): 1/0 signal.signal(signum, handler) tid = threading.get_ident() try: signal.pthread_kill(tid, signum) except ZeroDivisionError: pass else: raise Exception("ZeroDivisionError not raised") z-c)r)rr_rrrtest_pthread_killsz%PendingSignalsTests.test_pthread_killcCsd||f}td|dS)zo test: body of the "def test(signum):" function. blocked: number of the blocked signal awif 1: import signal import sys from signal import Signals def handler(signum, frame): 1/0 %s blocked = %s signum = signal.SIGALRM # child: block and wait the signal try: signal.signal(signum, handler) signal.pthread_sigmask(signal.SIG_BLOCK, [blocked]) # Do the tests test(signum) # The handler must not be called on unblock try: signal.pthread_sigmask(signal.SIG_UNBLOCK, [blocked]) except ZeroDivisionError: print("the signal handler has been called", file=sys.stderr) sys.exit(1) except BaseException as err: print("error: {}".format(err), file=sys.stderr) sys.stderr.flush() sys.exit(1) z-cN)stripr)rZblockedtestr_rrr wait_helpers'zPendingSignalsTests.wait_helpersigwaitzneed signal.sigwait()cCs|tjddS)Na  def test(signum): signal.alarm(1) received = signal.sigwait([signum]) assert isinstance(received, signal.Signals), received if received != signum: raise Exception('received %s, not %s' % (received, signum)) )rr rg)rrrr test_sigwaitHsz PendingSignalsTests.test_sigwait sigwaitinfozneed signal.sigwaitinfo()cCs|tjddS)Nz def test(signum): signal.alarm(1) info = signal.sigwaitinfo([signum]) if info.si_signo != signum: raise Exception("info.si_signo != %s" % signum) )rr rg)rrrrtest_sigwaitinfoTsz$PendingSignalsTests.test_sigwaitinfo sigtimedwaitzneed signal.sigtimedwait()cCs|tjddS)Nz def test(signum): signal.alarm(1) info = signal.sigtimedwait([signum], 10.1000) if info.si_signo != signum: raise Exception('info.si_signo != %s' % signum) )rr rg)rrrrtest_sigtimedwait_sz%PendingSignalsTests.test_sigtimedwaitcCs|tjddS)Nz def test(signum): import os os.kill(os.getpid(), signum) info = signal.sigtimedwait([signum], 0) if info.si_signo != signum: raise Exception('info.si_signo != %s' % signum) )rr rg)rrrrtest_sigtimedwait_polljsz*PendingSignalsTests.test_sigtimedwait_pollcCs|tjddS)Nz def test(signum): received = signal.sigtimedwait([signum], 1.0) if received is not None: raise Exception("received=%r" % (received,)) )rr rg)rrrrtest_sigtimedwait_timeoutwsz-PendingSignalsTests.test_sigtimedwait_timeoutcCstj}|ttj|gddS)Ng)r rgr!r"r)rrDrrr"test_sigtimedwait_negative_timeoutsz6PendingSignalsTests.test_sigtimedwait_negative_timeoutcCstdddS)Nz-caif True: import os, threading, sys, time, signal # the default handler terminates the process signum = signal.SIGUSR1 def kill_later(): # wait until the main thread is waiting in sigwait() time.sleep(1) os.kill(os.getpid(), signum) # the signal must be blocked by all the threads signal.pthread_sigmask(signal.SIG_BLOCK, [signum]) killer = threading.Thread(target=kill_later) killer.start() received = signal.sigwait([signum]) if received != signum: print("sigwait() received %s, not %s" % (received, signum), file=sys.stderr) sys.exit(1) killer.join() # unblock the signal, which should have been cleared by sigwait() signal.pthread_sigmask(signal.SIG_UNBLOCK, [signum]) )r)rrrrtest_sigwait_threads z'PendingSignalsTests.test_sigwait_threadcCsH|ttj|ttjd|ttjddd|ttjdgdS)Nrrxryi)r!r%r rkrG)rrrrtest_pthread_sigmask_argumentssz2PendingSignalsTests.test_pthread_sigmask_argumentscCsd}td|dS)Na- if 1: import signal import os; import threading def handler(signum, frame): 1/0 def kill(signum): os.kill(os.getpid(), signum) def check_mask(mask): for sig in mask: assert isinstance(sig, signal.Signals), repr(sig) def read_sigmask(): sigmask = signal.pthread_sigmask(signal.SIG_BLOCK, []) check_mask(sigmask) return sigmask signum = signal.SIGUSR1 # Install our signal handler old_handler = signal.signal(signum, handler) # Unblock SIGUSR1 (and copy the old mask) to test our signal handler old_mask = signal.pthread_sigmask(signal.SIG_UNBLOCK, [signum]) check_mask(old_mask) try: kill(signum) except ZeroDivisionError: pass else: raise Exception("ZeroDivisionError not raised") # Block and then raise SIGUSR1. The signal is blocked: the signal # handler is not called, and the signal is now pending mask = signal.pthread_sigmask(signal.SIG_BLOCK, [signum]) check_mask(mask) kill(signum) # Check the new mask blocked = read_sigmask() check_mask(blocked) if signum not in blocked: raise Exception("%s not in %s" % (signum, blocked)) if old_mask ^ blocked != {signum}: raise Exception("%s ^ %s != {%s}" % (old_mask, blocked, signum)) # Unblock SIGUSR1 try: # unblock the pending signal calls immediately the signal handler signal.pthread_sigmask(signal.SIG_UNBLOCK, [signum]) except ZeroDivisionError: pass else: raise Exception("ZeroDivisionError not raised") try: kill(signum) except ZeroDivisionError: pass else: raise Exception("ZeroDivisionError not raised") # Check the new mask unblocked = read_sigmask() if signum in unblocked: raise Exception("%s in %s" % (signum, unblocked)) if blocked ^ unblocked != {signum}: raise Exception("%s ^ %s != {%s}" % (blocked, unblocked, signum)) if old_mask != unblocked: raise Exception("%s != %s" % (old_mask, unblocked)) z-c)r)rr_rrrtest_pthread_sigmasksJz(PendingSignalsTests.test_pthread_sigmaskc CsJd}td|2}|\}}|}|dkr)lenappendrZ perf_counterr rr)rDframe)Ntimesrrr@'s z5StressTest.measure_itimer_resolution..handlerrgMbP?cs g|]}|d|qS)rr).0i)rrr 5sz8StressTest.measure_itimer_resolution..rz,detected median itimer() resolution: %.6f s.)NN)rPr rrrrgrrrrange statisticsZmedianrverboseprint)rr@Z durationsZmedr)rrrmeasure_itimer_resolution#s z$StressTest.measure_itimer_resolutioncCs4|}|dkrdS|dkr dS|d|fdS)Ng-C6?i'g{Gz?dz^detected itimer resolution (%.3f s.) too high (> 10 ms.) on this platform (or system too busy))rrc)rZresorrrdecide_itimer_count;szStressTest.decide_itimer_countrztest needs setitimer()cs|}gdd}d fdd }|tj||tj||tj|d}td}x||krt t tj|d7}x&t |krt|krt d qWt t tj|d7}x&t |krt|krt d qWq^W| t |d dS) z; This test uses dependent signal handlers. cSsttjdtddS)Ngư>gh㈵>)r rrrandom)rDrrrr first_handlerRsz@StressTest.test_stress_delivery_dependent..first_handlerNcs|dS)N)r)rDr)sigsrrsecond_handlerZszAStressTest.test_stress_delivery_dependent..second_handlerrg.@rgh㈵>zSome signals were lost)NN)rrr rr&rgrrr*r~getpidrrr)rrrr expected_sigsdeadliner)rrtest_stress_delivery_dependentIs&  z)StressTest.test_stress_delivery_dependentcs|}gfdd}|tj||tj|d}td}xh||krttjdt dt t tj|d7}x&t |krt|krtdqWqFW|t |dd S) z> This test uses simultaneous signal handlers. cs|dS)N)r)rDr)rrrr@sz=StressTest.test_stress_delivery_simultaneous..handlerrg.@gư>gh㈵>rxzSome signals were lostN)rrr r&rgrrrrrr*r~rrrr)rrr@rrr)rr!test_stress_delivery_simultaneousws   z,StressTest.test_stress_delivery_simultaneousN) rrrrrrrrWrorQr rrrrrrrs - rcCs tdS)N)rZ reap_childrenrrrrtearDownModulesr__main__)#r*rr rJrr}rZ threadingrrWrrZtest.support.script_helperrrrn ImportErrorZTestCaserrXrrror0rCrYrQrprwrrrrrmainrrrrsN     M 7 @ T eB