接口写完的那一刻,是信心最虚的时候——「我觉得能跑」。这一章教你用三件武器把「我觉得」换成「验证过」:Postman 手工走通业务链路、pytest 自动回归、curl 现场取证;再补上Swagger 自动生成接口文档,让前端不用再追着你要参数说明。
测试金字塔:底层的单元测试要多、快、稳;中层的接口测试覆盖主流程;顶层的端到端测试少而慢。本课重点在「接口测试 + pytest 单元/集成测试」。
把登录请求和各业务请求放进同一个 Postman 集合,登录后把 token 写入环境变量 {{token}};后续请求在 Authorization 头里引用它,省去每次手动复制。
{
"name": "古典文学网-本地",
"values": [
{ "key": "base_url", "value": "http://127.0.0.1:5000/api", "enabled": true },
{ "key": "token", "value": "", "enabled": true },
{ "key": "username", "value": "zhangsan", "enabled": true },
{ "key": "password", "value": "123456", "enabled": true },
{ "key": "article_id", "value": "1", "enabled": true }
]
}
// 1)先确认这次登录本身是成功的
pm.test("业务码 code = 200", function () {
const body = pm.response.json();
pm.expect(body.code).to.eql(200);
});
// 2)把 token 写进当前环境变量
// environment 是「环境级」,collection 是「集合级」,
// 需要跨集合复用时用 global 级
const body = pm.response.json();
if (body.code === 200) {
pm.environment.set("token", body.data.token);
pm.environment.set("user_id", body.data.user.id);
}
// 3)顺便把登录耗时打印到控制台,方便排查慢查询
console.log("登录耗时(ms):" + pm.response.responseTime);
Postman 断言让用例自动判断:检查状态码、业务 code、关键字段类型。失败会标红并给出期望与实际值,回归时一键跑完整个集合即可定位问题接口。
// ① HTTP 层:网络是通的
pm.test("HTTP 状态码为 200", function () {
pm.response.to.have.status(200);
});
// ② 业务层:统一响应结构对得上
const body = pm.response.json();
pm.test("统一响应结构正确", function () {
pm.expect(body).to.have.all.keys("code", "msg", "data");
pm.expect(body.code).to.eql(200);
pm.expect(body.msg).to.be.a("string");
});
// ③ 数据层:分页字段齐全、条数不超过 page_size
pm.test("返回了标准分页结构", function () {
pm.expect(body.data).to.have.property("list");
pm.expect(body.data).to.have.property("total");
pm.expect(body.data).to.have.property("page");
pm.expect(body.data).to.have.property("page_size");
});
pm.test("列表长度不超过 page_size", function () {
pm.expect(body.data.list).to.be.an("array");
pm.expect(body.data.list.length).to.be.at.most(body.data.page_size);
});
pm.test("每条作品都有标题和作者", function () {
body.data.list.forEach(function (item) {
pm.expect(item.title).to.be.a("string").and.not.empty;
pm.expect(item).to.have.property("author");
pm.expect(item).to.have.property("views");
});
});
// ④ 性能层:响应时间兜底
pm.test("响应时间小于 500ms", function () {
pm.expect(pm.response.responseTime).to.be.below(500);
});
// ⑤ 顺带把第一条作品的 id 存起来,供后续「详情 / 点赞」用例复用
if (body.data.list.length > 0) {
pm.environment.set("article_id", body.data.list[0].id);
}
pytest 按命名约定(test_*.py / test_*)自动收集并执行用例,assert 判断结果,最后给出通过/失败报告。配合 fixture 准备测试数据,用例之间互不干扰。
$ pip install pytest requests
$ pip freeze > requirements.txt
# 目录约定:测试文件放在项目根的 tests/ 下,名字以 test_ 开头
hiflask/
├── app.py
├── extensions.py
├── models/
├── views/
└── tests/
├── conftest.py # 放公共 fixture
├── test_user_api.py
└── test_article_api.py
# client 这个 fixture 来自 conftest.py,本节末尾就会写
def test_health(client):
"""冒烟测试:服务活着、能返回统一响应结构"""
resp = client.get("/api/health")
assert resp.status_code == 200
body = resp.get_json()
assert body["code"] == 200
assert body["data"]["status"] == "up"
def test_register_then_login(client):
"""主流程:注册 → 登录 → 拿到 token"""
resp = client.post("/api/user/register",
json={"username": "zhangsan",
"password": "123456",
"nickname": "张三"})
body = resp.get_json()
assert body["code"] == 200
assert body["msg"] == "注册成功"
assert body["data"]["user_id"] > 0
resp = client.post("/api/user/login",
json={"username": "zhangsan",
"password": "123456"})
body = resp.get_json()
assert body["code"] == 200
assert len(body["data"]["token"]) > 20
assert body["data"]["user"]["username"] == "zhangsan"
# 顺手确认敏感字段没有泄漏
assert "password" not in body["data"]["user"]
def test_register_duplicate_username(client):
"""异常分支:用户名重复必须被拦住"""
payload = {"username": "zhangsan", "password": "123456"}
client.post("/api/user/register", json=payload)
resp = client.post("/api/user/register", json=payload)
body = resp.get_json()
assert body["code"] == 400
assert body["msg"] == "用户名已存在"
def test_register_short_password(client):
"""异常分支:密码长度校验"""
resp = client.post("/api/user/register",
json={"username": "lisi", "password": "123"})
assert resp.get_json()["code"] == 400
def test_profile_requires_token(client):
"""鉴权分支:不带 token 访问需登录接口"""
resp = client.get("/api/user/profile")
body = resp.get_json()
assert body["code"] == 401
def test_profile_with_token(client):
"""鉴权主流程:注册 → 登录 → 拿 token → 查个人信息"""
client.post("/api/user/register",
json={"username": "tester", "password": "123456"})
token = client.post("/api/user/login",
json={"username": "tester",
"password": "123456"}) \
.get_json()["data"]["token"]
resp = client.get("/api/user/profile",
headers={"Authorization": f"Bearer {token}"})
body = resp.get_json()
assert body["code"] == 200
assert body["data"]["username"] == "tester"
assert body["data"]["stat"]["favorite_count"] == 0
pytest 跑测试时切换到独立的测试库(或开启事务回滚),用例产生的脏数据在跑完就被清掉。这样自动化测试可以反复跑,又不会弄脏你手点调试用的开发库。
class TestingConfig(Config):
"""pytest 专用:内存 SQLite,跑完即销毁"""
TESTING = True
SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"
JWT_SECRET_KEY = "test-only-secret"
# Flask-SQLAlchemy 3.x 对内存 SQLite 会自动使用 StaticPool,
# 并设置 check_same_thread=False,所以同一个内存库在多个连接间是共享的。
# 不需要自己写 SQLALCHEMY_ENGINE_OPTIONS。
import pytest
from app import create_app
from extensions import db
@pytest.fixture()
def app():
"""每个用例一个全新的应用实例 + 一个全新的空库"""
app = create_app("testing")
with app.app_context():
db.create_all()
yield app # 用例在这一行执行
db.session.remove() # 先断开会话,避免删表时连接被占用
db.drop_all()
@pytest.fixture()
def client(app):
"""测试客户端:不用真的起服务器,直接在进程内发请求"""
return app.test_client()
@pytest.fixture()
def auth_headers(client):
"""通用 fixture:注册一个测试用户并登录,返回可直接用的请求头
需要登录的用例只要在参数里写上 auth_headers,就自动带上 token。
"""
client.post("/api/user/register",
json={"username": "tester", "password": "123456"})
token = client.post("/api/user/login",
json={"username": "tester",
"password": "123456"}) \
.get_json()["data"]["token"]
return {"Authorization": f"Bearer {token}"}
# pip install flasgger
from flasgger import Swagger
SWAGGER_CONFIG = {
"headers": [],
"specs": [{
"endpoint": "apispec",
"route": "/apidocs.json",
"rule_filter": lambda rule: "/api/" in rule.rule,
"model_filter": lambda tag: "api" in (tag or []),
}],
"static_url_path": "/flasgger_static",
"swagger_ui": True,
"specs_route": "/apidocs/", # 文档页面地址
}
def create_app(config_name=None):
app = Flask(__name__)
# ... 加载配置、初始化扩展、注册蓝图 ...
Swagger(app, config=SWAGGER_CONFIG, merge=True)
return app
@article_bp.get("/list")
def article_list():
"""作品列表(分页 / 筛选 / 搜索)
---
tags:
- 作品
parameters:
- name: page
in: query
type: integer
default: 1
description: 页码,从 1 开始
- name: page_size
in: query
type: integer
default: 10
description: 每页条数
- name: category_id
in: query
type: integer
description: 分类 ID,如 1 表示唐诗
- name: keyword
in: query
type: string
description: 标题或摘要的模糊关键词
- name: order_by
in: query
type: string
enum: [latest, hot, likes]
default: latest
responses:
200:
description: 统一响应,data 内含 list / total / page / page_size
"""
page = request.args.get("page", 1, type=int)
# ... 其余实现不变 ...
# ① 健康检查:确认服务活着
$ curl -s http://127.0.0.1:5000/api/health
# ② 作品列表:带分页与排序参数(注意 URL 里的 & 要放在引号里)
$ curl -s "http://127.0.0.1:5000/api/article/list?page=1&page_size=2&order_by=hot"
# ③ 登录:-X 指定方法,-H 设请求头,-d 传 JSON 请求体
$ curl -s -X POST http://127.0.0.1:5000/api/user/login \
-H "Content-Type: application/json" \
-d '{"username": "zhangsan", "password": "123456"}'
# ④ 把 token 存进 shell 变量,再带着它查个人信息
$ TOKEN=$(curl -s -X POST http://127.0.0.1:5000/api/user/login \
-H "Content-Type: application/json" \
-d '{"username": "zhangsan", "password": "123456"}' \
| python -c "import sys, json; print(json.load(sys.stdin)['data']['token'])")
$ curl -s http://127.0.0.1:5000/api/user/profile \
-H "Authorization: Bearer $TOKEN"
# ⑤ 只看状态码,不看响应体(写脚本时常用)
$ curl -s -o /dev/null -w "%{http_code}\n" http://127.0.0.1:5000/api/health
# ============ 第 1 题:Postman → 「注册」请求的 Tests 标签 ============
pm.test("注册成功,code = 200", function () {
pm.expect(pm.response.json().code).to.eql(200);
});
pm.test("返回了数字类型的 user_id", function () {
const uid = pm.response.json().data.user_id;
pm.expect(uid).to.be.a("number");
pm.expect(uid).to.be.above(0);
pm.environment.set("user_id", uid);
});
# 「重复注册」请求的 Tests 标签(Body 里用户名和上面保持一致)
pm.test("重复注册被拒绝", function () {
const body = pm.response.json();
pm.expect(body.code).to.eql(400);
pm.expect(body.msg).to.eql("用户名已存在");
pm.expect(body.data).to.be.null;
});
# ============ 第 2 题:tests/test_user_api.py ============
def test_login_wrong_password(client):
"""密码错误必须返回统一的模糊提示,不能暴露账号是否存在"""
client.post("/api/user/register",
json={"username": "zhangsan", "password": "123456"})
resp = client.post("/api/user/login",
json={"username": "zhangsan",
"password": "wrong-pwd"})
body = resp.get_json()
assert body["code"] == 400
assert body["msg"] == "用户名或密码错误"
# 为什么统一提示?如果「用户名不存在」单独提示,攻击者就能拿一份
# 用户名字典逐个试:返回「密码错误」说明账号存在,返回「用户不存在」说明不存在。
# 这叫「用户名枚举漏洞」,所以要统一提示、统一耗时。
# ============ 第 3 题:tests/test_movie_api.py ============
def test_movie_detail(client):
resp = client.get("/api/movie/detail/1")
body = resp.get_json()
assert body["code"] == 200
assert body["data"]["id"] == 1
assert "title" in body["data"]
def test_movie_detail_views_plus_one(client):
first = client.get("/api/movie/detail/1").get_json()["data"]["views"]
second = client.get("/api/movie/detail/1").get_json()["data"]["views"]
assert second == first + 1
def test_movie_detail_not_found(client):
resp = client.get("/api/movie/detail/999999")
body = resp.get_json()
assert body["code"] == 404
assert body["msg"] == "电影不存在"
# ============ 第 3 题:curl 验证(真机上跑)============
# 正常访问
$ curl -s http://127.0.0.1:5000/api/movie/detail/1 | python -m json.tool
# 连跑两次看 views 是否 +1
$ curl -s http://127.0.0.1:5000/api/movie/detail/1 | grep -o '"views": [0-9]*'
$ curl -s http://127.0.0.1:5000/api/movie/detail/1 | grep -o '"views": [0-9]*'
# id 不存在
$ curl -s http://127.0.0.1:5000/api/movie/detail/999999
第 3 题的关键:换掉的是「资源名 movie 和它的字段」,不变的是「正常 / 边界 / 异常三条用例 + 一条命令验证」这套骨架。第 2 题那个「为什么统一提示」的思考,是接口安全里最容易被忽略的一课。