o
    bo$                     @   s   d Z ddlZddlZddlmZmZmZmZm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 d
Zd
ZzejdrMddlmZmZ dZdZW n	 eyW   Y nw G dd deeZdS )z/
Tests for L{twisted.internet.asyncioreactor}.
    N)DefaultEventLoopPolicyFutureSelectorEventLoopset_event_loopset_event_loop_policy)skipIf)AsyncioSelectorReactor)platform)SynchronousTestCase   )ReactorBuilderFwin32)WindowsProactorEventLoopPolicyWindowsSelectorEventLoopPolicyTc                	   @   s*  e Zd ZdZe  ZeeeZ	dd Z
ee	 deeejjejje dd Zee	 deeejjejje dd Zee	d	eeejjejje d
d Zee ddd Zee ddd Zee ddd Zee ddd Zdd Zdd Zdd Zdd Z d S )!AsyncioSelectorReactorTestsz*
    L{AsyncioSelectorReactor} tests.
    c                    sX   t  }g  fdd}|| |d | g  | j dd | dg dS )zn
        Ensure that C{reactor} has an event loop that works
        properly with L{asyncio.Future}.
        c                    s    |      d S N)appendresultstop)futurereactorr    K/usr/lib/python3/dist-packages/twisted/internet/test/test_asyncioreactor.py	completed7   s   zRAsyncioSelectorReactorTests.assertReactorWorksWithAsyncioFuture.<locals>.completedTr   )timeoutN)r   add_done_callback
set_resultassertEqual
runReactor)selfr   r   r   r   r   r   #assertReactorWorksWithAsyncioFuture/   s   

z?AsyncioSelectorReactorTests.assertReactorWorksWithAsyncioFuturezLdefault event loop: {}
is not of type SelectorEventLoop on Python {}.{} ({})c                 C   s   t  }| | dS )a  
        L{AsyncioSelectorReactor} wraps the global policy's event loop
        by default.  This ensures that L{asyncio.Future}s and
        coroutines created by library code that uses
        L{asyncio.get_event_loop} are bound to the same loop.
        N)r   r!   r    r   r   r   r   -test_defaultSelectorEventLoopFromGlobalPolicyB   s   zIAsyncioSelectorReactorTests.test_defaultSelectorEventLoopFromGlobalPolicyc                 C   0   t   }t|}t| | | td dS )z
        If we use the L{asyncio.DefaultLoopPolicy} to create a new event loop,
        and then pass that event loop to a new L{AsyncioSelectorReactor},
        this reactor should work properly with L{asyncio.Future}.
        N)r   new_event_loopr   r   r!   r   r    
event_loopr   r   r   r   3test_newSelectorEventLoopFromDefaultEventLoopPolicyV   s
   

zOAsyncioSelectorReactorTests.test_newSelectorEventLoopFromDefaultEventLoopPolicyzHdefault event loop: {}
is of type SelectorEventLoop on Python {}.{} ({})c                 C   s   |  tt dS )ay  
        On Windows Python 3.5 to 3.7, L{get_event_loop()} returns a
        L{WindowsSelectorEventLoop} by default.
        On Windows Python 3.8+, L{get_event_loop()} returns a
        L{WindowsProactorEventLoop} by default.
        L{AsyncioSelectorReactor} should raise a
        L{TypeError} if the default event loop is not a
        L{WindowsSelectorEventLoop}.
        N)assertRaises	TypeErrorr   r    r   r   r   1test_defaultNotASelectorEventLoopFromGlobalPolicyl   s   zMAsyncioSelectorReactorTests.test_defaultNotASelectorEventLoopFromGlobalPolicyz&WindowsProactorEventLoop not availablec                 C   s   t   }| tt| dS )z
        L{AsyncioSelectorReactor} will raise a L{TypeError}
        if instantiated with a L{asyncio.WindowsProactorEventLoop}
        N)r   r%   r)   r*   r   )r    r'   r   r   r   test_WindowsProactorEventLoop   s   
z9AsyncioSelectorReactorTests.test_WindowsProactorEventLoopz(WindowsSelectorEventLoop only on Windowsc                 C   r$   )zR
        L{WindowsSelectorEventLoop} works with L{AsyncioSelectorReactor}
        N)r   r%   r   r   r!   r   r&   r   r   r   test_WindowsSelectorEventLoop   s
   

z9AsyncioSelectorReactorTests.test_WindowsSelectorEventLoopz.WindowsProactorEventLoopPolicy only on Windowsc                 C   sF   t t  | t t  W d   n1 sw   Y  t d dS )z
        L{AsyncioSelectorReactor} will raise a L{TypeError}
        if L{asyncio.WindowsProactorEventLoopPolicy} is default.
        N)r   r   r)   r*   r   r+   r   r   r   #test_WindowsProactorEventLoopPolicy   s
   
	z?AsyncioSelectorReactorTests.test_WindowsProactorEventLoopPolicyz.WindowsSelectorEventLoopPolicy only on Windowsc                 C   s&   t t  t }| | t d dS )zy
        L{AsyncioSelectorReactor} will work if
        if L{asyncio.WindowsSelectorEventLoopPolicy} is default.
        N)r   r   r   r!   r"   r   r   r   #test_WindowsSelectorEventLoopPolicy   s   
	
z?AsyncioSelectorReactorTests.test_WindowsSelectorEventLoopPolicyc                 C   sH   t rtt  t }| }| |d | |d t r"td dS dS )z0L{seconds} should return a plausible epoch time.i ^l    #G4 N)!hasWindowsSelectorEventLoopPolicyr   r   r   secondsassertGreater
assertLess)r    r   r   r   r   r   test_seconds   s   
z(AsyncioSelectorReactorTests.test_secondsc                    s   t rtt  t  dg fdd}  } d|}|d  d j    | 	d  | 
d | d t rGtd dS dS )zQ
        L{DelayedCall.reset()} properly reschedules timer to later time
        Nc                            d< d S Nr   r2   r   r   timer_called_atr   r   on_timer      zJAsyncioSelectorReactorTests.test_delayedCallResetToLater.<locals>.on_timerr         ?r   皙?)r1   r   r   r   r2   	callLaterresetr   runassertIsNotNoner3   )r    r;   
start_timedcr   r9   r   test_delayedCallResetToLater   s   

z8AsyncioSelectorReactorTests.test_delayedCallResetToLaterc                    s   t rtt  t  dg fdd}  } d|}|d  d j ddl}ddl	m
} | }||    W d   n1 sKw   Y  | | d | d  | d | d	 t rqtd dS dS )
zS
        L{DelayedCall.reset()} properly reschedules timer to earlier time
        Nc                      r6   r7   r8   r   r9   r   r   r;      r<   zLAsyncioSelectorReactorTests.test_delayedCallResetToEarlier.<locals>.on_timerr=   r   r   )redirect_stderr r>   )r1   r   r   r   r2   r?   r@   r   io
contextlibrF   StringIOrA   r   getvaluerB   r4   )r    r;   rC   rD   rH   rF   stderrr   r9   r   test_delayedCallResetToEarlier   s*   



z:AsyncioSelectorReactorTests.test_delayedCallResetToEarlierc                 C   s   t rtt  t }t  z6tt }d}t }t	|D ]
}|
ddd  q|  tt }| || | d W |rEt  n|rMt  w w t rVtd dS dS )zW
        L{AsyncioSelectorReactor.callLater()} doesn't leave cyclic references
        i  r   c                   S   s   d S r   r   r   r   r   r   <lambda>  s    zOAsyncioSelectorReactorTests.test_noCycleReferencesInCallLater.<locals>.<lambda>r   N)r1   r   r   gc	isenableddisablelenget_objectsr   ranger?   runUntilCurrentr4   enable)r    gc_was_enabledobjects_beforetimer_countr   _objects_afterr   r   r   !test_noCycleReferencesInCallLater   s,   

z=AsyncioSelectorReactorTests.test_noCycleReferencesInCallLaterN)!__name__
__module____qualname____doc__r   r%   _defaultEventLoop
isinstancer   _defaultEventLoopIsSelectorr!   r   formattypesysversion_infomajorminorr	   getTyper#   r(   r,   !hasWindowsProactorEventLoopPolicyr-   r1   r.   r/   r0   r5   rE   rM   r\   r   r   r   r   r   '   sz    















r   )r`   rO   rf   asyncior   r   r   r   r   unittestr   twisted.internet.asyncioreactorr   twisted.python.runtimer	   twisted.trial.unittestr
   reactormixinsr   rk   r1   
startswithr   r   ImportErrorr   r   r   r   r   <module>   s*   