Pytest 详细介绍

Pytest 是什么?

Pytest 是一个功能强大的 Python 测试框架,主要用于:

  • 编写简单的单元测试

  • 复杂的集成测试

  • 功能测试

  • API 测试

主要功能和优点

1. 简单易用

  • 使用 assert 语句进行断言,无需学习特定 API

  • 自动发现测试文件和测试函数

2. 丰富的功能

  • Fixture:测试夹具,管理测试资源

  • 参数化测试:用不同数据运行同一测试

  • 标记(Mark):选择性运行测试

  • 插件系统:丰富的扩展生态

3. 强大的断言

  • 自动显示详细的失败信息

  • 支持各种数据类型比较

  • 自定义断言消息

4. 兼容性好

  • 兼容 unittest 测试用例

  • 支持多种测试报告格式

安装

bash

# 基础安装
pip install pytest

# 常用扩展
pip install pytest-mock     # mock支持
pip install pytest-cov      # 测试覆盖率
pip install pytest-html     # HTML报告
pip install pytest-xdist    # 并行测试
pip install pytest-asyncio  # 异步测试

基础使用示例

1. 基本测试结构

被测代码 calculator.py:

python

def add(a, b):
    return a + b

def divide(a, b):
    if b == 0:
        raise ValueError("Cannot divide by zero")
    return a / b

class Calculator:
    def __init__(self):
        self.result = 0
    
    def accumulate(self, value):
        self.result += value
        return self.result

测试代码 test_calculator.py:

python

import pytest
from calculator import add, divide, Calculator

# 测试函数
def test_add():
    assert add(2, 3) == 5
    assert add(-1, 1) == 0
    assert add(0, 0) == 0

# 测试异常
def test_divide_by_zero():
    with pytest.raises(ValueError) as exc_info:
        divide(10, 0)
    assert "Cannot divide by zero" in str(exc_info.value)

# 测试类
class TestCalculator:
    def setup_method(self):
        self.calc = Calculator()
    
    def test_accumulate(self):
        assert self.calc.accumulate(5) == 5
        assert self.calc.accumulate(3) == 8
        assert self.calc.accumulate(-2) == 6

2. 运行测试

bash

# 运行所有测试
pytest

# 运行指定文件
pytest test_calculator.py

# 运行指定测试类
pytest test_calculator.py::TestCalculator

# 运行指定测试函数
pytest test_calculator.py::test_add

# 详细输出
pytest -v

# 显示打印输出
pytest -s

# 只运行上次失败的测试
pytest --lf

核心功能详解

1. Fixture(测试夹具)

python

import pytest

# 定义fixture
@pytest.fixture
def sample_data():
    """提供测试数据"""
    return [1, 2, 3, 4, 5]

@pytest.fixture(scope="module")
def database_connection():
    """模块级fixture,整个模块只执行一次"""
    conn = create_database_connection()
    yield conn  # 测试前执行,yield后清理
    conn.close()

@pytest.fixture(autouse=True)
def setup_teardown():
    """自动使用的fixture"""
    print("\n测试开始...")
    yield
    print("测试结束...")

# 使用fixture
def test_sum(sample_data):
    assert sum(sample_data) == 15

def test_max(sample_data):
    assert max(sample_data) == 5

2. 参数化测试

python

import pytest

@pytest.mark.parametrize("a,b,expected", [
    (1, 2, 3),
    (0, 0, 0),
    (-1, 1, 0),
    (100, 200, 300),
])
def test_add_parametrized(a, b, expected):
    from calculator import add
    assert add(a, b) == expected

# 多个参数组合
@pytest.mark.parametrize("a", [1, 2, 3])
@pytest.mark.parametrize("b", [4, 5, 6])
def test_multiple_params(a, b):
    assert a + b == a + b  # 实际上测试了9种组合

3. 标记(Mark)

python

import pytest

@pytest.mark.slow
def test_slow_operation():
    import time
    time.sleep(5)
    assert True

@pytest.mark.skip(reason="功能尚未实现")
def test_unimplemented():
    assert False

@pytest.mark.skipif(
    sys.version_info < (3, 8),
    reason="需要Python 3.8+"
)
def test_python38_feature():
    assert True

@pytest.mark.xfail(reason="已知问题")
def test_buggy_feature():
    assert False

# 自定义标记
@pytest.mark.integration
def test_api_integration():
    assert True

4. Mock 使用详解

安装 pytest-mock

bash

pip install pytest-mock
基本 Mock 示例

被测代码 user_service.py:

python

import requests
import json

class UserService:
    def __init__(self, api_url):
        self.api_url = api_url
    
    def get_user(self, user_id):
        """从API获取用户信息"""
        response = requests.get(f"{self.api_url}/users/{user_id}")
        if response.status_code == 200:
            return response.json()
        return None
    
    def create_user(self, name, email):
        """创建新用户"""
        data = {"name": name, "email": email}
        response = requests.post(
            f"{self.api_url}/users",
            json=data,
            headers={"Content-Type": "application/json"}
        )
        return response.status_code == 201
    
    def process_users(self):
        """处理用户数据"""
        users = self.get_all_users()
        return [self._process_user(user) for user in users]
    
    def get_all_users(self):
        response = requests.get(f"{self.api_url}/users")
        return response.json()
    
    def _process_user(self, user):
        # 内部处理方法
        return {"id": user["id"], "name": user["name"].upper()}
Mock 测试示例

python

import pytest
from unittest.mock import Mock, call
import user_service

# 1. Mock 返回值
def test_get_user_success(mocker):
    """测试成功获取用户"""
    # 创建 Mock 响应
    mock_response = Mock()
    mock_response.status_code = 200
    mock_response.json.return_value = {
        "id": 1,
        "name": "Alice",
        "email": "alice@example.com"
    }
    
    # Mock requests.get 方法
    mock_get = mocker.patch('user_service.requests.get')
    mock_get.return_value = mock_response
    
    # 测试
    service = user_service.UserService("http://api.example.com")
    result = service.get_user(1)
    
    # 验证
    assert result == {
        "id": 1,
        "name": "Alice",
        "email": "alice@example.com"
    }
    mock_get.assert_called_once_with("http://api.example.com/users/1")

# 2. Mock 异常情况
def test_get_user_failure(mocker):
    """测试获取用户失败"""
    mock_response = Mock()
    mock_response.status_code = 404
    
    mock_get = mocker.patch('user_service.requests.get')
    mock_get.return_value = mock_response
    
    service = user_service.UserService("http://api.example.com")
    result = service.get_user(999)
    
    assert result is None

# 3. Mock 多个调用
def test_create_user(mocker):
    """测试创建用户"""
    mock_response = Mock()
    mock_response.status_code = 201
    
    mock_post = mocker.patch('user_service.requests.post')
    mock_post.return_value = mock_response
    
    service = user_service.UserService("http://api.example.com")
    result = service.create_user("Bob", "bob@example.com")
    
    assert result is True
    mock_post.assert_called_once_with(
        "http://api.example.com/users",
        json={"name": "Bob", "email": "bob@example.com"},
        headers={"Content-Type": "application/json"}
    )

# 4. 使用 side_effect 模拟复杂行为
def test_get_user_side_effect(mocker):
    """使用 side_effect 模拟不同返回值"""
    mock_get = mocker.patch('user_service.requests.get')
    
    # 第一次调用返回成功,第二次返回失败
    mock_response1 = Mock()
    mock_response1.status_code = 200
    mock_response1.json.return_value = {"id": 1, "name": "Alice"}
    
    mock_response2 = Mock()
    mock_response2.status_code = 404
    
    mock_get.side_effect = [mock_response1, mock_response2]
    
    service = user_service.UserService("http://api.example.com")
    
    # 第一次调用
    result1 = service.get_user(1)
    assert result1 == {"id": 1, "name": "Alice"}
    
    # 第二次调用
    result2 = service.get_user(999)
    assert result2 is None
    
    # 验证调用次数和参数
    assert mock_get.call_count == 2
    assert mock_get.call_args_list == [
        call("http://api.example.com/users/1"),
        call("http://api.example.com/users/999")
    ]

# 5. Mock 对象属性
def test_mock_object_property(mocker):
    """Mock 对象的属性"""
    mock_user = mocker.Mock()
    mock_user.id = 1
    mock_user.name = "Charlie"
    mock_user.get_email.return_value = "charlie@example.com"
    
    assert mock_user.id == 1
    assert mock_user.name == "Charlie"
    assert mock_user.get_email() == "charlie@example.com"

# 6. Mock 内部方法调用
def test_process_users(mocker):
    """测试内部方法调用"""
    # Mock 外部依赖
    mock_response = Mock()
    mock_response.json.return_value = [
        {"id": 1, "name": "Alice"},
        {"id": 2, "name": "Bob"}
    ]
    
    mock_get = mocker.patch('user_service.requests.get')
    mock_get.return_value = mock_response
    
    # Mock 内部方法 _process_user
    mock_process = mocker.patch.object(
        user_service.UserService,
        '_process_user',
        side_effect=lambda u: {"id": u["id"], "name": u["name"].lower()}
    )
    
    service = user_service.UserService("http://api.example.com")
    result = service.process_users()
    
    # 验证结果
    assert result == [
        {"id": 1, "name": "alice"},
        {"id": 2, "name": "bob"}
    ]
    assert mock_process.call_count == 2

# 7. 使用 spy 监视真实对象
def test_spy_real_object(mocker):
    """使用 spy 监视对象,部分方法被 mock,部分保持真实"""
    # 创建真实对象
    from calculator import Calculator
    real_calc = Calculator()
    
    # 使用 spy
    spy_calc = mocker.spy(real_calc, 'accumulate')
    
    # 调用方法
    result = real_calc.accumulate(5)
    
    # 验证 spy 记录了调用
    assert result == 5
    spy_calc.assert_called_once_with(5)

# 8. Mock 上下文管理器
def test_mock_context_manager(mocker):
    """Mock with 语句使用的对象"""
    mock_file = mocker.mock_open(read_data='{"name": "Alice"}')
    mocker.patch('builtins.open', mock_file)
    
    with open('test.json', 'r') as f:
        content = f.read()
    
    assert content == '{"name": "Alice"}'
    mock_file.assert_called_once_with('test.json', 'r')

# 9. 验证 Mock 调用
def test_verify_calls(mocker):
    """验证 Mock 对象的调用情况"""
    mock_func = mocker.Mock()
    
    # 模拟调用
    mock_func(1, 2, 3)
    mock_func(a=1, b=2)
    mock_func(4, 5, key='value')
    
    # 各种验证方式
    assert mock_func.call_count == 3
    mock_func.assert_any_call(1, 2, 3)
    mock_func.assert_any_call(a=1, b=2)
    
    # 获取最后一次调用
    last_call = mock_func.call_args
    assert last_call == call(4, 5, key='value')
    
    # 获取所有调用参数
    all_args = mock_func.call_args_list
    assert len(all_args) == 3

5. 高级 Mock 技巧

Mock 类方法

python

def test_mock_class_method(mocker):
    """Mock 类的静态方法或类方法"""
    
    class Database:
        @staticmethod
        def connect():
            return "real connection"
    
    # Mock 静态方法
    mocker.patch.object(Database, 'connect', return_value="mock connection")
    
    assert Database.connect() == "mock connection"

def test_mock_init(mocker):
    """Mock __init__ 方法"""
    class User:
        def __init__(self, name):
            self.name = name
            self.setup_complete = False
            self.setup()
        
        def setup(self):
            self.setup_complete = True
    
    # Mock setup 方法
    mocker.patch.object(User, 'setup')
    
    user = User("Alice")
    assert user.name == "Alice"
    assert user.setup_complete is False  # setup 被 mock 了,没有执行
    User.setup.assert_called_once()
Mock 链式调用

python

def test_mock_chained_calls(mocker):
    """Mock 链式调用的方法"""
    mock_obj = mocker.Mock()
    
    # 配置链式调用返回值
    mock_obj.method1.return_value.method2.return_value.method3.return_value = "final result"
    
    result = mock_obj.method1().method2().method3()
    assert result == "final result"

Pytest 配置文件

创建 pytest.ini:

ini

[pytest]
# 测试文件匹配模式
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*

# 标记定义
markers =
    slow: marks tests as slow (deselect with '-m "not slow"')
    integration: integration tests
    unit: unit tests

# 默认命令行选项
addopts = 
    -v
    --tb=short
    --strict-markers

# 覆盖率配置
# --cov=src --cov-report=term-missing

插件使用

测试覆盖率

bash

# 运行测试并检查覆盖率
pytest --cov=my_module tests/

# 生成HTML报告
pytest --cov=my_module --cov-report=html tests/

# 设置覆盖率阈值
pytest --cov=my_module --cov-fail-under=80 tests/

并行测试

bash

# 使用4个CPU核心并行运行测试
pytest -n 4

# 自动检测CPU数量
pytest -n auto

生成测试报告

bash

# 生成JUnit格式报告
pytest --junitxml=report.xml

# 生成HTML报告
pytest --html=report.html

最佳实践

1. 测试组织

bash

project/
├── src/
│   └── my_module/
│       ├── __init__.py
│       └── calculator.py
└── tests/
    ├── __init__.py
    ├── conftest.py          # 共享fixture
    ├── unit/                # 单元测试
    │   ├── test_calculator.py
    │   └── test_utils.py
    ├── integration/         # 集成测试
    │   └── test_api.py
    └── functional/          # 功能测试
        └── test_ui.py

2. 命名规范

  • 测试文件:test_*.py

  • 测试类:Test*

  • 测试函数:test_*

  • Fixture:有意义的名称

3. 测试独立性

  • 每个测试应该独立运行

  • 使用 fixture 管理测试数据

  • 避免测试间的依赖

4. Mock 使用原则

python

# 好的实践
def test_with_mock(mocker):
    # 在测试内部 mock,作用域最小化
    mock_func = mocker.patch('module.function')
    # ... 测试代码

# 避免的实践
# 不要在模块级别或fixture中mock,除非必要

常见问题解决

1. Mock 不生效

python

# 确保导入路径正确
# 错误:mocker.patch('requests.get')  # 如果requests在函数内导入
# 正确:mocker.patch('module.requests.get')  # 在模块级别导入

2. 异步测试

python

import pytest

@pytest.mark.asyncio
async def test_async_function():
    result = await async_function()
    assert result == expected

3. 数据库测试

python

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

@pytest.fixture
def db_session():
    engine = create_engine('sqlite:///:memory:')
    Session = sessionmaker(bind=engine)
    session = Session()
    yield session
    session.close()

Pytest 是一个功能强大且灵活的测试框架,配合 mocker 可以轻松进行各种类型的测试。建议从简单测试开始,逐步学习高级特性。

帮助文档和资源

官方文档

中文资源

学习教程

实用链接

书籍推荐

  • 《Python Testing with pytest》 - Brian Okken

  • 《Python测试驱动开发》 - Harry Percival

更多推荐