| 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/internet/test/__pycache__/ |
Upload File : |
o
�b� � @ s� d Z ddlmZ ddlmZ ddlmZ ddlmZ ddl m
Z
zddl m Z W n e
y3 dZ Y nw G d d
� d
�Zee d �G d d
� d
e
��ZdS )z-
Tests for L{twisted.internet.epollreactor}.
� )�skipIf��ConnectionDone)�_ContinuousPolling)�Clock)�TestCase)� epollreactorNc @ s8 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd S )
�
DescriptorzF
Records reads and writes, as if it were a C{FileDescriptor}.
c C s
g | _ d S �N)�events��self� r �I/usr/lib/python3/dist-packages/twisted/internet/test/test_epollreactor.py�__init__ s
zDescriptor.__init__c C s dS )N� r r r r r �fileno s zDescriptor.filenoc C � | j �d� d S )N�read�r �appendr r r r �doRead � zDescriptor.doReadc C r )N�writer r r r r �doWrite# r zDescriptor.doWritec C s |� t� | j�d� d S )N�lost)�trapr r r )r
�reasonr r r �connectionLost& s
zDescriptor.connectionLostN) �__name__�
__module__� __qualname__�__doc__r r r r r r r r r r s r z(epoll not supported in this environment.c @ sx e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
dd� Z dd� Z dd� Z
dd� Zdd� Zdd� Zd S ) �ContinuousPollingTestsza
L{_ContinuousPolling} can be used to read and write from C{FileDescriptor}
objects.
c C �v t t� �}| �|j� t� }| �|�|�� |�|� | �|j� | � |jj
� | � |jj |j
� | � |�|�� dS )zi
Adding a reader when there was previously no reader starts up a
C{LoopingCall}.
N)r r � assertIsNone�_loop�object� assertFalse� isReading� addReader�assertIsNotNone�
assertTrue�running�assertIs�clock�_reactor�r
�poller�readerr r r �test_addReader2 �
z%ContinuousPollingTests.test_addReaderc C r$ )zi
Adding a writer when there was previously no writer starts up a
C{LoopingCall}.
N)r r r% r& r' r( � isWriting� addWriterr+ r, r- r. r/ r0 �r
r2 �writerr r r �test_addWriterA r5 z%ContinuousPollingTests.test_addWriterc C �V t t� �}t� }|�|� |�|� | �|j� | �|j� � g � | �
|� |�� dS )z=
Removing a reader stops the C{LoopingCall}.
N) r r r' r* � removeReaderr% r& � assertEqualr0 �getDelayedCallsr( r) r1 r r r �test_removeReaderP �
z(ContinuousPollingTests.test_removeReaderc C r; )z=
Removing a writer stops the C{LoopingCall}.
N) r r r' r7 � removeWriterr% r& r= r0 r> r( r6 r8 r r r �test_removeWriter\ r@ z(ContinuousPollingTests.test_removeWriterc C s& t t� �}|�t� � |�t� � dS )zM
Removing unknown readers and writers silently does nothing.
N)r r rA r' r< )r
r2 r r r �test_removeUnknownh s
z)ContinuousPollingTests.test_removeUnknownc C s� t t� �}t� }|�|� | �|j� |�t� � | �|j� |�t� � | �|j� |�t� � |�|� | �|j� | �|jj � | �
t |j �
� �d� dS )za
Adding multiple readers and writers results in a single
C{LoopingCall}.
r N)r r r' r7 r+ r& r* rA r, r- r= �lenr0 r> r8 r r r � test_multipleReadersAndWritersp s
z5ContinuousPollingTests.test_multipleReadersAndWritersc C � t � }t|�}t� }|�|� | �|jg � |�d� | �|jdg� |�d� | �|jddg� |�d� | �|jg d�� dS )za
Adding a reader causes its C{doRead} to be called every 1
milliseconds.
g�h㈵��>r )r r r N)r r r r* r= r �advance�r
�reactorr2 �descr r r �test_readerPolling� �
z)ContinuousPollingTests.test_readerPollingc C rF )zb
Adding a writer causes its C{doWrite} to be called every 1
milliseconds.
���MbP?r )r r r N)r r r r7 r= r rG rH r r r �test_writerPolling� rL z)ContinuousPollingTests.test_writerPollingc C �T t � }t|�}t� }dd� |_|�|� | �|jg � |�d� | �|jdg� dS )zu
If a C{doRead} returns a value indicating disconnection,
C{connectionLost} is called on it.
c S � t � S r
r r r r r �<lambda>� � zBContinuousPollingTests.test_connectionLostOnRead.<locals>.<lambda>rM r N)r r r r r* r= r rG rH r r r �test_connectionLostOnRead� �
z0ContinuousPollingTests.test_connectionLostOnReadc C rO )zv
If a C{doWrite} returns a value indicating disconnection,
C{connectionLost} is called on it.
c S rP r
r r r r r rQ � rR zCContinuousPollingTests.test_connectionLostOnWrite.<locals>.<lambda>rM r N)r r r r r7 r= r rG rH r r r �test_connectionLostOnWrite� rT z1ContinuousPollingTests.test_connectionLostOnWritec C s� t t� �}t� }t� }t� }|�|� |�|� |�|� |�|� |�� }| �|�� g � | �|�� g � | �t |�d� | �t
|�|||h� dS )zv
L{_ContinuousPolling.removeAll} removes all descriptors and returns
the readers and writers.
� N) r r r' r* r7 � removeAllr= �
getReaders�
getWritersrD �set)r
r2 r3 r9 �both�removedr r r �test_removeAll� s
z%ContinuousPollingTests.test_removeAllc C �. t t� �}t� }|�|� | �||�� � dS )zb
L{_ContinuousPolling.getReaders} returns a list of the read
descriptors.
N)r r r' r* �assertInrX r1 r r r �test_getReaders� �
z&ContinuousPollingTests.test_getReadersc C r^ )zc
L{_ContinuousPolling.getWriters} returns a list of the write
descriptors.
N)r r r' r7 r_ rY r8 r r r �test_getWriters� ra z&ContinuousPollingTests.test_getWritersN)r r r! r" r4 r: r? rB rC rE rK rN rS rU r] r` rb r r r r r# + s
r# )r" �unittestr �twisted.internet.errorr �twisted.internet.posixbaser �twisted.internet.taskr �twisted.trial.unittestr �twisted.internetr � ImportErrorr r# r r r r �<module> s �