| 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[@% � @ s� d Z ddlmZmZ ddlmZ ddlmZ ddlm Z m
Z
ddl m Z dZ
z ddlmZ dd lmZ W n ek
r� d
Z
Y nX dd lmZ dd lmZ G d
d� de �ZG dd� de�ZG dd� de�ZG dd� de �ZG dd� de�ZG dd� de�ZdS )z>
Tests for L{twisted.internet.posixbase} and supporting code.
� )�division�absolute_import)�TestCase)�Deferred)�PosixReactorBase�_Waker)�
ServerFactoryN)�unix)� ClientProtoz)Platform does not support AF_UNIX sockets)�Port��reactorc @ s4 e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd S ) �TrivialReactorc C s i | _ i | _t�| � d S �N)�_readers�_writersr �__init__��self� r �F/usr/lib/python3/dist-packages/twisted/internet/test/test_posixbase.pyr s zTrivialReactor.__init__c C s d| j |< d S �NT�r �r �readerr r r � addReader# s zTrivialReactor.addReaderc C s | j |= d S r r r r r r � removeReader' s zTrivialReactor.removeReaderc C s d| j |< d S r �r �r �writerr r r � addWriter+ s zTrivialReactor.addWriterc C s | j |= d S r r r r r r � removeWriter/ s zTrivialReactor.removeWriterN)�__name__�
__module__� __qualname__r r r r r! r r r r r s
r c @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
S ) �PosixReactorBaseTestsz(
Tests for L{PosixReactorBase}.
c C s2 | � |jt� | �|j|j� | �|j|j� d S r )ZassertIsInstanceZwakerr �assertIn�_internalReadersr �r r
r r r � _checkWaker9 s z!PosixReactorBaseTests._checkWakerc C s t � }| �|� dS )z
When L{PosixReactorBase} is instantiated, it creates a waker and adds
it to its internal readers set.
N)r r) r( r r r �test_wakerIsInternalReader? s z0PosixReactorBaseTests.test_wakerIsInternalReaderc C s\ t � }t� }|j�|� |�|� |�|j|j� | �|� | � ||j� | � ||j� dS )z�
Any L{IReadDescriptors} in L{PosixReactorBase._internalReaders} are
left alone by L{PosixReactorBase._removeAll}.
N)
r �objectr' �addr �
_removeAllr r r) r& )r r
Zextrar r r �"test_removeAllSkipsInternalReadersH s
z8PosixReactorBaseTests.test_removeAllSkipsInternalReadersc C sn t � }t� }t� }|�|� |�|� |�|j|j�}| �t|�t||g�� | � ||j� | � ||j� dS )z�
L{PosixReactorBase._removeAll} returns a list of removed
L{IReadDescriptor} and L{IWriteDescriptor} objects.
N)
r r+ r r r- r r � assertEqual�setZ assertNotIn)r r
r r Zremovedr r r �'test_removeAllReturnsRemovedDescriptorsW s
�z=PosixReactorBaseTests.test_removeAllReturnsRemovedDescriptorsN)r" r# r$ �__doc__r) r* r. r1 r r r r r% 4 s
r% c @ s& e Zd ZdZeee�sdZdd� ZdS )� TCPPortTestsz1
Tests for L{twisted.internet.tcp.Port}.
zNon-posixbase reactorc C s, t dt� �}d|_dd� |_| �|�� t�S )z�
L{Port.stopListening} returns a L{Deferred} which errbacks if
L{Port.connectionLost} raises an exception.
i90 Tc S s dd S )N� r r )�reasonr r r �<lambda>x � z8TCPPortTests.test_connectionLostFailed.<locals>.<lambda>)r r Z connectedZconnectionLostZ
assertFailure�
stopListening�ZeroDivisionError)r �portr r r �test_connectionLostFailedq s
z&TCPPortTests.test_connectionLostFailedN) r" r# r$ r2 �
isinstancer
r �skipr; r r r r r3 i s
r3 c @ s8 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd S )
�TimeoutReportReactorz�
A reactor which is just barely runnable and which cannot monitor any
readers or writers, and which fires a L{Deferred} with the timeout
passed to its C{doIteration} method as soon as that method is invoked.
c C s t �| � t� | _d| _d S )N�d )r r r �iterationTimeout�nowr r r r r � s
z TimeoutReportReactor.__init__c C s dS )z�
Ignore the reader. This is necessary because the waker will be
added. However, we won't actually monitor it for any events.
Nr r r r r r � s z TimeoutReportReactor.addReaderc C s g S )z�
There are no readers or writers, so there is nothing to remove.
This will be called when the reactor stops, though, so it must be
implemented.
r r r r r � removeAll� s z TimeoutReportReactor.removeAllc C s | j S )zx
Override the real clock with a deterministic one that can be easily
controlled in a unit test.
)rA r r r r �seconds� s z TimeoutReportReactor.secondsc C s" | j }|d k r d | _ |�|� d S r )r@ �callback)r �timeout�dr r r � doIteration� s z TimeoutReportReactor.doIterationN) r" r# r$ r2 r r rB rC rG r r r r r> } s r> c @ sP e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
dd� Z dS )�IterationTimeoutTestsz�
Tests for the timeout argument L{PosixReactorBase.run} calls
L{PosixReactorBase.doIteration} with in the presence of various delayed
calls.
c s6 g }� j �|j� � j �� fdd�� � �� |d S )Nc s � � � S r )�stop)Zignoredr r r r6 � r7 z>IterationTimeoutTests._checkIterationTimeout.<locals>.<lambda>r )r@ Z addCallback�append�run�r r
rE r r r �_checkIterationTimeout� s
z,IterationTimeoutTests._checkIterationTimeoutc C s t � }| �|�}| �|� dS )zl
If there are no delayed calls, C{doIteration} is called with a
timeout of L{None}.
N)r> rM � assertIsNonerL r r r � test_noCalls� s
z"IterationTimeoutTests.test_noCallsc C s0 t � }|�ddd� � | �|�}| �|d� dS )z�
If there is a delayed call, C{doIteration} is called with a timeout
which is the difference between the current time and the time at
which that call is to run.
r? c S s d S r r r r r r r6 � r7 z8IterationTimeoutTests.test_delayedCall.<locals>.<lambda>N�r> � callLaterrM r/ rL r r r �test_delayedCall� s
z&IterationTimeoutTests.test_delayedCallc C s> t � }|�ddd� � | jd7 _| �|�}| �|d� dS )z�
If a delayed call is scheduled and then some time passes, the
timeout passed to C{doIteration} is reduced by the amount of time
which passed.
r? c S s d S r r r r r r r6 � r7 z7IterationTimeoutTests.test_timePasses.<locals>.<lambda>� �K N)r> rQ rA rM r/ rL r r r �test_timePasses� s
z%IterationTimeoutTests.test_timePassesc C sP t � }|�ddd� � |�ddd� � |�ddd� � | �|�}| �|d� dS ) z�
If there are several delayed calls, C{doIteration} is called with a
timeout which is the difference between the current time and the
time at which the earlier of the two calls is to run.
�2 c S s d S r r r r r r r6 � r7 zAIterationTimeoutTests.test_multipleDelayedCalls.<locals>.<lambda>�
c S s d S r r r r r r r6 � r7 r? c S s d S r r r r r r r6 � r7 NrP rL r r r �test_multipleDelayedCalls� s
z/IterationTimeoutTests.test_multipleDelayedCallsc C sH t � }|�ddd� �}| jd7 _|�d� | �|�}| �|d� dS )z�
If a delayed call is reset, the timeout passed to C{doIteration} is
based on the interval between the time when reset is called and the
new delay of the call.
rV c S s d S r r r r r r r6 � r7 z=IterationTimeoutTests.test_resetDelayedCall.<locals>.<lambda>rS � N)r> rQ rA �resetrM r/ �r r
�callrE r r r �test_resetDelayedCall� s
z+IterationTimeoutTests.test_resetDelayedCallc C sH t � }|�ddd� �}| jd7 _|�d� | �|�}| �|d� dS )z�
If a delayed call is re-delayed, the timeout passed to
C{doIteration} is based on the remaining time before the call would
have been made and the additional amount of time passed to the delay
method.
rV c S s d S r r r r r r r6 � r7 z=IterationTimeoutTests.test_delayDelayedCall.<locals>.<lambda>rW � �< N)r> rQ rA ZdelayrM r/ r[ r r r �test_delayDelayedCall� s
z+IterationTimeoutTests.test_delayDelayedCallc C s6 t � }|�ddd� �}|�� | �|�}| �|� dS )zp
If the only delayed call is canceled, L{None} is the timeout passed
to C{doIteration}.
rV c S s d S r r r r r r r6 r7 z>IterationTimeoutTests.test_cancelDelayedCall.<locals>.<lambda>N)r> rQ ZcancelrM rN r[ r r r �test_cancelDelayedCall s
z,IterationTimeoutTests.test_cancelDelayedCallN) r" r# r$ r2 rM rO rR rU rX r] r` ra r r r r rH � s
rH c @ s, e Zd ZdZedk reZdd� Zdd� ZdS )�ConnectedDatagramPortTestsz/
Test connected datagram UNIX sockets.
Nc s. � fdd�}t �dt� �}||_|�d� dS )z�
L{ConnectedDatagramPort} does not call the deprecated C{loseConnection}
in L{ConnectedDatagramPort.connectionFailed}.
c s � � d� dS )z�
Dummy C{loseConnection} method. C{loseConnection} is deprecated and
should not get called.
z7loseConnection is deprecated and should not get called.N)Zfailr r r r �loseConnection s z`ConnectedDatagramPortTests.test_connectionFailedDoesntCallLoseConnection.<locals>.loseConnectionN�goodbye)r �ConnectedDatagramPortr
rc �connectionFailed)r rc r: r r r �-test_connectionFailedDoesntCallLoseConnection s zHConnectedDatagramPortTests.test_connectionFailedDoesntCallLoseConnectionc s@ d� _ � fdd�}t�dt� �}||_|�d� � �� j � dS )z�
L{ConnectedDatagramPort} calls L{ConnectedDatagramPort.stopListening}
instead of the deprecated C{loseConnection} in
L{ConnectedDatagramPort.connectionFailed}.
Fc s
d� _ dS )z8
Dummy C{stopListening} method.
TN)�calledr r r r r8 3 s zYConnectedDatagramPortTests.test_connectionFailedCallsStopListening.<locals>.stopListeningNrd )rh r re r
r8 rf Z
assertTrue)r r8 r: r r r �'test_connectionFailedCallsStopListening+ s
zBConnectedDatagramPortTests.test_connectionFailedCallsStopListening)r" r# r$ r2 � skipSocketsr= rg ri r r r r rb s
rb ) r2 Z
__future__r r Ztwisted.trial.unittestr Ztwisted.internet.deferr Ztwisted.internet.posixbaser r Ztwisted.internet.protocolr rj Ztwisted.internetr Ztwisted.test.test_unixr
� ImportErrorZtwisted.internet.tcpr r
r r% r3 r> rH rb r r r r �<module> s&
5,i