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

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

它永远不会被等待，您的代码会在信号量进入调度程序之前返回协程。

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

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

import asyncio

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

如果您有一个当前正在运行的事件循环，您可能希望使用该事件循环。您可以通过asyncio.get_event_loop() 来实现，它将在运行循环中运行该函数。

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()