| 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 : /usr/lib/python3/dist-packages/twisted/test/__pycache__/ |
Upload File : |
o
�b�" � @ s� d Z ddlmZ ddlmZ ddlmZ ddlmZm Z ddl
m Z G dd� de �Z ee�G d d
� d
��Z
G d d � d e �ZG d
d� de �ZdS )z"
Tests for L{twisted.test.iosim}.
� )� implementer)�
IPushProducer)�Protocol)�
FakeTransport�connect)�TestCasec @ �( e Zd ZdZdd� Zdd� Zdd� ZdS ) �FakeTransportTestsz%
Tests for L{FakeTransport}.
c C sH t t� d�}t t� d�}| �|jt� | �|jt� | �|j|j� dS )ze
Each L{FakeTransport} receives a serial number that uniquely identifies
it.
TFN)r �object�assertIsInstance�serial�int�assertNotEqual)�self�a�b� r �9/usr/lib/python3/dist-packages/twisted/test/test_iosim.py�test_connectionSerial s
z(FakeTransportTests.test_connectionSerialc C s< t t� d�}|�d� |�g d�� | �d�|j�d� dS )zl
L{FakeTransport.writeSequence} will write a sequence of L{bytes} to the
transport.
F� a)� b� c� d� s abcdN)r r
�write�
writeSequence� assertEqual�join�stream�r r r r r �test_writeSequence! s
z%FakeTransportTests.test_writeSequencec C s@ t t� d�}|�d� |�� |�d� | �d�|j�d� dS )z�
L{FakeTransport.write} will accept writes after transport was closed,
but the data will be silently discarded.
Fs befores afterr N)r r
r �loseConnectionr r r r r r r �test_writeAfterClose- s
z'FakeTransportTests.test_writeAfterCloseN)�__name__�
__module__� __qualname__�__doc__r r r" r r r r r s
r c @ s, e Zd ZdZdZdd� Zdd� Zdd� Zd S )
�StrictPushProducerz�
An L{IPushProducer} implementation which produces nothing but enforces
preconditions on its state transition methods.
�runningc C s | j dkr td��d| _ d S )N�stoppedz)Cannot stop already-stopped IPushProducer��_state�
ValueError�r r r r �
stopProducingC s
z StrictPushProducer.stopProducingc C �& | j dkrtd| j � d���d| _ d S )Nr( z
Cannot pause � IPushProducer�pausedr* r- r r r �pauseProducingH �
z!StrictPushProducer.pauseProducingc C r/ )Nr1 zCannot resume r0 r( r* r- r r r �resumeProducingM r3 z"StrictPushProducer.resumeProducingN)r# r$ r% r& r+ r. r2 r4 r r r r r' : s r' c @ 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"d#� Zd$d%� Zd&d'� Zd(S ))�StrictPushProducerTestsz*
Tests for L{StrictPushProducer}.
c C s t � S )zp
@return: A new L{StrictPushProducer} which has not been through any state
changes.
)r' r- r r r �_initialX s z StrictPushProducerTests._initialc C � t � }|�� |S )z@
@return: A new, stopped L{StrictPushProducer}.
)r' r. �r �producerr r r �_stopped_ � z StrictPushProducerTests._stoppedc C r7 )z?
@return: A new, paused L{StrictPushProducer}.
)r' r2 r8 r r r �_pausedg r; z StrictPushProducerTests._pausedc C s t � }|�� |�� |S )zY
@return: A new L{StrictPushProducer} which has been paused and resumed.
)r' r2 r4 r8 r r r �_resumedo s z StrictPushProducerTests._resumedc C � | � |jd� dS )z�
Assert that the given producer is in the stopped state.
@param producer: The producer to verify.
@type producer: L{StrictPushProducer}
r) N�r r+ r8 r r r �
assertStoppedx � z%StrictPushProducerTests.assertStoppedc C r> )z�
Assert that the given producer is in the paused state.
@param producer: The producer to verify.
@type producer: L{StrictPushProducer}
r1 Nr? r8 r r r � assertPaused� rA z$StrictPushProducerTests.assertPausedc C r> )z�
Assert that the given producer is in the running state.
@param producer: The producer to verify.
@type producer: L{StrictPushProducer}
r( Nr? r8 r r r �
assertRunning� rA z%StrictPushProducerTests.assertRunningc C � | � t| �� j� dS )zz
L{StrictPushProducer.stopProducing} raises L{ValueError} if called when
the producer is stopped.
N)� assertRaisesr, r: r. r- r r r �test_stopThenStop� � z)StrictPushProducerTests.test_stopThenStopc C rD )z{
L{StrictPushProducer.pauseProducing} raises L{ValueError} if called when
the producer is stopped.
N)rE r, r: r2 r- r r r �test_stopThenPause� rG z*StrictPushProducerTests.test_stopThenPausec C rD )z|
L{StrictPushProducer.resumeProducing} raises L{ValueError} if called when
the producer is stopped.
N)rE r, r: r4 r- r r r �test_stopThenResume� rG z+StrictPushProducerTests.test_stopThenResumec C � | � � }|�� | �|� dS )zn
L{StrictPushProducer} is stopped if C{stopProducing} is called on a paused
producer.
N)r< r. r@ r8 r r r �test_pauseThenStop� � z*StrictPushProducerTests.test_pauseThenStopc C � | � � }| �t|j� dS )zs
L{StrictPushProducer.pauseProducing} raises L{ValueError} if called on a
paused producer.
N)r< rE r, r2 r8 r r r �test_pauseThenPause� � z+StrictPushProducerTests.test_pauseThenPausec C rJ )zp
L{StrictPushProducer} is resumed if C{resumeProducing} is called on a
paused producer.
N)r< r4 rC r8 r r r �test_pauseThenResume� rL z,StrictPushProducerTests.test_pauseThenResumec C rJ )zo
L{StrictPushProducer} is stopped if C{stopProducing} is called on a
resumed producer.
N)r= r. r@ r8 r r r �test_resumeThenStop� rL z+StrictPushProducerTests.test_resumeThenStopc C rJ )zo
L{StrictPushProducer} is paused if C{pauseProducing} is called on a
resumed producer.
N)r= r2 rB r8 r r r �test_resumeThenPause� rL z,StrictPushProducerTests.test_resumeThenPausec C rM )zu
L{StrictPushProducer.resumeProducing} raises L{ValueError} if called on a
resumed producer.
N)r= rE r, r4 r8 r r r �test_resumeThenResume� rO z-StrictPushProducerTests.test_resumeThenResumec C rJ )zn
L{StrictPushProducer} is stopped if C{stopProducing} is called in the
initial state.
N)r6 r. r@ r8 r r r � test_stop� rL z!StrictPushProducerTests.test_stopc C rJ )zn
L{StrictPushProducer} is paused if C{pauseProducing} is called in the
initial state.
N)r6 r2 rB r8 r r r �
test_pause� rL z"StrictPushProducerTests.test_pausec C rM )zz
L{StrictPushProducer} raises L{ValueError} if C{resumeProducing} is called
in the initial state.
N)r6 rE r, r4 r8 r r r � test_resume� rO z#StrictPushProducerTests.test_resumeN)r# r$ r% r& r6 r: r< r= r@ rB rC rF rH rI rK rN rP rQ rR rS rT rU rV r r r r r5 S s* r5 c @ r ) � IOPumpTestsz
Tests for L{IOPump}.
c C sr t � }t|dd�}t � }t|dd�}t||||dd�}t� }||d�| }|j|dd� |�� | �d|j� dS ) a�
Connect a couple protocol/transport pairs to an L{IOPump} and then pump
it. Verify that a streaming producer registered with one of the
transports does not receive invalid L{IPushProducer} method calls and
ends in the right state.
@param mode: C{u"server"} to test a producer registered with the
server transport. C{u"client"} to test a producer registered with
the client transport.
T)�isServerF)�greet)�server�client)� streamingr( N)r r r r' �registerProducer�pumpr r+ ) r �mode� serverProto�serverTransport� clientProto�clientTransportr^ r9 �victimr r r �_testStreamingProducer� s( ���z"IOPumpTests._testStreamingProducerc C � | j dd� dS )z�
L{IOPump.pump} does not call C{resumeProducing} on a L{IPushProducer}
(stream producer) registered with the server transport.
rZ �r_ N�re r- r r r � test_serverStreamingProducer � z(IOPumpTests.test_serverStreamingProducerc C rf )z�
L{IOPump.pump} does not call C{resumeProducing} on a L{IPushProducer}
(stream producer) registered with the client transport.
r[ rg Nrh r- r r r � test_clientStreamingProducer& rj z(IOPumpTests.test_clientStreamingProducerN)r# r$ r% r& re ri rk r r r r rW � s
# rW N)r& �zope.interfacer �twisted.internet.interfacesr �twisted.internet.protocolr �twisted.test.iosimr r �twisted.trial.unittestr r r' r5 rW r r r r �<module> s ) %