使用 pytest、tornado 和 aiopg 进行单元测试失败，任何查询都失败
Failure on unit tests with pytest, tornado and aiopg, any query fail
我在 Python 3.7 + Tornado 5 上有一个 REST API 运行，以 postgresql 作为数据库，使用 aiopg 和 SQLAlchemy 核心（通过 aiopg.sa 绑定).对于单元测试，我使用 py.test 和 pytest-tornado。
只要不涉及对数据库的查询，所有测试都会正常，我会得到这个：
运行时错误：任务 cb=[IOLoop.add_future..() 在 venv/lib/python3.7/site-packages/tornado/ioloop.py:719]> 将 Future 附加到另一个循环
相同的代码在测试中运行良好，到目前为止我能够处理 100 多个请求。
这是 @auth 装饰器的一部分，它将检查 JWT 令牌的授权 header，对其进行解码并获取用户数据并将其附加到请求；这是查询的部分：
partner_id = payload['partner_id']
provided_scopes = payload.get("scope", [])
for scope in scopes:
if scope not in provided_scopes:
logger.error(
'Authentication failed, scopes are not compliant - '
'required: {} - '
'provided: {}'.format(scopes, provided_scopes)
)
raise ForbiddenException(
"insufficient permissions or wrong user."
)
db = self.settings['db']
partner = await Partner.get(db, username=partner_id)
# The user is authenticated at this stage, let's add
# the user info to the request so it can be used
if not partner:
raise UnauthorizedException('Unknown user from token')
p = Partner(**partner)
setattr(self.request, "partner_id", p.uuid)
setattr(self.request, "partner", p)
Partner 的 .get() 异步方法来自应用程序中所有模型的基础 class。这是 .get 方法实现：
@classmethod
async def get(cls, db, order=None, limit=None, offset=None, **kwargs):
"""
Get one instance that will match the criteria
:param db:
:param order:
:param limit:
:param offset:
:param kwargs:
:return:
"""
if len(kwargs) == 0:
return None
if not hasattr(cls, '__tablename__'):
raise InvalidModelException()
tbl = cls.__table__
instance = None
clause = cls.get_clause(**kwargs)
query = (tbl.select().where(text(clause)))
if order:
query = query.order_by(text(order))
if limit:
query = query.limit(limit)
if offset:
query = query.offset(offset)
logger.info(f'GET query executing:\n{query}')
try:
async with db.acquire() as conn:
async with conn.execute(query) as rows:
instance = await rows.first()
except DataError as de:
[...]
return instance
上面的 .get() 方法将 return 模型实例（行表示）或 None。
它使用 db.acquire() 上下文管理器，如 aiopg 文档中所述：https://aiopg.readthedocs.io/en/stable/sa.html.
如同一文档中所述，sa.create_engine() 方法 return 是一个连接池，因此 db.acquire() 仅使用池中的一个连接。我将这个池共享给 Tornado 中的每个请求，他们在需要时使用它来执行查询。
所以这是我在 conftest.py:
中设置的固定装置
@pytest.fixture
async def db():
dbe = await setup_db()
return dbe
@pytest.fixture
def app(db, event_loop):
"""
Returns a valid testing Tornado Application instance.
:return:
"""
app = make_app(db)
settings.JWT_SECRET = 'its_secret_one'
return app
我找不到为什么会这样的解释； Tornado 的文档和源代码清楚地表明默认使用 asyncIO 事件循环，通过调试它我可以看到事件循环确实是同一个事件循环，但由于某种原因它似乎突然关闭或停止。
这是一项失败的测试：
@pytest.mark.gen_test(timeout=2)
def test_score_returns_204_empty(app, http_server, 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 = yield http_client.fetch(score_url, headers=headers, raise_error=False)
assert response.code == 204
此测试失败，因为它 returns 401 而不是 204，假设 auth 装饰器上的查询由于 RuntimeError 而失败，returns 然后是未经授权的响应。
非常感谢这里的异步专家的任何想法，我对此很迷茫！！！
好吧，经过大量的挖掘、测试，当然还有很多关于 asyncio 的知识，我自己让它工作了。感谢到目前为止的建议。
问题是来自 asyncio 的 event_loop 不是 运行ning；正如@hoefling 提到的，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_until_complete() 否则，循环永远不会 运行ning，因为这是 asyncio 默认工作的方式（并且 pytest 没有做任何改变它的工作方式） .
使用 pytest-asyncio，您的测试可以使用众所周知的异步/等待部分：
async def test_this_is_an_async_test(event_loop):
result = await my_async_function(param1, param2, param3)
assert result == 'expected'
pytest-asyncio 在这种情况下包装了上面的 run_until_complete() 调用，对其进行了大量总结，因此事件循环将 运行 并可供您的异步代码使用。
请注意：第二种情况下的 event_loop 参数在这里甚至不是必需的，pytest-asyncio 提供了一个可用于您的测试。
另一方面，当您测试 Tornado 应用程序时，您通常需要在测试期间启动并 运行ning 一个 http 服务器，监听一个众所周知的端口等，所以通常的方法是通过编写固定装置来获取 http 服务器，base_url（通常是 http://localhost:，带有未使用的端口等）。
pytest-tornado 是一个非常有用的工具，因为它为您提供了其中的几个固定装置：http_server、http_client、unused_port、base_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
但是，这样它不支持 async / await，实际上只有 Tornado 的 ioloop 可以通过 io_loop fixture 使用（尽管 Tornado 的 ioloop 默认使用 Tornado 5.0 下的 asyncio），所以你' d 需要组合 pytest.mark.gen_test 和 pytest.mark.asyncio， 但顺序正确！（我确实失败了）。
一旦我更好地理解可能是什么问题，这是下一个方法：
@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 的协程然后被包装在 pytest-tornado yield gen.coroutine 中，它不会得到事件循环 运行ning... 所以我的测试仍然失败同样的问题。对数据库的任何查询都返回一个等待事件循环的 Future 运行ning。
一旦我弥补了这个愚蠢的错误，我更新了代码：
@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 协同程序中，event_loop 运行s 是预期的协同程序！
但是还有一个小问题我也花了一点时间才意识到； pytest-asyncio 的 event_loop fixture 为每个测试创建一个新的事件循环，而 pytest-tornado 也创建一个新的 IOloop。测试仍然失败，但这次出现了不同的错误。
conftest.py 文件现在看起来像这样；请注意，我已经重新声明 event_loop fixture 以使用来自 pytest-tornado io_loop fixture 本身的 event_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
现在我所有的测试都成功了，我回到了一个快乐的人并且为我现在对 asyncio 生活方式有了更好的理解而感到自豪。酷！
我在 Python 3.7 + Tornado 5 上有一个 REST API 运行，以 postgresql 作为数据库，使用 aiopg 和 SQLAlchemy 核心（通过 aiopg.sa 绑定).对于单元测试，我使用 py.test 和 pytest-tornado。
只要不涉及对数据库的查询，所有测试都会正常，我会得到这个：
运行时错误：任务 cb=[IOLoop.add_future..() 在 venv/lib/python3.7/site-packages/tornado/ioloop.py:719]> 将 Future 附加到另一个循环
相同的代码在测试中运行良好，到目前为止我能够处理 100 多个请求。
这是 @auth 装饰器的一部分，它将检查 JWT 令牌的授权 header，对其进行解码并获取用户数据并将其附加到请求；这是查询的部分：
partner_id = payload['partner_id']
provided_scopes = payload.get("scope", [])
for scope in scopes:
if scope not in provided_scopes:
logger.error(
'Authentication failed, scopes are not compliant - '
'required: {} - '
'provided: {}'.format(scopes, provided_scopes)
)
raise ForbiddenException(
"insufficient permissions or wrong user."
)
db = self.settings['db']
partner = await Partner.get(db, username=partner_id)
# The user is authenticated at this stage, let's add
# the user info to the request so it can be used
if not partner:
raise UnauthorizedException('Unknown user from token')
p = Partner(**partner)
setattr(self.request, "partner_id", p.uuid)
setattr(self.request, "partner", p)
Partner 的 .get() 异步方法来自应用程序中所有模型的基础 class。这是 .get 方法实现：
@classmethod
async def get(cls, db, order=None, limit=None, offset=None, **kwargs):
"""
Get one instance that will match the criteria
:param db:
:param order:
:param limit:
:param offset:
:param kwargs:
:return:
"""
if len(kwargs) == 0:
return None
if not hasattr(cls, '__tablename__'):
raise InvalidModelException()
tbl = cls.__table__
instance = None
clause = cls.get_clause(**kwargs)
query = (tbl.select().where(text(clause)))
if order:
query = query.order_by(text(order))
if limit:
query = query.limit(limit)
if offset:
query = query.offset(offset)
logger.info(f'GET query executing:\n{query}')
try:
async with db.acquire() as conn:
async with conn.execute(query) as rows:
instance = await rows.first()
except DataError as de:
[...]
return instance
上面的 .get() 方法将 return 模型实例（行表示）或 None。
它使用 db.acquire() 上下文管理器，如 aiopg 文档中所述：https://aiopg.readthedocs.io/en/stable/sa.html.
如同一文档中所述，sa.create_engine() 方法 return 是一个连接池，因此 db.acquire() 仅使用池中的一个连接。我将这个池共享给 Tornado 中的每个请求，他们在需要时使用它来执行查询。
所以这是我在 conftest.py:
中设置的固定装置@pytest.fixture
async def db():
dbe = await setup_db()
return dbe
@pytest.fixture
def app(db, event_loop):
"""
Returns a valid testing Tornado Application instance.
:return:
"""
app = make_app(db)
settings.JWT_SECRET = 'its_secret_one'
return app
我找不到为什么会这样的解释； Tornado 的文档和源代码清楚地表明默认使用 asyncIO 事件循环，通过调试它我可以看到事件循环确实是同一个事件循环，但由于某种原因它似乎突然关闭或停止。
这是一项失败的测试：
@pytest.mark.gen_test(timeout=2)
def test_score_returns_204_empty(app, http_server, 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 = yield http_client.fetch(score_url, headers=headers, raise_error=False)
assert response.code == 204
此测试失败，因为它 returns 401 而不是 204，假设 auth 装饰器上的查询由于 RuntimeError 而失败，returns 然后是未经授权的响应。
非常感谢这里的异步专家的任何想法，我对此很迷茫！！！
好吧，经过大量的挖掘、测试，当然还有很多关于 asyncio 的知识，我自己让它工作了。感谢到目前为止的建议。
问题是来自 asyncio 的 event_loop 不是 运行ning；正如@hoefling 提到的，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_until_complete() 否则，循环永远不会 运行ning，因为这是 asyncio 默认工作的方式（并且 pytest 没有做任何改变它的工作方式） .
使用 pytest-asyncio，您的测试可以使用众所周知的异步/等待部分：
async def test_this_is_an_async_test(event_loop):
result = await my_async_function(param1, param2, param3)
assert result == 'expected'
pytest-asyncio 在这种情况下包装了上面的 run_until_complete() 调用，对其进行了大量总结，因此事件循环将 运行 并可供您的异步代码使用。
请注意：第二种情况下的 event_loop 参数在这里甚至不是必需的，pytest-asyncio 提供了一个可用于您的测试。
另一方面，当您测试 Tornado 应用程序时，您通常需要在测试期间启动并 运行ning 一个 http 服务器，监听一个众所周知的端口等，所以通常的方法是通过编写固定装置来获取 http 服务器，base_url（通常是 http://localhost:，带有未使用的端口等）。
pytest-tornado 是一个非常有用的工具，因为它为您提供了其中的几个固定装置：http_server、http_client、unused_port、base_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
但是，这样它不支持 async / await，实际上只有 Tornado 的 ioloop 可以通过 io_loop fixture 使用（尽管 Tornado 的 ioloop 默认使用 Tornado 5.0 下的 asyncio），所以你' d 需要组合 pytest.mark.gen_test 和 pytest.mark.asyncio， 但顺序正确！（我确实失败了）。
一旦我更好地理解可能是什么问题，这是下一个方法：
@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 的协程然后被包装在 pytest-tornado yield gen.coroutine 中，它不会得到事件循环 运行ning... 所以我的测试仍然失败同样的问题。对数据库的任何查询都返回一个等待事件循环的 Future 运行ning。
一旦我弥补了这个愚蠢的错误，我更新了代码：
@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 协同程序中，event_loop 运行s 是预期的协同程序！
但是还有一个小问题我也花了一点时间才意识到； pytest-asyncio 的 event_loop fixture 为每个测试创建一个新的事件循环，而 pytest-tornado 也创建一个新的 IOloop。测试仍然失败，但这次出现了不同的错误。
conftest.py 文件现在看起来像这样；请注意，我已经重新声明 event_loop fixture 以使用来自 pytest-tornado io_loop fixture 本身的 event_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
现在我所有的测试都成功了，我回到了一个快乐的人并且为我现在对 asyncio 生活方式有了更好的理解而感到自豪。酷！