Nexus File Manager
v2.0
🏠
Dashboard
⚡
System
🌐
›
opt
›
alt
›
python38
›
lib64
›
python3.8
›
test
›
__pycache__
Quick:
⬆️ Parent
🌐 Root
🏠 Home
🌍 WWW
📁 Temp
⚙️ Etc
📤 Upload
📁 Create
⚡ WordPress Admin
🔄 Refresh
✏️ test_signal.cpython-38.pyc
← Back
U i�fv� � @ s� d dl Z d dlZd dlZd dlZd dlZd dlZd dlZd dlZd dlZd dl Z d dl Z d dl m Z d dl mZmZ z d dlZW n ek r� dZY nX G dd� de j�Ze �ejdkd�G dd � d e j��Ze �ejdkd �G d d � d e j��ZG d d� de j�Ze �ejdkd�G dd� de j��Ze �eed�d�G dd� de j��Z e �ejdkd�G dd� de j��Z e �ejdkd�G dd� de j��Z G dd� de j�Z G dd � d e j�Z G d d � d e j�Z!d d � Z"e#d!k�r�e �$� dS )"� N)�support)�assert_python_ok� spawn_pythonc @ s e Zd Zdd� ZdS )� GenericTestsc C s� t t�D ]�}tt|�}|dkr.| �|tj� q|dkrF| �|tj� q|�d�rj|�d�sj| �|tj� q|�d�r| �|tj� | �t j d� qd S )N> �SIG_IGN�SIG_DFL> � SIG_UNBLOCK� SIG_BLOCK� SIG_SETMASKZSIGZSIG_ZCTRL_�win32) �dir�signal�getattr�assertIsInstance�HandlersZSigmasks� startswith�Signals� assertEqual�sys�platform)�self�name�sig� r �5/opt/alt/python38/lib64/python3.8/test/test_signal.py� test_enums s zGenericTests.test_enumsN)�__name__� __module__� __qualname__r r r r r r s r r zNot valid on Windowsc @ sZ e Zd Zdd� Zdd� Zdd� Zdd� Zd d � Zd d � Zd d� Z e � e j d�dd� �ZdS )� PosixTestsc G s d S �Nr �r �argsr r r �trivial_signal_handler& s z!PosixTests.trivial_signal_handlerc C s8 | � ttjd� | � ttjd| j� | � ttjd� d S )Ni� )� assertRaises� ValueErrorr � getsignalr# � strsignal�r r r r �,test_out_of_range_signal_number_raises_error) s �z7PosixTests.test_out_of_range_signal_number_raises_errorc C s | � ttjtjd � d S r )r$ � TypeErrorr �SIGUSR1r( r r r �0test_setting_signal_handler_to_none_raises_error1 s �z;PosixTests.test_setting_signal_handler_to_none_raises_errorc C sZ t � t j| j�}| �|t j� | �t �t j�| j� t � t j|� | �t �t j�|� d S r )r �SIGHUPr# r r r r& )r Zhupr r r �test_getsignal5 s �zPosixTests.test_getsignalc C s@ | � dt�tj�� | � dt�tj�� | � dt�tj�� d S )NZ InterruptZ TerminatedZHangup)�assertInr r' �SIGINT�SIGTERMr- r( r r r �test_strsignal= s zPosixTests.test_strsignalc C s& t j�t�}t j�|d�}t|� d S )Nzsignalinterproctester.py)�os�path�dirname�__file__�joinr )r r5 Zscriptr r r �test_interprocess_signalC s z#PosixTests.test_interprocess_signalc C sd t �� }| �|t� | �t jj|� | �t jj|� | �d|� | �t j |� | � t |�t j � d S �Nr ) r � valid_signalsr �setr/ r r0 �SIGALRM� assertNotIn�NSIG� assertLess�len�r �sr r r �test_valid_signalsH s z PosixTests.test_valid_signals�sys.executable required.c C s< t jtjddgt jd�}| �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�PIPEr/ rG r � returncoder r0 )r �processr r r �!test_keyboard_interrupt_exit_codeQ s ��z,PosixTests.test_keyboard_interrupt_exit_codeN)r r r r# r) r, r. r2 r8 rC �unittest� skipUnlessr rK rO r r r r r $ s r zWindows specificc @ s2 e Zd Zdd� Zdd� Ze�ejd�dd� �Z dS ) �WindowsSignalTestsc C sd t �� }| �|t� | �t|�d� | �t jj|� | � d|� | � t j |� | � t|�t j � d S )N� r ) r r: r r; ZassertGreaterEqualr@ r/ r r0 r= r> r? rA r r r rC e s z%WindowsSignalTests.test_valid_signalsc C s� dd� }t � }tjtjtjtjtjtjtjfD ]0}t� |�d k r.t�|t�||�� |� |� q.| � |� | � t �� t�d|� W 5 Q R X | � t �� t�d|� W 5 Q R X d S )Nc S s d S r r )�x�yr r r �<lambda>p � z3WindowsSignalTests.test_issue9324.<locals>.<lambda>���� )r; r �SIGABRTZSIGBREAK�SIGFPE�SIGILLr0 �SIGSEGVr1 r& �add� assertTruer$ r% )r �handler�checkedr r r r �test_issue9324n s � z!WindowsSignalTests.test_issue9324rD c C s< t jtjddgt jd�}| �d|j� d}| �|j|� dS )z?KeyboardInterrupt triggers an exit using STATUS_CONTROL_C_EXIT.rE zraise KeyboardInterruptrF rH l : N) rI rJ r rK rL r/ rG r rM )r rN ZSTATUS_CONTROL_C_EXITr r r rO � s �z4WindowsSignalTests.test_keyboard_interrupt_exit_codeN) r r r rC rb rP rQ r rK rO r r r r rR b s rR c @ sN e Zd Zdd� Zdd� Zdd� Zdd� Zd d � Ze� e j d kd �d d� �Z dS )� WakeupFDTestsc C sL | � t�� tjtjd� W 5 Q R X | � t�� t�tjd� W 5 Q R X d S )N)�signumF)r$ r* r � set_wakeup_fdr0 r( r r r �test_invalid_call� s z WakeupFDTests.test_invalid_callc C s t �� }| �ttftj|� d S r )r Z make_bad_fdr$ r% �OSErrorr re )r �fdr r r �test_invalid_fd� s �z WakeupFDTests.test_invalid_fdc C s0 t � � }|�� }|�� | �ttftj|� d S r )�socket�fileno�closer$ r% rg r re )r Zsockrh r r r �test_invalid_socket� s �z!WakeupFDTests.test_invalid_socketc C s� t �� \}}| �t j|� | �t j|� t �� \}}| �t j|� | �t j|� tt d�rrt �|d� t �|d� t�|� | �t�|�|� | �t�d�|� | �t�d�d� d S )N� set_blockingFrX ) r3 �pipe� addCleanuprl �hasattrrn r re r )r Zr1Zw1Zr2Zw2r r r �test_set_wakeup_fd_result� s z'WakeupFDTests.test_set_wakeup_fd_resultc C s� t � � }| �|j� |�d� |�� }t � � }| �|j� |�d� |�� }t�|� | �t�|�|� | �t�d�|� | �t�d�d� d S )NFrX )rj rp rl Z setblockingrk r re r )r Zsock1Zfd1Zsock2Zfd2r r r � test_set_wakeup_fd_socket_result� s z.WakeupFDTests.test_set_wakeup_fd_socket_resultr ztests specific to POSIXc C s� t �� \}}| �t j|� | �t j|� t �|d� | �t��}t�|� W 5 Q R X | � t |j �d| � t �|d� t�|� t�d� d S )NTz&the fd %s must be in non-blocking modeFrX ) r3 ro rp rl rn r$ r% r re r �strZ exception)r ZrfdZwfd�cmr r r �test_set_wakeup_fd_blocking� s � z)WakeupFDTests.test_set_wakeup_fd_blockingN) r r r rf ri rm rr rs rP �skipIfr r rv r r r r rc � s rc c @ st e Zd Ze�edkd�dd�dd��Ze�edkd�dd� �Zd d � Zd d � Z d d� Z e� e e d�d�dd� �ZdS )�WakeupSignalTestsN�need _testcapiT��orderedc G s& d� ttt|��||�}td|� d S )Na if 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) rE )�format�tuple�map�intr )r Z test_bodyr{ Zsignals�coder r r � check_wakeup� s �"z WakeupSignalTests.check_wakeupc C sj d}t �� \}}z4zt �|d� W n tk r6 Y n X | �d� W 5 t �|� t �|� X td|� d S )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 piperE )r3 ro rl �writerg �skipTestr )r r� �r�wr r r �test_wakeup_write_error s " z)WakeupSignalTests.test_wakeup_write_errorc C s | � dtj� d S )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( r r r �test_wakeup_fd_early< s �z&WakeupSignalTests.test_wakeup_fd_earlyc C s | � dtj� d S )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( r r r �test_wakeup_fd_during^ s �z'WakeupSignalTests.test_wakeup_fd_duringc C s | � dtjtj� d S )Nz�def test(): signal.signal(signal.SIGUSR1, handler) signal.raise_signal(signal.SIGUSR1) signal.raise_signal(signal.SIGALRM) )r� r r+ r< r( r r r � test_signum| s �z WakeupSignalTests.test_signum�pthread_sigmask� need signal.pthread_sigmask()c C s | j dtjtjdd� d S )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)) Frz )r� r r+ �SIGUSR2r( r r r � test_pending� s �z WakeupSignalTests.test_pending)r r r rP rw � _testcapir� r� r� r� r� rQ rq r r� r r r r rx � s & 3" �rx Z socketpairzneed socket.socketpairc @ sT e Zd Ze�edkd�dd� �Ze�edkd�dd� �Ze�edkd�dd� �ZdS ) �WakeupSocketSignalTestsNry c C s d}t d|� d S )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() rE �r �r r� r r r � test_socket� s z#WakeupSocketSignalTests.test_socketc C s. t jdkrd}nd}dj|d�}td|� d S )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) ��actionrE �r3 r r| r �r r� r� r r r �test_send_error� s !�"z'WakeupSocketSignalTests.test_send_errorc C s. t jdkrd}nd}dj|d�}td|� d S )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� rE r� r� r r r �test_warn_on_full_buffer� s c�dz0WakeupSocketSignalTests.test_warn_on_full_buffer) r r r rP rw r� r� r� r� r r r r r� � s # *r� c @ s, e Zd Zdd� Zdd� Zdd� Zdd� Zd S ) �SiginterruptTestc C s� d|f }t d|���}z |j�� }|jdd�\}}W n* tjk r^ |�� Y W 5 Q R � dS X || }|�� }|dkr�td||f ��|dkW 5 Q R � S W 5 Q R X d 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) rE g @)ZtimeoutF)� � z Child error (exit code %s): %rr� N) r �stdout�readline� communicaterI ZTimeoutExpired�kill�wait� Exception)r Z interruptr� rN Z first_liner� rG �exitcoder r r �readpipe_interruptedX s" #�$ �z%SiginterruptTest.readpipe_interruptedc C s | � d �}| �|� d S r �r� r_ �r Z interruptedr r r �test_without_siginterrupt� s z*SiginterruptTest.test_without_siginterruptc C s | � d�}| �|� d S �NTr� r� r r r �test_siginterrupt_on� s z%SiginterruptTest.test_siginterrupt_onc C s | � d�}| �|� d S )NF)r� Z assertFalser� r r r �test_siginterrupt_off� s z&SiginterruptTest.test_siginterrupt_offN)r r r r� r� r� r� r r r r r� U s <r� c @ sn e Zd Zdd� Zdd� Zdd� Zdd� Zd d � Zd d � Zd d� Z e � e j dkd�dd� �Zdd� Zdd� ZdS )� ItimerTestc C s( d| _ d| _d | _t�tj| j�| _d S )NFr )� hndl_called� hndl_count�itimerr r< �sig_alrm� old_alarmr( r r r �setUp� s zItimerTest.setUpc C s, t � t j| j� | jd k r(t �| jd� d S r9 )r r< r� r� � setitimerr( r r r �tearDown� s zItimerTest.tearDownc G s d| _ d S r� )r� r! r r r r� � s zItimerTest.sig_alrmc G sF d| _ | jdkr t�d��n| jdkr4t�tjd� | jd7 _d S )NTr� z.setitimer didn't disable ITIMER_VIRTUAL timer.r � )r� r� r � ItimerErrorr� �ITIMER_VIRTUALr! r r r � sig_vtalrm� s zItimerTest.sig_vtalrmc G s d| _ t�tjd� d S )NTr )r� r r� � ITIMER_PROFr! r r r �sig_prof� s zItimerTest.sig_profc C s | � tjtjdd� d S )NrX r )r$ r r� r� r( r r r �test_itimer_exc� s zItimerTest.test_itimer_excc C s0 t j| _t �| jd� t �� | �| jd� d S )Ng �?T)r � ITIMER_REALr� r� �pauser r� r( r r r �test_itimer_real� s zItimerTest.test_itimer_real)Znetbsd5zDitimer not reliable (does not mix well with threading) on some BSDs.c C s� t j| _t � t j| j� t �| jdd� t�� }t�� | dk r`tddd�}t � | j�dkr0qjq0| � d� | � t � | j�d� | � | j d � d S ) Ng333333�?皙�����?� N@�90 �2 铖� �� r� �8timeout: likely cause: machine too slow or load too highT) r r� r� � SIGVTALRMr� r� �time� monotonic�pow� getitimerr� r r� �r Z start_time�_r r r �test_itimer_virtual� s z ItimerTest.test_itimer_virtualc C s� t j| _t � t j| j� t �| jdd� t�� }t�� | dk r`tddd�}t � | j�dkr0qjq0| � d� | � t � | j�d� | � | j d� d S ) Nr� r� r� r� r� r� r� T) r r� r� �SIGPROFr� r� r� r� r� r� r� r r� r� r r r �test_itimer_prof� s zItimerTest.test_itimer_profc C s2 t j| _t �| jd� t�d� | �| jd� d S )N���ư>r� T)r r� r� r� r� �sleepr r� r( r r r �test_setitimer_tiny s z ItimerTest.test_setitimer_tinyN)r r r r� r� r� r� r� r� r� rP rw r r r� r� r� r r r r r� � s � r� c @ s� e Zd ZdZe�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�d d � �Ze�eed�d�d d � �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 C s | � t�� t� � d S r )r r r� r; r( r r r �test_sigpending_empty s z)PendingSignalsTests.test_sigpending_emptyr� r� c C s d}t d|� d S )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") rE r� r� r r r �test_sigpending s z#PendingSignalsTests.test_sigpending� pthread_killzneed signal.pthread_kill()c C s d}t d|� d S )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") rE r� r� r r r �test_pthread_kill9 s z%PendingSignalsTests.test_pthread_killc C s d|� � |f }td|� dS )zo test: body of the "def test(signum):" function. blocked: number of the blocked signal aw if 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) rE N)�stripr )r Zblocked�testr� r r r � wait_helperR s �%z PendingSignalsTests.wait_helper�sigwaitzneed signal.sigwait()c C s | � tjd� d S )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( r r r � test_sigwait� s z PendingSignalsTests.test_sigwait� sigwaitinfozneed signal.sigwaitinfo()c C s | � tjd� d S )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( r r r �test_sigwaitinfo� s z$PendingSignalsTests.test_sigwaitinfo� sigtimedwaitzneed signal.sigtimedwait()c C s | � tjd� d S )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( r r r �test_sigtimedwait� s z%PendingSignalsTests.test_sigtimedwaitc C s | � tjd� d S )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( r r r �test_sigtimedwait_poll� s z*PendingSignalsTests.test_sigtimedwait_pollc C s | � tjd� d S )Nz� def test(signum): received = signal.sigtimedwait([signum], 1.0) if received is not None: raise Exception("received=%r" % (received,)) r� r( r r r �test_sigtimedwait_timeout� s z-PendingSignalsTests.test_sigtimedwait_timeoutc C s t j}| �tt j|gd� d S )Ng �)r r< r$ r% r� )r rd r r r �"test_sigtimedwait_negative_timeout� s z6PendingSignalsTests.test_sigtimedwait_negative_timeoutc C s t dd� d S )NrE a� 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( r r r �test_sigwait_thread� s z'PendingSignalsTests.test_sigwait_threadc C s� | � ttj� | � ttjd� | � ttjddd� | � ttjdg � | � t�� t�tjtjg� W 5 Q R X | � t�� t�tjdg� W 5 Q R X | � t�� t�tjdd> g� W 5 Q R X d S )Nr� r� r� i� r i� )r$ r* r r� rg r% r r> r( r r r � test_pthread_sigmask_arguments� s z2PendingSignalsTests.test_pthread_sigmask_argumentsc C sJ t �t jt �� �}| �t jt j|� t �t jt �� �}| �|t �� � d S r )r r� r r: rp r r ZassertLessEqualrA r r r �"test_pthread_sigmask_valid_signals� s z6PendingSignalsTests.test_pthread_sigmask_valid_signalsc C s d}t d|� d S )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)) rE r� r� r r r �test_pthread_sigmask� s Hz(PendingSignalsTests.test_pthread_sigmaskc C sJ d}t d|��2}|�� \}}|�� }|dkr<td||f ��W 5 Q R X d S )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) rE r� z Child error (exit code %s): %s)r r� r� r� )r r� rN r� rG r� r r r � test_pthread_kill_main_threadE s �z1PendingSignalsTests.test_pthread_kill_main_threadN)r r r �__doc__rP rQ rq r r� r� r� r� r� r� r� r� r� r� r� r� r� r� r� r r r r r� s� � � � � � , � � � � � � � � � � � K �r� c @ sv e Zd ZdZdd� Zdd� Zdd� Ze�e e d�d �d d � �Z e�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. c C s t � ||�}| �t j ||� d S r )r rp )r rd r` � old_handlerr r r �setsigf s zStressTest.setsigc s� d� g �d � �fdd� }| � tjtjd� | �tj|� |� t��� k rVt�d� q>�fdd�t t��d �D �}t � |�}t j r�td |f � |S ) N� c s, t ��� k r(��t�� � t�tjd� d S )Nr� )r@ �appendr� � perf_counterr r� r� �rd �frame��N�timesr r r` n s z5StressTest.measure_itimer_resolution.<locals>.handlerr g����MbP?c s g | ]}� |d � | �qS )r� r )�.0�i)r� r r � <listcomp>| s z8StressTest.measure_itimer_resolution.<locals>.<listcomp>r� z,detected median itimer() resolution: %.6f s.)NN)rp r r� r� r� r< r@ r� r� �range� statisticsZmedianr �verbose�print)r r` Z durationsZmedr r� r �measure_itimer_resolutionj s z$StressTest.measure_itimer_resolutionc C s4 | � � }|dkrdS |dkr dS | �d|f � d S )Ng-C ��6?i' g{�G�z�?�d z^detected itimer resolution (%.3f s.) too high (> 10 ms.) on this platform (or system too busy))r r� )r Zresor r r �decide_itimer_count� s �z StressTest.decide_itimer_countr� ztest needs setitimer()c s� | � � }g � dd� }d � fdd� }| �tj|� | �tj|� | �tj|� d}t�� d }||k r�t� t� � tj� |d7 }t � �|k r�t�� |k r�t� d � q~t� t� � tj� |d7 }t � �|k r\t�� |k r\t� d � q�q\| � t � �|d � dS ) z; This test uses dependent signal handlers. c S s t �t jdt�� d � d S )Nr� ��h㈵��>)r r� r� �randomr� r r r � first_handler� s z@StressTest.test_stress_delivery_dependent.<locals>.first_handlerNc s � � | � d S r �r� r� �Zsigsr r �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� r3 r� �getpidr@ r� r )r r� r r � expected_sigs�deadliner r r � test_stress_delivery_dependent� s& z)StressTest.test_stress_delivery_dependentc s� | � � }g � � fdd�}| �tj|� | �tj|� d}t�� d }||k r�t�tjdt � � d � t � t � � tj� |d7 }t � �|k rDt�� |k rDt�d� q�qD| �t � �|d� d S ) z> This test uses simultaneous signal handlers. c s � � | � d S r r r� r r r r` � s z=StressTest.test_stress_delivery_simultaneous.<locals>.handlerr r r� r r� r N)r r� r r+ r< r� r� r� r� r r3 r� r r@ r� r )r r� r` r r r r r �!test_stress_delivery_simultaneous� s z,StressTest.test_stress_delivery_simultaneousr+ ztest needs SIGUSR1c s t j�d�d�d��fdd�� ���fdd�}� ��fdd�}t � �� �}| �t j �|� tj|d �}z�d}t�� �Z}|�� |� d �|�� |j d k r�| � |j j t � | � d �� d �t|j j �� d }W 5 Q R X |s�| ��d� | ���� W 5 d �|�� X d S ) Nr Fc s � d7 � d S �Nr� r r� )�num_received_signalsr r �custom_handler� s zAStressTest.test_stress_modifying_handlers.<locals>.custom_handlerc s � st ��� �d7 �q d S r )r � raise_signalr )�do_stop�num_sent_signalsrd r r �set_interrupts� s zAStressTest.test_stress_modifying_handlers.<locals>.set_interruptsc s8 �dk r4t d�D ] } � tjfD ]}t��|� q qq d S )Nr i N )r� r r )r� r` )r r rd r r �cycle_handlers� s zAStressTest.test_stress_modifying_handlers.<locals>.cycle_handlers)�targetTzSignal z ignored due to race condition)r r+ rp � threadingZThreadr7 r Zcatch_unraisable_exception�startZ unraisabler � exc_valuerg r/ rt Z assertGreaterr? )r r r r� �tZignoredru r )r r r r rd r � test_stress_modifying_handlers� s: � z)StressTest.test_stress_modifying_handlersN)r r r r� r� r r rP rQ rq r r r r r r r r r� _ s � , � �r� c @ s6 e Zd Zdd� Ze�ejdkd�dd� �Zdd� Z d S ) �RaiseSignalTestc C s& | � t�� t�tj� W 5 Q R X d S r )r$ �KeyboardInterruptr r r0 r( r r r � test_sigint s zRaiseSignalTest.test_sigintr zWindows specific testc C sT z d}t �|� | �d� W n2 tk rN } z|jtjkr<n� W 5 d }~X Y nX d S )Nr� z#OSError (Invalid argument) expected)r r Zfailrg �errnoZEINVAL)r r- �er r r �test_invalid_argument s z%RaiseSignalTest.test_invalid_argumentc sJ d� � fdd�}t � t j|�}| �t j t j|� t �t j� | �� � d S )NFc s d� d S r� r )�a�b�Zis_okr r r` . s z-RaiseSignalTest.test_handler.<locals>.handler)r r0 rp r r_ )r r` Z old_signalr r' r � test_handler, s z RaiseSignalTest.test_handlerN) r r r r! rP rw r r r$ r( r r r r r s r c C s t �� d S r )r Z reap_childrenr r r r �tearDownModule8 s r) �__main__)%r" r3 r r rj r� rI r r r� rP r� r Ztest.support.script_helperr r r� � ImportErrorZTestCaser rw r r rQ rR rc rx rq r� r� r� r� r� r r) r �mainr r r r �<module> sR =/M 6 @Te Q <
💻
System Info
Current path
/opt/alt/python38/lib64/python3.8/test/__pycache__
Contents
0 folders, 1461 files
Disk free
100.89 GB
PHP version
8.3.30
🚀
Quick Actions
📍 Script location
🌐 Web root
🗑️ Temp
📁 /opt/alt/python38/lib64/python3.8/test/__pycache__
⚡ Nexus File Manager • 1461 items