D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
lib
/
python3.8
/
unittest
/
__pycache__
/
Filename :
async_case.cpython-38.pyc
back
Copy
U �������g�����������������������@���s0���d�dl�Z�d�dlZddlmZ�G�dd��de�ZdS�)�����N����)�TestCasec�����������������������s����e�Zd�Zd ��fdd� Zdd��Zdd��Zdd ��Zd d ��Zd d ��Zdd��Z dd��Z dd��Z dd��Z dd��Z dd��Zdd��Zd ��fd d � Z���ZS�)!�IsolatedAsyncioTestCase�runTestc��������������������s ���t����|��d�|�_d�|�_d�S��N)�super�__init__�_asyncioTestLoop�_asyncioCallsQueue)�selfZ methodName�� __class__���)/usr/lib/python3.8/unittest/async_case.pyr���"���s���� z IsolatedAsyncioTestCase.__init__c���������������������s���d�S�r���r����r ���r���r���r���� asyncSetUp'���s����z"IsolatedAsyncioTestCase.asyncSetUpc���������������������s���d�S�r���r���r���r���r���r���� asyncTearDown*���s����z%IsolatedAsyncioTestCase.asyncTearDownc����������������O���s���|�j�|f|�|��d�S�r���)Z addCleanup)r ����func�args�kwargsr���r���r����addAsyncCleanup-���s���� z'IsolatedAsyncioTestCase.addAsyncCleanupc�����������������C���s���|������|��|�j��d�S�r���)ZsetUp� _callAsyncr���r���r���r���r���� _callSetUp<���s����z"IsolatedAsyncioTestCase._callSetUpc�����������������C���s���|���|��d�S�r�����_callMaybeAsync)r ����methodr���r���r����_callTestMethod@���s����z'IsolatedAsyncioTestCase._callTestMethodc�����������������C���s���|���|�j��|�����d�S�r���)r���r���ZtearDownr���r���r���r���� _callTearDownC���s���� z%IsolatedAsyncioTestCase._callTearDownc�����������������O���s���|�j�|f|�|��d�S�r���r���)r ���Zfunctionr���r���r���r���r���� _callCleanupG���s����z$IsolatedAsyncioTestCase._callCleanupc����������������O���sL���|�j�d�k st�|||�}t�|�s&t�|�j����}|�j�||f��|�j��|�S�r����r ����AssertionError�inspectZ isawaitable� create_futurer ���� put_nowait�run_until_complete�r ���r���r���r����ret�futr���r���r���r���J���s ���� z"IsolatedAsyncioTestCase._callAsyncc����������������O���sP���|�j�d�k st�|||�}t�|�rH|�j����}|�j�||f��|�j��|�S�|S�d�S�r���r ���r%���r���r���r���r���R���s���� z'IsolatedAsyncioTestCase._callMaybeAsyncc�������������� �������s����t�����|�_}|�d���|���I�d�H�}|����|d�kr:d�S�|\}}z |I�d�H�}|���s`|�|��W�q�ttfk r|�����Y�q�t t�j fk r��}�z|���s�|� |��W�5�d�}~X�Y�qX�qd�S�r���) �asyncioZQueuer ���Z set_result�getZ task_done� cancelled� SystemExit�KeyboardInterrupt� BaseExceptionZCancelledErrorZ set_exception)r ���r'���ZqueueZqueryZ awaitabler&���Zexr���r���r����_asyncioLoopRunner\���s ���� z*IsolatedAsyncioTestCase._asyncioLoopRunnerc�����������������C���sX���|�j�d�kst�t���}t�|��|�d��||�_�|���}|�|��|��|�_ |� |��d�S�)NT) r ���r ���r(���Znew_event_loop�set_event_loopZ set_debugr"���Z create_taskr.���Z_asyncioCallsTaskr$���)r ����loopr'���r���r���r����_setupAsyncioLoopo���s���� z)IsolatedAsyncioTestCase._setupAsyncioLoopc�������������� ���C���s����|�j�d�k st�|�j�}d�|�_�|�j�d���|�|�j�����z�t� |�}|sNW��vd�S�|D�] }|� ���qR|�tj ||dd����|D�]0}|� ��r�q||� ��d�k r||�d|� ��|d���q||�|�����W�5�t�d���|����X�d�S�)NT)r0���Zreturn_exceptionsz(unhandled exception during test shutdown)�message� exception�task)r ���r ���r ���r#���r$����joinr(���r/����closeZ all_tasksZcancelZgatherr*���r3���Zcall_exception_handlerZshutdown_asyncgens)r ���r0���Z to_cancelr4���r���r���r����_tearDownAsyncioLoopy���s4���� � � z,IsolatedAsyncioTestCase._tearDownAsyncioLoopNc��������������������s(���|������zt���|�W��S�|�����X�d�S�r���)r1���r7���r����run)r ����resultr ���r���r���r8�������s����zIsolatedAsyncioTestCase.run)r���)N)�__name__� __module__� __qualname__r���r���r���r���r���r ���r ���r ���r���r���r.���r1���r7���r8���� __classcell__r���r���r ���r���r������s ��� "r���)r(���r!���Zcaser���r���r���r���r���r����<module>���s���