1. 前情提要:
- fastapi官方文档只给了普通异步请求demo,没有orm数据库读写的示例
- 而tortoise-orm的文档还停留在同步版的测试范例
2. 需求:
单元测试函数不仅能通过HTTP请求访问API,还能读写数据库
3. 实现:
- 安装依赖:pip install httpx fastapi tortoise-orm pytest
- 在tortoise-orm范例的基础上,引入httpx并把测试函数改成异步,并配置好pytest使其在执行测试函数之前,配置好数据库连接,执行完测试之后关闭数据库连接,关键代码如下:
- conftest.py
import pytest
from httpx import AsyncClient
from tortoise import Tortoise
from main import app
DB_URL = "sqlite://:memory:"
async def init_db(db_url, create_db: bool = False, schemas: bool = False) -> None:
"""Initial database connection"""
await Tortoise.init(
db_url=db_url, modules={"models": ["models"]}, _create_db=create_db
)
if create_db:
print(f"Database created! {db_url = }")
if schemas:
await Tortoise.generate_schemas()
print("Success to generate schemas")
async def init(db_url: str = DB_URL):
await init_db(db_url, True, True)
@pytest.fixture(scope="session")
def anyio_backend():
return "asyncio"
@pytest.fixture(scope="session")
async def client():
async with AsyncClient(app=app, base_url="http://test") as client:
print("Client is ready")
yield client
@pytest.fixture(scope="session", autouse=True)
async def initialize_tests():
await init()
yield
- tests/test_users.py
import pytest
from httpx import AsyncClient
from models.users import User
@pytest.mark.anyio
async def test_testpost(client: AsyncClient):
name, age = ["sam", 99]
assert await User.filter(username=name).count() == 0
data = {"username": name, "age": age}
response = await client.post("/testpost", json=data)
assert response.json() == dict(data, id=1)
assert response.status_code == 200
response = await client.get("/users")
assert response.status_code == 200
assert response.json() == [dict(data, id=1)]
assert await User.filter(username=name).count() == 1
完整Demo见:
python - Testing in FastAPI using Tortoise-ORM - Stack Overflow