File: //lib/python3/dist-packages/twisted/internet/test/__pycache__/test_udp_internals.cpython-310.pyc
o
�b� � @ s� d Z ddlZddlmZ ddlmZ ddlmZ ddlm Z edkr)ddl
m Z ndd l
m Z G d
d � d �Z
G d d
� d
e�ZG dd� de j�ZdS )zK
Tests for the internal implementation details of L{twisted.internet.udp}.
� N)�udp)�DatagramProtocol)� platformType)�unittest�win32)�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$ �
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
�size�retr 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 6 r zKeepReads.__init__c C s | j �|� d S r
)r �append)r
�datar r r r �datagramReceived9 s zKeepReads.datagramReceivedN)r r r r r r# r r r r r 1 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 g d�� 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��N� result� 123� 456)r% r& r' )
r �_sockErrReadIgnorer! �
addCleanup�remover �Portr r r �doRead� assertEqualr �r
�protocol�portr r r �test_socketReadNormalB 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>e s 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_readImmediateErrorX 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.
r2 c s
� � d�S )NT)r! r ��refusedr r r4 � s
z>ErrorsTests.test_connectedReadImmediateError.<locals>.<lambda>Nr5 r6 z 127.0.0.1i' T)r r7 r! r) r* r r8 r+ r r r r r r, r- r r. r r: r � test_connectedReadImmediateErrort 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 � assertRaisesr, r- r r. r r r �test_readUnknownError� s
z!ErrorsTests.test_readUnknownErrorN)r r r r r1 r9 r< r> r r r r r$ = s r$ )r r �twisted.internetr �twisted.internet.protocolr �twisted.python.runtimer �
twisted.trialr �errnor r r r �SynchronousTestCaser$ r r r r �<module> s