File: //usr/lib/python3/dist-packages/twisted/internet/test/__pycache__/test_epollreactor.cpython-38.pyc
U
W[� � @ s� d Z ddlmZmZ ddlmZ zddlmZ W n ek
rH dZY nX ddl m
Z
ddl m Z G dd � d e
�ZG d
d � d e�ZdS ) z-
Tests for L{twisted.internet.epollreactor}.
� )�division�absolute_import)�TestCase)�_ContinuousPollingN)�Clock��ConnectionDonec @ 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 s | j �d� d S )N�read�r �appendr r r r �doRead! s zDescriptor.doReadc C s | j �d� d S )N�writer r r r r �doWrite% s zDescriptor.doWritec C s |� t� | j�d� d S )N�lost)Ztrapr 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 c @ s� e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
d � Zd d
� Z dd� Z
dd� Z dd� Z dd� Z
dd� Zdd� Zdd� Zed kr�d Zd S ) �ContinuousPollingTestsza
L{_ContinuousPolling} can be used to read and write from C{FileDescriptor}
objects.
c C sv 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_addReader5 s
z%ContinuousPollingTests.test_addReaderc C sv t t� �}| �|j� t� }| �|�|�� |�|� | �|j� | � |jj
� | � |jj |j
� | � |�|�� dS )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+ r, �r
r. �writerr r r �test_addWriterE s
z%ContinuousPollingTests.test_addWriterc C sV t t� �}t� }|�|� |�|� | �|j� | �|j� � g � | �
|� |�� dS )z=
Removing a reader stops the C{LoopingCall}.
N) r r r# r& � removeReaderr! r" � assertEqualr, �getDelayedCallsr$ r% r- r r r �test_removeReaderU s
z(ContinuousPollingTests.test_removeReaderc C sV t t� �}t� }|�|� |�|� | �|j� | �|j� � g � | �
|� |�� dS )z=
Removing a writer stops the C{LoopingCall}.
N) r r r# r2 � removeWriterr! r" r7 r, r8 r$ r1 r3 r r r �test_removeWriterb s
z(ContinuousPollingTests.test_removeWriterc C s& t t� �}|�t� � |�t� � dS )zM
Removing unknown readers and writers silently does nothing.
N)r r r: r# r6 )r
r. r r r �test_removeUnknowno 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# r2 r' r" r& r: r( r) r7 �lenr, r8 r3 r r r � test_multipleReadersAndWritersx s
z5ContinuousPollingTests.test_multipleReadersAndWritersc C s� t � }t|�}t� }|�|� | �|jg � |�d� | �|jdg� |�d� | �|jddg� |�d� | �|jdddg� dS )za
Adding a reader causes its C{doRead} to be called every 1
milliseconds.
g�h㈵��>r N)r r r r& r7 r �advance�r
Zreactorr. Zdescr r r �test_readerPolling� s
z)ContinuousPollingTests.test_readerPollingc C s� t � }t|�}t� }|�|� | �|jg � |�d� | �|jdg� |�d� | �|jddg� |�d� | �|jdddg� dS )zb
Adding a writer causes its C{doWrite} to be called every 1
milliseconds.
���MbP?r N)r r r r2 r7 r r? r@ r r r �test_writerPolling� s
z)ContinuousPollingTests.test_writerPollingc C sT 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 s t � S r
r r r r r �<lambda>� � zBContinuousPollingTests.test_connectionLostOnRead.<locals>.<lambda>rB r N)r r r r r& r7 r r? r@ r r r �test_connectionLostOnRead� s
z0ContinuousPollingTests.test_connectionLostOnReadc C sT t � }t|�}t� }dd� |_|�|� | �|jg � |�d� | �|jdg� dS )zv
If a C{doWrite} returns a value indicating disconnection,
C{connectionLost} is called on it.
c S s t � S r
r r r r r rD � rE zCContinuousPollingTests.test_connectionLostOnWrite.<locals>.<lambda>rB r N)r r r r r2 r7 r r? r@ r r r �test_connectionLostOnWrite� s
z1ContinuousPollingTests.test_connectionLostOnWritec C s� t t� �}t� }t� }t� }|�|� |�|� |�|� |�|� |�� }| �|�� g � | �|�� g � | �t |�d� | �t
|�t
|||g�� dS )zv
L{_ContinuousPolling.removeAll} removes all descriptors and returns
the readers and writers.
� N) r r r# r& r2 Z removeAllr7 �
getReaders�
getWritersr= �set)r
r. r/ r4 ZbothZremovedr r r �test_removeAll� s
z%ContinuousPollingTests.test_removeAllc C s. t t� �}t� }|�|� | �||�� � dS )zb
L{_ContinuousPolling.getReaders} returns a list of the read
descriptors.
N)r r r# r& �assertInrI r- r r r �test_getReaders� s
z&ContinuousPollingTests.test_getReadersc C s. t t� �}t� }|�|� | �||�� � dS )zc
L{_ContinuousPolling.getWriters} returns a list of the write
descriptors.
N)r r r# r2 rM rJ r3 r r r �test_getWriters� s
z&ContinuousPollingTests.test_getWritersNz(epoll not supported in this environment.)r r r r r0 r5 r9 r; r< r> rA rC rF rG rL rN rO r �skipr r r r r / s
r )r Z
__future__r r Ztwisted.trial.unittestr Z twisted.internet.epollreactorr � ImportErrorZtwisted.internet.taskr Ztwisted.internet.errorr r# r r r r r r �<module> s