嗯,经过大量的挖掘、测试,当然,还学习了很多关于Asyncio的知识,我让它自己工作了。谢谢你到目前为止的建议。
问题是Asyncio的事件_循环没有运行;正如@hoefring提到的,pytest本身不支持协程,但是pytest asyncio为您的测试带来了如此有用的特性。这里很好地解释了这一点:
https://medium.com/ideas-at-igenius/testing-asyncio-python-code-with-pytest-a2f3628f82bc
因此,如果没有pytest asyncio,需要测试的异步代码将如下所示:
def test_this_is_an_async_test():
loop = asyncio.get_event_loop()
result = loop.run_until_complete(my_async_function(param1, param2, param3)
assert result == 'expected'
我们使用loop.run直到\u complete(),否则循环将永远不会运行,因为这是Asyncio默认工作的方式(而pytest不会使它以不同的方式工作)。
使用Pytest Asyncio,您的测试可以使用众所周知的Async/Await部分:
async def test_this_is_an_async_test(event_loop):
result = await my_async_function(param1, param2, param3)
assert result == 'expected'
在这种情况下,pytest asyncio会将run_包装到上面的_complete()调用,并对其进行大量总结,因此事件循环将运行,并可供异步代码使用。
请注意:这里甚至不需要第二种情况下的event_loop参数,pytest asyncio为您的测试提供了一个可用的参数。
另一方面,当您测试Tornado应用程序时,通常需要在测试期间启动和运行一个HTTP服务器,在一个众所周知的端口中监听,因此通常的方法是编写设备来获取一个HTTP服务器,基URL(通常是
http://localhost
,带有未使用的端口等)。
Pytest Tornado是一个非常有用的工具,因为它为您提供了以下几种设备:http_服务器、http_客户机、未使用的_端口、基_URL等。
此外,它还获得了pytest标记的gen_test()特性,该特性将任何标准测试转换为通过yield使用协程,甚至断言它将以给定的超时运行,如下所示:
@pytest.mark.gen_test(timeout=3)
def test_fetch_my_data(http_client, base_url):
result = yield http_client.fetch('/'.join([base_url, 'result']))
assert len(result) == 1000
但是,这样它就不支持异步/等待,实际上只有Tornado的IOLoop可以通过IO-Loop fixture(虽然Tornado的IOLoop默认使用Tornado 5.0下面的Asyncio),所以您需要将pytest.mark.gen\u test和pytest.mark.async io结合起来,
但顺序是正确的!
(我确实失败了)。
一旦我更好地理解了问题所在,这就是下一个方法:
@pytest.mark.gen_test(timeout=2)
@pytest.mark.asyncio
async def test_score_returns_204_empty(http_client, base_url):
score_url = '/'.join([base_url, URL_PREFIX, 'score'])
token = create_token('test', scopes=['score:get'])
headers = {
'Authorization': f'Bearer {token}',
'Accept': 'application/json',
}
response = await http_client.fetch(score_url, headers=headers, raise_error=False)
assert response.code == 204
但如果您理解Python的装饰包装器是如何工作的,那么这是完全错误的。通过上面的代码,pytest asyncio的coroutine随后被包装在pytest tornado yield gen.coroutine中,这不会使事件循环运行…所以我的测试还是因为同样的问题而失败了。对数据库的任何查询都返回了一个等待事件循环运行的未来。
一旦我弥补了这个愚蠢的错误,我更新了代码:
@pytest.mark.asyncio
@pytest.mark.gen_test(timeout=2)
async def test_score_returns_204_empty(http_client, base_url):
score_url = '/'.join([base_url, URL_PREFIX, 'score'])
token = create_token('test', scopes=['score:get'])
headers = {
'Authorization': f'Bearer {token}',
'Accept': 'application/json',
}
response = await http_client.fetch(score_url, headers=headers, raise_error=False)
assert response.code == 204
在这种情况下,gen.coroutine被包装在pytest asyncio coroutine中,并且事件_循环按预期运行coroutine!
但是还有一个小问题让我花了一点时间才意识到:pytest asyncio的event_loop fixture为每个测试创建了一个新的事件循环,而pytest tornado也创建了一个新的ioloop。测试仍然失败,但这次有一个不同的错误。
conftest.py文件现在看起来是这样的;请注意,我已经声明了Event_Loop fixture,以使用Pytest Tornado IO_Loop fixture本身的事件_Loop(请回忆一下Pytest Tornado在每个测试函数上创建了一个新的IO_Loop):
@pytest.fixture(scope='function')
def event_loop(io_loop):
loop = io_loop.current().asyncio_loop
yield loop
loop.stop()
@pytest.fixture(scope='function')
async def db():
dbe = await setup_db()
yield dbe
@pytest.fixture
def app(db):
"""
Returns a valid testing Tornado Application instance.
:return:
"""
app = make_app(db)
settings.JWT_SECRET = 'its_secret_one'
yield app
现在我所有的测试都成功了,我回到了一个快乐的人,并且为我现在更好地理解异步的生活方式而感到骄傲。酷!
