File: //usr/local/lib/python3.7/test/__pycache__/test_threaded_import.cpython-37.pyc
B
��g�# � @ s� d dl Zd dlZd dlZd dlZd dlZd dlZd dlZd dlZd dlm Z d dl
m Z m Z m
Z
mZmZmZmZmZmZ dd� Zdd� Zdd d
d d �ZG d
d� d�ZG dd� d�ZG dd� dej�Zedd� �Ze dkr�e� dS )� N)�mock) �verbose�
import_module� run_unittest�TESTFN� reap_threads�forget�unlink�rmtree�
start_threadsc
C s� zvy>t |�d r"dd l}dd l}ndd l}dd l}|�dd�}W n2 tk
rr } z|�|�d �� W d d }~X Y nX W d |�t�� � t |�| k}|r�|� � X d S )N� r � � )
�len� modulefinder�randomZ randrange� Exception�append�with_traceback� threading� get_ident�set) �N�done�
done_tasks�errorsr r �x�eZfinished� r �5/usr/local/lib/python3.7/test/test_threaded_import.py�task s
& r c C s t jddd�| �S )Nzos.register_at_forkT)Zcreate)r Zpatch)�funcr r r �mock_register_at_fork) 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 ZdZdd� Zddd�ZdS )�FinderzIA dummy finder to detect concurrent access to its find_spec()
method.c C s d| _ d| _t�� | _d S )Nr )�numcallsr r ZLock�lock)�selfr r r �__init__G s zFinder.__init__Nc C sJ t �� s t�| j� | jd7 _W d Q R X | j}t�d� |d | _d S )Nr
g{�G�z�?)�imp� lock_held�AssertionErrorr) r( r �timeZsleep)r* �name�path�targetr r r r � find_specL s
zFinder.find_spec)NN)�__name__�
__module__� __qualname__�__doc__r+ r3 r r r r r' C s r' c @ s e Zd ZdZddd�ZdS )�FlushingFinderzMA dummy finder which flushes sys.path_importer_cache when it gets
called.Nc C s t j�� d S )N)�sys�path_importer_cache�clear)r* r0 r1 r2 r r r r3 [ s zFlushingFinder.find_spec)NN)r4 r5 r6 r7 r3 r r r r r8 W s r8 c @ s\ e Zd Zdd� Zdd� Zedd� �Zdd� Zd d
� Zd d � Z d
d� Z
dd� Z edd� �Z dS )�ThreadedImportTestsc C s t j�dd �| _d S )Nr )r9 �modules�pop�
old_random)r* r r r �setUpa s zThreadedImportTests.setUpc C s | j d k r| j tjd< d S )Nr )r? r9 r= )r* r r r �tearDownd s
z ThreadedImportTests.tearDownc
s t �� rt�d��t�� �x�dD ]� tr8td� ddd� x.dD ]&}y tj |= W q> t
k
rb Y q>X q>W g �g ��� � t �
� }t� ���fdd �t� �D ��� W d Q R X ��d
�}t �
� | }tr�td |d d
dd� dt��� f }| ��|� | �||� tr td� q W d S )Nz"can't run when import lock is held)� �2 rB rC rB rC ZTryingz threads ...� )�end)r r c 3 s$ | ] }t jt� ���fd �V qdS ))r2 �argsN)r �Threadr )�.0�i)r r r r r r � <genexpr> s zAThreadedImportTests.check_parallel_module_init.<locals>.<genexpr>iX z%.1f msg @�@T)�flushrE z done: %s/%szOK.)r, r- �unittestZSkipTestr ZEventr �printr9 r= �KeyErrorr; r/ Z monotonicr �range�waitr � assertFalseZ
assertTrue)r* �mock_os�modnameZt0Z completedZdtZdbg_infor )r r r r r �check_parallel_module_initk s6
z.ThreadedImportTests.check_parallel_module_initc C s | � � d S )N)rT )r* r r r �test_parallel_module_init� s z-ThreadedImportTests.test_parallel_module_initc C sR t � }tj�d|� z*| �� | �|jd� | �|j|j� W d tj� |� X d S )Nr )
r' r9 � meta_path�insertrT �
assertGreaterr( � assertEqualr �remove)r* �finderr r r �test_parallel_meta_path� s z+ThreadedImportTests.test_parallel_meta_pathc s� t � � t� }� fdd�}tj�d|� tj�|� z4|�d� | �� }| � � j
d� | � � j � j
� W d tj�
|� tj�
|� X d S )Nc s � � d� t�d S )N� )r3 � ImportError)r1 )r[ r r � path_hook� s
z?ThreadedImportTests.test_parallel_path_hooks.<locals>.path_hookr r] )r' r8 r9 �
path_hooksrW rV r r3 rT rX r( rY r rZ )r* Zflushing_finderr_ Znumtestsr )r[ 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}| �|jj� d S )Nz test.threaded_import_hangersr )r9 r= rN Z test.threaded_import_hangersrQ Zthreaded_import_hangersr )r* Ztestr r r �test_import_hangers� s z'ThreadedImportTests.test_import_hangersc
s d}t �t� | �tjt� tj�dt� | �tjj t� x`t
� � D ]T\}}|d|i }t t j�
t|d �d��}|�|�d�� W d Q R X | �t|� qDW t�� g � � fdd�}� fd d
�}tj|d �}tj|d �}|�� |�� |�
� |�
� | �t� �d d
h� d S )Ng �?r �delayz.py�wbzutf-8c s dd l } � �t| 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 } � �t| dd �� d S )Nr r )r$ r re )r$ )rf r r � import_ba� s z<ThreadedImportTests.test_circular_imports.<locals>.import_ba)r2 �a�b)�os�mkdirr �
addCleanup�shutilr
r9 r1 rW rZ �circular_imports_modules�items�open�join�write�encoder � importlib�invalidate_cachesr rG �startrY r ) r* rc r0 �contents�frg rh Zt1Zt2r )rf r �test_circular_imports� s*
z)ThreadedImportTests.test_circular_importsc C s� d}t j�dtj� | �t jjtj� td }t|d��}|� |�
d�� W d Q R X | �t |� | �t t� | �t
d� t�� 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.pyrd zutf-8� __pycache__)r9 r1 rW rk �curdirrm rZ r rq rs rt r r r
ru rv �
__import__r= )r* rR �code�filenamery r r r �test_side_effect_import� s z+ThreadedImportTests.test_side_effect_importN)
r4 r5 r6 r@ rA r" rT rU r\ ra rb rz r� r r r r r< _ s "
&r< c C sV d } yt �� } t �d� W n tk
r. Y nX z tt� W d | d k rPt �| � X d S )Ng�h㈵��>)r9 �getswitchinterval�setswitchinterval�AttributeErrorr r< )Zold_switchintervalr r r � test_main� s r� �__main__) �_impr, rk ru r9 r/ rn r rL r Z test.supportr r r r r r r r
r r r" ro r' r8 ZTestCaser< r� r4 r r r r �<module> s, ,