File: //proc/thread-self/root/usr/local/lib/python3.7/test/__pycache__/test_asynchat.cpython-37.pyc
B
��g<% � @ s d dl mZ d dlZd dlZd dlZd dlZd dlZd dlZd dl Z d dl
Z
d dl Z d dl Z ej
Z
dZdZG dd� de j�ZG dd� dej�Zd d
� ZG d d � d e j�ZG d
d� de�ZG dd� de j�ZG dd� de j�ZG dd� de j�Zedkr�e � � dS )� )�supportNs QUIT
g @c @ s e Zd ZdZdd� Zdd� ZdS )� echo_server� c C s< t j�| � || _t�tjtj�| _t� | j�| _
d | _ d S )N) � threading�Thread�__init__�event�socket�AF_INET� SOCK_STREAM�sockr Z bind_port�port�start_resend_event)�selfr � r �./usr/local/lib/python3.7/test/test_asynchat.pyr s
zecho_server.__init__c C s� | j �� | j�� | j �� \}}d| _x*t| jkrR|�d�}|sDP | j| | _q*W | j�td�| _| j rt| j �
� y@x:| jr�|� | jd | j � �}t
�d� | j|d � | _qxW W n Y nX |�� | j �� d S )N� r g����MbP?)r Zlistenr �setZaccept�buffer� SERVER_QUIT�recv�replacer �wait�send�
chunk_size�time�sleep�close)r Zconn�client�data�nr r r �run# s*
zecho_server.runN)�__name__�
__module__� __qualname__r r r! r r r r r s r c @ s$ e Zd Zdd� Zdd� Zdd� ZdS )� echo_clientc C s^ t j�| � g | _| �tjtj� | �t |f� | �
|� d| _ dd� }t j
dkrZdd� }d S )Nr c S s d S )Nr )r r r r �handle_connectM s z,echo_client.__init__.<locals>.handle_connect�darwinc S s d S )Nr )r r r r � handle_exptS s z)echo_client.__init__.<locals>.handle_expt)�asynchat�
async_chatr �contentsZ
create_socketr r
r Zconnect�HOST�set_terminatorr �sys�platform)r �
terminatorZ server_portr& r( r r r r E s
zecho_client.__init__c C s | j |7 _ d S )N)r )r r r r r �collect_incoming_dataV s z!echo_client.collect_incoming_datac C s | j �| j� d| _d S )Nr )r+ �appendr )r r r r �found_terminatorY s z echo_client.found_terminatorN)r" r# r$ r r1 r3 r r r r r% C s r% c C s: t �� } t| �}|�� | �� | �� t�d� || fS )Ng{�G�z�?)r �Eventr �startr �clearr r )r �sr r r �start_echo_server] s
r8 c @ s� e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
dd� Z dd� Z dd� Z
dd� Zdd� Zdd� Zd d � Zd d � Zd S )!� TestAsynchatFc C s t �� | _d S )N)r Zthreading_setup�_threads)r r r r �setUpj s zTestAsynchat.setUpc C s t j| j� d S )N)r Zthreading_cleanupr: )r r r r �tearDownm s zTestAsynchat.tearDownc C s� t �� }t|�}||_|�� |�� |�� t�d� t ||j
�}|� d� |� d| � |� d| � |� t � t
j| jddd� tj|td� | �|jddg� d S ) Ng{�G�z�?s hello s worlds I'm not dead yet!i, )�use_poll�count�timeout)r? s hello world)r r4 r r r5 r r6 r r r% r
�pushr �asyncore�loop�usepollr � join_thread�TIMEOUT� assertEqualr+ )r ZtermZ server_chunkr r7 �cr r r �line_terminator_checkp s
z"TestAsynchat.line_terminator_checkc C s xdD ]}| � d|� qW d S )N)r � � �
)rH )r �lr r r �test_line_terminator1� s
z"TestAsynchat.test_line_terminator1c C s xdD ]}| � d|� qW d S )N)r rI rJ s
)rH )r rL r r r �test_line_terminator2� s
z"TestAsynchat.test_line_terminator2c C s xdD ]}| � d|� qW d S )N)r rI rJ s qqq)rH )r rL r r r �test_line_terminator3� s
z"TestAsynchat.test_line_terminator3c C sj t � \}}t||j�}d}|�|� |�t� tj| jddd� tj |t
d� | � |j |d |� g� d S )Ns hello world, I'm not dead yet!
i, g{�G�z�?)r= r> r? )r? )
r8 r% r
r@ r rA rB rC r rD rE rF r+ )r Ztermlenr7 r rG r r r r �numeric_terminator_check� s
z%TestAsynchat.numeric_terminator_checkc C s | � d� d S )Nr )rP )r r r r �test_numeric_terminator1� s z%TestAsynchat.test_numeric_terminator1c C s | � d� d S )N� )rP )r r r r �test_numeric_terminator2� s z%TestAsynchat.test_numeric_terminator2c C sn t � \}}td |j�}d}|�|� |�t� tj| jddd� tj |t
d� | � |j g � | � |j
|� d S )Ns hello world, I'm not dead yet!
i, g{�G�z�?)r= r> r? )r? )r8 r% r
r@ r rA rB rC r rD rE rF r+ r )r r7 r rG r r r r �test_none_terminator� s
z!TestAsynchat.test_none_terminatorc C sl t � \}}td|j�}d}tj|t dd�}|�|� tj| j ddd� t
j |t d� | �
|jd d
g� d S ) NrK s hello world
I'm not dead yet!
� )� buffer_sizei, g{�G�z�?)r= r> r? )r? s hello worlds I'm not dead yet!)r8 r% r
r) Zsimple_producerr �push_with_producerrA rB rC r rD rE rF r+ )r r7 r rG r �pr r r �test_simple_producer� s
z!TestAsynchat.test_simple_producerc C s^ t � \}}td|j�}d}|�|t � tj| jddd� tj |t
d� | � |j ddg� d S ) NrK s hello world
I'm not dead yet!
i, g{�G�z�?)r= r> r? )r? s hello worlds I'm not dead yet!)
r8 r% r
rW r rA rB rC r rD rE rF r+ )r r7 r rG r r r r �test_string_producer� s
z!TestAsynchat.test_string_producerc C sb t � \}}td|j�}|�d� |�t� tj| jddd� tj |t
d� | � |j ddd g� d S )
NrK s hello world
I'm not dead yet!
i, g{�G�z�?)r= r> r? )r? s hello worldr s I'm not dead yet!)
r8 r% r
r@ r rA rB rC r rD rE rF r+ )r r7 r rG r r r �test_empty_line� s
z TestAsynchat.test_empty_linec C s� t � \}}t�� |_td|j�}|�d� |�t� |�� t j
| j ddd� |j� � t
j|td� | �|jg � | �t|j�d� d S )NrK s hello world
I'm not dead yet!
i, g{�G�z�?)r= r> r? )r? r )r8 r r4 r r% r
r@ r Zclose_when_donerA rB rC r r rD rE rF r+ Z
assertGreater�lenr )r r7 r rG r r r �test_close_when_done� s
z!TestAsynchat.test_close_when_donec C s� t � \}}td|j�}d}|�|� |�t|�� |�t|�� | �t|jd� | �t|jd� |�t� t j
| j ddd� t j
|td� | �|jd d d g� d S )
NrK s bytes
�
Zunicodei, g{�G�z�?)r= r> r? )r? s bytes)r8 r% r
r@ � bytearray�
memoryview� assertRaises� TypeErrorr rA rB rC r rD rE rF r+ )r r7 r rG r r r r � test_push� s
zTestAsynchat.test_pushN)r" r# r$ rC r; r< rH rM rN rO rP rQ rS rT rY rZ r[ r] rc r r r r r9 g s
r9 c @ s e Zd ZdZdS )�TestAsynchat_WithPollTN)r" r# r$ rC r r r r rd � s rd c @ s e Zd Zdd� ZdS )�TestAsynchatMockedc C sj t j�� }ttj�|j_t� � }|�
|� | � |j � t jj
�|d��}|�� W d Q R X | �|j� d S )NZ handle_error)�unittestZmockZMock�BlockingIOError�errnoZEAGAINr Z side_effectr) r* Z
set_socketZ
addCleanupZ del_channelZpatch�objectZ handle_readZ assertFalseZcalled)r r Z
dispatcher�errorr r r �test_blockingioerror s
z'TestAsynchatMocked.test_blockingioerrorN)r" r# r$ rk r r r r re s re c @ s e Zd Zdd� ZdS )�TestHelperFunctionsc C s, | � t�dd�d� | � t�dd�d� d S )Nzqwerty
z
r Z
qwertydkjfr )rF r) Zfind_prefix_at_end)r r r r �test_find_prefix_at_end s z+TestHelperFunctions.test_find_prefix_at_endN)r" r# r$ rm r r r r rl s rl c @ s e Zd Zdd� ZdS )�TestNotConnectedc C s t �� }| �t|jd� d S )N���)r) r* ra �
ValueErrorr- )r r r r r �!test_disallow_negative_terminator s z2TestNotConnected.test_disallow_negative_terminatorN)r" r# r$ rq r r r r rn s rn �__main__) Ztestr r) rA rh r r. �_threadZthreadr r rf Z
unittest.mockr, r rE r r r* r% r8 ZTestCaser9 rd re rl rn r" �mainr r r r �<module> s0 .