FastAPI 测试实战——TestClient、pytest fixtures、覆盖率完整方案

适读人群：有 FastAPI 开发经验、想建立完整测试体系的工程师 | 阅读时长：约 15 分钟 | 核心价值：掌握 FastAPI 项目测试的完整技术栈，从单元测试到覆盖率报告

---

那个在周五下午提交代码的同学

小文是我们组里代码写得最快的人，功能开发效率没得说，但有个习惯让所有人头疼——他几乎不写测试。每次提交完说"我本地测过了，没问题的"，结果上了测试环境就出 bug。

有一次，他改了一个认证依赖的逻辑，忘了有三个接口依赖这个函数，全部静默失败了，发现的时候已经影响了当天晚上的用户。

那次事故之后，组里强制要求所有 FastAPI 接口必须有测试覆盖，覆盖率不达标不允许合并。小文花了一周学 pytest，从此再没出过类似事故。

这篇文章，就把我们组沉淀出来的 FastAPI 测试方案完整分享给你。

---

一、测试技术栈

pip install fastapi httpx pytest pytest-asyncio pytest-cov anyio

工具	作用	Java 类比
pytest	测试框架	JUnit 5
httpx.AsyncClient	异步 HTTP 测试客户端	MockMvc / RestAssured
TestClient	同步测试客户端（简单场景）	MockMvc
pytest-asyncio	让 pytest 支持 async 测试函数	-
pytest-cov	覆盖率报告	JaCoCo

---

二、被测应用

# app/main.py
from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

FAKE_USERS = {
    "user-token-123": {"id": 1, "name": "老张", "role": "admin"},
    "user-token-456": {"id": 2, "name": "小陈", "role": "user"},
}

ITEMS = {1: {"id": 1, "name": "键盘", "price": 299.0}}


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


async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = FAKE_USERS.get(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user


@asynccontextmanager
async def lifespan(app: FastAPI):
    print("App started")
    yield
    print("App stopped")


app = FastAPI(lifespan=lifespan)


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.get("/me")
async def get_me(user=Depends(get_current_user)):
    return user


@app.get("/items/{item_id}")
async def get_item(item_id: int, user=Depends(get_current_user)):
    if item_id not in ITEMS:
        raise HTTPException(status_code=404, detail="Not found")
    return ITEMS[item_id]


@app.post("/items", status_code=201)
async def create_item(body: ItemCreate, user=Depends(get_current_user)):
    if user["role"] != "admin":
        raise HTTPException(status_code=403, detail="Admin only")
    new_id = max(ITEMS.keys()) + 1
    item = {"id": new_id, **body.model_dump()}
    ITEMS[new_id] = item
    return item

---

三、pytest fixtures：测试的基石

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


@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"


@pytest_asyncio.fixture
async def client():
    """基础异步测试客户端（无认证）"""
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c


@pytest_asyncio.fixture
async def admin_client():
    """带管理员 token 的测试客户端"""
    headers = {"Authorization": "Bearer user-token-123"}
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
        headers=headers,
    ) as c:
        yield c


@pytest_asyncio.fixture
async def user_client():
    """带普通用户 token 的测试客户端"""
    headers = {"Authorization": "Bearer user-token-456"}
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
        headers=headers,
    ) as c:
        yield c


@pytest.fixture
def override_auth():
    """
    依赖覆盖：绕过真实认证，用于隔离测试业务逻辑
    用法：在测试函数参数里声明 override_auth 即可激活
    """
    def fake_user():
        return {"id": 99, "name": "测试用户", "role": "admin"}

    app.dependency_overrides[get_current_user] = fake_user
    yield
    app.dependency_overrides.clear()

---

四、编写测试

4.1 基础接口测试

# tests/test_health.py
import pytest


@pytest.mark.anyio
async def test_health_check(client):
    resp = await client.get("/health")
    assert resp.status_code == 200
    assert resp.json() == {"status": "ok"}

4.2 认证接口测试

# tests/test_auth.py
import pytest


@pytest.mark.anyio
async def test_get_me_without_token(client):
    """未认证访问应该返回 401"""
    resp = await client.get("/me")
    assert resp.status_code == 401


@pytest.mark.anyio
async def test_get_me_with_invalid_token(client):
    """无效 token 应该返回 401"""
    resp = await client.get("/me", headers={"Authorization": "Bearer invalid"})
    assert resp.status_code == 401


@pytest.mark.anyio
async def test_get_me_as_admin(admin_client):
    """管理员可以获取自己的信息"""
    resp = await admin_client.get("/me")
    assert resp.status_code == 200
    data = resp.json()
    assert data["name"] == "老张"
    assert data["role"] == "admin"

4.3 业务逻辑测试

# tests/test_items.py
import pytest


@pytest.mark.anyio
async def test_get_item_exists(admin_client):
    resp = await admin_client.get("/items/1")
    assert resp.status_code == 200
    assert resp.json()["name"] == "键盘"


@pytest.mark.anyio
async def test_get_item_not_found(admin_client):
    resp = await admin_client.get("/items/999")
    assert resp.status_code == 404
    assert "Not found" in resp.json()["detail"]


@pytest.mark.anyio
async def test_create_item_as_admin(admin_client):
    payload = {"name": "显示器", "price": 1299.0}
    resp = await admin_client.post("/items", json=payload)
    assert resp.status_code == 201
    data = resp.json()
    assert data["name"] == "显示器"
    assert data["price"] == 1299.0
    assert "id" in data


@pytest.mark.anyio
async def test_create_item_as_user_forbidden(user_client):
    """普通用户创建商品应该返回 403"""
    payload = {"name": "测试商品", "price": 99.0}
    resp = await user_client.post("/items", json=payload)
    assert resp.status_code == 403


@pytest.mark.anyio
async def test_create_item_with_dependency_override(client, override_auth):
    """
    使用依赖覆盖，不需要真实 token
    测试纯粹的业务逻辑（管理员创建商品）
    """
    payload = {"name": "依赖覆盖测试", "price": 50.0}
    resp = await client.post("/items", json=payload)
    assert resp.status_code == 201

---

五、完整可运行示例：参数化测试 + 边界覆盖

#!/usr/bin/env python3
"""
pytest 参数化测试示例
运行：pytest tests/ -v --cov=app --cov-report=html
"""
import pytest
from httpx import ASGITransport, AsyncClient

# 被测应用（内联，方便独立运行）
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
USERS = {"tok-admin": {"id": 1, "role": "admin"}, "tok-user": {"id": 2, "role": "user"}}

app = FastAPI()


async def auth(token: str = Depends(oauth2_scheme)):
    u = USERS.get(token)
    if not u:
        raise HTTPException(401, "Unauthorized")
    return u


@app.get("/ping")
async def ping():
    return "pong"


@app.post("/items", status_code=201)
async def create(name: str, price: float, u=Depends(auth)):
    if u["role"] != "admin":
        raise HTTPException(403, "Forbidden")
    return {"name": name, "price": price}


# ===== 测试 =====
@pytest.fixture
def anyio_backend():
    return "asyncio"


@pytest.fixture
async def http():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c


@pytest.mark.anyio
async def test_ping(http):
    r = await http.get("/ping")
    assert r.status_code == 200
    assert r.json() == "pong"


@pytest.mark.anyio
@pytest.mark.parametrize("token,expected_status", [
    ("tok-admin", 201),   # 管理员 → 成功
    ("tok-user", 403),    # 普通用户 → 禁止
    ("invalid", 401),     # 无效 token → 未授权
])
async def test_create_item_permissions(http, token, expected_status):
    """参数化测试：不同角色访问创建接口的权限验证"""
    r = await http.post(
        "/items",
        params={"name": "测试", "price": "100"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert r.status_code == expected_status, f"Token={token}, got {r.status_code}"


@pytest.mark.anyio
@pytest.mark.parametrize("name,price", [
    ("键盘", 299.0),
    ("鼠标", 99.5),
    ("显示器", 2999.99),
    ("充电器", 0.01),   # 边界值：最小价格
])
async def test_create_various_items(http, name, price):
    """参数化测试：创建各种商品"""
    r = await http.post(
        "/items",
        params={"name": name, "price": str(price)},
        headers={"Authorization": "Bearer tok-admin"},
    )
    assert r.status_code == 201
    data = r.json()
    assert data["name"] == name
    assert abs(data["price"] - price) < 0.001


if __name__ == "__main__":
    import subprocess
    subprocess.run(["pytest", __file__, "-v"])

---

六、踩坑实录 1：async test 忘加 @pytest.mark.anyio

# 错误：async 测试函数没有标记，pytest 不会执行它
async def test_something(client):
    resp = await client.get("/health")
    assert resp.status_code == 200
# 结果：pytest 报告 "passed" 但实际上根本没运行！

# 正确：必须加 @pytest.mark.anyio（或 @pytest.mark.asyncio）
@pytest.mark.anyio
async def test_something(client):
    resp = await client.get("/health")
    assert resp.status_code == 200

这是非常隐蔽的坑——测试"通过"了但实际代码从来没有执行。

---

七、踩坑实录 2：依赖覆盖忘记清理

# 错误：测试后没有清理依赖覆盖，影响后续测试
def test_with_override():
    app.dependency_overrides[get_current_user] = lambda: {"id": 1}
    # ... 测试逻辑 ...
    # 忘记清理！下一个测试也会用这个假依赖

# 正确：用 fixture 的 yield + teardown 自动清理
@pytest.fixture
def mock_auth():
    app.dependency_overrides[get_current_user] = lambda: {"id": 1, "role": "admin"}
    yield
    app.dependency_overrides.clear()  # teardown，测试完自动清理

---

八、踩坑实录 3：数据库测试隔离

# 错误：多个测试共享同一个数据库连接，测试顺序影响结果
# 一个测试创建的数据，影响了另一个测试的断言

# 正确：每个测试用事务回滚隔离
@pytest_asyncio.fixture
async def db_session():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
        async_session = async_sessionmaker(bind=conn)
        async with async_session() as session:
            # 使用 SAVEPOINT，测试结束后回滚
            await session.begin_nested()
            yield session
            await session.rollback()

---

九、覆盖率配置

# pytest.ini
[pytest]
asyncio_mode = auto  # 自动处理 async 测试，不需要手动标记

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"

[tool.coverage.run]
source = ["app"]
omit = ["*/tests/*", "*/migrations/*"]

[tool.coverage.report]
min_coverage = 80  # 覆盖率低于 80% 则失败

# 运行测试并生成覆盖率报告
pytest tests/ -v --cov=app --cov-report=html --cov-report=term-missing

# CI 中要求最低覆盖率
pytest tests/ --cov=app --cov-fail-under=80

---

总结

FastAPI 测试体系的核心：

1. 用 AsyncClient + ASGITransport 做异步集成测试，不需要起真实服务
2. conftest.py 统一管理 fixtures，按角色准备测试客户端
3. dependency_overrides 隔离依赖，测试纯业务逻辑
4. 参数化测试覆盖边界条件和权限矩阵
5. CI 强制覆盖率门禁，让测试真正成为质量保障