| Server IP : 209.38.156.173 / Your IP : 216.73.216.122 [ Web Server : Apache/2.4.52 (Ubuntu) System : Linux lakekumayuhotel 5.15.0-136-generic #147-Ubuntu SMP Sat Mar 15 15:53:30 UTC 2025 x86_64 User : root ( 0) PHP Version : 8.1.2-1ubuntu2.22 Disable Function : NONE Domains : 2 Domains MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /lib/python3/dist-packages/twisted/internet/test/__pycache__/ |
Upload File : |
o
�b�A � @ sL d Z ddlZddlmZ ddlmZ ddlmZmZ ddl m
Z
m Z ddl m
Z
mZ ddlmZmZmZmZ dd lmZ dd
lmZmZ dd lmZ dd lm Z dd
l m Z m Z ddl m!Z!m"Z" ddl#m$Z$ dd� Z%e%� Z&dd� Z'G dd� de�Z(G dd� d�Z)G dd� dee)e(�Z*G dd� dee)e(�Z+e,� �-e*�.� � e,� �-e+�.� � dS ) zU
Tests for implementations of L{IReactorUDP} and the UDP parts of
L{IReactorSocket}.
� N)� implementer)� verifyObject)�defer�error)� IPv4Address� IPv6Address)�Deferred�
maybeDeferred)�IListeningPort�ILoggingContext�IReactorSocket� IReactorUDP)�DatagramProtocol)�LogObserverMixin� findFreePort)�ReactorBuilder)�context)� ILogContext�err)�
GoodClient�Server)�SkipTestc C sJ d} d}zt � t j�} | �d� d}W n ty Y nw | r#| �� |S )z4Returns True if the system can bind an IPv6 address.NF)�::1r T)�socket�AF_INET6�bind�OSError�close)�sock�has_ipv6� r �@/usr/lib/python3/dist-packages/twisted/internet/test/test_udp.py� _has_ipv6! s
�r" c C s t sd| _| S )Nz.Does not work on systems without IPv6 support.)�HAS_IPV6�skip)�fr r r! �skipWithoutIPv65 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! � logPrefixJ s z[DatagramTransportTestsMixin.test_startedListeningLogMessage.<locals>.SomeProtocol.logPrefixN)�__name__�
__module__� __qualname__r* r r r r! � SomeProtocolH 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_startedListeningLogMessage@ s z;DatagramTransportTestsMixin.test_startedListeningLogMessagec sz | � � � | �� �| ��t� ��d��� j� d�}�fdd��� ��fdd�}��|� | ��� | �|f� d d � d S )
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 z Closed)c s � � � d S �N��stop��ignored�r7 r r! � stopReactor] � zNDatagramTransportTestsMixin.test_connectionLostLogMessage.<locals>.stopReactorc s � d d �= t �j���� d S r< )r �
stopListening� addCallbackr )r6 r9 rB r r! �doStopListening` s
zRDatagramTransportTestsMixin.test_connectionLostLogMessage.<locals>.doStopListeningr r/ N) r0 r1 r2 r r3 r4 �callWhenRunning�
runReactorr5 )r) r: rF r )r6 r9 r7 rB r! � test_connectionLostLogMessageS 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� transportrD r( r r r! �
startProtocolw s
zdDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocol.startProtocolc s d| _ | j| _� �� d S )NT)�stoppedrK �stoppedInStartr> r( rA r r! � stopProtocol} s zcDatagramTransportTestsMixin.test_stopProtocolScheduling.<locals>.DisconnectingProtocol.stopProtocolN) r+ r, r- rJ rN rK rO rM rP r rA r r! �DisconnectingProtocolp s rQ N) r r1 r2 rH �
assertTruerJ rN � assertFalserO )r) rQ r8 r rA r! �test_stopProtocolSchedulingi s
z7DatagramTransportTestsMixin.test_stopProtocolSchedulingN)r+ r, r- �__doc__r; rI rT 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)r1 r2 r rR r r
�r) r7 r4 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)r4 � interface�UDPN) r r �
SOCK_DGRAMr1 r2 r r5 r3 r )r) �host�
portNumberr7 r4 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 �rZ N)r1 r2 r r3 r5 r] �assertIsInstancer )r) r7 r4 �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.comr` N)r1 � assertRaisesr �InvalidAddressError� listenUDPr )r) r7 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< )rh r( r r r! r* � s zSUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.logPrefixc S s2 | j d ur| j }d | _ |�t�t�d � d S d S )Nri )ri �callbackr �getr )r) �bytesrb ri r r r! �datagramReceived� s
�zZUDPPortTestsMixin.test_logPrefix.<locals>.CustomLogPrefixDatagramProtocol.datagramReceivedN)r+ r, r- rk r* ro r r r r! � CustomLogPrefixDatagramProtocol� s rp zCustom Datagramsc s � � d| � d S )NzCustom Datagrams (UDP))r5 )ri r( r r! � gotSystem� � z3UDPPortTestsMixin.test_logPrefix.<locals>.gotSystemc � � � � S r< r= r? rA r r! �<lambda>� � z2UDPPortTestsMixin.test_logPrefix.<locals>.<lambda>s
some bytes� 127.0.0.1N) r r1 ri r2 r3 rE �
addErrbackr �writer4 rH )r) rp r8 �dr4 �addressrq r )r7 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! rk � rC zMUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.__init__c S s | j �|� d S r< )r rl )r) �datarb r r r! ro � rr zUUDPPortTestsMixin.test_writeSequence.<locals>.SimpleDatagramProtocol.datagramReceivedN)r+ r, r- rk ro r r r r! �SimpleDatagramProtocol� s r} )s somes bytess tos writec s �� d�� �| � d S )N� )r5 �join)r| )� dataToWriter) r r! �gotData� s z5UDPPortTestsMixin.test_writeSequence.<locals>.gotDatac rs r< r= r? rA r r! rt � ru z6UDPPortTestsMixin.test_writeSequence.<locals>.<lambda>rv N) r r1 r r2 r3 rE rw r �
writeSequencer4 rH )r) r} r8 r r4 rz r� r )r� r7 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)r1 r2 r �assertIn�strr3 r4 rW r r r! �test_str � zUDPPortTestsMixin.test_strc C s4 | � � }| �|t� �}| �t|�� j�t|�� dS )zR
C{repr()} on the listening port object includes the port number.
N)r1 r2 r r� �reprr3 r4 r� rW r r r! � test_repr r� zUDPPortTestsMixin.test_reprc �� | � � �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 r` 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 )rL rx r3 r4 r r �packetReceived�r@ �serverReceived��client�serverr r! �cbClientStarted s zDUDPPortTestsMixin.test_writeToIPv6Interface.<locals>.cbClientStartedc � � � � dS �z�
Stop the reactor after a datagram is received.
@param ignored: L{None}, which is ignored
@returns: L{None}
Nr= r? rA r r! �cbServerReceived, s zEUDPPortTestsMixin.test_writeToIPv6Interface.<locals>.cbServerReceivedr r� N�r1 r r r �startedDeferredr2 r rL r3 �
gatherResultsrE rw r rH �packetsr5 r] r4 �r) �
serverStarted�
clientStarted�cAddrr� r� ry �packetr �r� r7 r� r! �test_writeToIPv6Interface s"
z+UDPPortTestsMixin.test_writeToIPv6Interfacec r� )
z�
An IPv6 address can be passed as the C{interface} argument to
L{listenUDP}. The resulting Port accepts IPv6 datagrams.
r r` c s4 � j �d�j �� j� � j �d� t�� }�_|S )r� r r� )rL �connectr3 r4 rx r r r� r� r� r r! r� N s zMUDPPortTestsMixin.test_connectedWriteToIPv6Interface.<locals>.cbClientStartedc r� r� r= r? rA r r! r� \ s zNUDPPortTestsMixin.test_connectedWriteToIPv6Interface.<locals>.cbServerReceivedr r� Nr� r� r r� r! �"test_connectedWriteToIPv6Interface> s"
z4UDPPortTestsMixin.test_connectedWriteToIPv6Interfacec C �. | � � }| �|t� �}| �tj|jdd� dS )zn
Writing to a hostname instead of an IP address will raise an
L{InvalidAddressError}.
�spam)�example.invalid� N�r1 r2 r rd r re rx rW r r r! �/test_writingToHostnameRaisesInvalidAddressErroro s
�zAUDPPortTestsMixin.test_writingToHostnameRaisesInvalidAddressErrorc C �2 | � � }| j|t� dd�}| �tj|jdd� dS )�l
Writing to an IPv6 address on an IPv4 socket will raise an
L{InvalidAddressError}.
rv r` r� )r r� Nr� rW r r r! �1test_writingToIPv6OnIPv4RaisesInvalidAddressErrorz s zCUDPPortTestsMixin.test_writingToIPv6OnIPv4RaisesInvalidAddressErrorc C r� )r� r r` r� )rv r� Nr� rW r r r! �1test_writingToIPv4OnIPv6RaisesInvalidAddressError� s
�zCUDPPortTestsMixin.test_writingToIPv4OnIPv6RaisesInvalidAddressErrorc C r� )zq
Connecting to a hostname instead of an IP address will raise an
L{InvalidAddressError}.
r� r� N)r1 r2 r rd r re r� rW 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)r1 r2 r �setBroadcastAllowedrR �getBroadcastAllowedrW r r r! �test_allowBroadcast� s
z%UDPPortTestsMixin.test_allowBroadcastN)r+ r, r- rU rX r_ r& rc rg r{ r� r� r� r� r� r� r� r� r� r� r r r r! rV � s,
$
,
0
rV c @ �" e Zd ZdZefZ ddd�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.
)rZ �
maxPacketSize)rf )r) r7 r8 r4 rZ r� r r r! r2 � s
�z&UDPServerTestsBuilder.getListeningPortN�r r� r� )r+ r, r- rU r
�requiredInterfacesr2 r r r r! r� � s
�r� c @ r� ) �UDPFDServerTestsBuilderzC
Run L{UDPPortTestsMixin} tests using adopted UDP sockets.
r r� r� c C s� t �|�rMd|v rtj}t�||�d d }ntj}||f}t�|tj�}|�|� |�d� z|� |�
� |j ||�W |�
� |� � S |�
� |� � w t
d��)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 IReactorSocket)r �
providedByr r � getaddrinfo�AF_INETr\ r � setblocking�adoptDatagramPort�fileno�familyr r ) r) r7 r8 r4 rZ r� �domainrz �portSockr r r! r2 � s$
�
�
z(UDPFDServerTestsBuilder.getListeningPortNr� )r+ r, r- rU r r� r2 r r r r! r� � s
�r� )/rU r �zope.interfacer �zope.interface.verifyr �twisted.internetr r �twisted.internet.addressr r �twisted.internet.deferr r �twisted.internet.interfacesr
r r r
�twisted.internet.protocolr �&twisted.internet.test.connectionmixinsr r �#twisted.internet.test.reactormixinsr �twisted.pythonr �twisted.python.logr r �twisted.test.test_udpr r �twisted.trial.unittestr r" r# r& r' rV r� r� �globals�update�makeTestCaseClassesr r r r! �<module> s< Q
�
�9