如何从 Pytest 的同一个测试中调用多个 SQLAlchemy `asyncio.run()` 函数？

Question

在 Postgres 后端使用 Pytest 6.2.5 和 SQLAlchemy 1.4 异步引擎。

我有两个测试简单更新和读取功能的测试。它们是 SQLAlchemy asyncio 方法。更新测试更新数据库中的值，读取函数断言新写入的值存在。目前，这是两个单独的测试，都通过了。我想将它们移到一个测试中，因为它们是相互依赖的。

将这两个函数放在一个测试中会导致测试失败。我可以看到它与异步功能有关，但我认为 @pytest.mark.asyncio 装饰器和 asyncio.run() 异步特定功能可以解决这个问题。

这两个测试通过：

@pytest.mark.asyncio
def test_update():
    return asyncio.run(update('appuser', {'name': 'Alice'}, {'org': 'Wonderland'}))


@pytest.mark.asyncio
def test_read():
    read = asyncio.run(read('appuser', query={'name': 'Alice'}, mods={'fields': ['org']}))
    assert read == [('Wonderland',)]

这失败了

@pytest.mark.asyncio
def test_read():
    asyncio.run(update('appuser', {'name': 'Alice'}, {'org': 'Wonderland'}))
    read = asyncio.run(read('appuser', query={'name': 'Alice'}, mods={'fields': ['org']}))
    assert read == [('Wonderland',)]

出现此错误：

>   ???
E   RuntimeError: Task <Task pending coro=<PostgresDb.read() running at file.py:262> cb=[_run_until_complete_cb() at /usr/lib/python3.7/asyncio/base_events.py:158]> got Future <Future pending cb=[Protocol._on_waiter_completed()]> attached to a different loop

asyncpg/protocol/protocol.pyx:338: RuntimeError

我尝试将 test_update 作为一个函数放在 test_read 函数中，并像这样从那里调用它，但错误是一样的：

@pytest.mark.asyncio
def test_read():
    def test_update():
        asyncio.run(DB.update('appuser', {'name': 'Alice'}, {'org': 'Wonderland'}))
    test_update()
    read = asyncio.run(DB.read('appuser', query={'name': 'Alice'}, mods={'fields': ['org']}))
    assert read == [('Wonderland',)]

我们如何在单个 Pytest 中运行多个 asyncio.run() 函数？

Answer 1

您将常规函数与异步函数混合在一起。

使用 @pytest.mark.asyncio 需要异步函数 (async def)。

此外，不要使用创建新循环且只能使用一次的 asyncio.run()，而是像这样尝试 awaiting：

@pytest.mark.asyncio
async def test_read():
    await update('appuser', {'name': 'Alice'}, {'org': 'Wonderland'})
    read = await read('appuser', query={'name': 'Alice'}, mods={'fields': ['org']})
    assert read == [('Wonderland',)]

如何从 Pytest 的同一个测试中调用多个 SQLAlchemy `asyncio.run()` 函数？

How to call multiple SQLAlchemy `asyncio.run()` functions from within the same test in Pytest?

python

sqlalchemy

pytest

python-asyncio