pytest 完全指南

官方文档：https://docs.pytest.org/
适用版本：pytest 8.x（2026-05-07 核实）

相关文档：FastAPI完全指南 SQLModel完全指南 asyncio异步编程完全指南

1. 基础概念

pytest 是什么

pytest 是 Python 最主流的测试框架，相比标准库 unittest 更简洁灵活：

安装

pip install pytest

# 常用插件
pip install pytest-asyncio      # 异步测试支持
pip install pytest-cov          # 覆盖率报告
pip install pytest-mock         # mock 便捷封装
pip install httpx               # FastAPI 测试客户端
pip install anyio[trio]         # pytest-asyncio 依赖

2. 基础测试写法

测试函数

# tests/test_math.py

def add(a: int, b: int) -> int:
    return a + b

def test_add():
    assert add(1, 2) == 3

def test_add_negative():
    assert add(-1, -2) == -3

def test_add_zero():
    result = add(0, 0)
    assert result == 0

pytest                      # 运行所有测试
pytest tests/               # 运行指定目录
pytest tests/test_math.py   # 运行指定文件
pytest tests/test_math.py::test_add  # 运行指定测试函数
pytest -v                   # 详细输出
pytest -s                   # 显示 print 输出
pytest -x                   # 第一个失败后停止
pytest -k "add"             # 只运行名称含 "add" 的测试
pytest --tb=short           # 简短的堆栈信息

断言失败信息

pytest 会自动展示详细的断言差异，无需使用特殊方法：

def test_list():
    result = [1, 2, 3]
    assert result == [1, 2, 4]
    # AssertionError: assert [1, 2, 3] == [1, 2, 4]
    #   At index 2 diff: 3 != 4

异常测试

import pytest

def divide(a: int, b: int) -> float:
    if b == 0:
        raise ZeroDivisionError("除数不能为 0")
    return a / b

def test_divide_by_zero():
    with pytest.raises(ZeroDivisionError, match="除数不能为 0"):
        divide(1, 0)

def test_divide_by_zero_type():
    with pytest.raises(ZeroDivisionError) as exc_info:
        divide(1, 0)
    assert "除数不能为 0" in str(exc_info.value)

3. Fixture — 测试夹具

Fixture 是可复用的测试前置/清理逻辑，通过函数参数自动注入。

基础 Fixture

import pytest

@pytest.fixture
def sample_user() -> dict:
    return {"id": 1, "name": "Alice", "email": "[email protected]"}

def test_user_name(sample_user):
    assert sample_user["name"] == "Alice"

def test_user_email(sample_user):
    assert sample_user["email"] == "[email protected]"

带清理逻辑的 Fixture（yield）

import pytest
import tempfile
import os

@pytest.fixture
def temp_file():
    # setup：创建临时文件
    fd, path = tempfile.mkstemp()
    os.write(fd, b"test content")
    os.close(fd)
    yield path  # 将文件路径传给测试
    # teardown：测试完成后清理
    os.unlink(path)

def test_read_file(temp_file):
    with open(temp_file, "rb") as f:
        assert f.read() == b"test content"

Fixture 作用域

@pytest.fixture(scope="session")
def db_engine():
    """整个测试会话只创建一次数据库引擎"""
    engine = create_engine("sqlite:///test.db")
    SQLModel.metadata.create_all(engine)
    yield engine
    SQLModel.metadata.drop_all(engine)
    engine.dispose()

@pytest.fixture(scope="function")
def db_session(db_engine):
    """每个测试函数创建一个独立 session，测试后回滚"""
    with Session(db_engine) as session:
        yield session
        session.rollback()  # 回滚，下一个测试从干净状态开始

conftest.py — 共享 Fixture

放在测试目录的 conftest.py 中，其中的 fixture 无需导入即可在同级及子目录的测试中使用：

tests/
  conftest.py          ← 这里的 fixture 全局可用
  test_users.py
  api/
    conftest.py        ← 这里的 fixture 在 api/ 下可用
    test_routes.py

# tests/conftest.py
import pytest
from sqlmodel import SQLModel, Session, create_engine

@pytest.fixture(scope="session")
def engine():
    engine = create_engine("sqlite:///./test.db")
    SQLModel.metadata.create_all(engine)
    yield engine
    SQLModel.metadata.drop_all(engine)

@pytest.fixture
def session(engine):
    with Session(engine) as s:
        yield s
        s.rollback()

4. 参数化测试

import pytest

@pytest.mark.parametrize("a, b, expected", [
    (1, 2, 3),
    (-1, -2, -3),
    (0, 0, 0),
    (100, -50, 50),
])
def test_add(a: int, b: int, expected: int):
    assert add(a, b) == expected

# 嵌套参数化（笛卡尔积）
@pytest.mark.parametrize("x", [1, 2])
@pytest.mark.parametrize("y", [3, 4])
def test_multiply(x, y):
    assert multiply(x, y) == x * y
    # 共运行 4 次：(1,3), (1,4), (2,3), (2,4)

参数化 + 自定义 ID

@pytest.mark.parametrize("username,password,expected", [
    pytest.param("alice", "correct", True, id="正确密码"),
    pytest.param("alice", "wrong", False, id="错误密码"),
    pytest.param("", "pass", False, id="空用户名"),
], )
def test_login(username, password, expected):
    assert login(username, password) == expected

5. 标记（Marks）

内置标记

# 跳过测试
@pytest.mark.skip(reason="功能尚未实现")
def test_future_feature():
    pass

# 条件跳过
import sys
@pytest.mark.skipif(sys.platform == "win32", reason="不支持 Windows")
def test_unix_only():
    pass

# 预期失败（已知 bug，防止测试阻塞流水线）
@pytest.mark.xfail(reason="Issue #123 待修复")
def test_known_bug():
    assert buggy_function() == expected

# 超时（需要 pytest-timeout 插件）
@pytest.mark.timeout(5)
def test_slow_operation():
    pass

自定义标记

# pytest.ini 或 pyproject.toml 中注册标记
# [pytest]
# markers =
#     slow: 运行慢的测试
#     integration: 集成测试，需要真实数据库

@pytest.mark.slow
@pytest.mark.integration
def test_big_query():
    ...

# 运行时过滤
# pytest -m "not slow"          运行所有非 slow 测试
# pytest -m "integration"       只运行集成测试
# pytest -m "slow and not integration"

6. 异步测试（pytest-asyncio）

pip install pytest-asyncio

配置

# pytest.ini 或 pyproject.toml
[pytest]
asyncio_mode = auto   # 自动识别 async 测试函数，无需手动加 @pytest.mark.asyncio

异步测试函数

import pytest
import httpx
from fastapi.testclient import TestClient

# asyncio_mode=auto 时，直接用 async def 即可
async def test_async_fetch():
    async with httpx.AsyncClient() as client:
        r = await client.get("https://httpbin.org/get")
    assert r.status_code == 200

# 异步 Fixture
@pytest.fixture
async def async_db_session():
    async with AsyncSessionLocal() as session:
        yield session
        await session.rollback()

7. Mock（pytest-mock / unittest.mock）

基础 Mock

from unittest.mock import MagicMock, AsyncMock, patch

# patch 装饰器：替换指定路径的对象
@patch("myapp.services.send_email")
def test_create_user_sends_email(mock_send_email):
    create_user({"name": "Alice", "email": "[email protected]"})
    mock_send_email.assert_called_once_with("[email protected]", subject="欢迎注册")

pytest-mock（更简洁的 mocker fixture）

def test_create_user(mocker):
    mock_send = mocker.patch("myapp.services.send_email")
    mock_send.return_value = True

    create_user({"name": "Alice", "email": "[email protected]"})

    mock_send.assert_called_once()
    args, kwargs = mock_send.call_args
    assert args[0] == "[email protected]"

Mock 异步函数

from unittest.mock import AsyncMock

async def test_async_service(mocker):
    mock_fetch = mocker.patch("myapp.services.fetch_data", new_callable=AsyncMock)
    mock_fetch.return_value = {"id": 1, "name": "Alice"}

    result = await get_user_profile(user_id=1)

    mock_fetch.assert_awaited_once_with(user_id=1)
    assert result["name"] == "Alice"

8. FastAPI 测试

同步测试客户端（TestClient）

from fastapi.testclient import TestClient
from myapp.main import app

client = TestClient(app)

def test_create_user():
    response = client.post("/users", json={"name": "Alice", "email": "[email protected]"})
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Alice"
    assert "id" in data

def test_get_user_not_found():
    response = client.get("/users/99999")
    assert response.status_code == 404

异步测试客户端（httpx.AsyncClient）

import pytest
import httpx
from myapp.main import app

@pytest.fixture
async def async_client():
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        yield client

async def test_list_users(async_client: httpx.AsyncClient, db_session):
    response = await async_client.get("/users")
    assert response.status_code == 200
    assert isinstance(response.json(), list)

覆盖依赖（测试数据库隔离）

import pytest
from fastapi.testclient import TestClient
from sqlmodel import SQLModel, Session, create_engine
from myapp.main import app
from myapp.database import get_session

# 测试专用数据库
test_engine = create_engine("sqlite:///./test.db")

@pytest.fixture(scope="session", autouse=True)
def create_tables():
    SQLModel.metadata.create_all(test_engine)
    yield
    SQLModel.metadata.drop_all(test_engine)

@pytest.fixture
def session():
    with Session(test_engine) as s:
        yield s
        s.rollback()

@pytest.fixture
def client(session: Session):
    # 用测试 session 替换生产 session
    def override_get_session():
        yield session

    app.dependency_overrides[get_session] = override_get_session
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

9. 覆盖率报告

# 生成覆盖率报告
pytest --cov=src --cov-report=term-missing  # 终端显示，含未覆盖行号
pytest --cov=src --cov-report=html           # 生成 HTML 报告（htmlcov/index.html）
pytest --cov=src --cov-fail-under=80        # 覆盖率低于 80% 时失败

# .coveragerc 或 pyproject.toml
[coverage:run]
source = src
omit =
    */migrations/*
    */tests/*
    */__init__.py

[coverage:report]
exclude_lines =
    pragma: no cover
    if TYPE_CHECKING:
    raise NotImplementedError

10. pyproject.toml 配置

[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
addopts = "-v --tb=short"
markers = [
    "slow: 运行较慢的测试",
    "integration: 需要真实服务的集成测试",
    "unit: 纯单元测试",
]

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

11. 最佳实践

一个测试只验证一件事

# 不推荐：一个测试做了多件事
def test_user_operations():
    user = create_user(...)
    assert user.id is not None
    updated = update_user(user.id, ...)
    assert updated.name == "Bob"
    delete_user(user.id)
    assert get_user(user.id) is None

# 推荐：拆分为独立测试
def test_create_user_returns_id(session):
    user = create_user(session, ...)
    assert user.id is not None

def test_update_user_name(session, existing_user):
    updated = update_user(session, existing_user.id, {"name": "Bob"})
    assert updated.name == "Bob"

测试命名表达意图

# 不好
def test_user():
    ...

# 好：test_[被测功能]_[场景]_[预期结果]
def test_create_user_with_duplicate_email_raises_error():
    ...

def test_get_user_by_id_returns_none_when_not_found():
    ...

---

## 最佳实践

**用 fixture 管理测试依赖，而非 setUp/tearDown**：pytest fixture 支持参数化、作用域（function/class/module/session）、依赖注入，比 unittest 的 setUp/tearDown 灵活得多。

```python
@pytest.fixture(scope="session")
def db_engine():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    yield engine
    engine.dispose()

@pytest.fixture
def db_session(db_engine):
    with Session(db_engine) as session:
        yield session
        session.rollback()

测试命名遵循 test_<功能>_<场景>_<预期结果> 模式：test_create_user_with_duplicate_email_raises_error 比 test_user_2 能在失败时立即定位问题。

用 @pytest.mark.parametrize 替代重复测试代码：同一逻辑不同输入的测试用参数化一次写完，避免复制粘贴。

@pytest.mark.parametrize("email,is_valid", [
    ("[email protected]", True),
    ("invalid-email", False),
    ("user@", False),
    ("@domain.com", False),
])
def test_email_validation(email, is_valid):
    assert validate_email(email) == is_valid

mock 外部依赖（HTTP / 数据库），只测自己的代码：单元测试不应依赖真实网络或数据库，用 pytest-mock 或 unittest.mock 隔离外部系统。

def test_fetch_user(mocker):
    mocker.patch("httpx.get", return_value=Mock(json=lambda: {"id": 1, "name": "Alice"}))
    user = fetch_user(1)
    assert user.name == "Alice"

使用 conftest.py 共享 fixture 和配置：项目级 conftest.py 在所有测试文件中自动生效，无需 import。

常见陷阱

陷阱：fixture 作用域错误导致测试互相污染

现象： 某个测试单独运行通过，全量运行时随机失败，测试顺序影响结果。

原因： session 或 module 作用域的 fixture 在多个测试间共享状态，一个测试修改了状态后影响后续测试。

解决： 数据库 session 类 fixture 用 function 作用域并在 teardown 中 rollback；或每个测试独立创建数据。

@pytest.fixture(scope="function")  # 每个测试函数独立
def db_session(db_engine):
    with Session(db_engine) as session:
        yield session
        session.rollback()   # 测试结束后回滚，不影响下一个测试

陷阱：异步测试不加 @pytest.mark.asyncio

现象： 异步测试函数静默跳过或直接通过，没有实际执行异步代码。

原因： pytest 默认不知道如何运行 async def 测试函数，需要 pytest-asyncio 插件并标记。

解决： pip install pytest-asyncio，并对异步测试加 @pytest.mark.asyncio，或在 pytest.ini 中设置 asyncio_mode = auto。

# 方法 A：逐个标记
@pytest.mark.asyncio
async def test_async_fetch():
    result = await fetch_data()
    assert result is not None

# 方法 B：pytest.ini 全局启用
# [pytest]
# asyncio_mode = auto

陷阱：mock patch 路径错误导致 mock 无效

现象： mock.patch("requests.get") 后，被测代码仍然发出真实 HTTP 请求。

原因： mock.patch 替换的是被测模块导入的那个名称，不是原始模块中的函数。如果被测代码用 from requests import get，需要 patch mymodule.get 而不是 requests.get。

解决： patch 的路径应指向被测模块中使用该函数的路径。

# 被测模块：mymodule.py
from requests import get   # 引入到本模块命名空间

# 测试中必须 patch mymodule.get，不是 requests.get
mocker.patch("mymodule.get", return_value=Mock(json=lambda: {}))

参见

• FastAPI完全指南
• SQLModel完全指南
• asyncio异步编程完全指南
• GitHub Actions完全指南