Skip to content

第 19 章 · 测试 Web 服务与集成测试

本章目标:明确集成层在测试金字塔中的位置,掌握进程内驱动 ASGI/WSGI 应用、mock 外部 HTTP 依赖、数据库事务回滚三大实战技术,并建立契约测试的意识。

19.1 集成测试的正确打开方式

单元测试验证零件,集成测试验证接线:路由→依赖注入→业务逻辑→数据库这一整条链路是否真的通。测试金字塔的经典配比(多单元、适中集成、少量端到端)对 Web 服务依然成立,但现代实践有一个关键升级:

不要为每个接口都起真实服务器。 用 TestClient/httpx 在进程内直接调用 app 对象,速度接近单元测试,覆盖却是全链路。

以 FastAPI + Starlette 的 TestClient(内部封装 httpx)为例:

python
# conftest.py
import pytest
from fastapi.testclient import TestClient
from myapp.main import app          # 你的 FastAPI 实例


@pytest.fixture
def client():
    # lifespan/startup 事件会在首次请求时触发,with 写法保证关闭事件执行
    with TestClient(app) as c:
        yield c


# test_users_api.py
def test_create_and_get_user(client):
    resp = client.post("/users", json={"name": "高同学", "age": 28})
    assert resp.status_code == 201
    uid = resp.json()["id"]

    resp = client.get(f"/users/{uid}")
    assert resp.status_code == 200
    assert resp.json()["name"] == "高同学"

注意 with TestClient(app) 与直接实例化的区别:with 会完整走一遍 lifespan(启动时连数据库/缓存,退出时清理),无状态接口则可以直接 TestClient(app).get(...)

19.2 隔离外部 HTTP:responses 与 respx

被测代码调第三方 API 时,绝不能真的发请求——慢、贵、不稳定还不可复现。

  • 被测代码用 requests → mock 库选 responses
  • 被测代码用 httpx → 选 respx
python
# test_payment.py —— requests 场景
import responses
from myapp.pay import create_order          # 内部 requests.post 到支付网关


@responses.activate                          # 激活后所有真实出网请求都会报错
def test_create_order():
    responses.post(
        "https://api.pay.example/v1/orders",
        json={"order_id": "P123", "status": "created"},
        status=201,
    )

    order = create_order(amount=9900)

    assert order["order_id"] == "P123"
    assert len(responses.calls) == 1         # 还能断言"只调了一次"
python
# test_weather.py —— httpx 场景
import respx
from httpx import Response
from myapp.weather import get_forecast       # 内部用 httpx.get


@respx.mock
def test_get_forecast():
    respx.get("https://api.weather.example/today").respond(
        json={"temp_c": 21}, status_code=200
    )
    assert get_forecast() == 21

两个库都支持按 URL 匹配顺序返回不同响应(第一次 500、第二次 200),用来测试重试逻辑;也都能断言请求体/请求头,防止"参数悄悄改了但测试还绿着"。若项目已经全面使用 pytest-mock/unittest.mock,patch 到 HTTP 边界也可以,但专用库的声明式写法可读性好得多。

19.3 数据库测试策略:事务回滚

集成测试碰真库时,最大的工程难题是用例间隔离与速度的平衡。三种常见策略对比:

策略隔离性速度适用
每个用例建/删全部表最强最慢小项目
事务包裹 + 回滚绝大多数场景
每个会话独立 schema/database需要跨连接可见的场景

事务回滚是性价比最高的默认选择:

python
# conftest.py —— pytest-asyncio 版本见第 17 章,这里展示同步版
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from myapp.models import Base


@pytest.fixture(scope="session")
def engine():
    eng = create_engine("sqlite:///./test.db")
    Base.metadata.create_all(eng)        # 整个会话建一次表
    yield eng
    eng.dispose()


@pytest.fixture
def db(engine):
    conn = engine.connect()
    tx = conn.begin()                    # 开启外层事务
    Session = sessionmaker(bind=conn)
    session = Session()
    yield session
    session.close()
    tx.rollback()                        # 无论测试做了什么,一律回滚
    conn.close()


# test_repo.py
def test_user_roundtrip(db):
    from myapp.repository import save_user, get_user
    save_user(db, name="张三")
    assert get_user(db, "张三") is not None
    # 函数结束后数据自动消失,下个用例拿到干净的库

被测代码如果自己 commit,会把外层事务一起提交,回滚就失效了。解法是使用 SAVEPOINT(嵌套事务):conn.begin_nested() 并在每次 commit 后重启 savepoint,SQLAlchemy 官方文档"Joining a Session into an External Transaction"(将会话加入外部事务)一节有现成配方。

19.4 契约测试思想一瞥

当服务 A 依赖服务 B 的 API 时,B 改个字段名就能把 A 打挂——mock 测不出来,联调又太晚。契约测试(Consumer-Driven Contracts)的做法是:

  1. 消费方 A 把"B 必须满足的请求/响应结构"写成契约文件;
  2. 提供方 B 的 CI 里跑契约验证,违反即失败;
  3. 双方各自本地测试,靠同一份契约对齐,无需联调环境。

Python 生态的代表工具是 Pact(pact-python)。哪怕暂时不引入工具,思想也能落地:把第三方 API 的响应用 Pydantic 模型校验(第 4 章),模型本身就是一份活文档式的弱契约——上游字段漂移时你的测试会立刻红给你看。

本章小结

  • 集成测试测"接线",进程内 TestClient/ASGITransport 是速度与覆盖的最优解;
  • 外部 HTTP 一律 mock:requests 配 responses,httpx 配 respx;
  • 数据库隔离首选事务回滚 fixture;被测代码自行 commit 时需上 SAVEPOINT;
  • 契约测试让上下游解耦演进,Pydantic 校验是其轻量替代。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. with TestClient(app) 相比直接 TestClient(app),额外保证了什么?

2. 被测代码内部使用 httpx 调第三方 API,最合适的 mock 工具是?

3. 数据库测试的事务回滚策略为什么会失效?

4. 契约测试(如 Pact)解决的核心问题是?

🛠️ 动手实践

  1. 给第 13 章的数据库应用补一套集成测试:一个事务回滚 fixture + 一个"故意 commit 后仍能隔离"的 SAVEPOINT 版本。
  2. 为某个调用外部 API 的函数分别写 responses 和 respx 两个版本的测试,并加上"第一次超时第二次成功"的重试路径。
  3. 用 Pydantic 定义一份第三方 API 的响应模型作为轻量契约,写一个测试断言真实样例 JSON 通过校验。

最后一步:把所有武器装进 CI 流水线与生产治理