File: //usr/lib/python3/dist-packages/twisted/internet/test/__pycache__/test_protocol.cpython-310.pyc
o
�bXH � @ s� d Z ddlmZ ddlmZ ddlmZ ddlmZ ddl m
Z
m Z m Z m
Z
ddlmZmZmZmZmZmZ ddlmZmZ dd lmZ dd
lmZm Z dd l m Z G d d
� d
e �Z G dd� de �Z G dd� de �Z!G dd� de �Z"G dd� de �Z#dS )z)
Tests for L{twisted.internet.protocol}.
� )�BytesIO)� implementer)� verifyObject)�CancelledError)� IConsumer�ILoggingContext� IProtocol�IProtocolFactory)�
ClientCreator�ConsumerToProtocolAdapter�Factory� FileWrapper�Protocol�ProtocolToConsumerAdapter)�LogLevel�globalLogPublisher)�Failure)�MemoryReactorClock�StringTransport)�TestCasec @ s� e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
dd� Z dd� Z dd� Z
dd� Zdd� Zdd� Zd d � Zd d � Zd d!� Zd"S )#�ClientCreatorTestsz?
Tests for L{twisted.internet.protocol.ClientCreator}.
c C sD G dd� dt �}t� }t||�}|||�}|�d�}| �||� dS )a
Helper for implementing a test to verify that one of the I{connect}
methods of L{ClientCreator} passes the right arguments to the right
reactor method.
@param check: A function which will be invoked with a reactor and a
L{ClientCreator} instance and which should call one of the
L{ClientCreator}'s I{connect} methods and assert that all of its
arguments except for the factory are passed on as expected to the
reactor. The factory should be returned.
c @ � e Zd ZdS )z:ClientCreatorTests._basicConnectTest.<locals>.SomeProtocolN��__name__�
__module__� __qualname__� r r �E/usr/lib/python3/dist-packages/twisted/internet/test/test_protocol.py� SomeProtocol5 � r N)r r r
�
buildProtocol�assertIsInstance)�self�checkr �reactor�cc�factory�protocolr r r �_basicConnectTest( s
z$ClientCreatorTests._basicConnectTestc � � fdd�}� � |� dS )z�
L{ClientCreator.connectTCP} calls C{reactor.connectTCP} with the host
and port information passed to it, and with a factory which will
construct the protocol passed to L{ClientCreator.__init__}.
c sX |� dddd� | j�� \}}}}}� �|d� � �|d� � �|d� � �|d� |S )N� example.com�� �� )z1.2.3.4i�& )�
connectTCP�
tcpClients�pop� assertEqual)r$ r% �host�portr& �timeout� bindAddress�r" r r r# E s z1ClientCreatorTests.test_connectTCP.<locals>.checkN�r( �r" r# r r5 r �test_connectTCP> s z"ClientCreatorTests.test_connectTCPc r) )z�
L{ClientCreator.connectUNIX} calls C{reactor.connectUNIX} with the
filename passed to it, and with a factory which will construct the
protocol passed to L{ClientCreator.__init__}.
c sF |� ddd� | j�� \}}}}� �|d� � �|d� � �|� |S )N�/foo/bar�{ T)� connectUNIX� unixClientsr/ r0 �
assertTrue)r$ r% �addressr& r3 �checkPIDr5 r r r# W s
z2ClientCreatorTests.test_connectUNIX.<locals>.checkNr6 r7 r r5 r �test_connectUNIXP s z#ClientCreatorTests.test_connectUNIXc r) )z�
L{ClientCreator.connectSSL} calls C{reactor.connectSSL} with the host,
port, and context factory passed to it, and with a factory which will
construct the protocol passed to L{ClientCreator.__init__}.
c sn t � }|�dd|dd� | j�� \}}}}}}� �|d� � �|d� � �||� � �|d� � �|d� |S )Nr* r+ r, )z4.3.2.1i. )�object�
connectSSL�
sslClientsr/ r0 �assertIs) r$ r% �expectedContextFactoryr1 r2 r& �contextFactoryr3 r4 r5 r r r# h s$
�
� z1ClientCreatorTests.test_connectSSL.<locals>.checkNr6 r7 r r5 r �test_connectSSLa s z"ClientCreatorTests.test_connectSSLc C sN t � }t|t�}||�}|j�� }| �|j� |�� | �|j� | � |t
�S )a
Helper for implementing a test to verify that cancellation of the
L{Deferred} returned by one of L{ClientCreator}'s I{connect} methods is
implemented to cancel the underlying connector.
@param connect: A function which will be invoked with a L{ClientCreator}
instance as an argument and which should call one its I{connect}
methods and return the result.
@return: A L{Deferred} which fires when the test is complete or fails if
there is a problem.
) r r
r �
connectorsr/ � assertFalse�
_disconnected�cancelr= �
assertFailurer �r" �connectr$ r% �d� connectorr r r �_cancelConnectTest~ s
z%ClientCreatorTests._cancelConnectTestc C � dd� }| � |�S )��
The L{Deferred} returned by L{ClientCreator.connectTCP} can be cancelled
to abort the connection attempt before it completes.
c S s | � dd�S �Nr* r+ )r- �r% r r r rN � s z9ClientCreatorTests.test_cancelConnectTCP.<locals>.connect�rQ �r" rN r r r �test_cancelConnectTCP� �
z(ClientCreatorTests.test_cancelConnectTCPc C rR )rS c S s
| � d�S �Nr9 )r; rU r r r rN � s
z:ClientCreatorTests.test_cancelConnectUNIX.<locals>.connectrV rW r r r �test_cancelConnectUNIX� rY z)ClientCreatorTests.test_cancelConnectUNIXc C rR )rS c S s | � ddt� �S rT )rB rA rU r r r rN � � z9ClientCreatorTests.test_cancelConnectSSL.<locals>.connectrV rW r r r �test_cancelConnectSSL� rY z(ClientCreatorTests.test_cancelConnectSSLc C sh t � }t|t�}|||�}|j�� }| �t|�� �d� |�� | �|�� g � | � |j
� | � |t �S )z�
Like L{_cancelConnectTest}, but for the case where the L{Deferred} is
cancelled after the connection is set up but before it is fired with the
resulting protocol instance.
� )
r r
r rH r/ r0 �len�getDelayedCallsrK r= rJ rL r rM r r r �_cancelConnectTimeoutTest� s
z,ClientCreatorTests._cancelConnectTimeoutTestc C rR )a�
L{ClientCreator.connectTCP} inserts a very short delayed call between
the time the connection is established and the time the L{Deferred}
returned from one of its connect methods actually fires. If the
L{Deferred} is cancelled in this interval, the established connection is
closed, the timeout is cancelled, and the L{Deferred} fails with
L{CancelledError}.
c
S s> |� dd�}| j�� \}}}}}|�d �}t� } |�| � |S rT )r- r. r/ r r �makeConnection)
r$ r% rO r1 r2 r& r3 r4 r' � transportr r r rN � s
z@ClientCreatorTests.test_cancelConnectTCPTimeout.<locals>.connect�ra rW r r r � test_cancelConnectTCPTimeout� �
z/ClientCreatorTests.test_cancelConnectTCPTimeoutc C rR )a�
L{ClientCreator.connectUNIX} inserts a very short delayed call between
the time the connection is established and the time the L{Deferred}
returned from one of its connect methods actually fires. If the
L{Deferred} is cancelled in this interval, the established connection is
closed, the timeout is cancelled, and the L{Deferred} fails with
L{CancelledError}.
c S s: |� d�}| j�� \}}}}|�d �}t� }|�|� |S rZ )r; r< r/ r r rb ) r$ r% rO r> r& r3 r4 r' rc r r r rN � s
zAClientCreatorTests.test_cancelConnectUNIXTimeout.<locals>.connectrd rW r r r � test_cancelConnectUNIXTimeout� rf z0ClientCreatorTests.test_cancelConnectUNIXTimeoutc C rR )a�
L{ClientCreator.connectSSL} inserts a very short delayed call between
the time the connection is established and the time the L{Deferred}
returned from one of its connect methods actually fires. If the
L{Deferred} is cancelled in this interval, the established connection is
closed, the timeout is cancelled, and the L{Deferred} fails with
L{CancelledError}.
c S sD |� ddt� �}| j�� \}}}}}}|�d �} t� }
| �|
� |S rT )rB rA rC r/ r r rb ) r$ r% rO r1 r2 r& rF r3 � bindADdressr' rc r r r rN s �
z@ClientCreatorTests.test_cancelConnectSSLTimeout.<locals>.connectrd rW r r r � test_cancelConnectSSLTimeout� s
z/ClientCreatorTests.test_cancelConnectSSLTimeoutc C st t � }t|t�}|||�\}}|j�� }|�|ttd��� | �t |�
� �d� |� � | �|�
� g � | � |t
�S )z�
Like L{_cancelConnectTest}, but for the case where the L{Deferred} is
cancelled after the connection attempt has failed but before it is fired
with the resulting failure.
zSimulated failurer^ )r r
r rH r/ �clientConnectionFailedr � Exceptionr0 r_ r` rK rL r )r" rN r$ r% rO r& rP r r r � _cancelConnectFailedTimeoutTest s
� z2ClientCreatorTests._cancelConnectFailedTimeoutTestc C rR )zz
Similar to L{test_cancelConnectTCPTimeout}, but for the case where the
connection attempt fails.
c S s( |� dd�}| j�� \}}}}}||fS rT )r- r. r/ )r$ r% rO r1 r2 r& r3 r4 r r r rN 2 s zFClientCreatorTests.test_cancelConnectTCPFailedTimeout.<locals>.connect�rl rW r r r �"test_cancelConnectTCPFailedTimeout, �
z5ClientCreatorTests.test_cancelConnectTCPFailedTimeoutc C rR )z{
Similar to L{test_cancelConnectUNIXTimeout}, but for the case where the
connection attempt fails.
c S s$ |� d�}| j�� \}}}}||fS rZ )r; r< r/ )r$ r% rO r> r& r3 r4 r r r rN ? s
zGClientCreatorTests.test_cancelConnectUNIXFailedTimeout.<locals>.connectrm rW r r r �#test_cancelConnectUNIXFailedTimeout9 ro z6ClientCreatorTests.test_cancelConnectUNIXFailedTimeoutc C rR )zz
Similar to L{test_cancelConnectSSLTimeout}, but for the case where the
connection attempt fails.
c S s. |� ddt� �}| j�� \}}}}}}||fS rT )rB rA rC r/ ) r$ r% rO r1 r2 r& rF r3 rh r r r rN L s �zFClientCreatorTests.test_cancelConnectSSLFailedTimeout.<locals>.connectrm rW r r r �"test_cancelConnectSSLFailedTimeoutF s
z5ClientCreatorTests.test_cancelConnectSSLFailedTimeoutN)r r r �__doc__r( r8 r@ rG rQ rX r[ r] ra re rg ri rl rn rp rq r r r r r # s$
r c @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) �
ProtocolTestsz:
Tests for L{twisted.internet.protocol.Protocol}.
c C �* t � }| �tt|�� | �tt|�� dS )zT
L{Protocol} instances provide L{IProtocol} and L{ILoggingContext}.
N)r r= r r r )r" �protor r r �test_interfaces` s z ProtocolTests.test_interfacesc C �& G dd� dt �}| �d|� �� � dS )zJ
L{Protocol.logPrefix} returns the protocol class's name.
c @ r )z/ProtocolTests.test_logPrefix.<locals>.SomeThingNr r r r r � SomeThingm r rx N)r r0 � logPrefix)r" rx r r r �test_logPrefixh � z ProtocolTests.test_logPrefixc s@ g � G � fdd�dt �}t� }|� }|�|� | �� |g� dS )zz
L{Protocol.makeConnection} sets the given transport on itself, and
then calls C{connectionMade}.
c � e Zd Z� fdd�ZdS )z7ProtocolTests.test_makeConnection.<locals>.SomeProtocolc s � � | j� d S �N)�appendrc r5 ��resultr r �connectionMadez r\ zFProtocolTests.test_makeConnection.<locals>.SomeProtocol.connectionMadeN)r r r r� r r r r r y s r N)r rA rb r0 )r" r rc r' r r r �test_makeConnectionr s
z!ProtocolTests.test_makeConnectionN)r r r rr rv rz r� r r r r rs [ s
rs c @ s@ e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dS )� FactoryTestsz(
Tests for L{protocol.Factory}.
c C rt )zg
L{Factory} instances provide both L{IProtocolFactory} and
L{ILoggingContext}.
N)r r= r r r )r" r& r r r rv � s z FactoryTests.test_interfacesc C rw )zM
L{Factory.logPrefix} returns the name of the factory class.
c @ r )z6FactoryTests.test_logPrefix.<locals>.SomeKindOfFactoryNr r r r r �SomeKindOfFactory� r r� N)r r0 ry )r" r� r r r rz � r{ zFactoryTests.test_logPrefixc C sD G dd� dt �}t� }||_|�d�}| �||� | �|j|� dS )z�
L{Factory.buildProtocol} by default constructs a protocol by calling
its C{protocol} attribute, and attaches the factory to the result.
c @ r )z<FactoryTests.test_defaultBuildProtocol.<locals>.SomeProtocolNr r r r r r � r r N)r r r' r r! rD r& )r" r �fr' r r r �test_defaultBuildProtocol� s
z&FactoryTests.test_defaultBuildProtocolc C sT G dd� dt �}|jtdddd�}| �|jt� | �|jd� | �|jddi� d S )
z�
L{Factory.forProtocol} constructs a Factory, passing along any
additional arguments, and sets its C{protocol} attribute to the given
Protocol subclass.
c @ s e Zd Zdd� ZdS )z7FactoryTests.test_forProtocol.<locals>.ArgTakingFactoryc _ s ||| _ | _d S r} )�args�kwargs)r" r� r� r r r �__init__� s z@FactoryTests.test_forProtocol.<locals>.ArgTakingFactory.__init__N)r r r r� r r r r �ArgTakingFactory� s r� r^ � � )�foo)r^ r� r� N)r � forProtocolr r0 r' r� r� )r" r� r& r r r �test_forProtocol� s
z FactoryTests.test_forProtocolc sr g � t �� j� | �� fdd�� t� }|�� | �� d d |� | �� d d tj � | �� d d d� dS ) z�
L{Factory.doStart} logs that it is starting a factory, followed by
the L{repr} of the L{Factory} instance that is being started.
c � t �� j�S r} �r �removeObserverr~ r ��eventsr r �<lambda>� � z;FactoryTests.test_doStartLoggingStatement.<locals>.<lambda>r r& � log_level�
log_formatz Starting factory {factory!r}N)
r � addObserverr~ �
addCleanupr �doStartrD r0 r �info)r" r� r r� r � test_doStartLoggingStatement� s z)FactoryTests.test_doStartLoggingStatementc s� g � t �� j� | �� fdd�� G dd� dt�}|� }|�� | �� d d |� | �� d d tj � | �� d d d � d
S ) z�
L{Factory.doStop} logs that it is stopping a factory, followed by
the L{repr} of the L{Factory} instance that is being stopped.
c r� r} r� r r� r r r� � r� z:FactoryTests.test_doStopLoggingStatement.<locals>.<lambda>c @ s e Zd ZdZdS )z;FactoryTests.test_doStopLoggingStatement.<locals>.MyFactoryr^ N)r r r �numPortsr r r r � MyFactory� s r� r r& r� r� z Stopping factory {factory!r}N)
r r� r~ r� r �doStoprD r0 r r� )r" r� r� r r� r �test_doStopLoggingStatement� s z(FactoryTests.test_doStopLoggingStatementN)
r r r rr rv rz r� r� r� r� r r r r r� � s
r� c @ � e Zd ZdZdd� Zdd� ZdS )� AdapterTestszR
Tests for L{ProtocolToConsumerAdapter} and L{ConsumerToProtocolAdapter}.
c C sB g }t � }|j|_t|�}|�d� | �|dg� | �|t� dS )z}
L{IProtocol} providers can be adapted to L{IConsumer} providers using
L{ProtocolToConsumerAdapter}.
� helloN)r r~ � dataReceivedr �writer0 r! r )r" r� �p�consumerr r r �test_protocolToConsumer� s
z$AdapterTests.test_protocolToConsumerc sT g � t t�G � fdd�d��}|� }t|�}|�d� | �� dg� | �|t� dS )z}
L{IConsumer} providers can be adapted to L{IProtocol} providers using
L{ProtocolToConsumerAdapter}.
c r| )z6AdapterTests.test_consumerToProtocol.<locals>.Consumerc s � � |� d S r} )r~ )r" rO r r r r� � s z<AdapterTests.test_consumerToProtocol.<locals>.Consumer.writeN)r r r r� r r r r �Consumer� s r� r� N)r r r r� r0 r! r )r" r� �cr' r r r �test_consumerToProtocol� s
z$AdapterTests.test_consumerToProtocolN)r r r rr r� r� r r r r r� � s
r� c @ r� )�FileWrapperTestsz2
L{twisted.internet.protocol.FileWrapper}
c C sP t t� �}|�d� | �|j�� d� t t� �}|�d� | �|j�� d� dS )z@
L{twisted.internet.protocol.FileWrapper.write}
� test1�stuffN)r
r r� r0 �file�getvalue�assertNotEqual�r" �wrapperr r r �
test_write s
zFileWrapperTests.test_writec C sL t t� �}|�ddg� | �|j�� d� t t� �}| �t|jddg� dS )zH
L{twisted.internet.protocol.FileWrapper.writeSequence}
r� s test2s
test1test2�test3�test4N)r
r �
writeSequencer0 r� r� � assertRaises� TypeErrorr� r r r �test_writeSequence s
z#FileWrapperTests.test_writeSequenceN)r r r rr r� r� r r r r r� s r� N)$rr �ior �zope.interfacer �zope.interface.verifyr �twisted.internet.deferr �twisted.internet.interfacesr r r r �twisted.internet.protocolr
r r r
r r �twisted.loggerr r �twisted.python.failurer �twisted.test.proto_helpersr r �twisted.trial.unittestr r rs r� r� r� r r r r �<module> s$ :([%