✘✘ GRAYBYTE WORDPRESS FILE MANAGER ✘✘

a

XC?h���@sdd
lZdd
l
Z
dd
lZdd
lZdd
lZdd
lZdd
lZdd
lZdd
lZdd
l	Z	dd
l
Z
dd
lZddlm
Z

ddlmZmZ
zdd
lZWney�


d
ZYn0Gdd�dej�Ze�ejdkd�Gdd	�d	ej��
Ze�ejdkd
�Gdd�dej��
ZGd
d�dej�Ze�ejdkd�Gdd�dej��
Ze�eed�d�Gdd�dej��
Ze�ejdkd�Gdd�dej��
Ze�ejdkd�Gdd�dej��
ZGdd�dej�Z Gdd�dej�Z!Gdd�dej�Z"Gdd �d ej�Z#d!d"�Z$e%d#k�
r�e�&�
d
S)$�N)
�support)�assert_python_ok�spawn_pythonc@seZ
dZd
d�Zdd�ZdS)�GenericTestsc
Cs�tt
�
D]�}
tt
|
�}|
d
vr.|�|t
j�
q|
dvrF|�|t
j�
q|
�d�
rj|
�d�
sj|�|t
j�
q|
�d�
r|�|t
j�
|�t	j
d�
qdS)N>�SIG_IGN�SIG_DFL>�	SIG_BLOCK�SIG_SETMASK�SIG_UNBLOCKZSIGZSIG_ZCTRL_�win32)�dir�signal�getattr�assertIsInstance�HandlersZSigmasks�
startswith�Signals�assertEqual�sys�platform)�self�name�sig�r�5/opt/alt/python39/lib64/python3.9/test/test_signal.py�
test_enumss












zGenericTests.test_enumsc
Cs>tt
�
D]0}
tt
|
�}t�|�
rt�|�
s|�|jd
�
qdS)Nr
)rr
r�inspectZ	isroutineZ	isbuiltinr�
__module__)rr�valuerrr�test_functions_module_attr$s



z'GenericTests.test_functions_module_attrN)�__name__r�__qualname__rrrrrrrs
rrzNot valid on Windowsc@sZeZ
dZd
d�Zdd�Zdd�Zdd�Zd	d
�Zdd�Zd
d�Z	e
�ej
d�dd��
ZdS)�
PosixTestsc

GsdS�
Nr�r�argsrrr�trivial_signal_handler/s
z!PosixTests.trivial_signal_handlerc

Cs8|�t
tjd
�
|�t
tjd
|j�
|�t
tjd
�
dS)Ni�)�assertRaises�
ValueErrorr
�	getsignalr&�	strsignal�
rrrr�,test_out_of_range_signal_number_raises_error2s


�z7PosixTests.test_out_of_range_signal_number_raises_errorc

Cs|�t
tjtjd�
dSr#)r'�	TypeErrorr
�SIGUSR1r+rrr�0test_setting_signal_handler_to_none_raises_error:s


�z;PosixTests.test_setting_signal_handler_to_none_raises_errorc
CsZt�tj
|j�}
|�|
tj�
|�t�tj
�
|j�
t�tj
|
�
|�t�tj
�
|
�
dSr#)r
�SIGHUPr&rrrr))rZhuprrr�test_getsignal>s



�
zPosixTests.test_getsignalc

Cs@|�d
t
�t
j�
�
|�dt
�t
j�
�
|�dt
�t
j�
�
dS)NZ	InterruptZ
TerminatedZHangup)�assertInr
r*�SIGINT�SIGTERMr0r+rrr�test_strsignalFs


zPosixTests.test_strsignalc
Cs&tj
�t�
}
tj
�|
d
�}t|�

dS)Nzsignalinterproctester.py)�os�path�dirname�__file__�joinr)rr8Zscriptrrr�test_interprocess_signalLs


z#PosixTests.test_interprocess_signalc
Csdt�
�}
|�|
t�
|�tjj|
�
|�tjj|
�
|�d
|
�
|�tj	|
�
|�
t|
�
tj	�
dS�Nr
)r
�
valid_signalsr�setr2rr3�SIGALRM�assertNotIn�NSIG�
assertLess�len�r�
srrr�test_valid_signalsQs






zPosixTests.test_valid_signals�sys.executable required.c
Cs<tj
tjd
dgtjd�}
|�d|
j�
|�|
jt	j
�
dS)z+KeyboardInterrupt triggers exit via SIGINT.�-czaimport os, signal, time
os.kill(os.getpid(), signal.SIGINT)
for _ in range(999): time.sleep(0.01)�
�stderr�KeyboardInterruptN)�
subprocess�runr�
executable�PIPEr2rJr�
returncoder
r3)r�processrrr�!test_keyboard_interrupt_exit_codeZs

��
z,PosixTests.test_keyboard_interrupt_exit_codeN)r rr!r&r,r/r1r5r;rF�unittest�
skipUnlessrrNrRrrrrr"-s	
r"zWindows specificc@s2eZ
dZd
d�Zdd�Ze�ejd�dd��
Z	dS)	�WindowsSignalTestsc
Csdt�
�}
|�|
t�
|�t|
�
d
�
|�tjj|
�
|�	d|
�
|�	tj
|
�
|�t|
�
tj
�
dS)N�r
)r
r=rr>ZassertGreaterEqualrCr2rr3r@rArBrDrrrrFns






z%WindowsSignalTests.test_valid_signalsc
Cs�d
d�}
t�}t
jt
jt
jt
jt
jt
jt
jfD]0}t
�	|�
du
r.t
�
|t
�
||
��
|�
|�

q.|�|�

|�t
�
�
t
�
d|
�
Wd�
n1s�0


Y
|�t
�
�
t
�
d|
�
Wd�
n1s�0


Y
dS)Nc
SsdSr#r)�
x�
yrrr�<lambda>y�z3WindowsSignalTests.test_issue9324.<locals>.<lambda>����)r>r
�SIGABRTZSIGBREAK�SIGFPE�SIGILLr3�SIGSEGVr4r)�add�
assertTruer'r()r�handler�checkedrrrr�test_issue9324ws



�



*
z!WindowsSignalTests.test_issue9324rGc
Cs<tj
tjd
dgtjd�}
|�d|
j�
d}|�|
j|�
dS)z?KeyboardInterrupt triggers an exit using STATUS_CONTROL_C_EXIT.rHzraise KeyboardInterruptrIrKl:
N)	rLrMrrNrOr2rJrrP)rrQZSTATUS_CONTROL_C_EXITrrrrR�s


�

z4WindowsSignalTests.test_keyboard_interrupt_exit_codeN)
r rr!rFrerSrTrrNrRrrrrrUks	
rUc@sNeZ
dZd
d�Zdd�Zdd�Zdd�Zd	d
�Ze�	e
jdkd�d
d��
ZdS)�
WakeupFDTestsc

Cst|�t
�
�
tjtjd
�

Wd�
n1s.0


Y
|�t
�
�
t�tjd�
Wd�
n1sf0


Y
dS)N)
�signumF)r'r-r
�
set_wakeup_fdr3r+rrr�test_invalid_call�s
,
zWakeupFDTests.test_invalid_callc
Cs t�
�}
|�ttftj|
�
dSr#)rZmake_bad_fdr'r(�OSErrorr
rh)r�fdrrr�test_invalid_fd�s



�zWakeupFDTests.test_invalid_fdc
Cs0t��}
|
�
�}|
��
|�ttftj|�
dSr#)�socket�fileno�closer'r(rjr
rh)rZsockrkrrr�test_invalid_socket�s





�z!WakeupFDTests.test_invalid_socketc
Cs�t�
�\}
}|�tj|
�
|�tj|�
t�
�\}}|�tj|�
|�tj|�
ttd
�rrt�|d�
t�|d�
t�|�

|�t�|�
|�
|�t�d�
|�
|�t�d�
d�
dS)N�set_blockingFr[)	r6�pipe�
addCleanupro�hasattrrqr
rhr)rZr1Zw1Zr2Zw2rrr�test_set_wakeup_fd_result�s












z'WakeupFDTests.test_set_wakeup_fd_resultc
Cs�t��}
|�
|
j�

|
�d
�

|
��}t��}|�
|j�

|�d
�

|��}t�|�

|�t�|�
|�
|�t�d�
|�
|�t�d�
d�
dS)NFr[)rmrsroZsetblockingrnr
rhr)rZsock1Zfd1Zsock2Zfd2rrr� test_set_wakeup_fd_socket_result�s












z.WakeupFDTests.test_set_wakeup_fd_socket_resultrztests specific to POSIXc
Cs�t�
�\}
}|�tj|
�
|�tj|�
t�|d
�
|�t�
�}t�|�

Wd�
n1s^0


Y
|�	t
|j�
d|�
t�|d�
t�|�

t�d�

dS)NTz&the fd %s must be in non-blocking modeFr[)r6rrrsrorqr'r(r
rhr�str�	exception)rZrfdZwfd�cmrrr�test_set_wakeup_fd_blocking�s



(

�


z)WakeupFDTests.test_set_wakeup_fd_blockingN)
r rr!rirlrprurvrS�skipIfrrrzrrrrrf�s	
rfc@steZ
dZe�ed
ud�dd�
dd��
Ze�ed
ud�dd��
Zd	d
�Zdd�Z	d
d�Z
e�ee
d�d�dd��
Zd
S)�WakeupSignalTestsN�need _testcapiT�
�orderedc
Gs&d
�t
tt|��
||
�}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)
        rH)�format�tuple�map�intr)rZ	test_bodyrZsignals�coderrr�check_wakeup�s �"zWakeupSignalTests.check_wakeupc
	Cs|d
}
t�
�\}}zFzt�|d�
Wnty4


Yn0|�d�

Wt�|�

t�|�

nt�|�

t�|�

0td|
�
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:
                signal.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 piperH)r6rr�writerj�skipTestror)rr��
r�
wrrr�test_wakeup_write_error
s"





�

z)WakeupSignalTests.test_wakeup_write_errorc

Cs|�d
t
j�
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
r?r+rrr�test_wakeup_fd_earlyE
s
�z&WakeupSignalTests.test_wakeup_fd_earlyc

Cs|�d
t
j�
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+rrr�test_wakeup_fd_duringg
s
�z'WakeupSignalTests.test_wakeup_fd_duringc

Cs|�d
t
jt
j�
dS)Nz�def test():
            signal.signal(signal.SIGUSR1, handler)
            signal.raise_signal(signal.SIGUSR1)
            signal.raise_signal(signal.SIGALRM)
        )r�r
r.r?r+rrr�test_signum�
s
�zWakeupSignalTests.test_signum�pthread_sigmask�need signal.pthread_sigmask()c

Cs|jd
t
jt
jdd�
dS)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))
            signal.raise_signal(signum1)
            signal.raise_signal(signum2)
            # Unblocking the 2 signals calls the C signal handler twice
            signal.pthread_sigmask(signal.SIG_UNBLOCK, (signum1, signum2))
        Fr~)r�r
r.�SIGUSR2r+rrr�test_pending�
s
�zWakeupSignalTests.test_pending)r rr!rSr{�	_testcapir�r�r�r�r�rTrtr
r�rrrrr|�s
&

3"
�r|Z
socketpairzneed socket.socketpairc@sTeZ
dZe�ed
ud�dd��
Ze�ed
ud�dd��
Ze�ed
ud�dd��
Zd
S)	�WakeupSocketSignalTestsNr}c
Csd
}
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())

        signal.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()
        rH�
r�rr�rrr�test_socket�
sz#WakeupSocketSignalTests.test_socketc
Cs.tj
d
krd}
nd}
dj|
d�
}td|�
dS)N�nt�sendr�a+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:
            signal.raise_signal(signum)

        err = err.getvalue()
        if ('Exception ignored when trying to {action} to the signal wakeup fd'
            not in err):
            raise AssertionError(err)
        �
�actionrH�r6rr�r�rr�r�rrr�test_send_error�
s


!�"z'WakeupSocketSignalTests.test_send_errorc
Cs.tj
d
krd}
nd}
dj|
d�
}td|�
dS)Nr�r�r�a�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:
            signal.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:
            signal.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:
            signal.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:
            signal.raise_signal(signum)

        err = err.getvalue()
        if msg not in err:
            raise AssertionError("second set_wakeup_fd() test failed, "
                                 "stderr: %r" % err)

        r�rHr�r�rrr�test_warn_on_full_buffer�
s


c�dz0WakeupSocketSignalTests.test_warn_on_full_buffer)	r rr!rSr{r�r�r�r�rrrrr��
s

#

*
r�c@s,eZ
dZd
d�Zdd�Zdd�Zdd�Zd	S)
�SiginterruptTestc	Cs�d
|
f
}td|���}z |j
��}|jtjd�
\}}Wn*tjy`


|��
YWd�
dS0||}|�	�}|dv
r�t
d||f�
�
|dkWd�
SWd�
n1s�0


Y
dS)	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)
        rH)
ZtimeoutNF)��zChild error (exit code %s): %rr�)r�stdout�readline�communicater�
SHORT_TIMEOUTrLZTimeoutExpired�kill�wait�	Exception)rZ	interruptr�rQZ
first_liner�rJ�exitcoderrr�readpipe_interruptedas"#�$








�z%SiginterruptTest.readpipe_interruptedc
Cs|�d�
}
|�
|
�

dSr#�r�rb�rZinterruptedrrr�test_without_siginterrupt�s

z*SiginterruptTest.test_without_siginterruptc
Cs|�d
�
}
|�
|
�

dS�NTr�r�rrr�test_siginterrupt_on�s

z%SiginterruptTest.test_siginterrupt_onc
Cs|�d
�
}
|�
|
�

dS)NF)r�ZassertFalser�rrr�test_siginterrupt_off�s

z&SiginterruptTest.test_siginterrupt_offN)r rr!r�r�r�r�rrrrr�^s<r�c@sneZ
dZd
d�Zdd�Zdd�Zdd�Zd	d
�Zdd�Zd
d�Z	e
�ej
dvd�dd��
Zdd�Zdd�ZdS)�
ItimerTestc

Cs(d
|_d|_
d|_t�tj|j�|_dS)NFr
)�hndl_called�
hndl_count�itimerr
r?�sig_alrm�	old_alarmr+rrr�setUp�s



zItimerTest.setUpc

Cs,t�tj
|j�
|jdu
r(t�|jd
�
dSr<)r
r?r�r��	setitimerr+rrr�tearDown�s


zItimerTest.tearDownc
Gs
d
|_dSr�)
r�r$rrrr��s
zItimerTest.sig_alrmc
GsFd
|_|j
dkrt�d�
�
n|j
dkr4t�tjd�
|j
d7_
dS)NTr�z.setitimer didn't disable ITIMER_VIRTUAL timer.r
�
)r�r�r
�ItimerErrorr��ITIMER_VIRTUALr$rrr�
sig_vtalrm�s


zItimerTest.sig_vtalrmc
Gsd
|_t
�t
jd�
dS)NTr
)r�r
r��ITIMER_PROFr$rrr�sig_prof�s

zItimerTest.sig_profc

Cs|�t
jt
jd
d�
dS)Nr[r
)r'r
r�r�r+rrr�test_itimer_exc�szItimerTest.test_itimer_excc

Cs0tj
|_t�|jd
�
t��
|�|jd�
dS)Ng�?T)r
�ITIMER_REALr�r��pauserr�r+rrr�test_itimer_real�s



zItimerTest.test_itimer_real)
Znetbsd5zDitimer not reliable (does not mix well with threading) on some BSDs.c
Cs�tj
|_t�tj|j�
t�|jd
d�
t��}
t��|
dkr`tddd�}t�	|j�
dkr0qjq0|�
d�

|�t�	|j�
d�
|�|jd	�
dS)
Ng333333�?皙�����?�N@�90�2	
铖���r��8timeout: likely cause: machine too slow or load too highT)
r
r�r��	SIGVTALRMr�r��time�	monotonic�pow�	getitimerr�rr��rZ
start_time�
_rrr�test_itimer_virtual�s





zItimerTest.test_itimer_virtualc
Cs�tj
|_t�tj|j�
t�|jd
d
�
t��}
t��|
dkr`tddd�}t�	|j�
dkr0qjq0|�
d�

|�t�	|j�
d�
|�|jd�
dS)	Nr�r�r�r�r�r�r�T)
r
r�r��SIGPROFr�r�r�r�r�r�r�rr�r�rrr�test_itimer_prof�s






zItimerTest.test_itimer_profc

Cs2tj
|_t�|jd
�
t�d�

|�|jd�
dS)N���ư>r�T)r
r�r�r�r��sleeprr�r+rrr�test_setitimer_tinys



zItimerTest.test_setitimer_tinyN)r rr!r�r�r�r�r�r�r�rSr{rrr�r�r�rrrrr��s
	
�
r�c@s�
eZ
dZd
Ze�eed�d�dd��
Ze�eed�d�e�eed�d�dd	��
�
Z	e�eed
�d�dd
��
Z
e�eed�d�dd��
Ze�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)��
Ze�eed
�d�d*d+��
Zd,S)-�PendingSignalsTestsz[
    Test pthread_sigmask(), pthread_kill(), sigpending() and sigwait()
    functions.
    �
sigpendingzneed signal.sigpending()c

Cs|�t
��t��
dSr#)rr
r�r>r+rrr�test_sigpending_emptysz)PendingSignalsTests.test_sigpending_emptyr�r�c
Csd
}
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")
        rHr�r�rrr�test_sigpending#sz#PendingSignalsTests.test_sigpending�pthread_killzneed signal.pthread_kill()c
Csd
}
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")
        rHr�r�rrr�test_pthread_killBsz%PendingSignalsTests.test_pthread_killcCsd
|��|
f}t
d|�
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)
        rHN)�stripr)rZblocked�testr�rrr�wait_helper[s 
�%zPendingSignalsTests.wait_helper�sigwaitzneed signal.sigwait()c

Cs|�t
jd
�
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
r?r+rrr�test_sigwait�sz PendingSignalsTests.test_sigwait�sigwaitinfozneed signal.sigwaitinfo()c

Cs|�t
jd
�
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+rrr�test_sigwaitinfo�sz$PendingSignalsTests.test_sigwaitinfo�sigtimedwaitzneed signal.sigtimedwait()c

Cs|�t
jd
�
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+rrr�test_sigtimedwait�sz%PendingSignalsTests.test_sigtimedwaitc

Cs|�t
jd
�
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+rrr�test_sigtimedwait_poll�sz*PendingSignalsTests.test_sigtimedwait_pollc

Cs|�t
jd
�
dS)Nz�
        def test(signum):
            received = signal.sigtimedwait([signum], 1.0)
            if received is not None:
                raise Exception("received=%r" % (received,))
        r�r+rrr�test_sigtimedwait_timeout�sz-PendingSignalsTests.test_sigtimedwait_timeoutc
Cstj
}
|�ttj|
g
d
�
dS)Ng�)r
r?r'r(r�)rrgrrr�"test_sigtimedwait_negative_timeout�s
z6PendingSignalsTests.test_sigtimedwait_negative_timeoutc

Cstd
d�
dS)NrHa�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�r+rrr�test_sigwait_thread�s	z'PendingSignalsTests.test_sigwait_threadc

Cs�|�t
tj�
|�t
tjd
�
|�t
tjd
dd�
|�ttjdg�
|�t�
�"
t�tjtjg
�
Wd�
n1sv0


Y
|�t�
� 
t�tjdg
�
Wd�
n1s�0


Y
|�t�
�$
t�tjd
d>g
�
Wd�
n1s�0


Y
dS)Nr�r�r�i�r
i�)r'r-r
r�rjr(rrAr+rrr�test_pthread_sigmask_arguments�s




0

.

z2PendingSignalsTests.test_pthread_sigmask_argumentsc
CsJt�
tjt���}
|�tj
tj|
�
t�
tjt���}
|�|
t���
dSr#)r
r�rr=rsr	r
ZassertLessEqualrDrrr�"test_pthread_sigmask_valid_signals�s

z6PendingSignalsTests.test_pthread_sigmask_valid_signalsc
Csd
}
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))
        rHr�r�rrr�test_pthread_sigmask
sHz(PendingSignalsTests.test_pthread_sigmaskc
Cs^d
}
td|
��<}|�
�\}}|��}|dkr<td||f�
�
Wd�
n1sP0


Y
dS)Na7
if True:
            import threading
            import signal
            import sys

            def handler(signum, frame):
                sys.exit(3)

            signal.signal(signal.SIGUSR1, handler)
            signal.pthread_kill(threading.get_ident(), signal.SIGUSR1)
            sys.exit(2)
        rHr�zChild error (exit code %s): %s)rr�r�r�)rr�rQr�rJr�rrr�test_pthread_kill_main_threadNs





�z1PendingSignalsTests.test_pthread_kill_main_threadN)r rr!�__doc__rSrTrtr
r�r�r�r�r�r�r�r�r�r�r�r�r�r�r�rrrrr�s�

�

�
�
�

�
,
�


�
	
�
	
�

�

�

�
�
�

�

�
K
�r�c@sveZ
dZd
Zdd�Zdd�Zdd�Ze�e	e
d�d	�d
d��
Ze�e	e
d�d	�dd
��
Ze�e	e
d�d�dd��
Z
dS)�
StressTestz�
    Stress signal delivery, especially when a signal arrives in
    the middle of recomputing the signal state or executing
    previously tripped signal handlers.
    cCs t�|
|�}|�
tj|
|�
dSr#)r
rs)rrgrc�old_handlerrrr�setsigos

zStressTest.setsigc
s�d
�g�
d
��
fdd�	}
|�t
jt
jd�
|�t
j|
�
|
�
t�
�
�krVt�d�

q>�
f
dd�t	t�
�
d�
D�
}t
�|�
}tj
r�td	|f
�

|S)N�cs,t�
�
�kr(�
�
t���

t�tjd
�
dS)Nr�)rC�appendr��perf_counterr
r�r��rg�frame��
N�timesrrrcws

z5StressTest.measure_itimer_resolution.<locals>.handlerr
g����MbP?c
s g|]}
�|
d�|
�qS)
r�r)�.0�
i)
r�rr�
<listcomp>�rZz8StressTest.measure_itimer_resolution.<locals>.<listcomp>r�z,detected median itimer() resolution: %.6f s.)NN)rsr
r�r�r�r?rCr�r��range�
statisticsZmedianr�verbose�print)rrcZ	durationsZmedrr�r�measure_itimer_resolutionss










z$StressTest.measure_itimer_resolutionc
Cs4|��}
|
d
k
rdS|
dk
r 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
r�)rZresorrr�decide_itimer_count�s



�zStressTest.decide_itimer_countr�ztest needs setitimer()c
s�|��}
g�d
d�}d
�f
dd�	}|�
tj|�
|�
tj|�
|�
tj|�
d}t��tj	}||
kr�t
�t
��tj�
|d7}t
��
|kr�t��|kr�t�d�

q�t
�t
��tj�
|d7}t
��
|kr^t��|kr^t�d�

q�q^|�t
��
|
d	�
dS)z;
        This test uses dependent signal handlers.
        cSst�
tjd
t��d�
dS)Nr���h㈵��>)r
r�r��randomr�rrr�
first_handler�sz@StressTest.test_stress_delivery_dependent.<locals>.first_handlerNcs��|�

dSr#�
r�r��
Zsigsrr�second_handler�s
zAStressTest.test_stress_delivery_dependent.<locals>.second_handlerr
r�r	
�Some signals were lost)NN)r
r�r
r�r.r?r�r�rr�r6r��getpidrCr�r)rr�r
r
�
expected_sigs�deadlinerr

r�test_stress_delivery_dependent�s&









z)StressTest.test_stress_delivery_dependentc
s�|��}
g��f
d
d�}|�
tj|�
|�
tj|�
d}t��tj}||
kr�t�	tj
dt��d�
t�
t��tj�
|d7}t��
|krFt��|krFt�d�

q�qF|�t��
|
d�
dS)	z>
        This test uses simultaneous signal handlers.
        cs��|�

dSr#r
r�r

rrrc�s
z=StressTest.test_stress_delivery_simultaneous.<locals>.handlerr
r�r	
r�r
N)r
r�r
r.r?r�r�rr�r�r�r

r6r�r
rCr�r)rr�rcr
r
rr

r�!test_stress_delivery_simultaneous�s




z,StressTest.test_stress_delivery_simultaneousr.ztest needs SIGUSR1c
s&
tj
�d
�d
�d�
�f
dd���
��fdd�}
���fdd�}t����}|�tj�|�
tj|
d	�
}z�d}t���d}|��
|�
d
�
|��
|j	du
r�|�
|j	jt�
|�
d��d�t|j	j�
�
d
}Wd�
n1s�0


Y
|s�|��d
�
|����
Wd
�
|��
nd
�
|��
0dS)
Nr
Fcs�d
7�dS�Nr�rr�)
�num_received_signalsrr�custom_handler�szAStressTest.test_stress_modifying_handlers.<locals>.custom_handlercs�st�
��

�
d
7�
qdSr
)r
�raise_signalr)�do_stop�num_sent_signalsrgrr�set_interrupts�s


zAStressTest.test_stress_modifying_handlers.<locals>.set_interruptscs8�
d
kr4td�
D] }�t
jfD]}
t
�
�|
�
qqqdS)Nr
i N)r
r
r)r
rc)r
r
rgrr�cycle_handlers�s


zAStressTest.test_stress_modifying_handlers.<locals>.cycle_handlers)
�targetTzSignal z ignored due to race condition)r
r.rs�	threadingZThreadrZcatch_unraisable_exception�startr:Z
unraisabler�	exc_valuerjr2rwZ
assertGreaterrB)rr
r
r��
tZignoredryr)r
r
r
r
rgr�test_stress_modifying_handlers�s>

















�"


�
z)StressTest.test_stress_modifying_handlersN)r rr!r�r�r
r
rSrTrtr
r
r
r"
rrrrr�hs 

�
,
�

�r�c@s6eZ
dZd
d�Ze�ejdkd�dd��
Zdd�Z	d	S)
�RaiseSignalTestc

Cs:|�t
�
�
t�tj�

Wd�
n1s,0


Y
dSr#)r'�KeyboardInterruptr
r
r3r+rrr�test_sigint%s

zRaiseSignalTest.test_sigintrzWindows specific testc

CsVzd
}
t�
|
�

|�d�

Wn4tyP
}
z|jtjkr:n�WYd}~n
d}~00dS)Nr�z#OSError (Invalid argument) expected)r
r
Zfailrj�errnoZEINVAL)rr0�
errr�test_invalid_argument)s






z%RaiseSignalTest.test_invalid_argumentc
sJd
��f
dd�}
t�tj
|
�}|�tjtj
|�
t�tj
�

|���

dS)NFc
sd
�dSr�r)�
a�
b�
Zis_okrrrc7sz-RaiseSignalTest.test_handler.<locals>.handler)r
r3rsr
rb)rrcZ
old_signalrr+
r�test_handler5s



zRaiseSignalTest.test_handlerN)
r rr!r%
rSr{rrr(
r,
rrrrr#
#s

r#
c@s&eZ
dZe�eed
�d�dd��
ZdS)�PidfdSignalTest�pidfd_send_signalzpidfd support not built inc
Cs 
|�t
�
�}
t�d
tj�
Wd�
n1s.0


Y
|
jjtjkrR|�d�

n|
jjtj	krj|�d�

|�
|
jjtj�
t�
dt����tj�}|�tj|�
|�td��$
t�|tjt�d
�
Wd�
n1s�0


Y
|�t�
�
t�|tj�
Wd�
n1�
s0


Y
dS)Nr
zkernel does not support pidfdsz"Not enough privileges to use pidfsz/proc/z^siginfo must be None$)r'rjr
r.
r3rxr&
ZENOSYSr�ZEPERMrZEBADFr6�openr
�O_DIRECTORYrsroZassertRaisesRegexr-�objectr$
)rryZmy_pidfdrrr�test_pidfd_send_signalCs
,









2

z&PidfdSignalTest.test_pidfd_send_signalN)r rr!rSrTrtr
r2
rrrrr-
As


�r-
cCst�
�
dSr#)rZ
reap_childrenrrrr�tearDownModuleVs
r3
�__main__)'r&
rr6r

r
rmr
rLrr
r�rSr�rZtest.support.script_helperrrr��ImportErrorZTestCaserr{rr"rTrUrfr|rtr�r�r�r�r�r#
r-
r3
r �mainrrrr�<module>
sV


















=
/M
6
@
T
eQ<