File: //lib64/python3.6/test/__pycache__/test_timeout.cpython-36.pyc
3
\o, � @ s� d Z ddlZddlZddlmZ ejd� ZddlZddlZddl Z ej
� dd� �Z G dd� dej �Z
G d d
� d
ej �ZG d d � d e�ZG d
d� de�Zdd� Zedkr�e� dS )z&Unit tests for socket timeout feature.� N)�support�networkc C s2 t j| �� tj| |tjtj�d d S Q R X dS )z�Resolve an (host, port) to an address.
We must perform name resolution before timeout tests, otherwise it will be
performed by connect().
r � N)r �transient_internet�socketZ getaddrinfo�AF_INET� SOCK_STREAM)�hostZport� r
�$/usr/lib64/python3.6/test_timeout.py�resolve_address s r c @ sX e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
dd� Z dd� Z dS )�CreationTestCasez9Test case for socket.gettimeout() and socket.settimeout()c C s t j t jt j�| _d S )N)r r r �sock)�selfr
r
r �setUp s zCreationTestCase.setUpc C s | j j� d S )N)r �close)r r
r
r �tearDown! s zCreationTestCase.tearDownc C s | j | jj� d d� d S )Nz timeout not disabled by default)� assertEqualr �
gettimeout)r r
r
r �testObjectCreation$ s z#CreationTestCase.testObjectCreationc C s^ | j jd� | j| j j� d� | j jd� | j| j j� d� | j jd � | j| j j� d � d S )Ng�z�Ga @� )r �
settimeoutr r )r r
r
r �testFloatReturnValue) s z%CreationTestCase.testFloatReturnValuec C sP | j jd� | jt| j j� �td�� | j jd� | jt| j j� �td�� d S )N� g �?g333333@)r r r �typer )r r
r
r �testReturnType4 s z CreationTestCase.testReturnTypec C s� | j jd� | j jd� | j jd� | j jd � | jt| j jd� | jt| j jd� | jt| j jf � | jt| j jg � | jt| j ji � | jt| j jd� d S )Nr g � y )r r � assertRaises� TypeError)r r
r
r �
testTypeCheck< s z CreationTestCase.testTypeCheckc C s: | j t| jjd� | j t| jjd� | j t| jjd� d S )Nr g �?���r g �)r �
ValueErrorr r )r r
r
r �testRangeCheckI s z CreationTestCase.testRangeCheckc C s� | j jd� | j jd� | j| j j� d � | j jd� | j| j j� d� | j jd� | j jd� | j| j j� d� | j jd� | j| j j� d � d S )N�
r r g )r r � setblockingr r )r r
r
r �testTimeoutThenBlockingO s z(CreationTestCase.testTimeoutThenBlockingc C sX | j jd� | j jd� | j| j j� d� | j jd� | j jd� | j| j j� d� d S )Nr r )r r$ r r r )r r
r
r �testBlockingThenTimeout] s z(CreationTestCase.testBlockingThenTimeoutN)
�__name__�
__module__� __qualname__�__doc__r r r r r r r" r% r& r
r
r
r r
s
r
c @ s* e Zd ZdZejZdd� ZeZdd� Z dS )�TimeoutTestCaseg @c C s
t � �d S )N)�NotImplementedError)r r
r
r r s s zTimeoutTestCase.setUpc G s� | j j|� t| j |�}xbt|�D ]L}tj� }y ||� W q" tjk
rl } ztj� | }P W Y dd}~X q"X q"W | jd� | j||| j � | j
||d � dS )z�
Test the specified socket method.
The method is run at most `count` times and must raise a socket.timeout
within `timeout` + self.fuzz seconds.
Nz socket.timeout was not raisedg �?) r r �getattr�range�timer �timeoutZfailZ
assertLess�fuzzZ
assertGreater) r �countr0 �method�args�iZt1�eZdeltar
r
r �_sock_operationx s
z TimeoutTestCase._sock_operationN)
r' r( r) r1 r ZHOST� localhostr r r7 r
r
r
r r+ h s
r+ c @ sP e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
dd� Z dS )�TCPTimeoutTestCasez3TCP test case for socket.socket() timeout functionsc C s" t j t jt j�| _tdd�| _d S )Nzwww.python.org.�P )r r r r r � addr_remote)r r
r
r r � s zTCPTimeoutTestCase.setUpc C s | j j� d S )N)r r )r r
r
r r � s zTCPTimeoutTestCase.tearDownc C s t dd�}t dd�}d}tjtjtj�}d}|j|� zZy|j|� W nF tjk
r\ Y n2 tk
r� } z|jtj kr|d}W Y d d }~X nX W d |j
� ~X |r�| j dj |d |d
||d |d
�� || _
tj| j
d �� | jd
d d | j
� W d Q R X d S )
Nzblackhole.snakebite.netiZ� zwhitehole.snakebite.neti[� Tr Fz�We didn't receive a connection reset (RST) packet from {}:{} within {} seconds, so we're unable to test connect timeout against the corresponding {}:{} (which is configured to silently drop packets).r r g����MbP?�connect)r r r r r r< r0 �OSError�errnoZ ECONNREFUSEDr ZskipTest�formatr; r r r7 )r Z blackholeZ whitehole�skipr r0 �errr
r
r �testConnectTimeout� s4
z%TCPTimeoutTestCase.testConnectTimeoutc C s> t j| jd ��$ | jj| j� | jdddd� W d Q R X d S )Nr r g �?Zrecvi )r r r; r r< r7 )r r
r
r �testRecvTimeout� s z"TCPTimeoutTestCase.testRecvTimeoutc C s, t j| j| j� | jj� | jddd� d S )Nr g �?Zaccept)r � bind_portr r8 �listenr7 )r r
r
r �testAcceptTimeout� s
z$TCPTimeoutTestCase.testAcceptTimeoutc
C sZ t j t jt j��@}tj|| j� |j� | jj|j � � | j
ddddd � W d Q R X d S )N�d g �?�send� Xi@
) r r r r rD r8 rE r r< � getsocknamer7 )r �servr
r
r �testSend� s
zTCPTimeoutTestCase.testSendc
C s` t j t jt j��F}tj|| j� |j� | jj|j � � | j
ddddd |j � � W d Q R X d S )NrG g �?ZsendtorI i@
) r r r r rD r8 rE r r< rJ r7 )r rK r
r
r �
testSendto s z TCPTimeoutTestCase.testSendtoc
C sZ t j t jt j��@}tj|| j� |j� | jj|j � � | j
ddddd � W d Q R X d S )NrG g �?ZsendallrI i@
) r r r r rD r8 rE r r< rJ r7 )r rK r
r
r � testSendall s
z TCPTimeoutTestCase.testSendallN) r' r( r) r* r r rB rC rF rL rM rN r
r
r
r r9 � s T
r9 c @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) �UDPTimeoutTestCasez3UDP test case for socket.socket() timeout functionsc C s t j t jt j�| _d S )N)r r Z
SOCK_DGRAMr )r r
r
r r s zUDPTimeoutTestCase.setUpc C s | j j� d S )N)r r )r r
r
r r s zUDPTimeoutTestCase.tearDownc C s$ t j| j| j� | jdddd� d S )Nr g �?Zrecvfromi )r rD r r8 r7 )r r
r
r �testRecvfromTimeout s z&UDPTimeoutTestCase.testRecvfromTimeoutN)r' r( r) r* r r rP r
r
r
r rO s rO c C s t jd� t jttt� d S )Nr )r ZrequiresZ run_unittestr
r9 rO r
r
r
r � test_main& s
rQ �__main__)r* � functoolsZunittestZtestr Zis_resource_enabledZ
skip_expectedr/ r> r � lru_cacher ZTestCaser
r+ r9 rO rQ r' r
r
r
r �<module> s M'