File: //lib/python3/dist-packages/twisted/internet/test/__pycache__/test_iocp.cpython-38.pyc
U
W[2 �
@ sJ d Z ddlZddlmZ ddlmZ ddlmZmZmZm Z m
Z
mZ ddl m Z ddl
mZ ddlmZ dd lmZ zPdd
lmZmZmZ dd lmZmZmZ dd l m Z dd
l m Z ddl m!Z! W n e"k
r� dZ#Y nX zeee��$� W n, e
k
�r Z% z e&e%�Z'W 5 dZ%[%X Y nX dZ'G dd� dej(�Z)G dd� dej(�Z*dS )z,
Tests for L{twisted.internet.iocpreactor}.
� N)�array)�pack)�AF_INET6�AF_INET� SOCK_STREAM�
SOL_SOCKET�error�socket)� verifyClass)�unittest)�msg)�
IPushProducer)� iocpsupport�tcp�udp)� IOCPReactor�EVENTS_PER_LOOP�
KEY_NORMAL)�IReadWriteHandle)�SO_UPDATE_ACCEPT_CONTEXT)�
FileHandlez%This test only applies to IOCPReactorc @ s6 e Zd ZdZdd� Zdd� Zdd� Zedk r2ee_dS ) � SupportTestszn
Tests for L{twisted.internet.iocpreactor.iocpsupport}, low-level reactor
implementation helpers.
c
C s> t d|f � t|t�}| �|j� |�d� |�d� t|t�}| �|j� |�d� z|�||� � d f� W n8 t
k
r� } z| � |j t j
t jf� W 5 d}~X Y nX t|t�}| �|j� tdd�}| �dt�|�� |�� |d�� |�tttd |�� �� | �||�� dd
� |� � dd
� ft�|�� |�� dS ) a
Create a C{SOCK_STREAM} connection to localhost using a socket with an
address family of C{family} and assert that the result of
L{iocpsupport.get_accept_addrs} is consistent with the result of
C{socket.getsockname} and C{socket.getpeername}.
z family = %r)� r � FN�Bs r �P� )r r r Z
addCleanup�closeZbindZlistenZ setblockingZconnectZ getsocknamer ZassertIn�errnoZ EINPROGRESSZ EWOULDBLOCKr � assertEqual�_iocpZaccept�filenoZ
setsockoptr r r Z getpeernameZget_accept_addrs)�selfZfamilyZ localhost�portZclient�eZserverZbuff� r% �A/usr/lib/python3/dist-packages/twisted/internet/test/test_iocp.py�_acceptAddressTest( s8
(
� � �z SupportTests._acceptAddressTestc C s | � td� dS )a
L{iocpsupport.get_accept_addrs} returns a three-tuple of address
information about the socket associated with the file descriptor passed
to it. For a connection using IPv4:
- the first element is C{AF_INET}
- the second element is a two-tuple of a dotted decimal notation IPv4
address and a port number giving the peer address of the connection
- the third element is the same type giving the host address of the
connection
z 127.0.0.1N)r' r �r" r% r% r& �test_ipv4AcceptAddressH s z#SupportTests.test_ipv4AcceptAddressc C s | � td� dS )a�
Like L{test_ipv4AcceptAddress}, but for IPv6 connections. In this case:
- the first element is C{AF_INET6}
- the second element is a two-tuple of a hexadecimal IPv6 address
literal and a port number giving the peer address of the connection
- the third element is the same type giving the host address of the
connection
z::1N)r' r r( r% r% r& �test_ipv6AcceptAddressW s
z#SupportTests.test_ipv6AcceptAddressN) �__name__�
__module__� __qualname__�__doc__r' r) r* �ipv6Skip�skipr% r% r% r& r # s r c @ s, e Zd Zdd� Zdd� Zdd� Zdd� Zd S )
�IOCPReactorTestsc C s" t � }|�� | �|�d�� dS )zd
Test reactor behavior (doIteration) when there are no pending time
events.
N)r ZwakeUpZ assertFalse� doIteration)r" �irr% r% r& �test_noPendingTimerEventsh s z*IOCPReactorTests.test_noPendingTimerEventsc C s( | � tttj�� | � tttj�� dS )zY
Verify that IOCP socket-representing classes implement IReadWriteHandle
N)�
assertTruer
r r Z
Connectionr ZPortr( r% r% r& �test_reactorInterfacesr s z'IOCPReactorTests.test_reactorInterfacesc C s | � ttt�� dS )zH
Verify that L{Filehandle} implements L{IPushProducer}.
N)r5 r
r
r r( r% r% r& �test_fileHandleInterfacesz s z*IOCPReactorTests.test_fileHandleInterfacesc C s� G dd� d�}t � }|� }t�|j|�}ttd �D ]}|j�dt|� q4|� d� | �
|j t� |� d� | �
|j td � dS )z�
Verify that we don't lose an event when more than EVENTS_PER_LOOP
events occur in the same reactor iteration
c @ s e Zd ZdZdd� Zdd� ZdS )z;IOCPReactorTests.test_maxEventsPerIteration.<locals>.FakeFDr c S s dS )N�FakeFDr% r( r% r% r& � logPrefix� s zEIOCPReactorTests.test_maxEventsPerIteration.<locals>.FakeFD.logPrefixc S s | j d7 _ d S )Nr )�counter)r" Zrc�bytesZevtr% r% r& �cb� s z>IOCPReactorTests.test_maxEventsPerIteration.<locals>.FakeFD.cbN)r+ r, r- r: r9 r< r% r% r% r& r8 � s r8 r r N) r r ZEventr< �ranger r# Z postEventr r2 r r: )r" r8 r3 �fdZevent�_r% r% r& �test_maxEventsPerIteration� s
z+IOCPReactorTests.test_maxEventsPerIterationN)r+ r, r- r4 r6 r7 r@ r% r% r% r& r1 g s
r1 )+r. r r Zstructr r r r r r r Zzope.interface.verifyr
Z
twisted.trialr Ztwisted.python.logr Ztwisted.internet.interfacesr
Z twisted.internet.iocpreactorr r r r Z$twisted.internet.iocpreactor.reactorr r r Z'twisted.internet.iocpreactor.interfacesr Z"twisted.internet.iocpreactor.constr Z%twisted.internet.iocpreactor.abstractr � ImportErrorr0 r r$ �strr/ ZTestCaser r1 r% r% r% r& �<module> s.
D