异步测试¶
前几章介绍了如何使用 TestClient
测试 FastAPI 应用,但 TestClient
不能在(同步的)Pytest 函数中测试或运行任何异步(async
)函数。
测试异步函数是非常必要的,例如,异步查询数据库。假设向 FastAPI 应用发送请求时,测试后端能否使用异步数据库支持库把数据写入数据库。
接下来,我们学习如何进行异步测试。
pytest-asyncio¶
测试时如需调用异步函数,测试函数必须支持异步。Pytest 的 pytest-asyncio
支持库允许异步调用指定的测试函数。
安装命令:
$ pip install pytest-asyncio
---> 100%
HTTPX¶
即便 FastAPI 应用使用普通函数(def
)代替了异步函数( async def
),但它在底层仍是异步应用。
TestClient
使用标准 pytest 在普通测试函数中调用异步 FastAPI 应用。但在异步函数内部,这种方式就不起作用了。要运行异步测试,不能再在测试函数中使用 TestClient
。
好在还有一个不错的备选方案 - HTTPX。
HTTPX 是支持 Python 3 的 HTTP 客户端,与 TestClient
类似,它也可以查询 FastAPI 应用。
如果您了解 Requests 支持库,就会发现 HTTPX 与 Request 的 API 几乎一样。
Request 与 HTTPX 最大的区别就是 HTTPX 不局限于同步,还能使用异步请求。
示例¶
main.py
模块的示例代码如下:
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Tomato"}
test_main.py
模块用于测试 main.py
,代码如下:
import pytest
from httpx import AsyncClient
from .main import app
@pytest.mark.anyio
async def test_root():
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Tomato"}
运行¶
使用以下命令运行测试:
$ pytest
---> 100%
技术细节¶
@pytest.mark.asyncio
标记告诉 pytest 这个测试函数以异步方式调用:
import pytest
from httpx import AsyncClient
from .main import app
@pytest.mark.anyio
async def test_root():
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Tomato"}
提示
注意,测试函数现在使用 async def
取代了之前使用 TestClient
时的 def
。
然后,用 app
创建 AsyncClient
,并使用 await
向它发送异步请求。
import pytest
from httpx import AsyncClient
from .main import app
@pytest.mark.anyio
async def test_root():
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Tomato"}
等效于:
response = client.get('/')
……使用 TestClient
时用于生成请求的这段代码。
提示
注意,asnyc/await
要与新的 AsyncClient
一起使用 - 因为请求是异步的。
其他异步函数调用¶
因为测试函数是异步的,与在代码中调用其他异步函数一样,除了能在测试中向 FastAPI 应用发送请求外,还可以调用(并等待 await
)其他异步函数。
提示
测试调用的异步函数时,如果遇到 RuntimeError: Task attached to a different loop
(例如,使用 MongoDB 的 MotorClient),请参阅 pytest-asyncio 仓库中的这个 Issue。