如何在单元测试时异步从数据库中获取记录？

Question

在此单元测试中，我想检查设备是否已创建以及到期日期是否为未来 7 天。

database.py

import databases
database = databases.Database(settings.sqlalchemy_database_uri)

单元测试：

from database.database import database

def test_successful_register_expiry_set_to_seven_days():
    response = client.post(
        "/register/",
        headers={},
        json={"device_id": "u1"},
    )
    assert response.status_code == 201
    query = device.select(whereclause=device.c.id == "u1")
    d = database.fetch_one(query)
    assert d.expires_at == datetime.utcnow().replace(microsecond=0) + timedelta(days=7)

因为 d 是协程对象，所以失败并显示消息：

AttributeError: 'coroutine' object has no attribute 'expires_at'

而且我不能在单元测试中使用 await。

d = await database.fetch_one(query)

请问我遗漏了什么？

Answer 1

好吧，它永远不会等待，你的代码 returns 信号量进入调度程序之前的协程。

如果您使用的是异步驱动程序，则需要等待它。有什么解决方法吗？是的

您可以使用 asyncio.run(awaitable) 到运行事件循环中的协程。

import asyncio

d = asyncio.run(database.fetch_one(query))

如果您目前有一个运行ning 事件循环，您可能想改用该事件循环。您可以通过 asyncio.get_event_loop() 实现，这将运行运行ning 循环中的函数。

import asyncio

asyncio.get_event_loop().run_until_complete(database.fetch_one(query))

您还可以使用 @pytest.mark.asyncio 装饰器（参见 documentation）。

@pytest.mark.asyncio
async def dummy():
    await some_awaitable()

如何在单元测试时异步从数据库中获取记录？

How to fetch a record from db asynchronously while unit testing?

python

sqlalchemy

fastapi