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_threadsignals.cpython-38.pyc
← Back
U i�f^( � @ s� d Z ddlZddlZddlZddlZddlmZ ddlZddl Z ej dd� dkrbe� dej ��e� � Z e�� Zejjdko�ejjdkZd d � Zd d � Zd d� ZG dd� dej�Zdd� Zedkr�e� dS )z6PyUnit testing that threads honor our signal semantics� N)�support� �winzCan't test signal on %sZpthreadz mutex+condc C s4 t � t j| �}t � t j|�}t � t j|�}|||fS �N)�signal�SIGUSR1�SIGUSR2�SIGALRM)Zfor_usr1Zfor_usr2Zfor_alrmZusr1Zusr2Zalrm� r �</opt/alt/python38/lib64/python3.8/test/test_threadsignals.py�registerSignals s r c C s( t | d d7 < t�� t | d< d S )N�tripped� � tripped_by)�signal_blackboard�thread� get_ident)�sig�framer r r �handle_signals s r c C s( t �ttj� t �ttj� t�� d S r )�os�kill� process_pidr r r � signalled_all�releaser r r r � send_signals" s r c @ s� e Zd Zdd� Zdd� Zdd� Ze�ed�e�e j � d�oBe j j d �e�e j � d �d �d d � ���Ze�ed�e�e j � d�o�e j j d �e�e j � d �d �dd� ���Zdd� Zdd� Zdd� Zdd� ZdS )� ThreadSignalsc C s� t �� � t�� | �� t�� W 5 Q R X ttj d dksPttj d dkrtzt� d� t� � W 5 t� d� X | � ttj d d� | � ttj d t � � � | � ttj d d� | � ttj d t � � � t�� d S )Nr r r r )r �wait_threads_exitr �acquire�spawnSignallingThreadr r r r �alarm�pauseZ assertEqualr r r ��selfr r r � test_signals) s( � ��zThreadSignals.test_signalsc C s t �td� d S )Nr )r �start_new_threadr r" r r r r I s z#ThreadSignals.spawnSignallingThreadc C s t �d S r )�KeyboardInterrupt)r# r r r r r �alarm_interruptL s z ThreadSignals.alarm_interruptz/POSIX condition variables cannot be interruptedZlinuxzBIssue 34004: musl does not allow interruption of locks by signals.Zopenbsdz%lock cannot be interrupted on OpenBSDc C s� t � t j| j�}zPt�� }|�� t �d� t�� }| j t |jdd� t�� | }| � |d� W 5 t �d� t � t j|� X d S )Nr r � �Ztimeout� @) r r r' r r � allocate_lockr �time� monotonic� assertRaisesr& � assertLess)r# �oldalrm�lock�t1�dtr r r � test_lock_acquire_interruptionO s z,ThreadSignals.test_lock_acquire_interruptionc s� t � t j| j�}z�t�� � � fdd�}t�� �n t�|d� � j dd�r\� � � t � d� q<t �d� t � � }| jt� j d d � t � � | }| �|d � W 5 Q R X W 5 t �d� t � t j|� X d S ) Nr c s � � � d S r )r r �Zrlockr r � other_thread� s zCThreadSignals.test_rlock_acquire_interruption.<locals>.other_threadr F�Zblocking�{�G�z�?r r( r) r* )r r r' r r �RLockr r r% r r r, �sleepr- r. r& r/ )r# r0 r6 r2 r3 r r5 r � test_rlock_acquire_interruptionn s z-ThreadSignals.test_rlock_acquire_interruptionc s� d�_ �fdd�}t�tj|�}zn� fdd�}t�� �P t�|d� � jdd�rd� �� t � d� qD� �� }�� �j � �� |� W 5 Q R X W 5 t�tj|� X d S ) NFc s d� _ d S )NT)� sig_recvd)r r r" r r � my_handler� s z9ThreadSignals.acquire_retries_on_intr.<locals>.my_handlerc s6 � � � t�d� t�ttj� t�d� � �� d S )N� �?) r r, r: r r r r r r r )r1 r r r6 � s z;ThreadSignals.acquire_retries_on_intr.<locals>.other_threadr r7 r8 ) r< r r r r r r% r r r, r: Z assertTrue)r# r1 r= � old_handlerr6 �resultr �r1 r# r �acquire_retries_on_intr� s z%ThreadSignals.acquire_retries_on_intrc C s | � t�� � d S r )rB r r+ r" r r r �!test_lock_acquire_retries_on_intr� s z/ThreadSignals.test_lock_acquire_retries_on_intrc C s | � t�� � d S r )rB r r9 r" r r r �"test_rlock_acquire_retries_on_intr� s z0ThreadSignals.test_rlock_acquire_retries_on_intrc s� d �_ d �_d�_t�� � � �� t�� ���� �fdd�}t�tj|�}z���fdd�}� fdd�}t� � �V t� |d� |� � �� �� �j�j d � �� �j�j d � �� �jd� W 5 Q R X W 5 t�tj|� X d S ) Nr c s � j d7 _ d S )Nr )� sigs_recvd)Zsignumr r" r r r= � s z@ThreadSignals.test_interrupted_timed_acquire.<locals>.my_handlerc s$ t �� �_� jdd� t �� �_d S )Nr> r) )r, r- �startr �endr rA r r � timed_acquire� s zCThreadSignals.test_interrupted_timed_acquire.<locals>.timed_acquirec s2 t d�D ] } t�d� t�ttj� q� �� d S )N�( g{�G�z�?) �ranger, r: r r r r r r )�_)�doner r r � s zBThreadSignals.test_interrupted_timed_acquire.<locals>.send_signalsr g @g333333�?) rF rG rE r r+ r r r r r r% r/ Z assertGreater)r# r= r? rH r r )rL r1 r# r � test_interrupted_timed_acquire� s( z,ThreadSignals.test_interrupted_timed_acquireN)�__name__� __module__� __qualname__r$ r r' �unittestZskipIf�USING_PTHREAD_COND�sys�platform� startswith� thread_info�versionr4 r; rB rC rD rM r r r r r ' s: ������ r c C sR t jddd�t jddd�t jddd�iatttt�} zt�t � W 5 t| � X d S )Nr )r r ) r r r r r r r r Z run_unittestr )Zoldsigsr r r � test_main� s � rX �__main__)�__doc__rQ r r rS Ztestr �_threadr r, rT ZSkipTest�getpidr r+ r rV �namer1 rR r r r ZTestCaser rX rN r r r r �<module> s, � C
💻
System Info
Current path
/opt/alt/python38/lib64/python3.8/test/__pycache__
Contents
0 folders, 1461 files
Disk free
101.24 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