File: //usr/local/lib/python3.7/test/test_asyncio/__pycache__/test_pep492.cpython-37.pyc
B
��g� � @ s� d Z ddlZddlZddlZddlmZ ddlmZ ddlZddlm Z
G dd� d�Z G dd � d e
j �Z
G d
d � d e
�ZG d d
� d
e
�ZG dd� de
�Zedkr�e�� dS )z3Tests support for new syntax introduced by PEP 492.� N)�support)�mock)�utilsc @ s. e Zd Zdd� Zd
dd�Zdd� Zdd � ZdS ) �FakeCoroc C s d S )N� )�self�valuer r �9/usr/local/lib/python3.7/test/test_asyncio/test_pep492.py�send s z
FakeCoro.sendNc C s d S )Nr )r �typ�val�tbr r r �throw s zFakeCoro.throwc C s d S )Nr )r r r r �close s zFakeCoro.closec c s
d V d S )Nr )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�setUp�asyncioZ
BaseEventLoop�loopr ZMockZ_process_eventsZ _selectorZselectZ return_valueZset_event_loop)r )� __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 sp t j� jd�t j� jd�t j� jd�t j� jd�g}� fdd�}x*|D ]"}� j�||�� � �|�� � qFW d S )N)r c
� s� t jd� jd�I d H � �| �� � | 4 I d H �D}� �|d � � �| �� � t jd� jd�I d H � �| �� � W d Q I d H R X � �| �� � d S )Ng{�G�z�?)r )r �sleepr � assertFalse�locked�assertIs�
assertTrue)�lock�_lock)r r r �test2 s z7LockTests.test_context_manager_async_with.<locals>.test) r �Lockr � Condition� Semaphore�BoundedSemaphore�run_until_completer r )r �
primitivesr# � primitiver )r r � test_context_manager_async_with* s
z)LockTests.test_context_manager_async_withc sp t j� jd�t j� jd�t j� jd�t j� jd�g}� fdd�}x*|D ]"}� j�||�� � �|�� � qFW d S )N)r c
� s� t jd� jd�I d H � �| �� � � �t��h | I d H �D}� �|d � � �| �� � t jd� jd�I d H � �| �� � W d Q R X � �| �� � W d Q R X d S )Ng{�G�z�?)r ) r r r r r Z assertWarns�DeprecationWarningr r )r! r" )r r r r# H s z7LockTests.test_context_manager_with_await.<locals>.test) r r$ r r% r& r' r( r r )r r) r# r* r )r r � test_context_manager_with_await@ s
z)LockTests.test_context_manager_with_awaitN)r r r r+ r- r r r r r ( s r c @ s e Zd Zdd� ZdS )�StreamReaderTestsc sT d}t j| jd�� � �|� � �� � fdd�}| j�|� �}| �|dddg� d S )Ns line1
line2
line3)r c � s> g } x4� 2 y 3 d H }W n
t k
s.X | �|� q
Y W | S )N)�StopAsyncIteration�append)�data�line)�streamr r �readera s z/StreamReaderTests.test_readline.<locals>.readers line1
s line2
s line3)r � StreamReaderr Z feed_dataZfeed_eofr( � assertEqual)r ZDATAr4 r1 r )r3 r �
test_readlineZ s
z StreamReaderTests.test_readlineN)r r r r7 r r r r r. X s r. c @ sL e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dS )�CoroutineTestsc C sD dd� }|� }z| � t�|�� W d |�� X | � t�t� �� d S )Nc � s d S )Nr r r r r �foon � z,CoroutineTests.test_iscoroutine.<locals>.foo)r r Z iscoroutiner r )r r9 �fr r r �test_iscoroutinem s
z CoroutineTests.test_iscoroutinec C s dd� }| � t�|�� d S )Nc � s d S )Nr r r r r r9 y r: z4CoroutineTests.test_iscoroutinefunction.<locals>.foo)r r Ziscoroutinefunction)r r9 r r r �test_iscoroutinefunctionx s z'CoroutineTests.test_iscoroutinefunctionc sD G dd� d�� t j� fdd��}|� }| �|�d �d� |�� d S )Nc @ s e Zd Zdd� ZdS )zCCoroutineTests.test_function_returning_awaitable.<locals>.Awaitablec S s dS )N)�spamr )r r r r r ~ s zMCoroutineTests.test_function_returning_awaitable.<locals>.Awaitable.__await__N)r r r r r r r r � Awaitable} s r? c s � � S )Nr r )r? r r �func� s z>CoroutineTests.test_function_returning_awaitable.<locals>.funcr> )r � coroutiner6 r
r )r r@ �coror )r? 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 )Nr> r r r r r �bar� s z5CoroutineTests.test_async_def_coroutines.<locals>.barc � s � � I d H S )Nr r )rD r r r9 � s z5CoroutineTests.test_async_def_coroutines.<locals>.foor> T)r r( r6 � set_debug)r r9 r1 r )rD 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)r6 rG rH r rE r( )r rI 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 r> r r r r r �gen� s
z0CoroutineTests.test_types_coroutine.<locals>.genc s � � S )Nr r )rK r r r@ � s z1CoroutineTests.test_types_coroutine.<locals>.funcc � s � � } �� | tj� | I d H S )N)ZassertIsInstance�types�_GeneratorWrapper)�wrapper)r@ r r r rB � s z1CoroutineTests.test_types_coroutine.<locals>.coror> )rL rA r r( r6 )r rB r1 r )r@ rK r r �test_types_coroutine� s
z#CoroutineTests.test_types_coroutinec s4 d � � �fdd��� ��fdd�}�j �|� � d S )Nc � s0 � j dd�} z��| d jjd� W d d } X d S )N� )�limitr r9 )Z get_stackr6 �f_code�co_name)r; )�Tr r r r9 � s z1CoroutineTests.test_task_print_stack.<locals>.fooc � s t j�� �jd�� � I d H d S )N)r )r Z
ensure_futurer r )rT r9 r r r �runner� s z4CoroutineTests.test_task_print_stack.<locals>.runner)r r( )r rU r )rT r9 r r �test_task_print_stack� s z$CoroutineTests.test_task_print_stackc sR �fdd�� � �fdd�}�j �d� �jtdd�� �j �|� � W d Q R X d S )Nc � s t jd� jd�I d H d S )Ng�������?)r )r r r r )r r r �afunc� s z/CoroutineTests.test_double_await.<locals>.afuncc � sJ � � } t j| �jd�}z$t jd�jd�I d H | I d H W d |�� X d S )N)r r )r ZTaskr r Zcancel)rB �t)rW r r r rU � s z0CoroutineTests.test_double_await.<locals>.runnerTz"coroutine is being awaited already)�msg)r rE Z assertRaises� RuntimeErrorr( )r rU r )rW r r �test_double_await� s
z CoroutineTests.test_double_awaitN) r r r r< r= rC rF rJ rO rV r[ r r r r r8 k s
r8 �__main__)�__doc__rG rL Zunittestr# r r r Ztest.test_asyncior Z
test_utilsr ZTestCaser r r. r8 r �mainr r r r �<module> s 0m