File: //usr/local/lib/python3.10/test/test_asyncio/__pycache__/test_buffered_proto.cpython-310.pyc
o
�i! � @ s� d dl Z d dlZd dlmZ dd� ZG dd� de j�ZG dd� dej�Z G d d
� d
e ej
�Z e� e
e d �d �G d
d� de ej
��ZedkrKe�� dS dS )� N)�
functionalc C s t �d � d S �N)�asyncioZset_event_loop_policy� r r �B/usr/local/lib/python3.10/test/test_asyncio/test_buffered_proto.py�tearDownModule s r c @ s, e Zd Zdd� Zdd� Zdd� Zdd� Zd S )
�ReceiveStuffProtoc C s || _ || _d S r )�cb� con_lost_fut)�selfr r
r r r �__init__ s
zReceiveStuffProto.__init__c C s t d�| _| jS )N�d )� bytearray�buffer)r �sizehintr r r �
get_buffer s
z ReceiveStuffProto.get_bufferc C s | � | jd |� � d S r )r r )r �nbytesr r r �buffer_updated s z ReceiveStuffProto.buffer_updatedc C s( |d u r | j �d � d S | j �|� d S r )r
�
set_result�
set_exception)r �excr r r �connection_lost s z!ReceiveStuffProto.connection_lostN)�__name__�
__module__� __qualname__r r r r r r r r r s
r c @ s e Zd Zdd� Zdd� ZdS )�BaseTestBufferedProtocolc C s t �r )�NotImplementedError�r r r r �new_loop s z!BaseTestBufferedProtocol.new_loopc sz dd � � �fdd�}� fdd�}�j �t�|dd��}|jd �� }�j �t�||�d �� |�� �j �|�� � d S )
Ns 12345678+i c � sV �d����fdd���j �� � �j j� �fdd�g| �R � I d H \�}� I d H d S )N� c s" �| 7 ��� kr�� d� d S d S )N� 1)�write)�buf)�NOISE�data�trr r �on_buf* s �z^BaseTestBufferedProtocol.test_buffered_proto_create_connection.<locals>.client.<locals>.on_bufc s
t �� �S r )r r )�
conn_lost_futr& r r �<lambda>3 s
z`BaseTestBufferedProtocol.test_buffered_proto_create_connection.<locals>.client.<locals>.<lambda>)�loopZ
create_future�create_connection)�addrZpr�r# r )r' r$ r& r% r �client' s �
��zNBaseTestBufferedProtocol.test_buffered_proto_create_connection.<locals>.clientc � s6 �|� � � | �d�I d H |�� |�� I d H d S )N� )r! Z readexactly�close� wait_closed)�reader�writer)r# r r �on_server_client7 s
�
zXBaseTestBufferedProtocol.test_buffered_proto_create_connection.<locals>.on_server_clientz 127.0.0.1r � ) r) Zrun_until_completer Z start_serverZsockets� getsockname�wait_forr/ r0 )r r- r3 Zsrvr+ r r, r �%test_buffered_proto_create_connection# s ���z>BaseTestBufferedProtocol.test_buffered_proto_create_connectionN)r r r r r7 r r r r r s r c @ � e Zd Zdd� ZdS )� BufferedProtocolSelectorTestsc C � t �� S r )r ZSelectorEventLoopr r r r r L � z&BufferedProtocolSelectorTests.new_loopN�r r r r r r r r r9 I s r9 �ProactorEventLoopz Windows onlyc @ r8 )� BufferedProtocolProactorTestsc C r: r )r r= r r r r r T r; z&BufferedProtocolProactorTests.new_loopNr< r r r r r> P s r> �__main__)r ZunittestZtest.test_asyncior Z
func_testsr ZBufferedProtocolr ZFunctionalTestCaseMixinr ZTestCaser9 Z
skipUnless�hasattrr> r �mainr r r r �<module> s +� � �