跳至主要內容

pytest 插件开发实战——自定义插件、hooks、报告扩展

老张2026/4/30大约 7 分钟

pytest 插件开发实战——自定义插件、hooks、报告扩展

适读人群：有 pytest 使用经验、希望深度定制测试框架的工程师 | 阅读时长：约 16 分钟 | 核心价值：掌握 pytest 插件开发机制，让测试框架为你的团队量身定制

一个让全团队受益的自定义插件

我们团队做 API 测试，有个共同需求：每次测试失败，我们都希望自动记录请求和响应的详细信息，包括 URL、请求头、请求体、响应状态码、响应体。

最原始的做法是每个测试都写日志代码，代码重复，还容易遗漏。

后来我花了一天写了一个 pytest 插件，把这个逻辑内置进去，测试工程师一行代码都不需要额外写，失败时自动就有完整的请求/响应日志。

这就是 pytest 插件的价值——把团队共同需要的能力做进框架，让所有人受益。

pytest 插件的三种形式

形式一：conftest.py 插件（推荐入门）

最简单的插件形式，放在 conftest.py 里的 hook 函数，自动被 pytest 发现：

# conftest.py

def pytest_runtest_makereport(item, call):
    """pytest hook：在测试报告生成时调用"""
    if call.when == "call" and call.excinfo is not None:
        # 测试失败了
        print(f"\n测试失败: {item.name}")
        # 可以从 item 获取 fixture 数据
        if hasattr(item, "funcargs"):
            if "page" in item.funcargs:
                # 自动截图
                page = item.funcargs["page"]
                page.screenshot(path=f"failures/{item.name}.png")

形式二：本地插件包

在项目内创建独立的插件包，通过 conftest.py 导入：

my_project/
├── tests/
│   ├── conftest.py    # 导入插件
│   └── test_*.py
├── pytest_plugins/
│   ├── __init__.py
│   ├── api_logger.py  # 插件实现
│   └── retry.py       # 重试插件
└── setup.cfg

形式三：可发布的 pip 包

适合跨项目复用，发布到 PyPI 供其他项目使用（比如 pytest-django、pytest-cov 就是这种形式）。

核心 Hook 详解

pytest 定义了丰富的 hook，覆盖测试生命周期的每个阶段：

pytest_configure：插件初始化

def pytest_configure(config):
    """
    pytest 启动时调用，用于注册 marker、配置选项等
    """
    # 注册自定义 marker，避免 PytestUnknownMarkWarning
    config.addinivalue_line(
        "markers",
        "slow: 标记为慢速测试（运行时间 > 5 秒）"
    )
    config.addinivalue_line(
        "markers",
        "integration: 集成测试，需要真实数据库"
    )
    config.addinivalue_line(
        "markers",
        "smoke: 冒烟测试，CI 快速验证用"
    )

pytest_addoption：添加命令行选项

def pytest_addoption(parser):
    """添加自定义命令行参数"""
    parser.addoption(
        "--env",
        action="store",
        default="test",
        choices=["test", "staging", "prod"],
        help="指定运行环境（默认：test）"
    )
    parser.addoption(
        "--browser",
        action="store",
        default="chromium",
        choices=["chromium", "firefox", "webkit"],
        help="E2E 测试使用的浏览器"
    )
    parser.addoption(
        "--capture-failures",
        action="store_true",
        default=False,
        help="失败时自动截图和记录日志"
    )

# 在 fixture 中使用命令行参数
@pytest.fixture(scope="session")
def env(request):
    return request.config.getoption("--env")

@pytest.fixture(scope="session")  
def config(env):
    return load_config_for_env(env)

pytest_runtest_protocol：完全控制测试执行

def pytest_runtest_protocol(item, nextitem):
    """
    完全控制单个测试的执行协议
    返回 True 表示接管了测试执行，pytest 不再执行默认流程
    返回 None/False 表示继续使用默认流程
    """
    # 大多数情况下返回 None，让 pytest 按默认流程执行
    return None

pytest_runtest_makereport：访问测试结果

import pytest

@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
    """
    hookwrapper=True 让我们可以在默认实现前后插入逻辑
    这个 hook 在每个阶段（setup/call/teardown）都会调用
    """
    outcome = yield  # 执行默认的报告生成逻辑
    report = outcome.get_result()
    
    # 把报告对象存到 item 上，方便其他 fixture 访问
    item.stash[pytest.StashKey()] = report
    
    if report.when == "call" and report.failed:
        # 测试执行阶段失败了
        handle_test_failure(item, report)

def handle_test_failure(item, report):
    """处理测试失败"""
    # 获取失败信息
    failure_info = {
        "test_name": item.name,
        "test_file": str(item.fspath),
        "failure_message": report.longreprtext,
        "timestamp": datetime.now().isoformat(),
    }
    
    # 如果是 API 测试，附加请求/响应信息
    if "api_client" in item.funcargs:
        api_client = item.funcargs["api_client"]
        failure_info["last_request"] = api_client.last_request
        failure_info["last_response"] = api_client.last_response
    
    # 保存到文件
    failure_dir = Path("test-results/failures")
    failure_dir.mkdir(parents=True, exist_ok=True)
    with open(failure_dir / f"{item.name}_{int(time.time())}.json", "w") as f:
        json.dump(failure_info, f, ensure_ascii=False, indent=2)

实战：开发一个 API 测试日志插件

# pytest_plugins/api_logger.py

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

class APILoggerPlugin:
    """
    自动记录 API 测试请求/响应的 pytest 插件
    """
    
    def __init__(self, config):
        self.config = config
        self.enabled = config.getoption("--capture-failures", default=False)
        self.log_dir = Path("test-results/api-logs")
        if self.enabled:
            self.log_dir.mkdir(parents=True, exist_ok=True)
    
    @pytest.hookimpl(hookwrapper=True)
    def pytest_runtest_makereport(self, item, call):
        outcome = yield
        report = outcome.get_result()
        
        if not self.enabled:
            return
        
        if report.when == "call" and report.failed:
            self._save_failure_log(item, report)
    
    def _save_failure_log(self, item, report):
        log_data = {
            "test_name": item.name,
            "test_nodeid": item.nodeid,
            "failure_time": datetime.now().isoformat(),
            "failure_message": str(report.longrepr),
        }
        
        # 收集 API 相关的 fixture 数据
        for fixture_name in ["api_client", "http_client", "session"]:
            if fixture_name in item.funcargs:
                client = item.funcargs[fixture_name]
                if hasattr(client, "request_history"):
                    log_data["request_history"] = [
                        {
                            "method": req.method,
                            "url": req.url,
                            "headers": dict(req.headers),
                            "body": req.body,
                            "response_status": req.response.status_code if req.response else None,
                            "response_body": req.response.text if req.response else None,
                        }
                        for req in client.request_history[-5:]  # 最近 5 个请求
                    ]
        
        # 保存日志
        log_file = self.log_dir / f"{item.name.replace('/', '_')}_{int(time.time())}.json"
        with open(log_file, "w", encoding="utf-8") as f:
            json.dump(log_data, f, ensure_ascii=False, indent=2)
        
        print(f"\n失败日志已保存: {log_file}")


def pytest_configure(config):
    config.pluginmanager.register(APILoggerPlugin(config), "api_logger")


def pytest_addoption(parser):
    parser.addoption(
        "--capture-failures",
        action="store_true",
        default=False,
        help="失败时自动保存 API 请求/响应日志"
    )

实战：开发一个智能重试插件

# pytest_plugins/smart_retry.py

import pytest
import time

class SmartRetryPlugin:
    """
    智能重试插件：只重试被标记为可重试的测试，并记录重试统计
    """
    
    def __init__(self):
        self.retry_counts = {}
        self.max_retries = 2
    
    @pytest.hookimpl(tryfirst=True)
    def pytest_runtest_protocol(self, item, nextitem):
        """在测试执行前检查是否有 retry marker"""
        marker = item.get_closest_marker("retry")
        if marker is None:
            return None  # 不干预，按默认流程
        
        max_retries = marker.kwargs.get("max_retries", self.max_retries)
        delay = marker.kwargs.get("delay", 1)
        
        for attempt in range(max_retries + 1):
            # 执行测试
            reports = self._run_test(item)
            
            # 检查是否通过
            failed = any(r.failed for r in reports if r.when == "call")
            
            if not failed:
                self._set_reports(item, reports)
                return True
            
            if attempt < max_retries:
                print(f"\n[重试] {item.name} 第 {attempt + 1} 次失败，{delay}s 后重试...")
                time.sleep(delay)
        
        # 所有重试都失败了，返回最后一次的报告
        self._set_reports(item, reports)
        self.retry_counts[item.nodeid] = max_retries
        return True
    
    def _run_test(self, item):
        """执行单次测试并收集报告"""
        reports = []
        for when in ["setup", "call", "teardown"]:
            report = item.runtest()
            reports.append(report)
        return reports


def pytest_configure(config):
    config.pluginmanager.register(SmartRetryPlugin(), "smart_retry")

# 用法：
# @pytest.mark.retry(max_retries=3, delay=2)
# def test_flaky_external_api():
#     ...

自定义报告扩展

# pytest_plugins/custom_reporter.py

import pytest
from pathlib import Path

class HTMLSummaryReporter:
    """生成简洁的 HTML 摘要报告"""
    
    def __init__(self):
        self.results = []
    
    def pytest_runtest_logreport(self, report):
        if report.when == "call":
            self.results.append({
                "nodeid": report.nodeid,
                "outcome": report.outcome,  # "passed" / "failed" / "skipped"
                "duration": report.duration,
                "failure": str(report.longrepr) if report.failed else None,
            })
    
    def pytest_sessionfinish(self, session, exitstatus):
        """测试会话结束时生成报告"""
        total = len(self.results)
        passed = sum(1 for r in self.results if r["outcome"] == "passed")
        failed = sum(1 for r in self.results if r["outcome"] == "failed")
        skipped = sum(1 for r in self.results if r["outcome"] == "skipped")
        
        html = self._generate_html(total, passed, failed, skipped)
        
        report_path = Path("test-results/summary.html")
        report_path.parent.mkdir(parents=True, exist_ok=True)
        report_path.write_text(html, encoding="utf-8")
        print(f"\nHTML 报告生成：{report_path}")
    
    def _generate_html(self, total, passed, failed, skipped):
        pass_rate = passed / total * 100 if total > 0 else 0
        
        return f"""<!DOCTYPE html>
<html>
<head><title>测试报告</title></head>
<body>
<h1>测试摘要</h1>
<p>通过率: {pass_rate:.1f}% ({passed}/{total})</p>
<p>通过: {passed} | 失败: {failed} | 跳过: {skipped}</p>
</body>
</html>"""

踩坑实录

坑一：hookwrapper 中忘记 yield 导致 hook 不执行

现象： 用了 @pytest.hookimpl(hookwrapper=True) 但实际 hook 逻辑没有被执行。

原因： hookwrapper=True 的函数必须有 yield，yield 之前是"前置"逻辑，yield 之后是"后置"逻辑。如果没有 yield，pytest 会报错或跳过后续执行。

# 错误：没有 yield
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(self, item, call):
    outcome = yield  # 必须有这个 yield！
    report = outcome.get_result()
    # 后置逻辑

坑二：插件注册顺序导致 hook 优先级问题

现象： 两个插件都 hook 了同一个事件，执行顺序不符合预期。

解法： 使用 tryfirst 或 trylast 控制优先级：

@pytest.hookimpl(tryfirst=True)   # 最先执行
def pytest_runtest_setup(item):
    ...

@pytest.hookimpl(trylast=True)    # 最后执行  
def pytest_runtest_setup(item):
    ...

坑三：在 conftest.py 中修改 item.funcargs 导致其他 fixture 异常

现象： 在 hook 函数中尝试访问 item.funcargs 修改某个 fixture，导致其他 fixture 的清理代码报错。

解法： hook 函数中只读取 funcargs，不要修改。如果需要传递数据，用 item.stash：

# 在 hook 中存储数据
item.stash[my_key] = {"some": "data"}

# 在另一个 hook 中读取
data = item.stash.get(my_key, None)

小结

pytest 插件开发的核心是 hook 机制。常用场景：

  • 失败时自动截图/日志：pytest_runtest_makereport + hookwrapper=True
  • 自定义命令行选项：pytest_addoption
  • 注册自定义 marker：pytest_configure
  • 智能重试：pytest_runtest_protocol
  • 自定义报告：pytest_runtest_logreport + pytest_sessionfinish

开发一个好的团队内部插件，能让每个测试工程师都少写大量重复代码，收益很高。

编辑此页
最近更新:
贡献者: xiaozhangge