Pytest源码底层解析:从使用者到框架二次开发者的进阶之路

import pluggy# 创建标记器hookspec = pluggy.HookspecMarker("demo")hookimpl = pluggy.HookimplMarker("demo")# 1. 定义钩子规范——声明"能做什么"class Spec:    @hookspec    def calculate(self, a, b):        """这是一个钩子，实现这个钩子就可以计算"""# 2. 实现钩子函数——定义"具体怎么做"class Impl1:    @hookimpl    def calculate(self, a, b):        print(f"Impl1: {a} + {b} = {a + b}")        return a + bclass Impl2:    @hookimpl    def calculate(self, a, b):        print(f"Impl2: {a} × {b} = {a * b}")        return a * b# 3. 创建PluginManager，连接规范和实现pm = pluggy.PluginManager("demo")pm.add_hookspecs(Spec)      # 注册钩子规范pm.register(Impl1())        # 注册插件1pm.register(Impl2())        # 注册插件2# 4. 调用钩子，所有插件都会被执行results = pm.hook.calculate(a=2, b=3)print(f"所有结果: {results}")

Impl1: 2 + 3 = 5Impl2: 2 × 3 = 6

# pytest_collection_modifyitems —— 测试收集后执行def pytest_collection_modifyitems(config, items):    """修改测试项的顺序、增删用例、动态添加mark"""    for item in items:        if "slow" in item.name:            item.add_marker(pytest.mark.slow)# pytest_configure —— 插件初始化时执行def pytest_configure(config):    """注册自定义标记、初始化全局数据"""    config.addinivalue_line("markers", "smoke: 冒烟测试")# pytest_runtest_setup —— 每个测试用例执行前执行def pytest_runtest_setup(item):    """测试用例前置处理"""    print(f"正在执行: {item.name}")

@hookimpl(tryfirst=True)   # 尽可能早执行def pytest_runtest_call(item):    ...@hookimpl(trylast=True)    # 尽可能晚执行def pytest_runtest_call(item):    ...@hookimpl(hookwrapper=True)  # 包装模式，可在执行前后注入逻辑def pytest_runtest_call(item):    outcome = yield    # 在yield之后可以获取执行结果并做后置处理‍

def main(args=None, plugins=None) -> Union[int, ExitCode]:    try:        # 1. 准备核心配置对象 Config        config = _prepareconfig(args, plugins)        # 2. 触发命令行主钩子，启动后续流程        return config.hook.pytest_cmdline_main(config=config)    finally:        if 'config' in locals():            config._ensure_unconfigure()

Session (根收集器，代表一次完整会话)├── Module (测试文件，如 test_user.py)│   ├── Class (测试类，如 TestUserAPI)│   │   ├── Function (测试方法，如 test_get_user)│   │   └── Function (测试方法，如 test_create_user)│   └── Function (独立的测试函数)└── Directory (目录)

def test_user_data(subtests):    user_info = {"name": "张三", "age": 18, "email": "invalid"}    for key, value in user_info.items():        with subtests.test(msg=f"校验字段: {key}", field=key):            if key == "email":                assert "@" in value            elif key == "age":                assert value >= 18            else:                assert value is not None

pytest_runtest_makereport：在每个测试执行完毕后获取执行结果pytest_runtest_setup：在重试前重置测试状态pytest_configure：添加自定义命令行参数和标记

# smart_retry_plugin.pyimport pytestimport timeclass SmartRetryPlugin:    def __init__(self, max_retries=2, base_delay=1):        self.max_retries = max_retries        self.base_delay = base_delay        self.retry_counts = {}    @pytest.hookimpl(tryfirst=True)    def pytest_runtest_makereport(self, item, call):        """在测试执行后检查是否失败，决定是否需要重试"""        if call.when == "call" and call.excinfo is not None:            nodeid = item.nodeid            retry_count = self.retry_counts.get(nodeid, 0)            if retry_count < self.max_retries:                self.retry_counts[nodeid] = retry_count + 1                # 递增等待策略                wait_time = self.base_delay * (retry_count + 1)                print(f"⚠️ {nodeid} 执行失败，{wait_time}秒后进行第{retry_count+1}次重试")                time.sleep(wait_time)                # 标记需要重试                item._retry_needed = True    @pytest.hookimpl(tryfirst=True)    def pytest_runtest_setup(self, item):        """重试前重置测试状态"""        if hasattr(item, '_retry_needed') and item._retry_needed:            item._retry_needed = False            # 重置测试执行状态            if hasattr(item, '_previous_failed'):                delattr(item, '_previous_failed')def pytest_addoption(parser):    """注册自定义命令行参数"""    parser.addoption("--smart-retry", action="store", default="2",                     help="失败重试次数，默认为2")def pytest_configure(config):    """配置插件并读取参数"""    max_retries = int(config.getoption("--smart-retry") or 2)    plugin = SmartRetryPlugin(max_retries=max_retries)    config.pluginmanager.register(plugin, "smart_retry_plugin")

# 在测试目录下执行，启用智能重试pytest tests/ --smart-retry=3

⚠️ test_user.py::test_get_user 执行失败，2秒后进行第1次重试⚠️ test_user.py::test_get_user 执行失败，3秒后进行第2次重试✅ test_user.py::test_get_user 重试后成功‍