| Server IP : 138.197.176.125 / Your IP : 216.73.217.54 Web Server : Apache/2.4.41 (Ubuntu) System : Linux SuiteCRM-8 5.4.0-216-generic #236-Ubuntu SMP Fri Apr 11 19:53:21 UTC 2025 x86_64 User : root ( 0) PHP Version : 8.3.19 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : ON Directory : /lib/python3/dist-packages/twisted/internet/test/__pycache__/ |
Upload File : |
U
W[�B � @ sb d Z ddlmZmZ eZddlZddlmZ ddl m
Z
ddl m Z ddl
mZmZ ddlmZ dd lmZmZ dd
lmZmZmZmZ dd lmZm Z dd l m Z dd
l m Z m!Z! ddl"m#Z#m$Z$ ddl%m&Z&m'Z' ddl(m)Z) dd� Z*e*� Z+dd� Z,G dd� de �Z-G dd� de.�Z/G dd� dee/e-�Z0G dd � d ee/e-�Z1e2� �3e0�4� � e2� �3e1�4� � dS ) zU
Tests for implementations of L{IReactorUDP} and the UDP parts of
L{IReactorSocket}.
� )�division�absolute_importN)� implementer)� verifyObject)�context)� ILogContext�err)�ReactorBuilder)�Deferred�
maybeDeferred)�ILoggingContext�IListeningPort� IReactorUDP�IReactorSocket)� IPv4Address� IPv6Address)�DatagramProtocol)�LogObserverMixin� findFreePort)�defer�error)�Server�
GoodClient)�SkipTestc C sN d} d}z t � t j�} | �d� d}W n t jk
r< Y nX | rJ| �� |S )z5 Returns True if the system can bind an IPv6 address.NF)�::1r T)�socket�AF_INET6�bindr �close)ZsockZhas_ipv6� r �@/usr/lib/python3/dist-packages/twisted/internet/test/test_udp.py� _has_ipv6" s
r! c C s t s
d| _| S )Nz.Does not work on systems without IPv6 support.)�HAS_IPV6�skip)�fr r r �skipWithoutIPv66 s r% c @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) �DatagramTransportTestsMixinzP
Mixin defining tests which apply to any port/datagram based transport.
c C sd | � � }| �� }tt�G dd� dt��}|� }| �||�}d|�� jf }| �|f|d d � dS )zu
When a port starts, a message including a description of the associated
protocol is logged.
c @ s e Zd Zdd� ZdS )zQDatagramTransportTestsMixin.test_startedListeningLogMessage.<locals>.SomeProtocolc S s dS )NzCrazy Protocolr ��selfr r r � logPrefixK s z[DatagramTransportTestsMixin.test_startedListeningLogMessage.<locals>.SomeProtocol.logPrefixN)�__name__�
__module__� __qualname__r) r r r r � SomeProtocolI s r- z Crazy Protocol starting on %dr �messageN) �observe� buildReactorr r r �getListeningPort�getHost�port� assertEqual)r( �loggedMessages�reactorr- �protocol�p�expectedMessager r r � test_startedListeningLogMessageA s z;DatagramTransportTestsMixin.test_startedListeningLogMessagec sx | � � � | �� �| ��t� ��d��� jf }�fdd��� ��fdd�}��|� | ��� | �|f� d d � dS ) z�
When a connection is lost a message is logged containing an
address identifying the port and the fact that it was closed.
z(UDP Port %s Closed)c s � � � d S �N��stop��ignored�r6 r r � stopReactor^ s zNDatagramTransportTestsMixin.test_connectionLostLogMessage.<locals>.stopReactorc s � d d �= t �j���� d S r; )r �
stopListening� addCallbackr )r5 r8 rA r r �doStopListeninga s
zRDatagramTransportTestsMixin.test_connectionLostLogMessage.<locals>.doStopListeningr r. N) r/ r0 r1 r r2 r3 ZcallWhenRunning�
runReactorr4 )r( r9 rD r )r5 r8 r6 rA r � test_connectionLostLogMessageT s
z9DatagramTransportTestsMixin.test_connectionLostLogMessagec s` G � fdd�dt �}| �� � |� }| �� |� | �� � | �|j� | �|j� | �|j� dS )z�
L{DatagramProtocol.stopProtocol} is called asynchronously (ie, not
re-entrantly) when C{stopListening} is used to stop the datagram
transport.
c s0 e Zd ZdZdZdZdZdd� Z� fdd�ZdS )zVDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocolFc S s d| _ d| _| j�� d| _d S )NTF)�started�inStartProtocol� transportrB r' r r r �
startProtocolx s
zdDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocol.startProtocolc s d| _ | j| _� �� d S )NT)�stoppedrH �stoppedInStartr= r' r@ r r � stopProtocol~ s zcDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocol.stopProtocolN) r* r+ r, rG rK rH rL rJ rM r r@ r r �DisconnectingProtocolq s rN N) r r0 r1 rE �
assertTruerG rK Z assertFalserL )r( rN r7 r r@ r �test_stopProtocolSchedulingk s
z7DatagramTransportTestsMixin.test_stopProtocolSchedulingN)r* r+ r, �__doc__r: rF rP r r r r r& = s r& c @ s� e Zd ZdZdd� Zdd� Zedd� �Zdd � Zd
d � Z d d
� Z
dd� Z dd� Z edd� �Z
edd� �Zdd� Zedd� �Zedd� �Zd d � Zd d � Zd S )!�UDPPortTestsMixinzY
Tests for L{IReactorUDP.listenUDP} and
L{IReactorSocket.adoptDatagramPort}.
c C s* | � � }| �|t� �}| �tt|�� dS )zY
L{IReactorUDP.listenUDP} returns an object providing L{IListeningPort}.
N)r0 r1 r rO r r
�r( r6 r3 r r r �test_interface� s z UDPPortTestsMixin.test_interfacec C sH t tjd�\}}| �� }| j|t� ||d�}| �|�� td||�� dS )z�
L{IListeningPort.getHost} returns an L{IPv4Address} giving a
dotted-quad of the IPv4 address the port is listening on as well as
the port number.
)�type)r3 � interfaceZUDPN) r r �
SOCK_DGRAMr0 r1 r r4 r2 r )r( �hostZ
portNumberr6 r3 r r r � test_getHost� s �
�z UDPPortTestsMixin.test_getHostc C s@ | � � }| j|t� dd�}|�� }| �|jd� | �|t� dS )zr
L{IListeningPort.getHost} returns an L{IPv6Address} when listening on
an IPv6 interface.
r �rV N)r0 r1 r r2 r4 rX ZassertIsInstancer )r( r6 r3 �addrr r r �test_getHostIPv6� s �z"UDPPortTestsMixin.test_getHostIPv6c C s& | � � }| jtj|jt� ddd� dS )z�
An L{InvalidAddressError} is raised when trying to listen on an address
that isn't a valid IPv4 or IPv6 address.
r z example.comrZ N)r0 � assertRaisesr �InvalidAddressError� listenUDPr )r( r6 r r r �test_invalidInterface� s �z'UDPPortTestsMixin.test_invalidInterfacec s� G dd� dt �}��� � |d�}|j}��� |�}|�� }�fdd�}|�|� |�t� |�� fdd�� |�dd |j f� ��
� � d
S ) z�
Datagram transports implement L{ILoggingContext.logPrefix} to return a
message reflecting the protocol they are running.
c @ s$ e Zd Zdd� Zdd� Zdd� ZdS )zIUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocolc S s || _ t� | _d S r; )�_prefixr
�system)r( �prefixr r r �__init__� s zRUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.__init__c S s | j S r; )ra r' r r r r) � s zSUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.logPrefixc S s. | j d k r*| j }d | _ |�t�t�d � d S )Nrb )rb �callbackr �getr )r( �bytesr[ rb r r r �datagramReceived� s
zZUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.datagramReceivedN)r* r+ r, rd r) rh r r r r � CustomLogPrefixDatagramProtocol� s ri zCustom Datagramsc s � � d| � d S )NzCustom Datagrams (UDP))r4 )rb r' r r � gotSystem� s z3UDPPortTestsMixin.test_logPrefix.<locals>.gotSystemc s � � � S r; r< r> r@ r r �<lambda>� � z2UDPPortTestsMixin.test_logPrefix.<locals>.<lambda>s
some bytes� 127.0.0.1N) r r0 rb r1 r2 rC �
addErrbackr �writer3 rE )r( ri r7 �dr3 �addressrj r )r6 r( r �test_logPrefix� s
z UDPPortTestsMixin.test_logPrefixc s� G dd� dt �}��� �|� }|j}���|�}|�� }d� � �fdd�}|�|� |�t� |��fdd�� |�� d|j f� ��
�� d S )
zH
Write a sequence of L{bytes} to a L{DatagramProtocol}.
c @ s e Zd Zdd� Zdd� ZdS )zDUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocolc S s t � | _d S r; )r
r r' r r r rd � s zMUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.__init__c S s | j �|� d S r; )r re )r( �datar[ r r r rh � s zUUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.datagramReceivedN)r* r+ r, rd rh r r r r �SimpleDatagramProtocol� s rt )s somes bytess tos writec s �� d�� �| � d S )Nrl )r4 �join)rs )� dataToWriter( r r �gotData� s z5UDPPortTestsMixin.test_writeSequence.<locals>.gotDatac s � � � S r; r< r> r@ r r rk � rl z6UDPPortTestsMixin.test_writeSequence.<locals>.<lambda>rm N) r r0 r r1 r2 rC rn r Z
writeSequencer3 rE )r( rt r7 r r3 rq rw r )rv r6 r( r �test_writeSequence� s
z$UDPPortTestsMixin.test_writeSequencec C s4 | � � }| �|t� �}| �t|�� j�t|�� dS )zQ
C{str()} on the listening port object includes the port number.
N)r0 r1 r �assertIn�strr2 r3 rS r r r �test_str s zUDPPortTestsMixin.test_strc C s4 | � � }| �|t� �}| �t|�� j�t|�� dS )zR
C{repr()} on the listening port object includes the port number.
N)r0 r1 r ry �reprr2 r3 rz rS r r r � test_repr s zUDPPortTestsMixin.test_reprc s� | � � �t� �t�� }�_| j��dd� t� � t�� }� _| j�� dd� � j�� }� �fdd�}�fdd�}t� ||g�}|�
|� |�
|� |� t � | �
�� �jd }| �|d|j|jff� d S )
zS
Writing to an IPv6 UDP socket on the loopback interface succeeds.
r rZ c s, � j �dd�j �� jf� t�� }�_|S )��
Send a datagram from the client once it's started.
@param ignored: a list of C{[None, None]}, which is ignored
@returns: a deferred which fires when the server has received a
datagram.
� spamr )rI ro r2 r3 r r
�packetReceived�r? ZserverReceived��client�serverr r �cbClientStarted# s �zDUDPPortTestsMixin.test_writeToIPv6Interface.<locals>.cbClientStartedc s � � � dS �z�
Stop the reactor after a datagram is received.
@param ignored: L{None}, which is ignored
@returns: L{None}
Nr< r> r@ r r �cbServerReceived0 s zEUDPPortTestsMixin.test_writeToIPv6Interface.<locals>.cbServerReceivedr r N�r0 r r r
ZstartedDeferredr1 r rI r2 Z
gatherResultsrC rn r rE Zpacketsr4 rX r3 �r( Z
serverStartedZ
clientStartedZcAddrr� r� rp Zpacketr �r� r6 r� r �test_writeToIPv6Interface s"
z+UDPPortTestsMixin.test_writeToIPv6Interfacec s� | � � �t� �t�� }�_| j��dd� t� � t�� }� _| j�� dd� � j�� }� �fdd�}�fdd�}t� ||g�}|�
|� |�
|� |� t � | �
�� �jd }| �|d|j|jff� d S )
z�
An IPv6 address can be passed as the C{interface} argument to
L{listenUDP}. The resulting Port accepts IPv6 datagrams.
r rZ c s4 � j �d�j �� j� � j �d� t�� }�_|S )r~ r r )rI �connectr2 r3 ro r r
r� r� r� r r r� S s zMUDPPortTestsMixin.test_connectedWriteToIPv6Interface.<locals>.cbClientStartedc s � � � dS r� r< r> r@ r r r� a s zNUDPPortTestsMixin.test_connectedWriteToIPv6Interface.<locals>.cbServerReceivedr r Nr� r� r r� r �"test_connectedWriteToIPv6InterfaceC s"
z4UDPPortTestsMixin.test_connectedWriteToIPv6Interfacec C s. | � � }| �|t� �}| �tj|jdd� dS )zn
Writing to a hostname instead of an IP address will raise an
L{InvalidAddressError}.
�spam)�example.invalid� N�r0 r1 r r] r r^ ro rS r r r �/test_writingToHostnameRaisesInvalidAddressErroru s �zAUDPPortTestsMixin.test_writingToHostnameRaisesInvalidAddressErrorc C s2 | � � }| j|t� dd�}| �tj|jdd� dS )�l
Writing to an IPv6 address on an IPv4 socket will raise an
L{InvalidAddressError}.
rm rZ r� )r r� Nr� rS r r r �1test_writingToIPv6OnIPv4RaisesInvalidAddressError� s � �zCUDPPortTestsMixin.test_writingToIPv6OnIPv4RaisesInvalidAddressErrorc C s2 | � � }| j|t� dd�}| �tj|jdd� dS )r� r rZ r� )rm r� Nr� rS r r r �1test_writingToIPv4OnIPv6RaisesInvalidAddressError� s � �zCUDPPortTestsMixin.test_writingToIPv4OnIPv6RaisesInvalidAddressErrorc C s. | � � }| �|t� �}| �tj|jdd� dS )zq
Connecting to a hostname instead of an IP address will raise an
L{InvalidAddressError}.
r� r� N)r0 r1 r r] r r^ r� rS r r r �2test_connectingToHostnameRaisesInvalidAddressError� s �zDUDPPortTestsMixin.test_connectingToHostnameRaisesInvalidAddressErrorc C s2 | � � }| �|t� �}|�d� | �|�� � dS )zk
L{IListeningPort.setBroadcastAllowed} sets broadcast to be allowed
on the socket.
TN)r0 r1 r ZsetBroadcastAllowedrO ZgetBroadcastAllowedrS r r r �test_allowBroadcast� s
z%UDPPortTestsMixin.test_allowBroadcastN)r* r+ r, rQ rT rY r% r\ r` rr rx r{ r} r� r� r� r� r� r� r� r r r r rR � s*
#
.
1
rR c @ s e Zd ZdZefZddd�ZdS ) �UDPServerTestsBuilderzM
Run L{UDPPortTestsMixin} tests using newly created UDP
sockets.
r � � c C s |j ||||d�S )aB
Get a UDP port from a reactor.
@param reactor: A reactor used to build the returned
L{IListeningPort} provider.
@type reactor: L{twisted.internet.interfaces.IReactorUDP}
@see: L{twisted.internet.IReactorUDP.listenUDP} for other
argument and return types.
)rV �
maxPacketSize)r_ )r( r6 r7 r3 rV r� r r r r1 � s
�z&UDPServerTestsBuilder.getListeningPortN)r r� r� )r* r+ r, rQ r �requiredInterfacesr1 r r r r r� � s
�r� c @ s e Zd ZdZefZddd�ZdS ) �UDPFDServerTestsBuilderzC
Run L{UDPPortTestsMixin} tests using adopted UDP sockets.
r r� r� c C s� t �|�r�d|kr.tj}t�||�d d }ntj}||f}t�|tj�}|�|� |�d� z|� |� � |j ||�W �S |� � |�
� X nt
d��dS )a�
Get a UDP port from a reactor, wrapping an already-initialized file
descriptor.
@param reactor: A reactor used to build the returned
L{IListeningPort} provider.
@type reactor: L{twisted.internet.interfaces.IReactorSocket}
@param port: A port number to which the adopted socket will be
bound.
@type port: C{int}
@param interface: The local IPv4 or IPv6 address to which the
adopted socket will be bound. defaults to '', ie all IPv4
addresses.
@type interface: C{str}
@see: L{twisted.internet.IReactorSocket.adoptDatagramPort} for other
argument and return types.
�:r � Fz'Reactor does not provide IReactorSocketN)r Z
providedByr r Z getaddrinfoZAF_INETrW r Z setblocking�filenor ZadoptDatagramPortZfamilyr ) r( r6 r7 r3 rV r� Zdomainrq ZportSockr r r r1 � s&
� z(UDPFDServerTestsBuilder.getListeningPortN)r r� r� )r* r+ r, rQ r r� r1 r r r r r� � s
�r� )5rQ Z
__future__r r rU Z
__metaclass__r Zzope.interfacer Zzope.interface.verifyr Ztwisted.pythonr Ztwisted.python.logr r Z#twisted.internet.test.reactormixinsr Ztwisted.internet.deferr
r Ztwisted.internet.interfacesr r
r r Ztwisted.internet.addressr r Ztwisted.internet.protocolr Z&twisted.internet.test.connectionmixinsr r Ztwisted.internetr r Ztwisted.test.test_udpr r Ztwisted.trial.unittestr r! r" r% r& �objectrR r� r� �globals�updateZmakeTestCaseClassesr r r r �<module> sB Q & � �7