File: //usr/local/lib/python3.10/test/test_importlib/__pycache__/test_threaded_import.cpython-310.pyc
o
�i�% � @ s� d dl Zd dlZd dlZd dlZd dlZd dlZd dlZd dlZd dlm Z d dl
m Z d dl m
Z
d dlmZmZmZ d dl
mZmZ dd� Zd d
� Zd d d
dd�ZG dd� d�ZG dd� d�ZG dd� dej�Zdd� Ze dkrxe� � dS dS )� N)�mock)�verbose)�forget)�TESTFN�unlink�rmtree)�
script_helper�threading_helperc
C s� zRz t |�d rdd l}dd l}ndd l}dd l}|�dd�}W n ty: } z|�|�d �� W Y d }~nd }~ww W |�t�� � t |�| k}|rQ|� � d S d S |�t�� � t |�| k}|rg|� � w w )N� r � � )
�len� modulefinder�random� randrange� Exception�append�with_traceback� threading� get_ident�set) �N�done�
done_tasks�errorsr r �x�e�finished� r �E/usr/local/lib/python3.10/test/test_importlib/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 �Lock�lock��selfr r r �__init__H s zFinder.__init__Nc C s^ t �� sJ �| j� | jd7 _W d � n1 sw Y | j}t�d� |d | _d S )Nr g{�G�z�?)�imp� lock_heldr+ r) r �time�sleep)r- �name�path�targetr r r r � find_specM s �
zFinder.find_spec�NN)�__name__�
__module__� __qualname__�__doc__r. r6 r r r r r' D 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- r3 r4 r5 r r r r6 \ s zFlushingFinder.find_specr7 )r8 r9 r: r; r6 r r r r r< X s r< c @ sl e Zd Zdd� Zdd� Zedd� �Zdd� Zd d
� Zd d � Z d
d� Z
dd� Z edd� �Z dd� Z
dd� ZdS )�ThreadedImportTestsc C s t j�dd �| _d S �Nr )r> �modules�pop�
old_randomr, r r r �setUpb s zThreadedImportTests.setUpc C s | j d ur
| j tjd< d S d S rB )rE r> rC r, r r r �tearDowne s
�z ThreadedImportTests.tearDownc
s" t �� r t�d��t�� �dD ]� trtd� ddd� dD ]}ztj |= W q t
y. Y q w g �g ��� � t �
� }t�� ���fdd �t� �D ��� W d � n1 sWw Y ��d
�}t �
� | }trttd |d d
dd� dt��� f }| ��|� | �||� tr�td� qd S )Nz"can't run when import lock is held)� �2 rH rI rH rI ZTryingz threads ...� )�end)r r c 3 s& � | ]}t jt� ���fd �V qdS ))r5 �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)�flushrK z done: %s/%szOK.)r/ r0 �unittestZSkipTestr �Eventr �printr> rC �KeyErrorr@ r1 � monotonicr Z
start_threads�range�waitr
� assertFalseZ
assertTrue)r- �mock_os�modname�t0Z completedZdtZdbg_infor rP r �check_parallel_module_initl s@
�
� �
��z.ThreadedImportTests.check_parallel_module_initc C s | � � d S r= )r^ r, r r r �test_parallel_module_init� s z-ThreadedImportTests.test_parallel_module_initc C s\ t � }tj�d|� z | �� | �|jd� | �|j|j� W tj� |� d S tj� |� w r( )
r' r> � meta_path�insertr^ �
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�|� z'|�d� | �� }| � � j
d� | � � j � j
� W tj�
|� tj�
|� d S tj�
|� tj�
|� w )Nc s � � d� t�)N� )r6 � ImportError)r4 �re r r � path_hook� s
z?ThreadedImportTests.test_parallel_path_hooks.<locals>.path_hookr rg )r' r< r> �
path_hooksra r` r r6 r^ rb r) rc r rd )r- Zflushing_finderrj Znumtestsr ri r �test_parallel_path_hooks� s
�z,ThreadedImportTests.test_parallel_path_hooksc C s< zt jd= W n ty Y nw dd l}| �|jjj� d S )Nz+test.test_importlib.threaded_import_hangersr )r> rC rV Z+test.test_importlib.threaded_import_hangersrZ Ztest_importlibZthreaded_import_hangersr )r- �testr r r �test_import_hangers� s �z'ThreadedImportTests.test_import_hangersc s d}t �t� | �tjt� tj�dt� | �tjj t� t
� � D ]4\}}|d|i }t t j�
t|d �d��}|�|�d�� W d � n1 sJw Y | �t|� q!t�� g � � fdd�}� fd d
�}tj|d �}tj|d �}|�� |�� |�
� |�
� | �t� �d d
h� d S )Ng �?r �delay�.py�wb�utf-8c � dd l } � �t| dd �� d S �Nr r )r# r �getattr)r# ��resultsr r � import_ab� � z<ThreadedImportTests.test_circular_imports.<locals>.import_abc rs rt )r$ r ru )r$ rv r r � import_ba� ry z<ThreadedImportTests.test_circular_imports.<locals>.import_ba)r5 �a�b)�os�mkdirr �
addCleanup�shutilr r> r4 ra rd �circular_imports_modules�items�open�join�write�encoder � importlib�invalidate_cachesr rM �startrc r ) r- ro r3 �contents�frx rz �t1�t2r rv r �test_circular_imports� s,
� z)ThreadedImportTests.test_circular_importsc C s� d}t j�dtj� | �t jjtj� td }t|d��}|� |�
d�� W d � n1 s/w Y | �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 rp rq rr � __pycache__)r> r4 ra r} �curdirr rd r r� r� r� r r r r� r� �
__import__rC )r- r[ �code�filenamer� r r r �test_side_effect_import� s � z+ThreadedImportTests.test_side_effect_importc C �& t j�t j�t�dd�}t�|� d S )N�partialz cfimport.py�r} r4 r� �dirname�__file__r Zassert_python_ok�r- �fnr r r �'test_concurrent_futures_circular_import� � �z;ThreadedImportTests.test_concurrent_futures_circular_importc C r� )Nr� zpool_in_threads.pyr� r� r r r �)test_multiprocessing_pool_circular_import� r� z=ThreadedImportTests.test_multiprocessing_pool_circular_importN)r8 r9 r: rF rG r" r^ r_ rf rl rn r� r� r� r� r r r r rA ` s
!
&
rA c C sX t �� } tjt jg| �R � zt�� }t�tj|� t�d� W d S ty+ Y d S w )Ng�h㈵��>) r Zthreading_setuprS ZaddModuleCleanupZthreading_cleanupr> �getswitchinterval�setswitchinterval�AttributeError)� thread_infoZold_switchintervalr r r � setUpModule s �r� �__main__) �_impr/ r} r� r> r1 r� r rS r Z test.supportr Ztest.support.import_helperr Ztest.support.os_helperr r r r r r r" r� r' r< ZTestCaserA r� r8 �mainr r r r �<module> s8
� & �