File: //lib/python3/dist-packages/twisted/internet/test/__pycache__/test_udp_internals.cpython-38.pyc
U
W[- � @ s� d Z ddlmZmZ ddlZddlmZ ddlmZ ddl m
Z
ddl m Z e dkrbdd l
mZ n dd
l
mZ G d d � d e�ZG d
d� de�ZG dd� dej�ZdS )zK
Tests for the internal implementation details of L{twisted.internet.udp}.
� )�division�absolute_importN)�unittest)�DatagramProtocol)�udp)� platformTypeZwin32)�WSAEWOULDBLOCK)� EWOULDBLOCKc @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) �StringUDPSocketa
A fake UDP socket object, which returns a fixed sequence of strings and/or
socket errors. Useful for testing.
@ivar retvals: A C{list} containing either strings or C{socket.error}s.
@ivar connectedAddr: The address the socket is connected to.
c C s || _ d | _d S �N)�retvals�
connectedAddr)�selfr � r �J/usr/lib/python3/dist-packages/twisted/internet/test/test_udp_internals.py�__init__" s zStringUDPSocket.__init__c C s
|| _ d S r )r
)r �addrr r r �connect' s zStringUDPSocket.connectc C s$ | j �d�}t|tj�r |�|dfS )zH
Return (or raise) the next value from C{self.retvals}.
r N)r �pop�
isinstance�socket�error)r �sizeZretr r r �recvfrom+ s zStringUDPSocket.recvfromN)�__name__�
__module__� __qualname__�__doc__r r r r r r r r
s r
c @ s e Zd ZdZdd� Zdd� ZdS )� KeepReadsz%
Accumulate reads in a list.
c C s
g | _ d S r )�reads)r r r r r ; s zKeepReads.__init__c C s | j �|� d S r )r �append)r �datar r r r �datagramReceived? s zKeepReads.datagramReceivedN)r r r r r r" r r r r r 6 s r c @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
S ) � ErrorsTestsz/
Error handling tests for C{udp.Port}.
c C s� t j�d� | �t jjd� t� }t �d|�}tddt� d�dt� d�g�|_|�
� | � |j ddg� |�
� | � |j dddg� dS )z�
Socket reads with some good data followed by a socket error which can
be ignored causes reading to stop, and no log messages to be logged.
i��Ns results 123s 456)
r Z_sockErrReadIgnorer �
addCleanup�remover �Portr
r r �doRead� assertEqualr �r ZprotocolZportr r r �test_socketReadNormalI s ��z!ErrorsTests.test_socketReadNormalc C s� t j�d� | �t jjd� t� }dd� |_t �d|�}tdt �
d�dt �
t �g�|_ |� � | �
|jdg� |� � | �
|jddg� dS )z�
If the socket is unconnected, socket reads with an immediate
connection refusal are ignored, and reading stops. The protocol's
C{connectionRefused} method is not called.
��c S s dd S )N� r r r r r r �<lambda>m � z5ErrorsTests.test_readImmediateError.<locals>.<lambda>N� a� b)r �_sockErrReadRefuser r$ r% r �connectionRefusedr& r
r r r r' r( r r) r r r �test_readImmediateError` s
�z#ErrorsTests.test_readImmediateErrorc s� t j�d� | �t jjd� t� }g � � fdd�|_t �d|�}tdt �
d�dt �
t �g�|_ |� dd� |�
� | �|jdg� | �� d g� |�
� | �|jddg� | �� d g� dS )
z�
If the socket connected, socket reads with an immediate
connection refusal are ignored, and reading stops. The protocol's
C{connectionRefused} method is called.
r+ c s
� � d�S )NT)r r �Zrefusedr r r- � r. z>ErrorsTests.test_connectedReadImmediateError.<locals>.<lambda>Nr/ r0 z 127.0.0.1i' T)r r1 r r$ r% r r2 r& r
r r r r r' r( r r) r r4 r � test_connectedReadImmediateError| s � z,ErrorsTests.test_connectedReadImmediateErrorc C sJ t � }t�d|�}tdt�d�g�|_| �tj|j� | �|j dg� dS )zG
Socket reads with an unknown socket error are raised.
Ns goodi���)
r r r& r
r r Z assertRaisesr' r( r r) r r r �test_readUnknownError� s
z!ErrorsTests.test_readUnknownErrorN)r r r r r* r3 r5 r6 r r r r r# D s
r# )r Z
__future__r r r Z
twisted.trialr Ztwisted.internet.protocolr Ztwisted.internetr Ztwisted.python.runtimer �errnor r �objectr
r ZSynchronousTestCaser# r r r r �<module> s