pytest-使用asyncio进行python测试

是否可以从函数将执行返回到事件循环.任务一旦完成将返回功能并继续执行？

我正在尝试使用pytest-asyncio插件

例：

@pytest.mark.asyncio
async def test_async1(event_loop):
    print('start 1')
    res = event_loop.create_task(
        send_async_request("http://test.com", limit=1000))) # here I need to return execution to event loop and continue only after getting response from send_async_request function 
    print('end1',res)



@pytest.mark.asyncio
async def test_async2(event_loop):
    print('start 2')
    res = event_loop.create_task(
        send_async_request("http://test2", limit=1000))) # here I need to return execution to event loop and continue only after getting response from send_async_request function
    print('end2', res)

send_async_request- aiohttp：

@asyncio.coroutine
def send_async_request(url, method='GET'):
    with aiohttp.ClientSession() as session:
        resp = yield from session.get(url, timeout=60)
        if resp.status == 200:
            return resp.status, response
        else:
            return resp.status, False

解决方法:

当您的测试用pytest.mark.asyncio标记时,它们将成为协程,因此您可以使用await语法：

@pytest.mark.asyncio
async def test_sleep(event_loop):
    result = await asyncio.sleep(1, result=3, loop=event_loop)
    assert result == 3

编辑：另一个示例,具有多个睡眠操作：

@pytest.mark.asyncio
async def test_multiple_sleep(event_loop):
    tasks = [event_loop.create_task(asyncio.sleep(1, result=x))
             for x in range(10)]
    results = await asyncio.gather(*tasks)
    assert results == list(range(10))

标签：python-3-x,pytest,python-asyncio,python
来源： https://codeday.me/bug/20191025/1931479.html