File: //usr/local/lib/python3.7/test/test_asyncio/__pycache__/test_windows_events.cpython-37.pyc
B
��gT � @ s� d dl Z d dlZd dlZd dlZd dlmZ ejdkr@e�d��d dlZd dlZd dl Z d dl m
Z
d dl m Z
G dd� de j�ZG d d
� d
e
j�ZG d d � d e
j�Zed
kr�e�� dS )� N)�mock�win32z Windows only)�windows_events)�utilsc @ s$ e Zd Zdd� Zdd� Zdd� ZdS )�
UpperProtoc C s
g | _ d S )N)�buf)�self� r �A/usr/local/lib/python3.7/test/test_asyncio/test_windows_events.py�__init__ s zUpperProto.__init__c C s
|| _ d S )N)�trans)r r r r r
�connection_made s zUpperProto.connection_madec C s: | j �|� d|kr6| j�d�| j ��� � | j�� d S )N�
� )r �appendr �write�join�upper�close)r �datar r r
�
data_received s zUpperProto.data_receivedN)�__name__�
__module__� __qualname__r r
r r r r r
r s r c sT e Zd Z� fdd�Zdd� Zdd� Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
� Z S )�
ProactorTestsc s$ t � �� t�� | _| �| j� d S )N)�super�setUp�asyncio�ProactorEventLoop�loopZset_event_loop)r )� __class__r r
r " s
zProactorTests.setUpc C sh t �� \}}| j�|t�� �}tj| j�|d�| jd�}|�� | j� |� | �
|� � d� |�� d S )N�d )r r ) �socketZ
socketpairr Z_make_socket_transportr �ProtocolZ
ensure_futureZ sock_recvr �run_until_complete� assertEqual�result)r �a�br �fr r r
�
test_close' s zProactorTests.test_closec C sB dt �� }t�|�}| �t�� t�|� W d Q R X |�� d S )Nz \\.\pipe\test_double_bind-%s)�os�getpidr �
PipeServer� assertRaises�PermissionErrorr )r �ADDRESSZserver1r r r
�test_double_bind0 s
z ProactorTests.test_double_bindc C s | j �| �� �}| �|d� d S )N�done)r r$ �
_test_piper% )r Zresr r r
� test_pipe7 s zProactorTests.test_pipec � s� dt �� }| �t�� | j�tj|�I d H W d Q R X | j�t |�I d H \}| �
|t j � g }xvt
d�D ]j}tj| jd�}tj|| jd�� | j�� fdd�|�I d H \}}| �
|tj� | �� |� |�||f� qjW x,t|�D ] \}\}} | �d�|��� � q�W xFt|�D ]:\}\}} |�� I d H }
| �|
d�|��� � | �� �qW |�� | �t�� | j�tj|�I d H W d Q R X dS ) Nz\\.\pipe\_test_pipe-%s� )r c s � S )Nr r )�protocolr r
�<lambda>L r z*ProactorTests._test_pipe.<locals>.<lambda>z lower-{}
z LOWER-{}
r2 )r+ r, r. �FileNotFoundErrorr Zcreate_pipe_connectionr r# Zstart_serving_piper �assertIsInstancer r- �range� StreamReaderZStreamReaderProtocolZ Transportr% r � enumerater �format�encode�readliner ) r r0 ZserverZclients�iZ
stream_readerr �proto�r�wZresponser )r6 r
r3 ; s8
zProactorTests._test_pipec
C sv t � }tj|_tjjtd|d��L}| jj� d�}| j�
|�}|� � | � t
j�� | j�|� W d Q R X W d Q R X d S )NZ ConnectPipe)Z side_effectZ pipe_address)�OSError� _overlappedZERROR_PIPE_BUSYZwinerrorr Zpatch�objectr � _proactorZ connect_pipeZ create_task�cancelr. r �CancelledErrorr$ )r �excZconnect�coroZtaskr r r
�test_connect_pipe_cancela s
z&ProactorTests.test_connect_pipe_cancelc C s t �d ddd �}| �tj|� | jj�|d�}| j�� }| j� |�}| j�� | }| �
|d� | � |� � � | �
d| ko�dkn |� t �|� | jj�|d�}| j�� }| j� |�}| j�� | }| �
|d� | �
|� � � | �
d| ko�dk n |� |�� d S ) NTFg �?g�������?g�������?�
r g333333�?)rE � CreateEvent�
addCleanup�_winapi� CloseHandler rG �wait_for_handle�timer$ r% Z assertFalser& �
assertTrueZSetEventrH )r �event�fut�startr2 �elapsedr r r
�test_wait_for_handlen s$
z"ProactorTests.test_wait_for_handlec C s� t �d ddd �}| �tj|� | jj�|d�}|�� | j� � }| �
t j �� | j�
|� W d Q R X | j� � | }| �d| ko�dk n |� | jj�|�}|�� |�� d S )NTFrM r g�������?)rE rN rO rP rQ r rG rR rH rS r. r rI r$ rT )r rU rV rW rX r r r
�test_wait_for_handle_cancel� s
z)ProactorTests.test_wait_for_handle_cancel) r r r r r* r1 r4 r3 rL rY rZ �
__classcell__r r )r r
r s &
"r c @ s e Zd Zdd� Zdd� ZdS )�WinPolicyTestsc sD � fdd�}t �� }z t �t �� � t �|� � W d t �|� X d S )Nc � s � � t�� tj� d S )N)r9 r �get_running_loopZSelectorEventLoopr )r r r
�main� s z5WinPolicyTests.test_selector_win_policy.<locals>.main)r �get_event_loop_policy�set_event_loop_policyZ WindowsSelectorEventLoopPolicy�run)r r^ �
old_policyr )r r
�test_selector_win_policy� s
z'WinPolicyTests.test_selector_win_policyc sD � fdd�}t �� }z t �t �� � t �|� � W d t �|� X d S )Nc � s � � t�� tj� d S )N)r9 r r] r r )r r r
r^ � s z5WinPolicyTests.test_proactor_win_policy.<locals>.main)r r_ r` Z WindowsProactorEventLoopPolicyra )r r^ rb r )r r
�test_proactor_win_policy� s
z'WinPolicyTests.test_proactor_win_policyN)r r r rc rd r r r r
r\ � s r\ �__main__)r+ r" �sysZunittestr �platformZSkipTestrE rP r r Ztest.test_asyncior Z
test_utilsr# r ZTestCaser r\ r r^ r r r r
�<module> s"