B 2*™\e¥ã@sÐddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m Z mZy ddlZWnek rdZYnXGdd„de jƒZe  ejdkd¡Gdd „d e jƒƒZe  ejdkd ¡Gd d „d e jƒƒZGd d„de jƒZe  ejdkd¡Gdd„de jƒƒZe  eedƒd¡Gdd„de jƒƒZe  ejdkd¡Gdd„de jƒƒZe  ejdkd¡Gdd„de jƒƒZGdd„de jƒZGdd„de jƒZdd„Z e!dkrÌe  "¡dS) éN)Úsupport)Úassert_python_okÚ spawn_pythonc@seZdZdd„ZdS)Ú GenericTestscCsšx”ttƒD]ˆ}tt|ƒ}|dkr0| |tj¡q |dkrH| |tj¡q | d¡rl| d¡sl| |tj¡q | d¡r | |tj¡| t j d¡q WdS)N>ÚSIG_DFLÚSIG_IGN>Ú SIG_SETMASKÚ SIG_BLOCKÚ SIG_UNBLOCKZSIGZSIG_ZCTRL_Úwin32) ÚdirÚsignalÚgetattrÚassertIsInstanceÚHandlersZSigmasksÚ startswithZSignalsÚ assertEqualÚsysÚplatform)ÚselfÚnameÚsig©rú;C:\ALexclude\prg\programme\Python37\Lib\test\test_signal.pyÚ test_enumss  zGenericTests.test_enumsN)Ú__name__Ú __module__Ú __qualname__rrrrrrsrr zNot valid on Windowsc@s4eZdZdd„Zdd„Zdd„Zdd„Zd d „Zd S) Ú PosixTestscGsdS)Nr)rÚargsrrrÚtrivial_signal_handler%sz!PosixTests.trivial_signal_handlercCs(| ttjd¡| ttjd|j¡dS)Ni’)Ú assertRaisesÚ ValueErrorr Ú getsignalr )rrrrÚ,test_out_of_range_signal_number_raises_error(s z7PosixTests.test_out_of_range_signal_number_raises_errorcCs| ttjtjd¡dS)N)r!Ú TypeErrorr ÚSIGUSR1)rrrrÚ0test_setting_signal_handler_to_none_raises_error.s z;PosixTests.test_setting_signal_handler_to_none_raises_errorcCsZt tj|j¡}| |tj¡| t tj¡|j¡t tj|¡| t tj¡|¡dS)N)r ZSIGHUPr rrrr#)rZhuprrrÚtest_getsignal2s zPosixTests.test_getsignalcCs&tj t¡}tj |d¡}t|ƒdS)Nzsignalinterproctester.py)ÚosÚpathÚdirnameÚ__file__Újoinr)rr+ZscriptrrrÚtest_interprocess_signal;s z#PosixTests.test_interprocess_signalN)rrrr r$r'r(r.rrrrr#s  rzWindows specificc@seZdZdd„ZdS)ÚWindowsSignalTestsc Cs¶dd„}tƒ}xTtjtjtjtjtjtjtjfD]0}t  |¡dk r0t |t ||¡¡|  |¡q0W|  |¡|  t ¡t d|¡WdQRX|  t ¡t d|¡WdQRXdS)NcSsdS)Nr)ÚxÚyrrrÚEóz3WindowsSignalTests.test_issue9324..éÿÿÿÿé)Úsetr ÚSIGABRTÚSIGBREAKÚSIGFPEÚSIGILLÚSIGINTÚSIGSEGVÚSIGTERMr#ÚaddÚ assertTruer!r")rÚhandlerÚcheckedrrrrÚtest_issue9324Cs     z!WindowsSignalTests.test_issue9324N)rrrrBrrrrr/Asr/c@sNeZdZdd„Zdd„Zdd„Zdd„Zd d „Ze  e j d kd ¡d d„ƒZ dS)Ú WakeupFDTestsc CsL| t¡tjtjdWdQRX| t¡t tjd¡WdQRXdS)N)ÚsignumF)r!r%r Ú set_wakeup_fdr;)rrrrÚtest_invalid_call[s  zWakeupFDTests.test_invalid_callcCs t ¡}| ttftj|¡dS)N)rZ make_bad_fdr!r"ÚOSErrorr rE)rÚfdrrrÚtest_invalid_fdds zWakeupFDTests.test_invalid_fdcCs0t ¡}| ¡}| ¡| ttftj|¡dS)N)ÚsocketÚfilenoÚcloser!r"rGr rE)rZsockrHrrrÚtest_invalid_socketis  z!WakeupFDTests.test_invalid_socketcCs¶t ¡\}}| tj|¡| tj|¡t ¡\}}| tj|¡| tj|¡ttdƒrrt |d¡t |d¡t |¡| t |¡|¡| t d¡|¡| t d¡d¡dS)NÚ set_blockingFr4) r)ÚpipeÚ addCleanuprLÚhasattrrNr rEr)rZr1Zw1Zr2Zw2rrrÚtest_set_wakeup_fd_resultps      z'WakeupFDTests.test_set_wakeup_fd_resultcCst ¡}| |j¡| d¡| ¡}t ¡}| |j¡| d¡| ¡}t |¡| t |¡|¡| t d¡|¡| t d¡d¡dS)NFr4)rJrPrLZ setblockingrKr rEr)rZsock1Zfd1Zsock2Zfd2rrrÚ test_set_wakeup_fd_socket_results     z.WakeupFDTests.test_set_wakeup_fd_socket_resultr ztests specific to POSIXc CsŽt ¡\}}| tj|¡| tj|¡t |d¡| t¡}t |¡WdQRX|  t |j ƒd|¡t |d¡t |¡t d¡dS)NTz&the fd %s must be in non-blocking modeFr4) r)rOrPrLrNr!r"r rErÚstrZ exception)rZrfdZwfdÚcmrrrÚtest_set_wakeup_fd_blocking“s       z)WakeupFDTests.test_set_wakeup_fd_blockingN) rrrrFrIrMrRrSÚunittestÚskipIfrrrVrrrrrCYs  rCc@steZdZe edkd¡ddœdd„ƒZe edkd¡dd„ƒZd d „Zd d „Z d d„Z e  e e dƒd¡dd„ƒZdS)ÚWakeupSignalTestsNzneed _testcapiT)ÚorderedcGs&d ttt|ƒƒ||¡}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)ÚformatÚtupleÚmapÚintr)rZ test_bodyrZZsignalsÚcoderrrÚ check_wakeup¨s#zWakeupSignalTests.check_wakeupc Csjd}t ¡\}}z4yt |d¡Wntk r6Yn X| d¡Wdt |¡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)rOÚwriterGÚskipTestrLr)rr_ÚrÚwrrrÚtest_wakeup_write_errorÏs&   z)WakeupSignalTests.test_wakeup_write_errorcCs| dtj¡dS)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) # 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)rrrrÚtest_wakeup_fd_earlysz&WakeupSignalTests.test_wakeup_fd_earlycCs| dtj¡dS)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)rrrrÚtest_wakeup_fd_during%sz'WakeupSignalTests.test_wakeup_fd_duringcCs| dtjtj¡dS)NzÊdef 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_signumÚpthread_sigmaskzneed signal.pthread_sigmask()cCs|jdtjtjdddS)Naìdef 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&ZSIGUSR2)rrrrÚ test_pendingKs zWakeupSignalTests.test_pending)rrrrWrXÚ _testcapir`rfrhrirjÚ skipUnlessrQr rlrrrrrY¦s&4" rYZ socketpairzneed socket.socketpairc@sTeZdZe edkd¡dd„ƒZe edkd¡dd„ƒZe edkd¡dd„ƒZdS) ÚWakeupSocketSignalTestsNzneed _testcapicCsd}td|ƒdS)Na·if 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)NÚntÚsendrba.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)rrsr_rrrÚtest_send_error„s  " z'WakeupSocketSignalTests.test_send_errorcCs.tjdkrd}nd}dj|d}td|ƒdS)Nrqrrrba× 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) )rsz-c)r)rr[r)rrsr_rrrÚtest_warn_on_full_buffer¯s  d z0WakeupSocketSignalTests.test_warn_on_full_buffer) rrrrWrXrmrprtrurrrrro]s$+roc@s,eZdZdd„Zdd„Zdd„Zdd„Zd S) ÚSiginterruptTestc Cs’d|f}td|ƒt}y|j ¡}|jdd\}}Wntjk rR| ¡dSX||}| ¡}|dkr|td||fƒ‚|dkSWd QRXd S) z¿Perform 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. aòif 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): %rrxN) rÚstdoutÚreadlineÚ communicateÚ subprocessZTimeoutExpiredÚkillÚwaitÚ Exception)rZ interruptr_ÚprocessZ first_lineryÚstderrÚexitcoderrrÚreadpipe_interrupted s*    z%SiginterruptTest.readpipe_interruptedcCs| d¡}| |¡dS)N)rƒr?)rÚ interruptedrrrÚtest_without_siginterrupt\s z*SiginterruptTest.test_without_siginterruptcCs| d¡}| |¡dS)NT)rƒr?)rr„rrrÚtest_siginterrupt_oncs z%SiginterruptTest.test_siginterrupt_oncCs| d¡}| |¡dS)NF)rƒZ assertFalse)rr„rrrÚtest_siginterrupt_offjs z&SiginterruptTest.test_siginterrupt_offN)rrrrƒr…r†r‡rrrrrvs<rvc@sneZdZdd„Zdd„Zdd„Zdd„Zd d „Zd d „Zd d„Z e   e j dkd¡dd„ƒZdd„Zdd„ZdS)Ú ItimerTestcCs(d|_d|_d|_t tj|j¡|_dS)NFr)Ú hndl_calledÚ hndl_countÚitimerr rgÚsig_alrmÚ old_alarm)rrrrÚsetUptszItimerTest.setUpcCs,t tj|j¡|jdk r(t |jd¡dS)Nr)r rgrr‹Ú setitimer)rrrrÚtearDownzs zItimerTest.tearDowncGs d|_dS)NT)r‰)rrrrrrŒ€szItimerTest.sig_alrmcGsFd|_|jdkrt d¡‚n|jdkr4t tjd¡|jd7_dS)NTrxz.setitimer didn't disable ITIMER_VIRTUAL timer.ré)r‰rŠr Ú ItimerErrorrÚITIMER_VIRTUAL)rrrrrÚ sig_vtalrmƒs    zItimerTest.sig_vtalrmcGsd|_t tjd¡dS)NTr)r‰r rÚ ITIMER_PROF)rrrrrÚsig_profszItimerTest.sig_profcCs| tjtjdd¡dS)Nr4r)r!r r’r)rrrrÚtest_itimer_exc”szItimerTest.test_itimer_exccCs0tj|_t |jd¡t ¡| |jd¡dS)Ngð?T)r Ú ITIMER_REALr‹rZpauserr‰)rrrrÚtest_itimer_realszItimerTest.test_itimer_real)Znetbsd5zDitimer not reliable (does not mix well with threading) on some BSDs.cCs”tj|_t tj|j¡t |jdd¡t ¡}xr‘T)r r˜r‹rršÚsleeprr‰)rrrrÚtest_setitimer_tinyÎs zItimerTest.test_setitimer_tinyN)rrrrŽrrŒr”r–r—r™rWrXrrr r¢r¤rrrrrˆrs   rˆc@s eZdZdZe eedƒd¡dd„ƒZe eedƒd¡e eedƒd¡dd „ƒƒZ e eed ƒd ¡d d „ƒZ e eedƒd¡dd„ƒZ e eedƒd¡dd„ƒZ e eedƒd¡dd„ƒZ e eedƒd¡dd„ƒZe eedƒd¡dd„ƒZe eedƒd¡dd„ƒZe eedƒd¡d d!„ƒZe eedƒd¡e eedƒd¡d"d#„ƒƒZe eedƒd¡d$d%„ƒZe eedƒd¡d&d'„ƒZe eed ƒd ¡d(d)„ƒZd*S)+ÚPendingSignalsTestsz[ Test pthread_sigmask(), pthread_kill(), sigpending() and sigwait() functions. Ú sigpendingzneed signal.sigpending()cCs| t ¡tƒ¡dS)N)rr r¦r6)rrrrÚtest_sigpending_emptyÝsz)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_rrrÚtest_sigpendingâsz#PendingSignalsTests.test_sigpendingZ pthread_killzneed signal.pthread_kill()cCsd}td|ƒdS)Naâif 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_rrrÚtest_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)rZblockedÚtestr_rrrÚ wait_helpers'zPendingSignalsTests.wait_helperZsigwaitzneed signal.sigwait()cCs| tjd¡dS)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)) )r¬r rg)rrrrÚ test_sigwaitHsz PendingSignalsTests.test_sigwaitZ sigwaitinfozneed signal.sigwaitinfo()cCs| tjd¡dS)Nz× def test(signum): signal.alarm(1) info = signal.sigwaitinfo([signum]) if info.si_signo != signum: raise Exception("info.si_signo != %s" % signum) )r¬r rg)rrrrÚtest_sigwaitinfoTsz$PendingSignalsTests.test_sigwaitinfoÚ sigtimedwaitzneed signal.sigtimedwait()cCs| tjd¡dS)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) )r¬r rg)rrrrÚtest_sigtimedwait_sz%PendingSignalsTests.test_sigtimedwaitcCs| tjd¡dS)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) )r¬r rg)rrrrÚtest_sigtimedwait_polljsz*PendingSignalsTests.test_sigtimedwait_pollcCs| tjd¡dS)Nz¿ def test(signum): received = signal.sigtimedwait([signum], 1.0) if received is not None: raise Exception("received=%r" % (received,)) )r¬r rg)rrrrÚtest_sigtimedwait_timeoutwsz-PendingSignalsTests.test_sigtimedwait_timeoutcCstj}| ttj|gd¡dS)Ngð¿)r rgr!r"r¯)rrDrrrÚ"test_sigtimedwait_negative_timeoutsz6PendingSignalsTests.test_sigtimedwait_negative_timeoutcCstddƒdS)Nz-ca­if 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)rrrrÚtest_sigwait_thread‡s z'PendingSignalsTests.test_sigwait_threadcCsH| ttj¡| ttjd¡| ttjddd¡| ttjdg¡dS)Nr‘rwrxi¤)r!r%r rkrG)rrrrÚtest_pthread_sigmask_arguments©sz2PendingSignalsTests.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_rrrÚtest_pthread_sigmask±sJz(PendingSignalsTests.test_pthread_sigmaskc CsJd}td|ƒ2}| ¡\}}| ¡}|dkr)ÚlenÚappendršZ perf_counterr rr˜)rDÚframe)ÚNÚtimesrrr@'s z5StressTest.measure_itimer_resolution..handlerrgü©ñÒMbP?cs g|]}ˆ|dˆ|‘qS)r‘r)Ú.0Úi)rÀrrú 5sz8StressTest.measure_itimer_resolution..r‘z,detected median itimer() resolution: %.6f s.)NN)rPr rr˜rºrgr¼ršr£ÚrangeÚ statisticsZmedianrÚverboseÚprint)rr@Z durationsZmedr)r¿rÀrÚmeasure_itimer_resolution#s z$StressTest.measure_itimer_resolutioncCs4| ¡}|dkrdS|dkr dS| d|f¡dS)Ng-Cëâ6?i'g{®Gáz„?édz^detected itimer resolution (%.3f s.) too high (> 10 ms.) on this platform (or system too busy))rÈrc)rZresorrrÚdecide_itimer_count;szStressTest.decide_itimer_countrztest needs setitimer()cs| ¡}g‰dd„}d ‡fdd„ }| tj|¡| tj|¡| tj|¡d}t ¡d}x||krìt  t  ¡tj¡|d7}x&t ˆƒ|kr¦t ¡|kr¦t  d ¡q‚Wt  t  ¡tj¡|d7}x&t ˆƒ|krèt ¡|krèt  d ¡qÄWq^W|  t ˆƒ|d ¡dS) z; This test uses dependent signal handlers. cSst tjdt ¡d¡dS)Ngíµ ÷Æ°>gñh㈵øä>)r rr˜Úrandom)rDr¾rrrÚ first_handlerRsz@StressTest.test_stress_delivery_dependent..first_handlerNcsˆ |¡dS)N)r½)rDr¾)ÚsigsrrÚsecond_handlerZszAStressTest.test_stress_delivery_dependent..second_handlerrg.@r‘gñh㈵øä>zSome signals were lost)NN)rÊrºr r¡r&rgršr›r)r}Úgetpidr¼r£r)rr¿rÌrÎÚ expected_sigsÚdeadliner)rÍrÚtest_stress_delivery_dependentIs&  z)StressTest.test_stress_delivery_dependentcsÄ| ¡}g‰‡fdd„}| tj|¡| tj|¡d}t ¡d}xh||kr¬t tjdt   ¡d¡t   t   ¡tj¡|d7}x&t ˆƒ|kr¨t ¡|kr¨t d¡q„WqFW| t ˆƒ|d¡d S) z> This test uses simultaneous signal handlers. csˆ |¡dS)N)r½)rDr¾)rÍrrr@€sz=StressTest.test_stress_delivery_simultaneous..handlerrg.@gíµ ÷Æ°>gñh㈵øä>rwzSome signals were lostN)rÊrºr r&rgršr›rr˜rËr)r}rÏr¼r£r)rr¿r@rÐrÑr)rÍrÚ!test_stress_delivery_simultaneousws   z,StressTest.test_stress_delivery_simultaneousN) rrrr¸rºrÈrÊrWrnrQr rÒrÓrrrrr¹s - r¹cCs t ¡dS)N)rZ reap_childrenrrrrÚtearDownModule™srÔÚ__main__)#r)rËr rJrÅr|rZ threadingršrWr«rZtest.support.script_helperrrrmÚ ImportErrorZTestCaserrXrrrnr/rCrYrQrorvrˆr¥r¹rÔrÚmainrrrrÚsN     M 7 @ T eB