File: //usr/lib/python3/dist-packages/twisted/internet/test/__pycache__/test_socket.cpython-310.pyc
o
�b�$ � @ s� d Z ddlZddlZddlmZ ddlmZ ddlmZ ddl m
Z
m Z ddl m
Z
mZ ddlmZ dd lmZ G d
d � d e
�ZG d d
� d
e
�ZG dd� de
�ZG dd� de
�Ze� �e�� � e� �e�� � e� �e�� � e� �e�� � dS )aD
Tests for implementations of L{IReactorSocket}.
Generally only tests for failure cases are found here. Success cases for
this interface are tested elsewhere. For example, the success case for
I{AF_INET} is in L{twisted.internet.test.test_tcp}, since that case should
behave exactly the same as L{IReactorTCP.listenTCP}.
� N)�verify)�UnsupportedAddressFamily)�IReactorSocket)�DatagramProtocol�
ServerFactory)�ReactorBuilder�needsRunningReactor)�err)�platformc @ � e Zd ZdZegZdd� ZdS )�&IReactorSocketVerificationTestsBuilderaO
Builder for testing L{IReactorSocket} implementations for required
methods and method signatures.
L{ReactorBuilder} already runs L{IReactorSocket.providedBy} to
ensure that these tests will only be run on reactor classes that
claim to implement L{IReactorSocket}.
These tests ensure that reactors which claim to provide the
L{IReactorSocket} interface actually have all the required methods
and that those methods have the expected number of arguments.
These tests will be skipped for reactors which do not claim to
provide L{IReactorSocket}.
c C s | � � }| �t�t|�� dS )zf
The reactor instance returned by C{buildReactor} provides
L{IReactorSocket}.
N)� buildReactor�
assertTruer � verifyObjectr )�self�reactor� r �C/usr/lib/python3/dist-packages/twisted/internet/test/test_socket.py�
test_provider- s z4IReactorSocketVerificationTestsBuilder.test_providerN)�__name__�
__module__� __qualname__�__doc__r �requiredInterfacesr r r r r r s r c @ �. e Zd ZdZegZdd� Zdd� Zdd� ZdS ) �!AdoptStreamPortErrorsTestsBuilderas
Builder for testing L{IReactorSocket.adoptStreamPort} implementations.
Generally only tests for failure cases are found here. Success cases for
this interface are tested elsewhere. For example, the success case for
I{AF_INET} is in L{twisted.internet.test.test_tcp}, since that case should
behave exactly the same as L{IReactorTCP.listenTCP}.
c C �r | � � }t�� }|�� }|�� | �tj|j|tjt� �}t �
� r-| � |j d t
j� dS | � |j d t
j� dS )z�
An implementation of L{IReactorSocket.adoptStreamPort} raises
L{socket.error} if passed an integer which is not associated with a
socket.
r N)r
�socket�fileno�close� assertRaises�error�adoptStreamPort�AF_INETr r
� isWindows� assertEqual�args�errno� WSAENOTSOCK�EBADF�r r �prober �excr r r �test_invalidDescriptorB � �z8AdoptStreamPortErrorsTestsBuilder.test_invalidDescriptorc C sR | � � }t�� }|�d� |�d� | �|j� d}| �t|j|� � |t
� � dS )z�
An implementation of L{IReactorSocket.adoptStreamPort} raises
L{UnsupportedAddressFamily} if passed an address family it does not
support.
�z 127.0.0.1r � � N) r
r �bind�listen�
addCleanupr r r r" r r �r r �port� arbitraryr r r �test_invalidAddressFamilyZ s
�z;AdoptStreamPortErrorsTestsBuilder.test_invalidAddressFamilyc s� �� � �t�� ����j� ��d� ��d� ��d� ����� �j t
� �}|� � � ��fdd�}� � |� � �
td� t�� �fdd�� ��� d S )
a9
When the L{IListeningPort} returned by
L{IReactorSocket.adoptStreamPort} is stopped using
C{stopListening}, the underlying socket is closed but not
shutdown. This allows another process which still has a
reference to it to continue accepting connections over it.
r/ r0 Fc sH �� tj� j�}t�� r��|jd tj � d S ��|jd tj
� d S )Nr ) r r r! �acceptr
r$ r% r& r' �WSAEWOULDBLOCK�EAGAIN��ignoredr, ��
portSocketr r r �stopped� s zFAdoptStreamPortErrorsTestsBuilder.test_stopOnlyCloses.<locals>.stoppedz"Failed to accept on original port.c � � � �fdd��S )Nc � � � � S �N��stop�r= �r r r �<lambda>� � zYAdoptStreamPortErrorsTestsBuilder.test_stopOnlyCloses.<locals>.<lambda>.<locals>.<lambda>�� addCallbackr ��dr r r rH � � zGAdoptStreamPortErrorsTestsBuilder.test_stopOnlyCloses.<locals>.<lambda>N)r
r r4 r r2 r3 � setblockingr" r �familyr �
stopListeningrK �
addErrbackr r �run�r r6 r@ r �rM r? r r r �test_stopOnlyClosesq s"
�
� z5AdoptStreamPortErrorsTestsBuilder.test_stopOnlyClosesN� r r r r r r r- r8 rV r r r r r 6 s r c @ r )�'AdoptStreamConnectionErrorsTestsBuildera}
Builder for testing L{IReactorSocket.adoptStreamConnection}
implementations.
Generally only tests for failure cases are found here. Success cases for
this interface are tested elsewhere. For example, the success case for
I{AF_INET} is in L{twisted.internet.test.test_tcp}, since that case should
behave exactly the same as L{IReactorTCP.listenTCP}.
c C s> | � � }t�� }| �|j� d}| �t|j|�� |t� � dS )z�
An implementation of L{IReactorSocket.adoptStreamConnection} raises
L{UnsupportedAddressFamily} if passed an address family it does not
support.
r1 N) r
r r4 r r r �adoptStreamConnectionr r )r r �
connectionr7 r r r r8 � s �zAAdoptStreamConnectionErrorsTestsBuilder.test_invalidAddressFamilyN)r r r r r r r8 r r r r rX � s
rX c @ r ) �#AdoptDatagramPortErrorsTestsBuilderzR
Builder for testing L{IReactorSocket.adoptDatagramPort} implementations.
c C r )z�
An implementation of L{IReactorSocket.adoptDatagramPort} raises
L{socket.error} if passed an integer which is not associated with a
socket.
r N)r
r r r r r! �adoptDatagramPortr# r r
r$ r% r& r' r( r) r* r r r r- � r. z:AdoptDatagramPortErrorsTestsBuilder.test_invalidDescriptorc C sF | � � }t�tjtj�}| �|j� d}| �t|j|� � |t
� � dS )z�
An implementation of L{IReactorSocket.adoptDatagramPort} raises
L{UnsupportedAddressFamily} if passed an address family it does not
support.
r1 N) r
r r# �
SOCK_DGRAMr4 r r r r\ r r r5 r r r r8 � s �z=AdoptDatagramPortErrorsTestsBuilder.test_invalidAddressFamilyc s� �� � �t�tjtj�����j� ��d� ��d� ���� � �j
t � �}|� � � ��fdd�}� �
|� � �td� t�� �fdd�� ��� dS ) a7
When the L{IListeningPort} returned by
L{IReactorSocket.adoptDatagramPort} is stopped using
C{stopListening}, the underlying socket is closed but not
shutdown. This allows another process which still has a
reference to it to continue reading and writing to it.
r/ Fc sJ �� tj� jd�}t�� r��|jd tj � d S ��|jd tj
� d S )Nr0 r ) r r r! �recvfromr
r$ r% r& r' r: r; r< r> r r r@ s zHAdoptDatagramPortErrorsTestsBuilder.test_stopOnlyCloses.<locals>.stoppedz Failed to read on original port.c rA )Nc rB rC rD rF rG r r rH rI z[AdoptDatagramPortErrorsTestsBuilder.test_stopOnlyCloses.<locals>.<lambda>.<locals>.<lambda>rJ r rL r r rH rN zIAdoptDatagramPortErrorsTestsBuilder.test_stopOnlyCloses.<locals>.<lambda>N)r
r r# r] r4 r r2 rO r\ r rP r rQ rK rR r r rS rT r rU r rV � s
�
� z7AdoptDatagramPortErrorsTestsBuilder.test_stopOnlyClosesNrW r r r r r[ � s r[ )r r' r �zope.interfacer �twisted.internet.errorr �twisted.internet.interfacesr �twisted.internet.protocolr r �#twisted.internet.test.reactormixinsr r �twisted.python.logr �twisted.python.runtimer
r r rX r[ �globals�update�makeTestCaseClassesr r r r �<module> s$ f#]