使用pytest、tornado和aiopg进行单元测试失败,任何查询都失败

我有一个RESTAPI在python 3.7+tornado 5上运行,PostgreSQL作为数据库,使用aiopg和sqlachemy核心(通过aiopg.sa绑定)。对于单元测试,我将py.test与py test tornado结合使用。

只要不涉及对数据库的查询,所有测试都会正常进行,在这里我会得到:

运行时错误:task cb=[ioloop.add_future….()at venv/lib/python3.7/site packages/tornado/ioloop.py:719]>将future附加到其他循环

同样的代码在测试中也能正常工作,到目前为止,我已经能够处理100个请求了。

这是@auth decorator的一部分,它将检查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()只使用来自该池的一个连接。我将此池共享给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

此测试失败,因为它返回401而不是204,因为对auth decorator的查询由于runtimeerror而失败,而runtimeerror则返回未经授权的响应。

这里的异步专家的任何想法都会非常感谢,我对此完全迷路了!!!!