接口测试,就是用程序发送请求,再检查接口有没有完成约定的操作。
以“新增笔记”为例:向 POST /notes 发送标题,接口把笔记保存到 PostgreSQL,并返回笔记 ID。测试要检查两件事:响应中有没有 ID,以及数据库里这个 ID 对应的标题是否正确。
整个过程是:
发送 {"title": "学习 Flask"}
→ 接口向数据库插入笔记
→ 接口返回笔记 ID
→ 测试用这个 ID 查询数据库
→ 检查标题是不是“学习 Flask”检查完成后,再清理测试产生的数据。本文使用事务回滚做清理。判断接口是否正确,靠的是响应和查询结果;回滚负责把测试数据撤销。
下面用这一个接口走完准备环境、编写测试、运行测试和清理数据的过程。
1. 先看一个接口测试怎么写
下面是测试的主体,后面的示例会补齐它使用的 client 和 connection:
def test_create_note(client, connection):
# 发送请求,让接口执行新增操作。
response = client.post("/notes", json={"title": "学习 Flask"})
# 检查接口的响应。
assert response.status_code == 201
note_id = response.get_json()["id"]
# 查询数据库,检查接口写入的内容。
title = connection.execute(
text("SELECT title FROM note WHERE id = :id"),
{"id": note_id},
).scalar_one()
assert title == "学习 Flask"client.post() 相当于在测试代码中发起一次 POST 请求,json 是请求体。Flask 的测试客户端会调用对应的接口函数,不需要先运行 flask run。它在进程内处理请求,不经过浏览器或网络服务器。参见 Flask 的 Testing Flask Applications。
assert 表示“检查这个条件是否成立”。如果接口返回的状态码不是 201,或者查出的标题不是“学习 Flask”,pytest 就会把这次测试标为失败。scalar_one() 要求查询返回一行,取出其中的标题;查不到记录也会使测试失败。
这里没有在测试中插入笔记。新增操作由接口完成,测试只发送请求和查询结果。如果测试自己插入记录,再查询自己插入的数据,就没有检查到新增接口。
2. 准备一个测试数据库
示例需要 Python 3.10 或以上版本,以及运行中的 Docker。四个文件放在同一个目录:
flask-api-test/
├── docker-compose.test.yml
├── app.py
├── conftest.py
└── test_notes.py在这个目录安装依赖:
python3 -m venv .venv
.venv/bin/python -m pip install 'Flask>=3,<4' 'SQLAlchemy>=2.0,<2.1' psycopg2-binary pytestFlask 提供接口,SQLAlchemy 管理数据库连接和事务,psycopg2 是连接 PostgreSQL 的驱动,pytest 负责执行测试函数。
docker-compose.test.yml 定义一个专供测试使用的 PostgreSQL:
services:
postgres-test:
image: postgres:17-alpine
environment:
POSTGRES_DB: example_api_test
POSTGRES_USER: example_test
POSTGRES_PASSWORD: example_test_only
ports:
- '127.0.0.1:55439:5432'
tmpfs:
- /var/lib/postgresql/data
healthcheck:
test: ['CMD-SHELL', 'pg_isready -U example_test -d example_api_test']
interval: 1s
timeout: 3s
retries: 30这里的库名、账号和密码都是示例值。程序通过本机的 55439 端口连接这个数据库。它与开发库分开,测试不会向开发库添加笔记。
启动数据库:
docker compose -p flask-api-article-test -f docker-compose.test.yml up -d --wait--wait 等待数据库通过健康检查后再结束命令。tmpfs 把数据库数据放在临时存储中,容器停止后数据不保留。
3. 编写被测试的新增接口
app.py:
from uuid import uuid4
from flask import Flask, jsonify, request
from sqlalchemy import text
def create_app(session_factory):
app = Flask(__name__)
@app.post("/notes")
def create_note():
body = request.get_json(silent=True)
title = body.get("title") if isinstance(body, dict) else None
if not isinstance(title, str) or not title.strip():
return jsonify(error="标题不能为空"), 400
note_id = uuid4().hex
with session_factory() as session:
session.execute(
text("INSERT INTO note (id, title) VALUES (:id, :title)"),
{"id": note_id, "title": title.strip()},
)
session.commit()
return jsonify(id=note_id), 201
return app这个接口先检查标题,再生成 ID、执行 INSERT,最后返回 ID。标题为空时返回 400,新增成功时返回 201。
session_factory() 创建一个 SQLAlchemy Session。这里可以把 Session 理解为执行 SQL、管理提交和回滚的对象。create_app() 把创建 Session 的函数作为参数接收,让测试可以指定它连接哪个数据库。
SQL 中的 :id 和 :title 是参数占位符,值通过后面的字典传入,不把用户输入拼接进 SQL 字符串。
4. 为每次测试准备连接
pytest 的 fixture 用来准备测试需要的对象。下面的 client fixture 返回 Flask 测试客户端,connection fixture 返回数据库连接。测试函数把这两个名字写在参数中,pytest 就会把对应对象传进来。
conftest.py 是 pytest 自动查找 fixture 的文件:
import os
import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.engine import make_url
from sqlalchemy.orm import sessionmaker
from app import create_app
@pytest.fixture(scope="session")
def engine():
url = make_url(os.environ["TEST_DATABASE_URL"])
if (
url.host != "127.0.0.1"
or url.port != 55439
or url.database != "example_api_test"
):
raise RuntimeError("请使用本例的测试数据库")
engine = create_engine(url)
try:
with engine.begin() as connection:
connection.execute(text(
"CREATE TABLE IF NOT EXISTS note "
"(id TEXT PRIMARY KEY, title TEXT NOT NULL)"
))
yield engine
finally:
engine.dispose()
@pytest.fixture()
def connection(engine):
with engine.connect() as connection:
transaction = connection.begin()
try:
yield connection
finally:
transaction.rollback()
@pytest.fixture()
def client(connection):
factory = sessionmaker(
bind=connection,
join_transaction_mode="create_savepoint",
)
app = create_app(factory)
app.config["TESTING"] = True
return app.test_client()三个 fixture 分别做一件事:
engine:设置数据库连接,创建note表。整次运行准备一次。connection:为每个测试打开连接、开始事务,结束时回滚。client:创建 Flask 应用,让接口使用这次测试的数据库连接。
yield connection 把连接交给测试函数。测试结束后,包括断言失败时,finally 中的回滚都会执行。fixture 的基础用法见 Pytest fixture 概念及用法。
这里先记住 join_transaction_mode="create_savepoint" 是配合回滚使用的设置。先运行测试,再看它为什么需要存在。
5. 检查正常输入和错误输入
test_notes.py:
from sqlalchemy import text
def test_create_note(client, connection):
response = client.post("/notes", json={"title": "学习 Flask"})
assert response.status_code == 201
note_id = response.get_json()["id"]
title = connection.execute(
text("SELECT title FROM note WHERE id = :id"),
{"id": note_id},
).scalar_one()
assert title == "学习 Flask"
def test_empty_title(client, connection):
response = client.post("/notes", json={"title": ""})
assert response.status_code == 400
assert response.get_json()["error"] == "标题不能为空"
count = connection.execute(
text("SELECT COUNT(*) FROM note")
).scalar_one()
assert count == 0第一个测试检查新增成功后,数据库有没有保存提交的标题。第二个测试检查空标题是否被拒绝,以及数据库里有没有多出记录。
pytest 会执行名字以 test_ 开头的测试函数。每个函数对应一个要检查的情况,不需要在文件末尾手动调用它。
在示例目录运行:
TEST_DATABASE_URL='postgresql+psycopg2://example_test:example_test_only@127.0.0.1:55439/example_api_test' \
.venv/bin/python -m pytest -qTEST_DATABASE_URL 指定测试库的连接地址。缺少这个变量时,fixture 会报错;地址中的主机、端口或库名不符合示例配置时,也会停止。
预期输出包含:
2 passed这表示两个测试函数的断言都成立。若把新增接口改成保存固定标题,而不是请求中的标题,第一个测试会在 assert title == "学习 Flask" 处失败。
如果出现连接拒绝,先检查测试容器是否已启动、55439 端口是否被其他程序占用。如果出现 ModuleNotFoundError,检查安装依赖和执行 pytest 是否使用同一个 .venv。
6. 回滚为什么放在测试结束后
第一个测试创建了一条笔记,但没有调用删除接口。如果这条数据留下来,第二个测试查询总数时会得到 1,即使空标题接口没有写入数据,也会失败。
因此,每个测试结束后,都要撤销它产生的数据。这里用事务回滚完成清理:
第一个测试开始:表中有 0 条笔记
→ POST /notes 创建笔记
→ 表中有 1 条笔记
→ SELECT 查到标题,断言通过
→ fixture 回滚
第一个测试结束:表中有 0 条笔记
第二个测试开始:表中有 0 条笔记
→ POST /notes 提交空标题
→ 接口返回 400,没有插入笔记
→ SELECT 查到总数为 0,断言通过
→ fixture 回滚还有一个问题:接口里调用了 session.commit(),为什么测试最后还能撤销新增记录?
普通事务提交之后,不能靠另一次 rollback() 撤销。为此,fixture 先在 Connection 上开启一个外层事务,再让接口的 Session 在这个事务里使用 SAVEPOINT(保存点)。
join_transaction_mode="create_savepoint" 就是这个设置。接口调用 session.commit() 时,完成的是 Session 管理的保存点,测试持有的外层事务还没有提交。测试查询后,再回滚外层事务,新增记录就被撤销。这个用法见 SQLAlchemy 的 外部事务测试示例。
接口和测试中的 SELECT 使用同一条 Connection,所以可以查到事务内新增的笔记。这种写法适合检查接口执行的 SQL 和写入内容;如果要检查提交后另一个进程能否读到数据,就需要允许事务提交,另用连接查询,不能沿用这里的回滚方式。
测试完成后,停止并删除示例容器:
docker compose -p flask-api-article-test -f docker-compose.test.yml down