File: //usr/local/lib/python3.10/test/test_asyncio/__pycache__/test_pep492.cpython-310.pyc
o
�i� � @ s� d Z ddlZddlZddlZddlmZ ddlZddlmZ dd� Z G dd� d�Z
G d d
� d
ej �Z G d d � d e �Z
G d
d� de �ZG dd� de �ZedkrTe�� dS dS )z3Tests support for new syntax introduced by PEP 492.� N)�mock)�utilsc C s t �d � d S �N)�asyncioZset_event_loop_policy� r r �:/usr/local/lib/python3.10/test/test_asyncio/test_pep492.py�tearDownModule
s r c @ s. e Zd Zdd� Zd
dd�Zdd� Zdd � ZdS ) �FakeCoroc C � d S r r )�self�valuer r r �send � z
FakeCoro.sendNc C r
r r )r �typ�val�tbr r r �throw r zFakeCoro.throwc C r
r r �r r r r �close r zFakeCoro.closec c s � d V d S r r r r r r � __await__ s �
zFakeCoro.__await__)NN)�__name__�
__module__� __qualname__r
r r r r r r r r s
r c s e Zd Z� fdd�Z� ZS )�BaseTestc sH t � �� t�� | _t�� | j_t�� | j_d| jjj _
| � | j� d S )Nr ) �super�setUpr Z
BaseEventLoop�loopr ZMockZ_process_events� _selector�selectZ return_valueZset_event_loopr �� __class__r r r " s
zBaseTest.setUp)r r r r �
__classcell__r r r r r s r c @ s e Zd Zdd� Zdd� ZdS )� LockTestsc �T t �� t �� t �� t �� g}� fdd�}|D ]}� j�||�� � �|�� � qd S )Nc � s� �t �d�I d H � �| �� � | 4 I d H �'}� �|d � � �| �� � t �d�I d H � �| �� � W d �I d H n 1 I d H sCw Y � �| �� � d S )N�{�G�z�?)r �sleep� assertFalse�lockedZassertIs�
assertTrue)�lock�_lockr r r �test5 s � (�z7LockTests.test_context_manager_async_with.<locals>.test� r �Lock� Condition� Semaphore�BoundedSemaphorer �run_until_completer& r' �r Z
primitivesr+ Z primitiver r r � test_context_manager_async_with- � �
�z)LockTests.test_context_manager_async_withc r# )Nc � s� �t �d�I d H � �| �� � � �td��& | I d H � W d � n1 s'w Y W d � d S W d � d S 1 s?w Y d S )Nr$ z#can't be used in 'await' expression)r r% r& r' ZassertRaisesRegex� TypeError)r) r r r r+ K s �� ��"�z7LockTests.test_context_manager_with_await.<locals>.testr, r2 r r r � test_context_manager_with_awaitC r4 z)LockTests.test_context_manager_with_awaitN)r r r r3 r6 r r r r r" + s r" c @ � e Zd Zdd� ZdS )�StreamReaderTestsc sR d}t j| jd�� � �|� � �� � fdd�}| j�|� �}| �|g d�� d S )Ns line1
line2
line3�r c � s( �g } � 2 z 3 d H W }| � |� q6 | S r )�append)�data�line��streamr r �readerc s � �z/StreamReaderTests.test_readline.<locals>.reader)s line1
s line2
s line3)r � StreamReaderr Z feed_dataZfeed_eofr1 � assertEqual)r ZDATAr? r; r r= r �
test_readline\ s
z StreamReaderTests.test_readlineN)r r r rB r r r r r8 Z s r8 c @ sL e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dS )�CoroutineTestsc C sL dd� }|� }z| � t�|�� W |�� n|�� w | � t�t� �� d S )Nc � � �d S r r r r r r �foop � � z,CoroutineTests.test_iscoroutine.<locals>.foo)r( r Z iscoroutiner r )r rE �fr r r �test_iscoroutineo s z CoroutineTests.test_iscoroutinec C s dd� }| � t�|�� d S )Nc � rD r r r r r r rE { rF z4CoroutineTests.test_iscoroutinefunction.<locals>.foo)r( r Ziscoroutinefunction)r rE r r r �test_iscoroutinefunctionz s z'CoroutineTests.test_iscoroutinefunctionc sn G dd� d�� | � t�� tj� fdd��}W d � n1 s w Y |� }| �|�d �d� |�� d S )Nc @ r7 )zCCoroutineTests.test_function_returning_awaitable.<locals>.Awaitablec S s dS )N)�spamr r r r r r � r zMCoroutineTests.test_function_returning_awaitable.<locals>.Awaitable.__await__N)r r r r r r r r � Awaitable s rK c � � � S r r r �rK r r �func� � z>CoroutineTests.test_function_returning_awaitable.<locals>.funcrJ )Z assertWarns�DeprecationWarningr � coroutinerA r
r )r rN �coror rM r �!test_function_returning_awaitable~ s � z0CoroutineTests.test_function_returning_awaitablec sX dd� � � fdd�}| j �|� �}| �|d� | j �d� | j �|� �}| �|d� d S )Nc � s �dS )NrJ r r r r r �bar� s �z5CoroutineTests.test_async_def_coroutines.<locals>.barc � s �� � I d H S r r r �rT r r rE � s � z5CoroutineTests.test_async_def_coroutines.<locals>.foorJ T)r r1 rA � set_debug)r rE r; r rU r �test_async_def_coroutines� s z(CoroutineTests.test_async_def_coroutinesc sJ � fdd�}� � t�� d� � j�d� � j�|� � � � t�� d� d S )Nc � s �� � t�� dk� d S �Nr )r( �sys�#get_coroutine_origin_tracking_depthr r r r �start� s �zOCoroutineTests.test_debug_mode_manages_coroutine_origin_tracking.<locals>.startr T)rA rY rZ r rV r1 )r r[ r r r �1test_debug_mode_manages_coroutine_origin_tracking� s
z@CoroutineTests.test_debug_mode_manages_coroutine_origin_trackingc sF dd� �t j�fdd��� � �fdd�}�j�|� �}��|d� d S )Nc s s � dE d H dS )Nr rJ r r r r r �gen� s �
z0CoroutineTests.test_types_coroutine.<locals>.genc rL r r r )r] r r rN � rO z1CoroutineTests.test_types_coroutine.<locals>.funcc � s �� � } �� | tj� | I d H S r )ZassertIsInstance�types�_GeneratorWrapper)�wrapper)rN r r r rR � s �
z1CoroutineTests.test_types_coroutine.<locals>.cororJ )r^ rQ r r1 rA )r rR r; r )rN r] r r �test_types_coroutine� s z#CoroutineTests.test_types_coroutinec s4 d � � �fdd��� ��fdd�}�j �|� � d S )Nc � s4 �� j dd�} z��| d jjd� W d } d S d } w )N� )�limitr rE )Z get_stackrA �f_code�co_name)rG )�Tr r r rE � s
� z1CoroutineTests.test_task_print_stack.<locals>.fooc � s" �t j�� �jd�� � I d H d S )Nr9 )r Z
ensure_futurer r �rf rE r r r �runner� s �z4CoroutineTests.test_task_print_stack.<locals>.runner)r r1 �r rh r rg r �test_task_print_stack� s z$CoroutineTests.test_task_print_stackc sd dd� � � �fdd�}�j �d� �jtdd�� �j �|� � W d � d S 1 s+w Y d S )Nc � s �t �d�I d H d S )Ng�������?)r r% r r r r �afunc� s �z/CoroutineTests.test_double_await.<locals>.afuncc � sH �� � } �j �| �}zt�d�I d H | I d H W |�� d S |�� w rX )r Z create_taskr r% �cancel)rR �t�rk r r r rh � s � z0CoroutineTests.test_double_await.<locals>.runnerTz"coroutine is being awaited already)�msg)r rV Z assertRaises� RuntimeErrorr1 ri r rn r �test_double_await� s �"�z CoroutineTests.test_double_awaitN) r r r rH rI rS rW r\ ra rj rq r r r r rC m s rC �__main__)�__doc__rY r^ Zunittestr r Ztest.test_asyncior Z
test_utilsr r ZTestCaser r" r8 rC r �mainr r r r �<module> s /n �