| 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 : /lib/python3/dist-packages/twisted/internet/test/__pycache__/ |
Upload File : |
o
�bo$ � @ s� d Z ddlZddlZddlmZmZmZmZmZ ddl m
Z
ddl m Z ddl
mZ ddlmZ dd lmZ d
Zd
Zzej�d �rMdd lmZmZ d
Zd
ZW n eyW Y nw G dd� dee�ZdS )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 Zd ZdZe� �� Zeee�Z dd� Z
e e d� e
e�ejjejje�� ��dd� �Ze e d� e
e�ejjejje�� ��dd� �Ze e d � e
e�ejjejje�� ��d
d � �Ze e d �d
d� �Ze e d�dd� �Ze e d�dd� �Ze e d�dd� �Z dd� Z dd� Z d d � Z d d � Z d S )!�AsyncioSelectorReactorTestsz*
L{AsyncioSelectorReactor} tests.
c sX t � }g �� �fdd�}|�|� |�d� | ��g � | j� dd� | ��dg� dS )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 � }| �|� dS )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|� | �|� td� dS )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 | � tt� dS )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 � �� }| �tt|� dS )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 � n1 sw Y t d� dS )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� dS )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 rtt� � t� }|�� }| �|d� | �|d� t r"td� dS dS )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 rtt� � t� � dg�� �fdd�}� �� }� �d|�}|�d� � �d� j� � �� | � �d � | �
�d | d� t rGtd� dS dS )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 rtt� � t� � dg�� �fdd�}� �� }� �d|�}|�d� � �d� j� ddl}ddl m
} |� � }||�� � � � W d � n1 sKw Y | �
|�� d� | ��d � | ��d | d � t rqtd� dS dS )
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 rtt� � t�� }t�� z6tt�� �}d}t� }t |�D ]
}|�
ddd� � q |� � tt�� �}| � || | d� W |rEt�
� n|rMt�
� w w t rVtd� dS dS )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* � �