今天的文章分享如下在 FastAPI 框架下,使用 pytest 来自动化测试数据库相关的接口,文章的最后给出全部代码。
最近越来越喜欢使用 FastAPI 来写后端服务了,因为它是 Python 领域性能最好的 Web 框架,它专注于提供高性能的 Web API,其他方面并不限制你的手脚,可以随意使用你喜欢的三方库,这点类似于 Flask,可以量身定制你的后端架构,以满足自己的需求。
需要说明的是,后端服务基本是离不开关系型数据库的,我之前是使用 Django,Django 的 ORM 太优秀了,以至于我从 Django 转 FastAPI 有点很不适应。在 ORM 领域,可以说除了 Django 的 ORM,就是 SQLAlchemy 了。所以不用 Django,就必须会用 SQLAlchemy,要快速了解,看看廖雪峰的官方网站的使用 SQLAlchemy[1] 来快速入门。
FastAPI 涉及数据库的接口写起来并不难,跟着官方文档sql_databases[2],5 分钟,我们就可以生成关于数据库的增删改查的 Restful 风格的 API,难的是如何自动化的测试,
通常情况下,我们会使用 pytest 进行自动化单元测试,根据数据库的记录数来断言,但是,每测试一次,数据库中的记录就保存了下来,你下次测试时如果不手动清理,那测试仍然可能失败。
那怎么解决呢?
那就是利用数据库的回滚功能,会改变数据库记录的接口测试完成后让事务回滚,这样每次测试完成后,数据库的记录数是不变的,每次运行 pytest,数据库的记录数是不变的,这样就可以进行自动化测试。
要想实现这一点,我们需要借助于 pytest 的 fixture 功能。
pytest.fixture 是一个装饰器,用于声明函数是一个 fixture。如果测试函数的参数列表中包含 fixture 名,那么 pytest 会检测到,并在测试函数运行之前执行 fixture。
比如:
代码语言:javascript复制import pytest
@pytest.fixture()
def some_data():
return 42
def test_some_data(some_data):
assert some_data==42
fixture 包含一个 scope 的可选参数,用于控制 fixture 执行配置和销毁逻辑的频率:
- scope='function' 函数级别的 fixture 每个测试函数只运行一次。配置代码在测试用例运行之前运行,销毁代码在测试用例运行之后执行。function 是 fixture 参数的默认值。
- scope='class' 类级别的 fixture 每个测试类只运行一次,不管测试类中有多少个类方法都可以共享这个 fixture
- scope='module' 模块级别的 fixture 每个模块只运行一次,不管模块里有多少个测试函数,类方法或其他 fixture 都可以共享这个fixture
- scope='session' 会话级别的 fixture 每次会话只运行一次。一次 pytest 会话中的所有测试函数、方法都可以共享这个 fixture
比如说让数据库回滚的,我们就可以写一个这样的 fixture:
代码语言:javascript复制@pytest.fixture(scope="function")
def db(db_engine):
connection = db_engine.connect()
# begin a non-ORM transaction
connection.begin()
# bind an individual Session to the connection
db = Session(bind=connection)
# db = Session(db_engine)
app.dependency_overrides[get_db] = lambda: db
yield db
db.rollback()
connection.close()
当然还有很多 fixture,比如说创建数据库引擎:
代码语言:javascript复制@pytest.fixture(scope="session")
def db_engine():
engine = create_engine(SQLALCHEMY_DATABASE_URL)
if not database_exists:
create_database(engine.url)
Base.metadata.create_all(bind=engine)
yield engine
再比如,在测试前,数据库中先插入 2 条数据:
代码语言:javascript复制@pytest.fixture
def items(db):
create_item(db, schemas.ItemCreate(title="item 1"))
create_item(db, schemas.ItemCreate(title="item 2"))
把这些 fixture 函数放在文件名conftest.py
中,pytest 会自动读取并执行。至于为什么放在 conftest.py中,请查阅 pytest 文档,这里不展开,
接下来,利用这些 fixture,编写单元测试用例,一个示例如下:
代码语言:javascript复制from fastapi.testclient import TestClient
from . import crud
from .main import app
def test_post_items(db):
client = TestClient(app)
client.post("/items/", json={"title": "Item 1"})
client.post("/items/", json={"title": "Item 2"})
client.post("/items/", json={"title": "Item 3"})
items = crud.get_items(db)
assert len(items) == 3
def test_list_items(items, client):
response = client.get("/items")
assert len(response.json()) == 2
其中 test_post_items,测试的是提交了 3 个数据,然后断言数据库中的记录数为 3。test_list_items 有个参数是 items,会调用之前的 fixture,提前往数据库插入了 2 条记录,因此断言记录数为 2。
每个测试函数执行时互不影响,执行完成后,数据库都会回滚,测试前 items 是空的,测试之后 表仍然是空的,这样就可以自动进行数据库的测试了。
完整代码
不能选择 sqlite 数据库进行测试,因为它不支持并发访问。
代码的数据库配置为 mysql,用户名、密码、数据库名请自行修改后执行。
最后的话
本文分享了如下在 FastAPI 框架下,使用 pytest 来自动化测试数据库相关的接口,希望对你的单元测试技能有所帮助。如果有帮助,请点点赞、在看、关注支持。如果有问题,可以留言讨论哈。
参考资料
[1]使用 SQLAlchemy: https://www.liaoxuefeng.com/wiki/1016959663602400/1017803857459008
[2]sql_databases: https://fastapi.tiangolo.com/tutorial/sql-databases/