12.pytest的自定义插件

作者

Pytest 自定义插件详解

1. 什么是 Pytest 自定义插件

1.1 插件的基本概念

插件（Plugin） 是 pytest 框架的核心扩展机制。你可以把插件理解为 pytest 的”扩展包”，它们可以：

  • 扩展功能：添加新的命令行选项、钩子函数、fixture 等
  • 改变行为：修改测试发现、执行、报告等行为
  • 集成工具：与其他工具（如数据库、API、监控系统等）集成
  • 定制化：根据项目需求定制测试框架的行为

1.2 为什么需要自定义插件

在实际项目中，你可能会遇到以下需求：

场景 1：需要自定义测试报告格式

# 没有自定义插件的情况
# 只能使用 pytest 默认的报告格式
# 无法添加项目特定的信息（如测试环境、版本号等）

# 使用自定义插件后
# 可以生成包含项目特定信息的报告
# 可以集成到公司的报告系统中

场景 2：需要在测试前后执行特定操作

# 没有自定义插件的情况
# 每个测试文件都要手动写前置和后置代码
# 代码重复，维护困难

# 使用自定义插件后
# 可以在插件中统一处理
# 所有测试自动应用，无需修改测试代码

场景 3：需要添加项目特定的命令行选项

# 没有自定义插件的情况
# 无法添加项目特定的命令行参数
# 例如：--test-env、--test-version 等

# 使用自定义插件后
# 可以添加任意命令行选项
# 可以在测试中使用这些选项

1.3 插件的类型

Pytest 插件主要分为以下几类：

  1. 内置插件（Built-in Plugins）：

    • pytest 自带的插件
    • 例如：测试发现插件、断言重写插件等

  2. 第三方插件（Third-party Plugins）：

    • 需要单独安装的插件
    • 例如：pytest-html、pytest-cov 等

  3. 自定义插件（Custom Plugins）：

    • 你自己编写的插件
    • 用于满足项目特定需求

1.4 插件的工作原理

插件加载流程：

  1. pytest 启动时，会扫描已安装的插件
  2. 插件通过 pytest_plugins 或 setup.py 注册
  3. pytest 调用插件的钩子函数（hooks）
  4. 插件可以修改 pytest 的行为

钩子函数（Hooks）：

钩子函数是 pytest 提供的扩展点，插件可以在这些点上插入自己的代码。例如：

  • pytest_configure：pytest 配置时调用
  • pytest_collection_modifyitems：修改收集的测试项
  • pytest_runtest_setup：测试运行前调用
  • pytest_runtest_teardown：测试运行后调用

2. 创建第一个自定义插件

2.1 最简单的插件示例

让我们从一个最简单的插件开始：

创建 pytest_custom_plugin.py：

"""
最简单的 pytest 自定义插件
"""

def pytest_configure(config):
    """pytest 配置时调用"""
    print("【插件】自定义插件已加载！")
    print("【插件】pytest 配置完成")

运行测试：

# 运行任何测试
pytest test_example.py -v

# 输出：
# 【插件】自定义插件已加载！
# 【插件】pytest 配置完成
# ========================= test session starts ==========================
# ...

说明：

  • pytest_configure 是一个钩子函数，在 pytest 配置时自动调用
  • 插件文件必须以 pytest_ 开头或包含在 conftest.py 中
  • pytest 会自动发现并加载插件

2.2 插件文件命名规则

Pytest 会自动发现以下文件作为插件：

  1. 以 pytest_ 开头的文件：

    pytest_my_plugin.py  # 会被发现
pytest_custom.py     # 会被发现

  2. conftest.py 文件：

    # conftest.py 中的钩子函数也会被当作插件
def pytest_configure(config):
   print("conftest.py 中的插件")

  3. 通过 pytest_plugins 显式声明：

    # conftest.py
pytest_plugins = ["my_module.my_plugin"]

2.3 插件的基本结构

一个典型的插件包含以下部分：

"""
Pytest 自定义插件示例
"""

def pytest_configure(config):
    """pytest 配置钩子"""
    pass

def pytest_collection_modifyitems(config, items):
    """修改收集的测试项"""
    pass

def pytest_runtest_setup(item):
    """测试运行前钩子"""
    pass

def pytest_runtest_teardown(item):
    """测试运行后钩子"""
    pass

3. 钩子函数详解

3.1 什么是钩子函数

钩子函数（Hook Functions） 是 pytest 提供的扩展点，允许插件在特定时机执行代码。你可以把钩子函数理解为”事件监听器”。

类比：

  • 想象你在参加一个会议
  • 会议有多个时间点：开始前、进行中、结束后
  • 钩子函数就像在这些时间点执行的”动作”
  • 例如：会议开始前播放音乐（pytest_configure）
  • 例如：会议结束后清理场地（pytest_unconfigure）

3.2 配置钩子（Configuration Hooks）

3.2.1 pytest_configure

作用：在 pytest 配置时调用，用于初始化插件。

示例：

def pytest_configure(config):
    """pytest 配置时调用"""
    print("【插件】开始配置 pytest")

    # 添加自定义标记
    config.addinivalue_line(
        "markers", "smoke: 标记为冒烟测试"
    )
    config.addinivalue_line(
        "markers", "regression: 标记为回归测试"
    )

    print("【插件】pytest 配置完成")

实际应用：

def pytest_configure(config):
    """配置测试环境"""
    import os

    # 读取环境变量
    test_env = os.getenv("TEST_ENV", "dev")
    print(f"【插件】测试环境：{test_env}")

    # 设置全局配置
    config.test_env = test_env

    # 添加自定义标记
    config.addinivalue_line("markers", "api: API 测试")
    config.addinivalue_line("markers", "ui: UI 测试")

3.2.2 pytest_unconfigure

作用：在 pytest 退出前调用，用于清理资源。

示例：

def pytest_configure(config):
    """配置时初始化"""
    print("【插件】初始化数据库连接")
    config.db_connection = connect_database()

def pytest_unconfigure(config):
    """退出时清理"""
    print("【插件】关闭数据库连接")
    if hasattr(config, 'db_connection'):
        config.db_connection.close()

3.3 收集钩子（Collection Hooks）

3.3.1 pytest_collection_modifyitems

作用：修改收集到的测试项，可以添加标记、修改名称等。

示例 1：自动添加标记

def pytest_collection_modifyitems(config, items):
    """自动为测试添加标记"""
    for item in items:
        # 如果测试名称包含 "slow"，添加 slow 标记
        if "slow" in item.name:
            item.add_marker(pytest.mark.slow)

        # 如果测试名称包含 "api"，添加 api 标记
        if "api" in item.name:
            item.add_marker(pytest.mark.api)

示例 2：修改测试名称

def pytest_collection_modifyitems(config, items):
    """修改测试名称，添加前缀"""
    for item in items:
        # 为所有测试名称添加前缀
        item.name = f"[AUTO] {item.name}"
        item._nodeid = f"[AUTO] {item._nodeid}"

示例 3：跳过特定测试

import pytest

def pytest_collection_modifyitems(config, items):
    """根据条件跳过测试"""
    skip_ui = config.getoption("--skip-ui", default=False)

    if skip_ui:
        skip_marker = pytest.mark.skip(reason="跳过 UI 测试")
        for item in items:
            if "ui" in item.name.lower():
                item.add_marker(skip_marker)

实际应用：根据环境跳过测试

import pytest
import os

def pytest_collection_modifyitems(config, items):
    """根据环境跳过测试"""
    test_env = os.getenv("TEST_ENV", "dev")

    # 如果是生产环境，跳过标记为 dev 的测试
    if test_env == "prod":
        skip_marker = pytest.mark.skip(reason="生产环境跳过开发测试")
        for item in items:
            if "dev" in item.name.lower():
                item.add_marker(skip_marker)

3.3.2 pytest_collect_file

作用：自定义文件收集逻辑，可以收集非 Python 文件。

示例：

def pytest_collect_file(parent, path):
    """收集 YAML 测试文件"""
    if path.ext == ".yaml" and path.basename.startswith("test_"):
        # 返回一个自定义的收集器
        return YamlFile.from_parent(parent, path=path)

3.4 运行钩子（Runtime Hooks）

3.4.1 pytest_runtest_setup

作用：在每个测试运行前调用。

示例 1：记录测试开始时间

import time

def pytest_runtest_setup(item):
    """测试运行前"""
    print(f"【插件】开始运行测试：{item.name}")
    # 记录开始时间
    item.start_time = time.time()

示例 2：检查前置条件

def pytest_runtest_setup(item):
    """检查测试前置条件"""
    # 检查数据库连接
    if not check_database_connection():
        pytest.skip("数据库连接失败")

    # 检查 API 服务
    if "api" in item.name and not check_api_service():
        pytest.skip("API 服务不可用")

实际应用：自动登录

def pytest_runtest_setup(item):
    """测试前自动登录"""
    # 如果测试需要登录，自动执行登录
    if "login_required" in item.keywords:
        login_user()
        print(f"【插件】已为用户 {get_current_user()} 自动登录")

3.4.2 pytest_runtest_teardown

作用：在每个测试运行后调用。

示例 1：记录测试执行时间

import time

def pytest_runtest_setup(item):
    """记录开始时间"""
    item.start_time = time.time()

def pytest_runtest_teardown(item, nextitem):
    """记录结束时间并计算耗时"""
    if hasattr(item, 'start_time'):
        duration = time.time() - item.start_time
        print(f"【插件】测试 {item.name} 耗时：{duration:.2f} 秒")

        # 如果测试超过 5 秒，记录警告
        if duration > 5:
            print(f"【警告】测试 {item.name} 执行时间过长！")

示例 2：清理测试数据

def pytest_runtest_teardown(item, nextitem):
    """测试后清理数据"""
    # 清理测试创建的临时数据
    cleanup_test_data()

    # 重置测试环境
    reset_test_environment()

实际应用：自动截图（失败时）

import allure
from selenium import webdriver

def pytest_runtest_teardown(item, nextitem):
    """测试失败时自动截图"""
    # 检查测试是否失败
    if hasattr(item, 'rep_call') and item.rep_call.failed:
        # 如果有 WebDriver，截图
        if hasattr(item, 'driver'):
            screenshot = item.driver.get_screenshot_as_png()
            allure.attach(
                screenshot,
                name="失败截图",
                attachment_type=allure.attachment_type.PNG
            )

3.4.3 pytest_runtest_call

作用：在测试函数实际执行时调用。

示例：

def pytest_runtest_call(item):
    """测试执行时"""
    print(f"【插件】正在执行测试：{item.name}")

    # 可以在这里包装测试执行
    # 例如：添加重试逻辑、超时控制等

3.5 报告钩子（Reporting Hooks）

3.5.1 pytest_runtest_makereport

作用：生成测试报告，可以获取测试结果。

示例 1：获取测试结果

def pytest_runtest_makereport(item, call):
    """生成测试报告"""
    # call.when 可以是 "setup", "call", "teardown"
    if call.when == "call":
        # 测试执行阶段
        if call.excinfo is not None:
            # 测试失败
            print(f"【插件】测试 {item.name} 失败")
            print(f"【插件】错误信息：{call.excinfo.value}")
        else:
            # 测试通过
            print(f"【插件】测试 {item.name} 通过")

示例 2：失败时发送通知

def pytest_runtest_makereport(item, call):
    """测试失败时发送通知"""
    if call.when == "call" and call.excinfo is not None:
        # 测试失败，发送通知
        send_notification(
            f"测试 {item.name} 失败",
            str(call.excinfo.value)
        )

实际应用：记录测试结果到数据库

def pytest_runtest_makereport(item, call):
    """记录测试结果到数据库"""
    if call.when == "call":
        result = {
            "test_name": item.name,
            "status": "passed" if call.excinfo is None else "failed",
            "duration": call.duration,
            "error": str(call.excinfo.value) if call.excinfo else None
        }
        save_test_result_to_database(result)

3.5.2 pytest_report_header

作用：在报告头部添加自定义信息。

示例：

def pytest_report_header(config, start_path):
    """在报告头部添加信息"""
    import platform
    import sys

    return [
        f"操作系统: {platform.system()} {platform.version()}",
        f"Python 版本: {sys.version}",
        f"测试环境: {config.getoption('--test-env', default='dev')}",
        f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
    ]

输出示例：

============================= test session starts =============================
platform win32 -- Python 3.9.0, pytest-7.0.0
操作系统: Windows 10.0.19045
Python 版本: 3.9.0
测试环境: dev
测试时间: 2024-01-01 10:00:00
...

3.5.3 pytest_report_teststatus

作用：修改测试状态报告。

示例：

def pytest_report_teststatus(report, config):
    """修改测试状态显示"""
    if report.when == "call":
        if report.outcome == "passed":
            # 自定义通过状态的显示
            return "passed", "✓", "PASSED"
        elif report.outcome == "failed":
            # 自定义失败状态的显示
            return "failed", "✗", "FAILED"

4. 添加命令行选项

4.1 为什么需要命令行选项

命令行选项允许用户在运行测试时传递参数，例如：

  • --test-env=prod：指定测试环境
  • --skip-ui：跳过 UI 测试
  • --test-version=1.0：指定测试版本

4.2 添加命令行选项

使用 pytest_addoption 钩子函数添加命令行选项。

示例 1：添加简单的选项

def pytest_addoption(parser):
    """添加命令行选项"""
    parser.addoption(
        "--test-env",
        action="store",
        default="dev",
        help="指定测试环境 (dev, test, prod)"
    )

    parser.addoption(
        "--skip-ui",
        action="store_true",
        default=False,
        help="跳过 UI 测试"
    )

使用选项：

def pytest_configure(config):
    """使用命令行选项"""
    test_env = config.getoption("--test-env")
    print(f"【插件】测试环境：{test_env}")

    skip_ui = config.getoption("--skip-ui")
    if skip_ui:
        print("【插件】将跳过 UI 测试")

运行测试：

# 使用自定义选项
pytest --test-env=prod --skip-ui test_example.py

# 查看帮助
pytest --help
# 会显示：
# --test-env=TEST_ENV  指定测试环境 (dev, test, prod)
# --skip-ui             跳过 UI 测试

4.3 选项类型详解

4.3.1 store（存储值）

示例：

def pytest_addoption(parser):
    parser.addoption(
        "--test-env",
        action="store",
        default="dev",
        help="测试环境"
    )

    parser.addoption(
        "--test-version",
        action="store",
        type=str,
        help="测试版本"
    )

使用：

pytest --test-env=prod --test-version=1.0

4.3.2 store_true / store_false（布尔标志）

示例：

def pytest_addoption(parser):
    parser.addoption(
        "--skip-ui",
        action="store_true",
        default=False,
        help="跳过 UI 测试"
    )

    parser.addoption(
        "--verbose-output",
        action="store_true",
        default=False,
        help="详细输出"
    )

使用：

# 启用选项（不需要值）
pytest --skip-ui

# 不启用（默认 False）
pytest

4.3.3 append（追加值）

示例：

def pytest_addoption(parser):
    parser.addoption(
        "--include-tag",
        action="append",
        default=[],
        help="包含的标签（可多次使用）"
    )

使用：

pytest --include-tag=api --include-tag=smoke

4.3.4 choice（选择值）

示例：

def pytest_addoption(parser):
    parser.addoption(
        "--test-env",
        action="store",
        choices=["dev", "test", "prod"],
        default="dev",
        help="测试环境"
    )

使用：

# 正确
pytest --test-env=prod

# 错误（会报错）
pytest --test-env=invalid

4.4 实际应用示例

完整的命令行选项示例：

def pytest_addoption(parser):
    """添加所有命令行选项"""
    # 测试环境
    parser.addoption(
        "--test-env",
        action="store",
        choices=["dev", "test", "prod"],
        default="dev",
        help="指定测试环境"
    )

    # 跳过 UI 测试
    parser.addoption(
        "--skip-ui",
        action="store_true",
        default=False,
        help="跳过 UI 测试"
    )

    # 测试版本
    parser.addoption(
        "--test-version",
        action="store",
        type=str,
        help="测试版本号"
    )

    # 详细输出
    parser.addoption(
        "--verbose-output",
        action="store_true",
        default=False,
        help="启用详细输出"
    )

    # 包含标签（可多次使用）
    parser.addoption(
        "--include-tag",
        action="append",
        default=[],
        help="包含的标签"
    )

def pytest_configure(config):
    """使用命令行选项"""
    # 获取选项值
    test_env = config.getoption("--test-env")
    skip_ui = config.getoption("--skip-ui")
    test_version = config.getoption("--test-version")
    verbose = config.getoption("--verbose-output")
    include_tags = config.getoption("--include-tag")

    # 存储到 config 对象
    config.test_env = test_env
    config.skip_ui = skip_ui
    config.test_version = test_version
    config.verbose_output = verbose
    config.include_tags = include_tags

    if verbose:
        print(f"【插件】测试环境：{test_env}")
        print(f"【插件】跳过 UI：{skip_ui}")
        print(f"【插件】测试版本：{test_version}")
        print(f"【插件】包含标签：{include_tags}")

def pytest_collection_modifyitems(config, items):
    """根据选项修改测试项"""
    # 如果跳过 UI 测试
    if config.skip_ui:
        skip_marker = pytest.mark.skip(reason="跳过 UI 测试")
        for item in items:
            if "ui" in item.name.lower():
                item.add_marker(skip_marker)

    # 如果指定了包含标签
    if config.include_tags:
        for item in items:
            # 检查测试是否有指定标签
            item_tags = [mark.name for mark in item.iter_markers()]
            if not any(tag in item_tags for tag in config.include_tags):
                skip_marker = pytest.mark.skip(reason=f"不包含标签 {config.include_tags}")
                item.add_marker(skip_marker)

5. 创建 Fixture

5.1 在插件中创建 Fixture

插件可以创建全局可用的 fixture。

示例 1：简单的 Fixture

import pytest

@pytest.fixture(scope="session")
def test_config():
    """测试配置 fixture"""
    return {
        "base_url": "https://api.example.com",
        "timeout": 30,
        "retry_count": 3
    }

示例 2：带清理的 Fixture

import pytest

@pytest.fixture(scope="session")
def database():
    """数据库连接 fixture"""
    print("【插件】连接数据库")
    db = connect_database()
    yield db
    print("【插件】关闭数据库连接")
    db.close()

5.2 根据命令行选项创建 Fixture

示例：

import pytest

def pytest_addoption(parser):
    parser.addoption(
        "--test-env",
        action="store",
        default="dev",
        help="测试环境"
    )

@pytest.fixture(scope="session")
def api_base_url(request):
    """根据测试环境返回 API 地址"""
    test_env = request.config.getoption("--test-env")

    urls = {
        "dev": "https://dev-api.example.com",
        "test": "https://test-api.example.com",
        "prod": "https://api.example.com"
    }

    return urls.get(test_env, urls["dev"])

使用：

def test_api(api_base_url):
    """使用插件提供的 fixture"""
    response = requests.get(f"{api_base_url}/users")
    assert response.status_code == 200

6. 完整插件示例

6.1 示例 1：测试环境管理插件

创建 pytest_env_plugin.py：

"""
测试环境管理插件
功能：
1. 管理测试环境配置
2. 根据环境自动选择配置
3. 在测试报告中显示环境信息
"""

import pytest
import os
from datetime import datetime

def pytest_addoption(parser):
    """添加命令行选项"""
    parser.addoption(
        "--test-env",
        action="store",
        choices=["dev", "test", "prod"],
        default="dev",
        help="指定测试环境"
    )

    parser.addoption(
        "--env-config",
        action="store",
        help="环境配置文件路径"
    )

def pytest_configure(config):
    """配置插件"""
    # 获取测试环境
    test_env = config.getoption("--test-env")
    env_config_path = config.getoption("--env-config")

    # 加载环境配置
    if env_config_path and os.path.exists(env_config_path):
        config.env_config = load_config(env_config_path)
    else:
        config.env_config = get_default_config(test_env)

    config.test_env = test_env

    print(f"【环境插件】测试环境：{test_env}")
    print(f"【环境插件】API 地址：{config.env_config['api_url']}")
    print(f"【环境插件】数据库：{config.env_config['database']}")

def pytest_report_header(config):
    """在报告头部显示环境信息"""
    return [
        f"测试环境: {config.test_env}",
        f"API 地址: {config.env_config['api_url']}",
        f"数据库: {config.env_config['database']}",
        f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
    ]

@pytest.fixture(scope="session")
def api_url(request):
    """API 地址 fixture"""
    return request.config.env_config['api_url']

@pytest.fixture(scope="session")
def database_config(request):
    """数据库配置 fixture"""
    return request.config.env_config['database']

def get_default_config(env):
    """获取默认配置"""
    configs = {
        "dev": {
            "api_url": "https://dev-api.example.com",
            "database": "dev_db"
        },
        "test": {
            "api_url": "https://test-api.example.com",
            "database": "test_db"
        },
        "prod": {
            "api_url": "https://api.example.com",
            "database": "prod_db"
        }
    }
    return configs.get(env, configs["dev"])

def load_config(config_path):
    """从文件加载配置"""
    import json
    with open(config_path, 'r', encoding='utf-8') as f:
        return json.load(f)

使用：

def test_api(api_url):
    """使用环境插件提供的 fixture"""
    response = requests.get(f"{api_url}/users")
    assert response.status_code == 200

运行：

pytest --test-env=prod test_api.py

6.2 示例 2：测试结果记录插件

创建 pytest_result_logger.py：

"""
测试结果记录插件
功能：
1. 记录测试执行结果
2. 统计测试通过率
3. 生成测试报告
"""

import pytest
import json
from datetime import datetime
from pathlib import Path

class TestResultLogger:
    """测试结果记录器"""

    def __init__(self):
        self.results = []
        self.start_time = None
        self.end_time = None

    def add_result(self, test_name, status, duration, error=None):
        """添加测试结果"""
        self.results.append({
            "test_name": test_name,
            "status": status,
            "duration": duration,
            "error": error,
            "timestamp": datetime.now().isoformat()
        })

    def get_summary(self):
        """获取测试摘要"""
        total = len(self.results)
        passed = sum(1 for r in self.results if r["status"] == "passed")
        failed = sum(1 for r in self.results if r["status"] == "failed")
        skipped = sum(1 for r in self.results if r["status"] == "skipped")

        return {
            "total": total,
            "passed": passed,
            "failed": failed,
            "skipped": skipped,
            "pass_rate": (passed / total * 100) if total > 0 else 0,
            "start_time": self.start_time.isoformat() if self.start_time else None,
            "end_time": self.end_time.isoformat() if self.end_time else None
        }

    def save_to_file(self, filepath):
        """保存结果到文件"""
        data = {
            "summary": self.get_summary(),
            "results": self.results
        }

        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(data, f, indent=2, ensure_ascii=False)

def pytest_configure(config):
    """初始化记录器"""
    config.test_logger = TestResultLogger()
    config.test_logger.start_time = datetime.now()

def pytest_unconfigure(config):
    """保存测试结果"""
    if hasattr(config, 'test_logger'):
        config.test_logger.end_time = datetime.now()

        # 保存结果
        output_dir = Path("test_results")
        output_dir.mkdir(exist_ok=True)

        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_file = output_dir / f"test_results_{timestamp}.json"
        config.test_logger.save_to_file(output_file)

        # 打印摘要
        summary = config.test_logger.get_summary()
        print("n【结果记录插件】测试摘要：")
        print(f"  总数: {summary['total']}")
        print(f"  通过: {summary['passed']}")
        print(f"  失败: {summary['failed']}")
        print(f"  跳过: {summary['skipped']}")
        print(f"  通过率: {summary['pass_rate']:.2f}%")
        print(f"  结果文件: {output_file}")

def pytest_runtest_makereport(item, call):
    """记录测试结果"""
    if call.when == "call":
        config = item.config
        if hasattr(config, 'test_logger'):
            status = "passed"
            error = None

            if call.excinfo is not None:
                status = "failed"
                error = str(call.excinfo.value)
            elif call.outcome == "skipped":
                status = "skipped"

            config.test_logger.add_result(
                test_name=item.name,
                status=status,
                duration=call.duration,
                error=error
            )

使用：

def test_example():
    """测试示例"""
    assert 1 + 1 == 2

运行：

pytest test_example.py

# 会自动生成 test_results/test_results_20240101_100000.json

6.3 示例 3：自动重试插件

创建 pytest_auto_retry.py：

"""
自动重试插件
功能：
1. 测试失败时自动重试
2. 可配置重试次数和延迟
3. 记录重试信息
"""

import pytest
import time

def pytest_addoption(parser):
    """添加命令行选项"""
    parser.addoption(
        "--retry-count",
        action="store",
        type=int,
        default=3,
        help="失败重试次数"
    )

    parser.addoption(
        "--retry-delay",
        action="store",
        type=float,
        default=1.0,
        help="重试延迟（秒）"
    )

def pytest_runtest_setup(item):
    """初始化重试计数器"""
    item.retry_count = 0
    item.max_retries = item.config.getoption("--retry-count")
    item.retry_delay = item.config.getoption("--retry-delay")

def pytest_runtest_makereport(item, call):
    """检查是否需要重试"""
    if call.when == "call" and call.excinfo is not None:
        # 测试失败
        item.retry_count += 1

        if item.retry_count < item.max_retries:
            # 需要重试
            print(f"n【重试插件】测试 {item.name} 失败，准备重试 ({item.retry_count}/{item.max_retries})")
            time.sleep(item.retry_delay)

            # 重新运行测试
            pytest.runtest.runtest(item)
        else:
            print(f"n【重试插件】测试 {item.name} 重试 {item.max_retries} 次后仍然失败")

使用：

# 失败后重试 3 次，每次延迟 1 秒
pytest --retry-count=3 --retry-delay=1.0 test_example.py

7. 插件注册和安装

7.1 自动发现插件

Pytest 会自动发现以下位置的插件：

  1. 项目根目录：以 pytest_ 开头的文件
  2. conftest.py：文件中的钩子函数
  3. 已安装的包：通过 setup.py 或 pyproject.toml 注册

7.2 在 conftest.py 中使用插件

方法 1：直接在 conftest.py 中定义

# conftest.py
def pytest_configure(config):
    """直接在 conftest.py 中定义插件"""
    print("conftest.py 中的插件")

方法 2：导入插件模块

# conftest.py
pytest_plugins = ["pytest_env_plugin", "pytest_result_logger"]

7.3 作为包安装插件

创建 setup.py：

from setuptools import setup, find_packages

setup(
    name="pytest-custom-plugins",
    version="1.0.0",
    packages=find_packages(),
    entry_points={
        "pytest11": [
            "env_plugin = pytest_env_plugin",
            "result_logger = pytest_result_logger",
            "auto_retry = pytest_auto_retry",
        ]
    },
    install_requires=[
        "pytest>=7.0.0",
    ],
)

安装：

# 开发模式安装
pip install -e .

# 或直接安装
pip install .

使用：

# 插件会自动加载
pytest test_example.py

7.4 使用 pyproject.toml

创建 pyproject.toml：

[build-system]
requires = ["setuptools>=45", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "pytest-custom-plugins"
version = "1.0.0"
requires-python = ">=3.7"
dependencies = [
    "pytest>=7.0.0",
]

[project.entry-points.pytest11]
env_plugin = "pytest_env_plugin"
result_logger = "pytest_result_logger"
auto_retry = "pytest_auto_retry"

8. 高级用法

8.1 插件之间的交互

示例：插件 A 使用插件 B 的功能

# pytest_plugin_a.py
def pytest_configure(config):
    """插件 A"""
    # 检查插件 B 是否已加载
    if hasattr(config, 'plugin_b_feature'):
        print("【插件 A】使用插件 B 的功能")
        config.plugin_b_feature.enable()
    else:
        print("【插件 A】插件 B 未加载")

# pytest_plugin_b.py
def pytest_configure(config):
    """插件 B"""
    class PluginBFeature:
        def enable(self):
            print("【插件 B】功能已启用")

    config.plugin_b_feature = PluginBFeature()

8.2 条件加载插件

示例：

def pytest_configure(config):
    """根据条件加载插件"""
    import os

    # 只在特定环境下加载插件
    if os.getenv("ENABLE_CUSTOM_PLUGIN") == "true":
        # 加载自定义插件
        config.pluginmanager.register(CustomPlugin())
        print("【插件】自定义插件已加载")
    else:
        print("【插件】自定义插件未加载（环境变量未设置）")

8.3 动态修改测试

示例：根据配置动态修改测试

def pytest_collection_modifyitems(config, items):
    """根据配置动态修改测试"""
    test_mode = config.getoption("--test-mode", default="normal")

    if test_mode == "quick":
        # 快速模式：只运行标记为 quick 的测试
        for item in items:
            if "quick" not in [mark.name for mark in item.iter_markers()]:
                item.add_marker(pytest.mark.skip(reason="快速模式跳过"))

    elif test_mode == "full":
        # 完整模式：运行所有测试
        pass

8.4 集成外部工具

示例：集成 Slack 通知

import requests

def pytest_runtest_makereport(item, call):
    """测试失败时发送 Slack 通知"""
    if call.when == "call" and call.excinfo is not None:
        # 测试失败，发送通知
        webhook_url = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
        message = {
            "text": f"测试失败: {item.name}",
            "attachments": [{
                "color": "danger",
                "text": str(call.excinfo.value)
            }]
        }

        try:
            requests.post(webhook_url, json=message)
            print("【插件】已发送 Slack 通知")
        except Exception as e:
            print(f"【插件】发送通知失败: {e}")

9. 调试和测试插件

9.1 调试插件

方法 1：使用 print 语句

def pytest_configure(config):
    """使用 print 调试"""
    print("【调试】插件配置开始")
    print(f"【调试】配置对象: {config}")
    print("【调试】插件配置完成")

方法 2：使用 logging

import logging

logger = logging.getLogger(__name__)

def pytest_configure(config):
    """使用 logging 调试"""
    logger.info("插件配置开始")
    logger.debug(f"配置对象: {config}")
    logger.info("插件配置完成")

方法 3：使用 pdb

def pytest_configure(config):
    """使用 pdb 调试"""
    import pdb
    pdb.set_trace()  # 设置断点
    print("插件配置")

9.2 测试插件

创建测试文件 test_plugin.py：

import pytest

def test_plugin_loaded():
    """测试插件是否加载"""
    # 检查插件功能是否可用
    assert True

def test_plugin_option(pytestconfig):
    """测试插件选项"""
    # 测试命令行选项
    test_env = pytestconfig.getoption("--test-env", default="dev")
    assert test_env in ["dev", "test", "prod"]

运行插件测试：

pytest test_plugin.py -v

10. 常见问题和最佳实践

10.1 常见问题

问题 1：插件未加载

原因：

  • 文件命名不正确
  • 插件未正确注册

解决方法：

# 确保文件以 pytest_ 开头
pytest_my_plugin.py  # ✓ 正确
my_plugin.py         # ✗ 错误

# 或在 conftest.py 中显式声明
pytest_plugins = ["my_plugin"]

问题 2：钩子函数未执行

原因：

  • 钩子函数名称错误
  • 参数不匹配

解决方法：

# 确保钩子函数名称正确
def pytest_configure(config):  # ✓ 正确
    pass

def pytest_config(config):  # ✗ 错误
    pass

问题 3：命令行选项未显示

原因：

  • pytest_addoption 未定义
  • 选项定义错误

解决方法：

def pytest_addoption(parser):
    """确保正确定义"""
    parser.addoption(
        "--my-option",
        action="store",
        default="default_value",
        help="选项说明"
    )

10.2 最佳实践

实践 1：模块化设计

好的做法：

# pytest_plugin.py
class PluginConfig:
    """插件配置类"""
    def __init__(self):
        self.test_env = "dev"
        self.verbose = False

class PluginManager:
    """插件管理类"""
    def __init__(self, config):
        self.config = config
        self.plugin_config = PluginConfig()

    def setup(self):
        """设置插件"""
        pass

def pytest_configure(config):
    """使用类组织代码"""
    manager = PluginManager(config)
    manager.setup()

实践 2：错误处理

好的做法：

def pytest_configure(config):
    """添加错误处理"""
    try:
        # 插件初始化代码
        initialize_plugin()
    except Exception as e:
        print(f"【警告】插件初始化失败: {e}")
        # 使用默认配置
        use_default_config()

实践 3：文档和注释

好的做法：

"""
Pytest 自定义插件

功能：
1. 管理测试环境
2. 记录测试结果
3. 发送测试通知

使用方法：
    pytest --test-env=prod test_example.py

作者：Your Name
日期：2024-01-01
"""

def pytest_configure(config):
    """
    pytest 配置钩子

    Args:
        config: pytest 配置对象
    """
    pass

实践 4：配置验证

好的做法：

def pytest_configure(config):
    """验证配置"""
    test_env = config.getoption("--test-env")

    if test_env not in ["dev", "test", "prod"]:
        raise ValueError(f"无效的测试环境: {test_env}")

    # 验证其他配置
    validate_config(config)

11. 完整实战项目

11.1 项目结构

myproject/
├── pytest_custom_plugins/
│   ├── __init__.py
│   ├── env_plugin.py          # 环境管理插件
│   ├── result_logger.py       # 结果记录插件
│   └── notification_plugin.py # 通知插件
├── tests/
│   ├── test_example.py
│   └── conftest.py
├── setup.py
├── pyproject.toml
└── README.md

11.2 环境管理插件

pytest_custom_plugins/env_plugin.py：

"""
环境管理插件
"""

import pytest
import os
import json

def pytest_addoption(parser):
    parser.addoption(
        "--test-env",
        action="store",
        choices=["dev", "test", "prod"],
        default="dev",
        help="测试环境"
    )

def pytest_configure(config):
    test_env = config.getoption("--test-env")
    config.test_env = test_env

    # 加载环境配置
    config_file = f"config/{test_env}.json"
    if os.path.exists(config_file):
        with open(config_file, 'r') as f:
            config.env_config = json.load(f)
    else:
        config.env_config = get_default_config(test_env)

@pytest.fixture(scope="session")
def api_base_url(request):
    return request.config.env_config['api_url']

def get_default_config(env):
    configs = {
        "dev": {"api_url": "https://dev-api.example.com"},
        "test": {"api_url": "https://test-api.example.com"},
        "prod": {"api_url": "https://api.example.com"}
    }
    return configs.get(env, configs["dev"])

11.3 结果记录插件

pytest_custom_plugins/result_logger.py：

"""
结果记录插件
"""

import pytest
import json
from datetime import datetime
from pathlib import Path

class ResultLogger:
    def __init__(self):
        self.results = []

    def add_result(self, name, status, duration, error=None):
        self.results.append({
            "name": name,
            "status": status,
            "duration": duration,
            "error": error,
            "timestamp": datetime.now().isoformat()
        })

    def save(self, filepath):
        with open(filepath, 'w') as f:
            json.dump(self.results, f, indent=2)

def pytest_configure(config):
    config.result_logger = ResultLogger()

def pytest_unconfigure(config):
    if hasattr(config, 'result_logger'):
        output_dir = Path("results")
        output_dir.mkdir(exist_ok=True)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        config.result_logger.save(output_dir / f"results_{timestamp}.json")

def pytest_runtest_makereport(item, call):
    if call.when == "call" and hasattr(item.config, 'result_logger'):
        status = "passed" if call.excinfo is None else "failed"
        error = str(call.excinfo.value) if call.excinfo else None
        item.config.result_logger.add_result(
            item.name, status, call.duration, error
        )

11.4 通知插件

pytest_custom_plugins/notification_plugin.py：

"""
通知插件
"""

import pytest
import requests

def pytest_addoption(parser):
    parser.addoption(
        "--enable-notifications",
        action="store_true",
        default=False,
        help="启用通知"
    )

def pytest_runtest_makereport(item, call):
    if call.when == "call" and call.excinfo is not None:
        if item.config.getoption("--enable-notifications"):
            send_notification(item.name, str(call.excinfo.value))

def send_notification(test_name, error):
    # 发送通知逻辑
    print(f"【通知】测试 {test_name} 失败: {error}")

11.5 注册插件

setup.py：

from setuptools import setup, find_packages

setup(
    name="pytest-custom-plugins",
    version="1.0.0",
    packages=find_packages(),
    entry_points={
        "pytest11": [
            "env = pytest_custom_plugins.env_plugin",
            "result_logger = pytest_custom_plugins.result_logger",
            "notification = pytest_custom_plugins.notification_plugin",
        ]
    },
    install_requires=["pytest>=7.0.0"],
)

11.6 使用插件

tests/test_example.py：

def test_api(api_base_url):
    """使用环境插件提供的 fixture"""
    import requests
    response = requests.get(f"{api_base_url}/users")
    assert response.status_code == 200

运行测试：

# 安装插件
pip install -e .

# 运行测试
pytest --test-env=prod --enable-notifications tests/

12. 总结

12.1 插件开发流程

  1. 确定需求：明确插件要实现的功能
  2. 选择钩子：选择合适的钩子函数
  3. 实现功能：编写插件代码
  4. 测试验证：测试插件功能
  5. 文档编写：编写使用文档
  6. 发布安装：打包和安装插件

12.2 常用钩子函数总结

钩子函数 调用时机 常用用途
pytest_addoption 添加命令行选项时 添加自定义命令行参数
pytest_configure pytest 配置时 初始化插件、添加标记
pytest_unconfigure pytest 退出前 清理资源
pytest_collection_modifyitems 收集测试项后 修改测试项、添加标记
pytest_runtest_setup 测试运行前 前置准备
pytest_runtest_teardown 测试运行后 后置清理
pytest_runtest_makereport 生成测试报告时 获取测试结果
pytest_report_header 生成报告头部时 添加自定义信息

12.3 插件开发建议

  1. 保持简单：插件应该专注于单一功能
  2. 文档完善：提供清晰的使用文档
  3. 错误处理：添加适当的错误处理
  4. 测试充分：为插件编写测试用例
  5. 向后兼容：保持 API 的稳定性

12.4 学习资源

通过本教程，你应该已经掌握了 pytest 自定义插件的开发方法。记住，插件开发是一个实践的过程，多写、多试、多思考，才能开发出优秀的插件！

发表评论