U
    n5{iš  ã                   @   s˜   d Z ddlmZ ddlmZ ddlmZ ddlmZ ddl	m
Z
 zddlmZ W n ek
rh   dZY nX G d	d
„ d
ƒZee dƒG dd„ de
ƒƒZdS )z-
Tests for L{twisted.internet.epollreactor}.
é    )ÚskipIf©ÚConnectionDone)Ú_ContinuousPolling)ÚClock)ÚTestCase)ÚepollreactorNc                   @   s8   e Zd ZdZdd„ Zdd„ Zdd„ Zdd	„ Zd
d„ ZdS )Ú
DescriptorzF
    Records reads and writes, as if it were a C{FileDescriptor}.
    c                 C   s
   g | _ d S ©N)Úevents©Úself© r   úK/tmp/pip-unpacked-wheel-mnhg7r0q/twisted/internet/test/test_epollreactor.pyÚ__init__   s    zDescriptor.__init__c                 C   s   dS )Né   r   r   r   r   r   Úfileno   s    zDescriptor.filenoc                 C   s   | j  d¡ d S )NÚread©r   Úappendr   r   r   r   ÚdoRead    s    zDescriptor.doReadc                 C   s   | j  d¡ d S )NÚwriter   r   r   r   r   ÚdoWrite#   s    zDescriptor.doWritec                 C   s   |  t¡ | j d¡ d S )NÚlost)Ztrapr   r   r   )r   Úreasonr   r   r   ÚconnectionLost&   s    
zDescriptor.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 Zd ZdZdd„ Zdd„ Zdd„ Zdd	„ Zd
d„ Zdd„ Z	dd„ Z
dd„ Zdd„ Zdd„ Zdd„ Zdd„ Zdd„ ZdS )ÚContinuousPollingTestsza
    L{_ContinuousPolling} can be used to read and write from C{FileDescriptor}
    objects.
    c                 C   sv   t tƒ ƒ}|  |j¡ tƒ }|  | |¡¡ | |¡ |  |j¡ |  	|jj
¡ |  |jj|j¡ |  	| |¡¡ dS )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   s    

z%ContinuousPollingTests.test_addReaderc                 C   sv   t tƒ ƒ}|  |j¡ tƒ }|  | |¡¡ | |¡ |  |j¡ |  	|jj
¡ |  |jj|j¡ |  	| |¡¡ dS )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_addWriterA   s    

z%ContinuousPollingTests.test_addWriterc                 C   sV   t tƒ ƒ}tƒ }| |¡ | |¡ |  |j¡ |  |j 	¡ g ¡ |  
| |¡¡ dS )z=
        Removing a reader stops the C{LoopingCall}.
        N)r   r   r#   r&   ÚremoveReaderr!   r"   ÚassertEqualr,   ÚgetDelayedCallsr$   r%   r-   r   r   r   Útest_removeReaderP   s    


z(ContinuousPollingTests.test_removeReaderc                 C   sV   t tƒ ƒ}tƒ }| |¡ | |¡ |  |j¡ |  |j 	¡ g ¡ |  
| |¡¡ dS )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_removeWriter\   s    


z(ContinuousPollingTests.test_removeWriterc                 C   s&   t tƒ ƒ}| tƒ ¡ | tƒ ¡ dS )zM
        Removing unknown readers and writers silently does nothing.
        N)r   r   r:   r#   r6   )r   r.   r   r   r   Útest_removeUnknownh   s    
z)ContinuousPollingTests.test_removeUnknownc                 C   s    t tƒ ƒ}tƒ }| |¡ |  |j¡ | tƒ ¡ |  |j¡ | tƒ ¡ |  |j¡ | tƒ ¡ | |¡ |  |j¡ |  |jj	¡ |  
t|j ¡ ƒd¡ dS )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_multipleReadersAndWritersp   s    


z5ContinuousPollingTests.test_multipleReadersAndWritersc                 C   s„   t ƒ }t|ƒ}tƒ }| |¡ |  |jg ¡ | d¡ |  |jdg¡ | d¡ |  |jddg¡ | d¡ |  |jdddg¡ dS )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   Zreactorr.   Údescr   r   r   Útest_readerPollingƒ   s    



z)ContinuousPollingTests.test_readerPollingc                 C   s„   t ƒ }t|ƒ}tƒ }| |¡ |  |jg ¡ | d¡ |  |jdg¡ | d¡ |  |jddg¡ | d¡ |  |jdddg¡ dS )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ƒ }dd„ |_| |¡ |  |jg ¡ | d¡ |  |jdg¡ dS )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>rC   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ƒ }dd„ |_| |¡ |  |jg ¡ | d¡ |  |jdg¡ dS )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   rE   »   rF   zCContinuousPollingTests.test_connectionLostOnWrite.<locals>.<lambda>rC   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
|ƒ|||h¡ dS )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   ZbothÚremovedr   r   r   Útest_removeAllÁ   s    




z%ContinuousPollingTests.test_removeAllc                 C   s.   t tƒ ƒ}tƒ }| |¡ |  || ¡ ¡ dS )zb
        L{_ContinuousPolling.getReaders} returns a list of the read
        descriptors.
        N)r   r   r#   r&   ÚassertInrJ   r-   r   r   r   Útest_getReadersÔ   s    

z&ContinuousPollingTests.test_getReadersc                 C   s.   t tƒ ƒ}tƒ }| |¡ |  || ¡ ¡ dS )zc
        L{_ContinuousPolling.getWriters} returns a list of the write
        descriptors.
        N)r   r   r#   r2   rO   rK   r3   r   r   r   Útest_getWritersÞ   s    

z&ContinuousPollingTests.test_getWritersN)r   r   r   r   r0   r5   r9   r;   r<   r>   rB   rD   rG   rH   rN   rP   rQ   r   r   r   r   r    +   s   
r    )r   Zunittestr   Ztwisted.internet.errorr   Ztwisted.internet.posixbaser   Ztwisted.internet.taskr   Ztwisted.trial.unittestr   Ztwisted.internetr   ÚImportErrorr	   r    r   r   r   r   Ú<module>   s   

