conftest.py 文件源码

def loop(request, loop_type):
    old_loop = asyncio.get_event_loop()
    asyncio.set_event_loop(None)
    if loop_type == 'uvloop':
        loop = uvloop.new_event_loop()
    elif loop_type == 'tokio':
        import tokio
        policy = tokio.TokioLoopPolicy()
        asyncio.set_event_loop_policy(policy)
        loop = tokio.new_event_loop()
    else:
        loop = asyncio.new_event_loop()

    yield loop

    loop.close()
    asyncio.set_event_loop(old_loop)
    gc.collect()