社区所有版块导航
Python
python开源   Django   Python   DjangoApp   pycharm  
DATA
docker   Elasticsearch  
aigc
aigc   chatgpt  
WEB开发
linux   MongoDB   Redis   DATABASE   NGINX   其他Web框架   web工具   zookeeper   tornado   NoSql   Bootstrap   js   peewee   Git   bottle   IE   MQ   Jquery  
机器学习
机器学习算法  
Python88.com
反馈   公告   社区推广  
产品
短视频  
印度
印度  
Py学习  »  tornado

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

mydaemon • 6 年前 • 1624 次点击  

我有一个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则返回未经授权的响应。

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

Python社区是高质量的Python/Django开发社区
本文地址:http://www.python88.com/topic/30822
 
1624 次点击  
文章 [ 1 ]  |  最新文章 6 年前
mydaemon
Reply   •   1 楼
mydaemon    6 年前

嗯,经过大量的挖掘、测试,当然,还学习了很多关于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

现在我所有的测试都成功了,我回到了一个快乐的人,并且为我现在更好地理解异步的生活方式而感到骄傲。酷!