|
Server : LiteSpeed System : Linux server51.dnsbootclub.com 4.18.0-553.62.1.lve.el8.x86_64 #1 SMP Mon Jul 21 17:50:35 UTC 2025 x86_64 User : nandedex ( 1060) PHP Version : 8.1.33 Disable Function : NONE Directory : /opt/cppython/lib/python3.8/test/__pycache__/ |
U
>��g�# � @ s� d dl Zd dlZd dlZd dlZd dlZd dlZd dlZd dlZd dlm Z d dl
m Z m Z m
Z
mZmZmZmZmZ dd� Zdd� Zdd d
d d �ZG d
d� d�ZG dd� d�ZG dd� dej�Zedd� �Zedkr�e� dS )� N)�mock)�verbose� run_unittest�TESTFN� reap_threads�forget�unlink�rmtree�
start_threadsc
C s� zvz>t|�d r"dd l}dd l}ndd l}dd l}|�dd�}W n2 tk
rr } z|� |� d �� W 5 d }~X Y nX W 5 |� t�� � t|�| k}|r�|�� X d S )N� r � � )
�append� threading� get_ident�len�set� modulefinder�random� randrange� Exception�with_traceback) �N�done�
done_tasks�errors�finishedr r �x�e� r �8/opt/cppython/lib/python3.8/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 �Lock�lock��selfr r r �__init__G s zFinder.__init__Nc C sJ t �� s t�| j� | jd7 _W 5 Q R X | j}t�d� |d | _d S )Nr g{�G�z�?)�imp� lock_held�AssertionErrorr, r* r �time�sleep)r. �name�path�targetr r r r � find_specL s
zFinder.find_spec)NN)�__name__�
__module__� __qualname__�__doc__r/ r8 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. r5 r6 r7 r r r r8 [ s zFlushingFinder.find_spec)NN)r9 r: r; r< r8 r r r r r= W s r= 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 )r? �modules�pop�
old_randomr- r r r �setUpa s zThreadedImportTests.setUpc C s | j d k r| j tjd< d S rC )rF r? rD r- r r r �tearDownd s
z ThreadedImportTests.tearDownc
s t �� rt�d��t�� �dD ]� tr6td� ddd� dD ]&}z tj |= W q: t
k
r^ Y q:X q:g �g ��� � t �
� }t� ���fdd �t� �D ��� W 5 Q R X ��d
�}t �
� | }tr�td |d d
dd� dt��� f }| ��|� | �||� tr td� q d S )Nz"can't run when import lock is held)� �2 rI rJ rI rJ ZTryingz threads ...� )�end)r r c 3 s$ | ] }t jt� ���fd �V qdS ))r7 �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)�flushrL z done: %s/%szOK.)r0 r1 �unittestZSkipTestr �Eventr �printr? rD �KeyErrorrA r3 � monotonicr
�range�waitr � assertFalseZ
assertTrue)r. �mock_os�modname�t0Z completedZdtZdbg_infor rQ r �check_parallel_module_initk s8
�
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 sR t � }tj�d|� z*| �� | �|jd� | �|j |j� W 5 tj�|� X d S r) )
r( r? � meta_path�insert�remover_ �
assertGreaterr* � assertEqualr )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 5 tj�|� tj�|� X d S )Nc s � � d� t�d S )N� )r8 � ImportError)r6 �rf r r � path_hook� s
z?ThreadedImportTests.test_parallel_path_hooks.<locals>.path_hookr rh )r( r= r? �
path_hooksrb ra r rc r8 r_ rd r* re r )r. Zflushing_finderrk Znumtestsr rj r �test_parallel_path_hooks� s
z,ThreadedImportTests.test_parallel_path_hooksc C s< z t jd= W n tk
r Y nX dd l}| �|jj� d S )Nz test.threaded_import_hangersr )r? rD rW Z test.threaded_import_hangersr[ Zthreaded_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 ]T\}}|d|i }t t j�
t|d �d��}|�|�d�� W 5 Q R X | �t|� qBt�� g � � fdd�}� fd d
�}tj|d �}tj|d �}|�� |�� |�
� |�
� | �t� �d d
h� d S )Ng �?r �delay�.py�wb�utf-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 rt )r% r ru )r% rv r r � import_ba� s z<ThreadedImportTests.test_circular_imports.<locals>.import_ba)r7 �a�b)�os�mkdirr �
addCleanup�shutilr r? r6 rb rc �circular_imports_modules�items�open�join�write�encoder � importlib�invalidate_cachesr rN �startre r ) r. rp r5 �contents�frx ry �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 5 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 rq rr rs � __pycache__)r? r6 rb r| �curdirr~ rc r r� r� r� r r r r� r� �
__import__rD )r. r\ �code�filenamer� r r r �test_side_effect_import� s z+ThreadedImportTests.test_side_effect_importN)
r9 r: r; rG rH r# r_ r` rg rm ro r� r� r r r r rB _ s
!
&rB c C sV d } zt �� } t �d� W n tk
r. Y nX z tt� W 5 | d k rPt �| � X d S )Ng�h㈵��>)r? �getswitchinterval�setswitchinterval�AttributeErrorr rB )Zold_switchintervalr r r � test_main� s r� �__main__) �_impr0 r| r� r? r3 r r rT r Z test.supportr r r r r r r r
r! r# r� r( r= ZTestCaserB r� r9 r r r r �<module> s0 (
�