| 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 : /proc/self/root/lib/python3/dist-packages/twisted/test/__pycache__/ |
Upload File : |
U
W[ � @ s� d Z ddlmZ ddlmZmZ ddlmZ ddlm Z m
Z
m Z m Z ddl
mZ G dd� dej�ZG d d
� d
ej�ZG d d � d ej�ZG d
d� dej�ZG dd� dej�ZdS )z+
Test cases for twisted.protocols package.
� )�unittest)�wire� portforward)� iterbytes)�reactor�defer�address�protocol)�
proto_helpersc @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
S ) � WireTestsz
Test wire protocols.
c C s` t �� }t�� }|�|� |�d� |�d� |�d� |�d� |�d� | �|�� d� dS )zT
Test wire.Echo protocol: send some data and check it send it back.
� hello� world� how� are� yous helloworldhowareyouN)r
�StringTransportr �Echo�makeConnection� dataReceived� assertEqual�value��self�t�a� r �=/usr/lib/python3/dist-packages/twisted/test/test_protocols.py� test_echo s
zWireTests.test_echoc C s. t �� }t�� }|�|� | �|�� d� dS )z)
Test wire.Who protocol.
s root
N)r
r r ZWhor r r r r r r �test_who# s
zWireTests.test_whoc C s. t �� }t�� }|�|� | �|�� d� dS )z*
Test wire.QOTD protocol.
s' An apple a day keeps the doctor away.
N)r
r r ZQOTDr r r r r r r � test_QOTD- s
�zWireTests.test_QOTDc C s` t �� }t�� }|�|� |�d� |�d� |�d� |�d� |�d� | �|�� d� dS )z-
Test wire.Discard protocol.
r r
r r r � N)r
r r ZDiscardr r r r r r r r � test_discard8 s
zWireTests.test_discardN)�__name__�
__module__� __qualname__�__doc__r r r r! r r r r r s
r c @ s e Zd ZdZdd� ZdS )�TestableProxyClientFactoryz�
Test proxy client factory that keeps the last created protocol instance.
@ivar protoInstance: the last instance of the protocol.
@type protoInstance: L{portforward.ProxyClient}
c C s t j�| |�}|| _|S )zE
Create the protocol instance and keeps track of it.
)r �ProxyClientFactory�
buildProtocol�
protoInstance�r �addr�protor r r r( P s z(TestableProxyClientFactory.buildProtocolN�r" r# r$ r% r( r r r r r& H s r& c @ s e Zd ZdZdd� ZdS )�TestableProxyFactoryal
Test proxy factory that keeps the last created protocol instance.
@ivar protoInstance: the last instance of the protocol.
@type protoInstance: L{portforward.ProxyServer}
@ivar clientFactoryInstance: client factory used by C{protoInstance} to
create forward connections.
@type clientFactoryInstance: L{TestableProxyClientFactory}
c s. t j�� |�}t� � _� fdd�|_|� _|S )z�
Create the protocol instance, keeps track of it, and makes it use
C{clientFactoryInstance} as client factory.
c s � j S �N)�clientFactoryInstancer �r r r �<lambda>n r z4TestableProxyFactory.buildProtocol.<locals>.<lambda>)r � ProxyFactoryr( r& r0 ZclientProtocolFactoryr) r* r r1 r r( f s
z"TestableProxyFactory.buildProtocolNr- r r r r r. Z s r. c @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
S ) �PortforwardingTestsz
Test port forwarding.
c C s t �� | _t�� | _g | _d S r/ )r r �serverProtocolr ZProtocol�clientProtocol� openPortsr1 r r r �setUpy s
zPortforwardingTests.setUpc C s� z| j jj�� W n tk
r& Y nX z| j jj}|j�� W n tk
rT Y nX z| jj�� W n tk
rz Y nX z| jj�� W n tk
r� Y nX t� dd� | j
D ��S )Nc S s g | ]}t �|j��qS r )r Z
maybeDeferredZ
stopListening)�.0�pr r r �
<listcomp>� s z0PortforwardingTests.tearDown.<locals>.<listcomp>) �proxyServerFactoryr) � transportZloseConnection�AttributeErrorr0 r6 r5 r Z
gatherResultsr7 )r Zpir r r �tearDown s(
�z PortforwardingTests.tearDownc s� t �� }�fdd�|_ tjd|dd�}�j�|� td|�� j��_ tjd�j dd�}�j�|� d�g �t
� � � � ���fdd�}|�j _
��fd d
�}|�j _t �� }�fd d�|_ t�d|�� j|� � S ) z=
Test port forwarding through Echo protocol.
c s � j S r/ )r5 r r1 r r r2 � r z6PortforwardingTests.test_portforward.<locals>.<lambda>r � 127.0.0.1)Z interfacei� c s> �� t| �� t���kr:��d���d� � � �d � d S )Nr � x)�extendr �lenr �join�callback)�data��d�nBytesZreceivedr r r �testDataReceived� s z>PortforwardingTests.test_portforward.<locals>.testDataReceivedc s �j j�d� � d S )NrA )r6 r= �writer )rI r r r �testConnectionMade� s z@PortforwardingTests.test_portforward.<locals>.testConnectionMadec s � j S r/ )r6 r r1 r r r2 � r )r Z
ServerFactoryr Z listenTCPr7 �appendr. ZgetHost�portr< r ZDeferredr6 r ZconnectionMadeZ
ClientFactoryZ
connectTCP)r ZrealServerFactoryZrealServerPortZproxyServerPortrJ rL �
clientFactoryr rG r �test_portforward� s8 � �
� �z$PortforwardingTests.test_portforwardc C s� t �ddd�}t�dd��|�}t�� }||_t�� }|� |� | �
t |j �d� |j d \}}}}} | �
|tj� |�|�}
t�� } |
� | � | �| j|� | �|j| � | �| j� | �|j� dS )zm
The proxy client registers itself as a producer of the proxy server and
vice versa.
ZTCPr@ r � N)r Z IPv4Addressr r3 r( r
Z
MemoryReactorr r r r rC Z
tcpClientsZassertIsInstancer' ZassertIsZproducerZ
assertTrueZ streaming) r r+ Zserverr ZserverTransportZhostrN rO Ztimeout�_ZclientZclientTransportr r r �test_registerProducers� s
z*PortforwardingTests.test_registerProducersN)r" r# r$ r% r8 r? rP rS r r r r r4 t s
)r4 c @ s e Zd ZdZdd� ZdS )�StringTransportTestszA
Test L{proto_helpers.StringTransport} helper behaviour.
c C s t �� }| �t|jd� dS )zY
Test that L{proto_helpers.StringTransport} doesn't accept unicode data.
ZfooN)r
r Z assertRaises� TypeErrorrK )r �sr r r �test_noUnicode� s z#StringTransportTests.test_noUnicodeN)r" r# r$ r% rW r r r r rT � s rT N)r% Z
twisted.trialr Ztwisted.protocolsr r Ztwisted.python.compatr Ztwisted.internetr r r r Z twisted.testr
ZTestCaser r' r& r3 r. r4 rT r r r r �<module> s 9o