今天的文章分享如下在 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.
比如:
- 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:
- @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, 比如说创建数据库引擎:
- @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 条数据:
- @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, 编写单元测试用例, 一个示例如下:
- 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, 用户名, 密码, 数据库名请自行修改后执行.
来源: http://database.51cto.com/art/202112/697341.htm