o }5h@s0ddlZddlZddlmZGdddeZdS)N)TestCasecseZdZd#fdd ZddZddZdd Zd d Zd d ZddZ ddZ ddZ ddZ ddZ ddZddZd$fdd Zfdd Zd!d"ZZS)%IsolatedAsyncioTestCaserunTestcst|d|_d|_dSN)super__init___asyncioTestLoop_asyncioCallsQueue)self methodName __class__*/usr/lib/python3.10/unittest/async_case.pyr!s  z IsolatedAsyncioTestCase.__init__cdSrrr rrr asyncSetUp&z"IsolatedAsyncioTestCase.asyncSetUpcrrrrrrr asyncTearDown)rz%IsolatedAsyncioTestCase.asyncTearDowncO|j|g|Ri|dSr) addCleanup)r funcargskwargsrrraddAsyncCleanup,s z'IsolatedAsyncioTestCase.addAsyncCleanupcCs|||jdSr)setUp _callAsyncrrrrr _callSetUp;sz"IsolatedAsyncioTestCase._callSetUpcCs||dSr_callMaybeAsync)r methodrrr_callTestMethod?sz'IsolatedAsyncioTestCase._callTestMethodcCs||j|dSr)rrtearDownrrrr _callTearDownBs  z%IsolatedAsyncioTestCase._callTearDowncOrrr)r functionrrrrr _callCleanupFsz$IsolatedAsyncioTestCase._callCleanupcOs^|jdus Jd||i|}t|sJ|d|j}|j||f|j|S)N$asyncio test loop is not initializedz returned non-awaitabler inspect isawaitable create_futurer put_nowaitrun_until_completer rrrretfutrrrrIs   z"IsolatedAsyncioTestCase._callAsynccOsT|jdus Jd||i|}t|r(|j}|j||f|j|S|S)Nr'r(r.rrrr Qs   z'IsolatedAsyncioTestCase._callMaybeAsyncc st|_}|d |IdH}||durdS|\}}z|IdH}|s2||Wn'ttfy=t tj fyZ}z|sP| |WYd}~nd}~wwqr) asyncioQueuer set_resultget task_done cancelled SystemExitKeyboardInterrupt BaseExceptionCancelledError set_exception)r r0queuequery awaitabler/exrrr_asyncioLoopRunner[s,    z*IsolatedAsyncioTestCase._asyncioLoopRunnercCs\|jdus Jdt}t||d||_|}||||_| |dS)Nz%asyncio test loop already initializedT) r r1new_event_loopset_event_loop set_debugr+ create_taskr@_asyncioCallsTaskr-)r loopr0rrr_setupAsyncioLoopns  z)IsolatedAsyncioTestCase._setupAsyncioLoopc Cs$|jdus Jd|j}d|_|jd||jzct|}|s8W||td| dS|D]}| q:|tj |ddi|D]}| rUqN| durf|d| |dqN||W||td| dS||td| w)Nr'return_exceptionsTz(unhandled exception during test shutdown)message exceptiontask)r r r,r-joinr1 all_tasksshutdown_default_executorrBclosecancelgatherr6rJcall_exception_handlershutdown_asyncgens)r rF to_cancelrKrrr_tearDownAsyncioLoopxsD          z,IsolatedAsyncioTestCase._tearDownAsyncioLoopNcs*|z t|W|S|wr)rGrrunrU)r resultr rrrVs zIsolatedAsyncioTestCase.runcs|t|dSr)rGrdebugrUrr rrrXs  zIsolatedAsyncioTestCase.debugcCs|jdur |dSdSr)r rUrrrr__del__s  zIsolatedAsyncioTestCase.__del__)rr)__name__ __module__ __qualname__rrrrrr"r$r&rr r@rGrUrVrXrY __classcell__rrr rrs"  $ r)r1r)caserrrrrrs