使用pytest、tor旋风和aiopg的单元测试失败，任何查询都失败

Question

我有一个REST运行在Python3.7+旋风5上，将postgresql作为数据库，使用带有SQLAlchemy内核的aiopg (通过aiopg.sa绑定)。在单元测试中，我使用py.test和pytest-龙卷风。

只要不涉及对数据库的查询，所有测试都会正常，我可以得到以下内容：

运行时错误: venv/lib/python3.7/site-packages/tornado/ioloop.py:719>的任务cb=IOLoop.add_future.()将未来附加到不同的循环中

在测试中，相同的代码运行良好，到目前为止，我能够处理100多个请求。

这是@auth装饰器的一部分，它将检查JWT令牌的授权头，对其进行解码并获取用户的数据并将其附加到请求中；这是查询的一部分：

                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()异步方法来自于应用程序中所有模型的基类。这是.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()方法将返回模型实例(行表示)或无。

它使用db.acquire()上下文管理器，如aiopg的文档中所述：https://aiopg.readthedocs.io/en/stable/sa.html。

正如在同一文档中所描述的，sa.create_engine()方法返回一个连接池，因此db.acquire()只使用池中的一个连接。我将这个池共享到“旋风”中的每个请求中，他们在需要时使用它来执行查询。

这就是我在我的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

我无法解释为什么会发生这种情况；“旋风”的文档和源代码清楚地表明，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

这个测试失败，因为它返回401而不是204，因为在auth装饰器上的查询由于RuntimeError而失败，然后返回一个未经授权的响应。

这里的异步专家的任何想法都将是非常感谢的，我对此非常迷茫！

Answer 1

嗯，经过大量的挖掘、测试，当然，我也学了很多关于异步的知识，我让它自己发挥作用。到目前为止，谢谢你的建议。

问题是来自异步的event_loop没有在运行；正如@hoefling所提到的，pytest本身不支持协同，但是pytest-异步给您的测试带来了这样一个有用的特性。这里很好地解释了这一点：https://medium.com/ideas-at-igenius/testing-asyncio-python-code-with-pytest-a2f3628f82bc

因此，如果没有pytest-异步，则需要测试的异步代码如下所示：

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()，否则，循环将永远不会运行，因为默认情况下这就是异步的工作方式(而且pytest没有使其工作方式不同)。

使用pytest-异步，您的测试可以使用众所周知的异步/等待部分：

async def test_this_is_an_async_test(event_loop):
   result = await my_async_function(param1, param2, param3)
   assert result == 'expected'

pytest-异步在本例中封装了上面的run_until_complete()调用，并对其进行了大量总结，因此事件循环将运行并可供您的异步代码使用。

请注意:第二种情况下的event_loop参数在这里甚至是不必要的，pytest-异步为您的测试提供了一个可用的参数。

另一方面，在测试龙卷风应用程序时，您通常需要在测试期间启动并运行http服务器，在著名端口监听等等，所以通常的方法是编写补丁以获得base_url服务器(通常是http://localhost:，带有未使用的端口等等)。

龙卷风是非常有用的，因为它为你提供了几个这样的装置: http_server，http_client，unused_port，base_url等等。

还值得一提的是，它获得了pytest标记的gen_test()特性，它将任何标准测试转换为通过产率使用协同测试，甚至断言它将在给定的超时情况下运行，如下所示：

    @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

但是，通过这种方式它不支持异步/等待，实际上只有龙卷风的ioloop可以通过io_loop夹具获得(尽管在龙卷风5.0下面默认使用的是龙卷风的ioloop )，所以您需要将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-异步的coroutine然后被包装在一个pytest-龙卷风产量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-异步协同器中，event_loop按预期运行协同！

但是还有一个小问题，我也花了一点时间才意识到；pytest-异步的event_loop夹具为每个测试创建了一个新的事件循环，而pytest-龙卷风创建了一个新的IOloop。测试仍然失败，但这次有不同的错误。

conftest.py文件现在看起来如下所示；请注意，我已经重新声明了event_loop夹具，以使用pytest- io_loop event_loop本身(请回忆一下，pytest-龙卷风在每个测试函数上创建了一个新的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

现在我所有的测试都成功了，我回到了一个快乐的人，并且为我现在对异步生活方式的更好的理解而感到自豪。凉爽的!