| Server IP : 118.27.122.248 / Your IP : 216.73.217.130 Web Server : Apache System : Linux web0264.sh.tyo1 4.18.0-553.79.1.lve.el7h.x86_64 #1 SMP Wed Oct 15 16:34:46 UTC 2025 x86_64 User : c9415830 ( 11735) PHP Version : 8.4.17 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : OFF | Pkexec : OFF Directory : /opt/alt/python36/lib64/python3.6/test/__pycache__/ |
Upload File : |
3
�
�^N" � @ s� d dl Zd dlZd dlZd dlZd dlZd dlZd dlZd dlm Z m
Z
m Z m Z m
Z
mZmZmZmZ e
d�Zdd� Zdddd d
�ZG d d � d �ZG d
d� d�ZG dd� dej�Ze
dd� �Zedkr�e� dS )� N) �verbose�
import_module� run_unittest�TESTFN� reap_threads�forget�unlink�rmtree�
start_threads� threadingc C s� zvy>t |�d r"dd l}dd l}ndd l}dd l}|jdd�}W n2 tk
rr } z|j|jd �� W Y d d }~X nX W d |jtj� � t |�| k}|r�|j � X d S )N� r � � )
�len� modulefinder�randomZ randrange� Exception�append�with_tracebackr � get_ident�set) �N�done�
done_tasks�errorsr r �x�eZfinished� r �9/opt/alt/python36/lib64/python3.6/test_threaded_import.py�task s
& r zaif 1:
import time
time.sleep(%(delay)s)
x = 'a'
import C
zaif 1:
import time
time.sleep(%(delay)s)
x = 'b'
import D
zimport Bzimport A)�A�B�C�Dc @ s e Zd Zdd� Zddd�ZdS )�Finderc C s d| _ d| _tj� | _d S )Nr )�numcallsr r ZLock�lock)�selfr r r �__init__@ s zFinder.__init__Nc
C s> | j � | jd7 _W d Q R X | j}tjd� |d | _d S )Nr
g{�G�z�?)r& r% r �timeZsleep)r' �name�path�targetr r r r � find_specE s
zFinder.find_spec)NN)�__name__�
__module__� __qualname__r( r- r r r r r$ < s r$ c @ s e Zd Zddd�ZdS )�FlushingFinderNc C s t jj� d S )N)�sys�path_importer_cache�clear)r' r* r+ r, r r r r- T s zFlushingFinder.find_spec)NN)r. r/ r0 r- r r r r r1 P s r1 c @ sT e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dd� Z dS )�ThreadedImportTestsc C s t jjdd �| _d S )Nr )r2 �modules�pop�
old_random)r' r r r �setUpZ s zThreadedImportTests.setUpc C s | j d k r| j tjd< d S )Nr )r8 r2 r6 )r' r r r �tearDown] s
z ThreadedImportTests.tearDownc
s t j� rtjd��tj� �� x�dD ]� tr:td� ddd� x.dD ]&}y tj |= W q@ t
k
rd Y q@X q@W g �g ��j � t j
� }t� ���fd d �t� �D ��� W d Q R X �jd�}t j
� | }tr�td|d ddd� dt��� f }| j�|� | j||� tr"td� q"W d S )Nz"can't run when import lock is held� �2 r ZTryingz threads ...� )�endr r c 3 s$ | ] }t jt� ���fd �V qdS ))r, �argsN)r �Threadr )�.0�i)r r r r r r � <genexpr>w s zAThreadedImportTests.check_parallel_module_init.<locals>.<genexpr>�
�< z%.1f msg @�@T)�flushr> z done: %s/%szOK.)r; r< )r; r< r; r< r; r< )r r iX )�imp� lock_held�unittestZSkipTestr ZEventr �printr2 r6 �KeyErrorr4 r) Z monotonicr
�range�waitr � assertFalseZ
assertTrue)r' �modnameZt0Z completedZdtZdbg_infor )r r r r r �check_parallel_module_initd s6
z.ThreadedImportTests.check_parallel_module_initc C s | j � d S )N)rP )r' r r r �test_parallel_module_init� s z-ThreadedImportTests.test_parallel_module_initc C sR t � }tjjd|� z*| j� | j|jd� | j|j|j� W d tjj |� X d S )Nr )
r$ r2 � meta_path�insertrP �
assertGreaterr% � assertEqualr �remove)r' �finderr r r �test_parallel_meta_path� s z+ThreadedImportTests.test_parallel_meta_pathc s� t � � t� }� fdd�}tjjd|� tjj|� z4|jd� | j� }| j � j
d� | j � j � j
� W d tjj
|� tjj
|� X d S )Nc s � j d� t�d S )N� )r- � ImportError)r+ )rW r r � path_hook� s
z?ThreadedImportTests.test_parallel_path_hooks.<locals>.path_hookr rY )r$ r1 r2 �
path_hooksrS rR r r- rP rT r% rU r rV )r' Zflushing_finderr[ Znumtestsr )rW r �test_parallel_path_hooks� s
z,ThreadedImportTests.test_parallel_path_hooksc C s< y t jd= W n tk
r Y nX dd l}| j|jj� d S )Nz test.threaded_import_hangersr )r2 r6 rK Z test.threaded_import_hangersrN Zthreaded_import_hangersr )r' Ztestr r r �test_import_hangers� s z'ThreadedImportTests.test_import_hangersc s d}t jt� | jtjt� tjjdt� | jtjj t� x`t
j � D ]T\}}|d|i }t t jj
t|d �d��}|j|jd�� W d Q R X | jt|� qDW tj� g � � fdd�}� fd d
�}tj|d �}tj|d �}|j� |j� |j
� |j
� | jt� �d d
h� d S )Ng �?r �delayz.py�wbzutf-8c s dd l } � jt| dd �� d S )Nr r )r r �getattr)r )�resultsr r � import_ab� s z<ThreadedImportTests.test_circular_imports.<locals>.import_abc s dd l } � jt| dd �� d S )Nr r )r! r ra )r! )rb r r � import_ba� s z<ThreadedImportTests.test_circular_imports.<locals>.import_ba)r, �a�b)�os�mkdirr �
addCleanup�shutilr r2 r+ rS rV �circular_imports_modules�items�open�join�write�encoder � importlib�invalidate_cachesr r@ �startrU r ) r' r_ r* �contents�frc rd Zt1Zt2r )rb r �test_circular_imports� s*
z)ThreadedImportTests.test_circular_importsc
C s� d}t jjdtj� | jt jjtj� td }t|d��}|j |j
d�� W d Q R X | jt |� | jt t� | jt
d� tj� tt� t jt= d S )Nz�if 1:
import threading
def target():
import random
t = threading.Thread(target=target)
t.start()
t.join()
t = Noner z.pyr` zutf-8� __pycache__)r2 r+ rS rg �curdirri rV r rm ro rp r r r rq rr �
__import__r6 )r' �code�filenameru r r r �test_side_effect_import� s z+ThreadedImportTests.test_side_effect_importN) r. r/ r0 r9 r: rP rQ rX r] r^ rv r| r r r r r5 X s !
&r5 c C sV d } yt j� } t jd� W n tk
r. Y nX z tt� W d | d k rPt j| � X d S )Ng�h㈵��>)r2 �getswitchinterval�setswitchinterval�AttributeErrorr r5 )Zold_switchintervalr r r � test_main� s r� �__main__)�_imprG rg rq r2 r) rj rI Z test.supportr r r r r r r r r
r r rk r$ r1 ZTestCaser5 r� r. r r r r �<module> s( ,