第十一章测试

本章目标：掌握 FastAPI 应用的自动化测试方法，包括 TestClient 同步测试、依赖覆盖（Mock）和异步测试。
预计时长：30 分钟

11.1 使用 TestClient 编写测试

准备工作

bash

pip install pytest httpx

TestClient 底层基于 httpx，所以需要安装它。

被测试的应用

python

# app/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

fake_db: dict[int, dict] = {}
next_id = 1


class ItemCreate(BaseModel):
    name: str
    price: float


class ItemResponse(BaseModel):
    id: int
    name: str
    price: float


@app.post("/items/", response_model=ItemResponse, status_code=201)
def create_item(item: ItemCreate):
    global next_id
    item_data = {"id": next_id, "name": item.name, "price": item.price}
    fake_db[next_id] = item_data
    next_id += 1
    return item_data


@app.get("/items/{item_id}", response_model=ItemResponse)
def get_item(item_id: int):
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="商品不存在")
    return fake_db[item_id]


@app.get("/items/", response_model=list[ItemResponse])
def list_items():
    return list(fake_db.values())

编写测试

python

# tests/test_items.py
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)


def test_create_item():
    """测试创建商品"""
    response = client.post("/items/", json={"name": "手机", "price": 4999.0})
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "手机"
    assert data["price"] == 4999.0
    assert "id" in data


def test_get_item():
    """测试获取商品"""
    # 先创建
    create_resp = client.post("/items/", json={"name": "耳机", "price": 299.0})
    item_id = create_resp.json()["id"]

    # 再获取
    response = client.get(f"/items/{item_id}")
    assert response.status_code == 200
    assert response.json()["name"] == "耳机"


def test_get_item_not_found():
    """测试获取不存在的商品"""
    response = client.get("/items/99999")
    assert response.status_code == 404
    assert response.json()["detail"] == "商品不存在"


def test_list_items():
    """测试列表接口"""
    response = client.get("/items/")
    assert response.status_code == 200
    assert isinstance(response.json(), list)


def test_create_item_validation_error():
    """测试参数校验：缺少必填字段"""
    response = client.post("/items/", json={"name": "手机"})
    assert response.status_code == 422  # Unprocessable Entity

运行测试

bash

pytest tests/ -v

输出示例：

tests/test_items.py::test_create_item PASSED
tests/test_items.py::test_get_item PASSED
tests/test_items.py::test_get_item_not_found PASSED
tests/test_items.py::test_list_items PASSED
tests/test_items.py::test_create_item_validation_error PASSED

========================= 5 passed in 0.32s =========================

TestClient 常用方法

方法	对应 HTTP 方法	示例
`client.get(url)`	GET	`client.get("/items/1")`
`client.post(url, json=...)`	POST	`client.post("/items/", json={...})`
`client.put(url, json=...)`	PUT	`client.put("/items/1", json={...})`
`client.delete(url)`	DELETE	`client.delete("/items/1")`
`client.patch(url, json=...)`	PATCH	`client.patch("/items/1", json={...})`

测试带认证的接口

python

def test_protected_endpoint():
    """测试需要 Token 的接口"""
    # 先登录获取 Token
    login_resp = client.post("/token", data={
        "username": "zhangsan",
        "password": "123456",
    })
    token = login_resp.json()["access_token"]

    # 携带 Token 请求
    response = client.get(
        "/users/me",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 200
    assert response.json()["username"] == "zhangsan"

11.2 依赖覆盖

问题：测试不应该连真实数据库

在真实项目中，接口依赖数据库 Session：

python

# app/dependencies/database.py
from sqlalchemy.orm import Session
from app.database import SessionLocal

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

测试时不想连真实数据库，怎么办？用 dependency_overrides 替换依赖。

使用 dependency_overrides

python

# tests/test_users.py
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.main import app
from app.dependencies.database import get_db
from app.database import Base

# ======== 创建测试专用的内存数据库 ========
SQLALCHEMY_TEST_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_TEST_URL, connect_args={"check_same_thread": False})
TestSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建表
Base.metadata.create_all(bind=engine)


# ======== 覆盖依赖 ========
def override_get_db():
    db = TestSessionLocal()
    try:
        yield db
    finally:
        db.close()


app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)


# ======== 测试用例 ========
def test_create_user():
    response = client.post("/users/", json={
        "username": "testuser",
        "password": "testpass",
    })
    assert response.status_code == 201
    assert response.json()["username"] == "testuser"


def test_list_users():
    response = client.get("/users/")
    assert response.status_code == 200

原理图解

正常运行时：
  路由函数 → Depends(get_db) → 真实数据库 Session

测试运行时：
  路由函数 → Depends(get_db) → override_get_db() → 测试数据库 Session
                 ↑
     dependency_overrides 替换了这里

覆盖认证依赖

测试时跳过认证，直接返回模拟用户：

python

from app.dependencies.auth import get_current_user
from app.schemas.user import User


def override_get_current_user():
    return User(username="test_admin", role="admin")


app.dependency_overrides[get_current_user] = override_get_current_user

client = TestClient(app)


def test_admin_dashboard():
    """不需要真正登录，直接测试业务逻辑"""
    response = client.get("/admin/dashboard")
    assert response.status_code == 200
    assert "test_admin" in response.json()["message"]

使用 pytest fixture 管理

更规范的做法是用 fixture 管理覆盖和清理：

python

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.dependencies.database import get_db


@pytest.fixture()
def client(tmp_path):
    """每个测试用例使用独立的测试数据库"""
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from app.database import Base

    db_path = tmp_path / "test.db"
    engine = create_engine(f"sqlite:///{db_path}")
    TestSession = sessionmaker(bind=engine)
    Base.metadata.create_all(bind=engine)

    def override_get_db():
        db = TestSession()
        try:
            yield db
        finally:
            db.close()

    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    app.dependency_overrides.clear()  # 测试结束后清理覆盖

python

# tests/test_users.py
def test_create_user(client):  # client 来自 fixture
    response = client.post("/users/", json={
        "username": "zhangsan",
        "password": "123456",
    })
    assert response.status_code == 201

11.3 异步测试

为什么需要异步测试？

TestClient 是同步的，对于使用了 async def 的路由和异步数据库操作，有时需要真正的异步测试客户端。

使用 httpx.AsyncClient

bash

pip install pytest-asyncio httpx

python

# tests/test_async.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app


@pytest.mark.asyncio
async def test_read_root():
    """异步测试示例"""
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/")
        assert response.status_code == 200
        assert response.json() == {"message": "Hello, FastAPI!"}


@pytest.mark.asyncio
async def test_create_item_async():
    """异步测试：创建商品"""
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post("/items/", json={
            "name": "键盘",
            "price": 599.0,
        })
        assert response.status_code == 201
        data = response.json()
        assert data["name"] == "键盘"

异步 fixture

python

# tests/conftest.py
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from app.main import app


@pytest_asyncio.fixture
async def async_client():
    """异步测试客户端 fixture"""
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        yield client

python

# tests/test_async.py
import pytest


@pytest.mark.asyncio
async def test_list_items(async_client):
    response = await async_client.get("/items/")
    assert response.status_code == 200
    assert isinstance(response.json(), list)

pytest 配置

在 pyproject.toml 或 pytest.ini 中配置异步模式：

toml

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"  # 自动识别异步测试，无需每个函数加 @pytest.mark.asyncio

配置后可以简写：

python

async def test_example(async_client):
    response = await async_client.get("/")
    assert response.status_code == 200

TestClient vs AsyncClient 对比

特性	`TestClient`（同步）	`AsyncClient`（异步）
导入	`from fastapi.testclient import TestClient`	`from httpx import AsyncClient`
函数定义	`def test_xxx()`	`async def test_xxx()`
发起请求	`client.get(...)`	`await client.get(...)`
需要额外依赖	仅 `httpx`	`httpx` + `pytest-asyncio`
适用场景	大多数场景	异步数据库、需要真正异步执行
推荐程度	⭐⭐⭐⭐⭐ 优先使用	⭐⭐⭐ 需要时使用

建议：大多数情况下 TestClient 就够了，它内部会自动处理异步。只有在测试异步依赖或需要精确控制异步执行流程时，才用 AsyncClient。

11.4 动手练习

练习 1：基础测试

为以下应用编写完整的测试用例（至少 5 个）：

python

# app/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

todos: list[dict] = []


class TodoCreate(BaseModel):
    title: str
    done: bool = False


@app.post("/todos/", status_code=201)
def create_todo(todo: TodoCreate):
    item = {"id": len(todos) + 1, **todo.model_dump()}
    todos.append(item)
    return item


@app.get("/todos/")
def list_todos():
    return todos


@app.get("/todos/{todo_id}")
def get_todo(todo_id: int):
    for todo in todos:
        if todo["id"] == todo_id:
            return todo
    raise HTTPException(status_code=404, detail="Todo 不存在")

测试要覆盖：创建成功、列表查询、单个查询、不存在返回 404、参数校验错误返回 422。

练习 2：依赖覆盖

假设 get_todo 接口依赖 get_current_user，用 dependency_overrides 编写不需要真实登录的测试。

练习 3：异步测试

使用 httpx.AsyncClient 重写练习 1 中的测试用例。

本章小结

概念	要点
`TestClient`	基于 `httpx` 的同步测试客户端，最常用
测试方法	`client.get()`、`client.post(url, json={})`，断言 `status_code` 和 `json()`
`dependency_overrides`	用测试替身替换真实依赖（数据库、认证等）
fixture 管理	用 `conftest.py` 中的 pytest fixture 管理客户端和数据库
`AsyncClient`	异步测试客户端，通过 `ASGITransport` 连接 FastAPI 应用
`pytest-asyncio`	pytest 异步支持插件，`asyncio_mode = "auto"` 简化使用

下一章预告：测试通过了，接下来该把应用部署到生产环境了。下一章我们将学习 Uvicorn/Gunicorn 生产配置、Docker 容器化部署，以及上线前的安全检查清单。

第十一章 测试 ​

11.1 使用 TestClient 编写测试 ​

准备工作 ​

被测试的应用 ​

编写测试 ​

运行测试 ​

TestClient 常用方法 ​

测试带认证的接口 ​

11.2 依赖覆盖 ​

问题：测试不应该连真实数据库 ​

使用 dependency_overrides ​

原理图解 ​

覆盖认证依赖 ​

使用 pytest fixture 管理 ​

11.3 异步测试 ​

为什么需要异步测试？ ​

使用 httpx.AsyncClient ​

异步 fixture ​

pytest 配置 ​

TestClient vs AsyncClient 对比 ​

11.4 动手练习 ​

练习 1：基础测试 ​

练习 2：依赖覆盖 ​

练习 3：异步测试 ​

本章小结 ​

第十一章测试

11.1 使用 TestClient 编写测试

准备工作

被测试的应用

编写测试

运行测试

TestClient 常用方法

测试带认证的接口

11.2 依赖覆盖

问题：测试不应该连真实数据库

使用 dependency_overrides

原理图解

覆盖认证依赖

使用 pytest fixture 管理

11.3 异步测试

为什么需要异步测试？

使用 httpx.AsyncClient

异步 fixture

pytest 配置

TestClient vs AsyncClient 对比

11.4 动手练习

练习 1：基础测试

练习 2：依赖覆盖

练习 3：异步测试

本章小结