1. 项目概述：深入理解pytest插件开发的两个核心钩子

如果你正在用pytest做自动化测试，并且已经不再满足于仅仅使用conftest.py来组织你的fixture和钩子，而是想开发一个独立的、可复用的插件来封装更复杂的测试逻辑，那么pytest_collection_modifyitems和pytest_runtest_setup这两个钩子函数绝对是你绕不开的核心。它们一个掌管着测试用例的“生杀大权”和“出场顺序”，另一个则深度介入每个测试用例执行前的“热身准备”。我见过不少团队在尝试插件开发时，对这两个钩子的使用场景和边界感到困惑，要么用错了地方导致插件行为诡异，要么因为性能问题让测试套件变得异常缓慢。今天，我就结合自己踩过的坑和实际项目经验，来彻底拆解这两个钩子，让你不仅能写出功能强大的pytest插件，更能写出高效、稳定的插件。

简单来说，pytest_collection_modifyitems是在pytest收集完所有测试用例（items）之后、任何测试执行之前被调用的。它给你一个机会去审视、筛选、排序甚至修改这些即将被执行的测试用例集合。而pytest_runtest_setup则是在每个独立的测试用例（item）执行其setup阶段（即执行测试函数本身之前）时被调用的，它是介入单个测试生命周期早期的一个关键节点。理解它们触发的时机和所能操作的对象，是正确使用它们的前提。一个常见的误区是把应该在收集阶段做的全局性工作（比如基于标签过滤）放到每个测试的setup阶段去做，这会导致不必要的重复计算和性能损耗。接下来，我们就从设计思路开始，一步步拆解如何用好这两个钩子。

2. 核心钩子函数的设计思路与职责边界

在动手写代码之前，我们必须先厘清每个钩子的设计初衷和职责边界。这就像盖房子要先画图纸，搞清楚承重墙在哪里，水管电线怎么走，否则代码堆起来就是一盘散沙，后期维护和调试会异常痛苦。

2.1pytest_collection_modifyitems：测试集的“总导演”

你可以把pytest_collection_modifyitems想象成一场演出的总导演。在演员（测试用例）海选（测试收集）结束后，正式彩排和演出（测试执行）开始前，导演需要做以下几件事：

1. 审核演员名单：看看收集到的演员是否符合要求（例如，只保留有特定标签的测试）。
2. 确定出场顺序：根据剧情需要或演员特点，安排谁先上场谁后上场（例如，让冒烟测试先执行，或者让资源密集型的测试最后执行）。
3. 给演员加戏：给某些演员临时增加一些道具或设定（例如，为所有测试动态添加一个自定义的marker，或者修改测试用例的nodeid）。

这个钩子函数接收两个关键参数：session和config，但最重要的是它接收并直接操作items列表。这个列表包含了所有被收集到的测试用例对象（pytest.Item或其子类，如pytest.Function）。你在这个钩子里对items列表所做的任何修改（如增、删、改、排序），都会直接影响后续测试执行的流程。

它的核心职责是批量处理和全局规划。所有只需要做一次、并且影响整个测试集的行为，都应该尽量放在这里。比如，根据命令行参数动态过滤用例、根据测试模块或类进行分组排序、为一批测试统一注入某种元数据等。

2.2pytest_runtest_setup：单个测试的“私人教练”

而pytest_runtest_setup则像是每个演员的私人教练，在演员即将上台表演前一刻，教练会进行最后的检查和准备。这个“前一刻”特指pytest执行测试的setup阶段，它位于标准的setup_module/setup_class/setup_methodfixture执行之后，但在测试函数体本身执行之前。

这个钩子函数接收一个参数：item，即当前即将执行的那个测试用例对象。它的职责是针对单个测试用例进行前置操作。你可以在这里：

1. 进行最后的条件检查：例如，检查某个外部服务是否可用，如果不可用则跳过当前测试（通过抛出pytest.skip）。
2. 执行特定的准备逻辑：这些逻辑可能不适合或无法通过fixture来实现。例如，某些遗留系统测试需要在执行前手动修改一个全局配置状态。
3. 动态修改测试上下文：基于当前测试的某些属性（如名称、所属模块），在运行时动态地为其绑定一些资源或数据。

它的核心特点是按需触发和精细控制。每个测试执行前都会调用一次，因此这里的代码执行频率很高，必须非常注意性能，避免耗时的操作。

2.3 关键区别与选用原则

为了避免混淆，我总结了一个简单的对照表：

选用原则：当你需要对整个测试集进行规划时，用pytest_collection_modifyitems；当你需要对单个测试执行前的瞬间进行干预时，用pytest_runtest_setup。一个简单的判断方法是：如果这个操作的结果对所有测试都是一样的，或者需要在测试开始前就决定好测试的执行序列，那么它很可能属于收集阶段；如果这个操作需要根据每个测试的具体情况来决定，那么它可能属于setup阶段。

3.pytest_collection_modifyitems实战：从过滤到排序的完整实现

理解了设计思路，我们来看具体怎么实现。我将通过一个完整的插件示例，展示pytest_collection_modifyitems最常见的几种用法。假设我们要开发一个叫pytest-smart-runner的插件，它需要实现：1）根据测试时长历史数据，优先运行快测；2）自动跳过标记为flaky（不稳定）的测试，除非显式要求运行它们。

3.1 插件基础结构与钩子声明

首先，创建一个pytest_smart_runner.py文件。一个pytest插件本质上就是一个可以导入的模块，其中定义了特定的钩子函数。

# pytest_smart_runner.py import pytest def pytest_addoption(parser): """添加插件专用的命令行参数。""" group = parser.getgroup("smart-runner") group.addoption( "--run-flaky", action="store_true", default=False, help="Run tests marked as 'flaky' (they are skipped by default)." ) group.addoption( "--seed", action="store", type=int, default=None, help="A seed to randomize test order for reproducibility." ) def pytest_configure(config): """在配置阶段初始化插件所需的数据或状态。""" # 我们可以在这里初始化一个“历史运行时长”的缓存字典。 # 实际项目中，这个数据可能来自文件或数据库。 config._smart_runner_history = {} # test_nodeid -> average_duration # 模拟一些历史数据 config._smart_runner_history.update({ "test_module.py::test_fast": 0.1, "test_module.py::test_slow": 2.5, "test_module.py::test_medium": 0.8, })

接下来是重头戏pytest_collection_modifyitems：

def pytest_collection_modifyitems(config, items): """修改收集到的测试用例列表。""" # 1. 过滤：默认跳过标记为‘flaky’的测试，除非用户指定--run-flaky if not config.getoption("--run-flaky"): non_flaky_items = [] for item in items[:]: # 遍历副本，因为我们要修改原列表 # 检查测试是否有‘flaky’标记 flaky_marker = item.get_closest_marker("flaky") if flaky_marker is None: non_flaky_items.append(item) else: # 给被跳过的测试添加一个skip标记，这样报告会更清晰 item.add_marker(pytest.mark.skip(reason="Flaky test skipped (use --run-flaky to run)")) # 替换items列表，只保留非flaky的测试 items[:] = non_flaky_items # 2. 排序：根据历史运行时长，短的先执行（让快速反馈更早） def get_duration(item): # 从模拟的历史缓存中获取时长，如果没有记录，返回一个较大的默认值（如10秒） return config._smart_runner_history.get(item.nodeid, 10.0) items.sort(key=get_duration) # 3. （可选）随机化：如果提供了--seed，则根据种子打乱顺序，用于发现顺序依赖的bug seed = config.getoption("--seed") if seed is not None: import random random.seed(seed) random.shuffle(items)

关键点解析与避坑指南：

1. 操作items列表本身：注意我们是直接修改items[:]（即原列表的切片副本赋值回原列表）或对items进行sort操作。这是修改测试集合的标准方式。
2. item.get_closest_marker：这是获取标记的推荐方法。直接访问item.keywords或item.own_markers可能不够准确，get_closest_marker会考虑从模块、类继承下来的标记。
3. 动态添加skip标记：我们不是直接return或pop掉item，而是给它加了一个skip标记。这样做的好处是，被跳过的测试仍然会出现在测试报告中（状态为SKIPPED），并附上原因，这比直接消失（不报告）要友好得多，便于统计和追溯。
4. 排序的稳定性：list.sort()是稳定的排序。这意味着如果两个测试的key（时长）相同，它们将保持原有的相对顺序。这通常是可接受的。
5. 性能考量：这里的排序操作时间复杂度是O(N log N)，对于成千上万的测试用例，这个开销在收集阶段是完全可以接受的，因为它只发生一次。但如果你的get_duration函数需要从网络或数据库读取数据，那就必须考虑缓存，否则会严重拖慢收集速度。

3.2 更复杂的场景：基于依赖关系的排序

在实际项目中，测试用例之间可能存在依赖关系。比如，test_B需要test_A创建的数据。虽然pytest不鼓励用例依赖，但某些遗留系统或集成测试中确实存在。我们可以利用pytest_collection_modifyitems来尝试进行拓扑排序。

假设我们通过一个自定义标记depends来声明依赖：

# test_module.py import pytest @pytest.mark.depends(on=["test_a"]) def test_b(): assert True def test_a(): assert True

在插件中，我们可以实现一个简单的拓扑排序（这里假设依赖关系无环）：

def pytest_collection_modifyitems(config, items): """实现基于依赖关系的测试排序。""" # 构建依赖图 dependency_graph = {} nodeid_to_item = {} for item in items: nodeid_to_item[item.nodeid] = item dep_marker = item.get_closest_marker("depends") if dep_marker: # depends标记可能这样使用：@pytest.mark.depends(on=["test_a"]) depends_on = dep_marker.kwargs.get("on", []) or dep_marker.args[0] if dep_marker.args else [] dependency_graph[item.nodeid] = depends_on else: dependency_graph[item.nodeid] = [] # 简单的拓扑排序（深度优先） visited = set() sorted_order = [] def visit(nodeid): if nodeid in visited: return visited.add(nodeid) for dep in dependency_graph.get(nodeid, []): # 确保依赖的测试也在本次收集的items中 if dep in nodeid_to_item: visit(dep) if nodeid in nodeid_to_item: sorted_order.append(nodeid_to_item[nodeid]) for nodeid in nodeid_to_item: visit(nodeid) # 用排序后的列表替换原列表 items[:] = sorted_order

注意：这是一个简化示例。真实的依赖管理要复杂得多，需要处理循环依赖、跨模块依赖、依赖测试失败后的处理策略等。通常建议使用专门的插件（如pytest-dependency）来处理复杂依赖，而不是自己从头造轮子。这里主要是展示pytest_collection_modifyitems在操作测试执行顺序上的强大能力。

4.pytest_runtest_setup实战：精细化的测试前置拦截与准备

现在我们把目光转向pytest_runtest_setup。这个钩子因为会在每个测试前运行，所以我们必须像对待热点代码一样对待它，确保其中的逻辑轻量、高效。

4.1 基础用法：运行时环境检查与跳过

一个最经典的场景是运行时环境检查。比如，某些测试需要特定的数据库版本、外部API可用，或者只有Linux环境下才能运行。

# pytest_env_check.py import pytest import sys import requests def pytest_runtest_setup(item): """ 在每个测试的setup阶段被调用。 可以在这里进行最后的条件检查，并跳过不满足条件的测试。 """ # 示例1：检查操作系统 linux_only_marker = item.get_closest_marker("linux_only") if linux_only_marker and not sys.platform.startswith("linux"): pytest.skip(f"Test requires Linux, current OS is {sys.platform}") # 示例2：检查外部服务是否可达 needs_external_api_marker = item.get_closest_marker("needs_external_api") if needs_external_api_marker: api_url = needs_external_api_marker.kwargs.get("url", "https://api.example.com/health") try: # 设置一个很短的超时，避免检查本身拖慢测试 resp = requests.get(api_url, timeout=2) if resp.status_code != 200: pytest.skip(f"External API is unhealthy (status: {resp.status_code})") except requests.exceptions.RequestException as e: pytest.skip(f"External API is unreachable: {e}") # 示例3：基于自定义fixture的复杂条件判断 # 假设有一个`fixture`叫`feature_toggle`，表示某个功能是否开启 # 我们可以通过item对象的`funcargs`属性（谨慎使用）来获取，但更推荐在测试函数内判断。 # 这里仅作演示，说明可以访问到已解析的fixture值（如果该fixture在测试参数中）。 if "feature_toggle" in item.fixturenames: # 注意：item.funcargs 可能在setup阶段还未完全填充，此方法并不完全可靠。 # 更稳健的做法是在fixture内部或测试函数开头进行判断。 pass

关键点与避坑指南：

1. 使用pytest.skip：这是在钩子函数中跳过测试的标准方式。它会抛出一个特殊的Skipped异常，pytest会捕获并处理它，将其记录为跳过状态。
2. 轻量级检查：pytest_runtest_setup中的代码会运行很多次。像网络请求、数据库查询等I/O操作，必须加上超时，并且要考虑缓存结果的可能性。例如，可以将外部服务的健康检查结果缓存在config对象中一段时间，避免每个测试都去请求。
3. 谨慎访问item.funcargs：在pytest_runtest_setup被调用时，并非所有fixture都已经完成解析和缓存。依赖于item.funcargs可能遇到KeyError。如果逻辑依赖某个fixture的值，更好的模式是：要么将该逻辑移入一个autouse的fixture中（在setup之前执行），要么在测试函数内部进行判断。

4.2 高级用法：动态Fixture注入与上下文修改

有时，我们可能想根据测试的某些属性，动态地改变它的执行上下文。虽然fixture是更推荐的方式，但在某些限制下，pytest_runtest_setup可以作为一种补充手段。

例如，一个测试可能需要根据其所在模块名称，连接不同的数据库：

# pytest_dynamic_db.py import os def pytest_runtest_setup(item): """根据测试模块动态设置环境变量，模拟连接不同数据库。""" # 获取测试用例所在的模块文件路径 module_path = item.location[0] # location 是一个元组 (文件路径, 行号, 测试名) module_name = os.path.splitext(os.path.basename(module_path))[0] # 假设模块名以‘test_customer_a’和‘test_customer_b’区分不同客户数据库 if module_name.startswith("test_customer_a"): original_db_url = os.environ.get("DATABASE_URL") os.environ["DATABASE_URL"] = "postgresql://user:pass@host_a/db" # 我们可以在这里将一个还原函数添加到item对象上，以便在teardown时恢复 if not hasattr(item, "_original_env"): item._original_env = {"DATABASE_URL": original_db_url} elif module_name.startswith("test_customer_b"): original_db_url = os.environ.get("DATABASE_URL") os.environ["DATABASE_URL"] = "postgresql://user:pass@host_b/db" if not hasattr(item, "_original_env"): item._original_env = {"DATABASE_URL": original_db_url} # 配套的teardown钩子，用于清理 def pytest_runtest_teardown(item, nextitem): """测试执行后清理动态设置的环境变量。""" if hasattr(item, "_original_env"): for key, value in item._original_env.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value

重要警告：上述修改全局环境变量的方式极具破坏性且不推荐在生产中使用，因为它不是线程/进程安全的，并且在pytest并行运行（pytest-xdist）时会完全失效甚至导致混乱。这里仅仅是为了演示pytest_runtest_setup和pytest_runtest_teardown的联动能力。

正确的做法是使用fixture和pytest的monkeypatchfixture来临时修改环境：

import pytest @pytest.fixture(autouse=True) def dynamic_db_url(request, monkeypatch): module_name = request.module.__name__ if module_name.startswith("test_customer_a"): monkeypatch.setenv("DATABASE_URL", "postgresql://user:pass@host_a/db") elif module_name.startswith("test_customer_b"): monkeypatch.setenv("DATABASE_URL", "postgresql://user:pass@host_b/db")

这个fixture方式更安全、更符合pytest的模式，并且兼容并行测试。pytest_runtest_setup更适合那些无法或很难通过fixture实现的、必须在setup时间点执行的一次性检查或操作。

5. 性能优化与常见陷阱排查

使用这两个钩子，尤其是pytest_runtest_setup，必须时刻警惕性能问题。同时，一些不当的使用会导致难以调试的问题。

5.1 性能优化要点

1. pytest_collection_modifyitems优化：

   • 避免重复计算：如果你需要根据测试的源代码或文件属性进行计算（如计算代码复杂度），尽量在钩子内缓存结果。或者，考虑使用pytest_collection_modifyitems的姊妹钩子pytest_pycollect_makeitem在更早的收集阶段进行计算和附加。
   • 懒加载外部数据：如果排序或过滤依赖于外部数据（如历史执行时间数据库），考虑在插件启动时（pytest_configure）一次性加载并缓存，而不是在pytest_collection_modifyitems中为每个item都去查询。

2. pytest_runtest_setup优化（重中之重）：

   • 将检查移出热路径：对于环境检查，如果所有测试都需要同样的检查（如检查同一个API），将其移至pytest_configure或一个session作用域的fixture中，只检查一次并存储结果。在pytest_runtest_setup中只读取缓存的结果。
   • 使用轻量级判断：优先使用标记（marker）判断、属性判断等内存操作，避免文件I/O、网络I/O和数据库查询。
   • 考虑使用pytest_runtest_protocol：如果你需要在每个测试前后执行操作，但又不严格限定在setup阶段，可以考虑使用pytest_runtest_protocol钩子，它包装了整个测试执行过程，有时可以合并一些逻辑。

5.2 常见问题与调试技巧

问题1：插件不生效

• 检查点：
  • 插件是否已安装或位于pytest能自动发现的目录（如测试根目录、conftest.py同目录、或通过setup.py/pyproject.toml声明）？
  • 钩子函数名是否拼写正确？pytest_collection_modifyitems和pytest_runtest_setup一个字母都不能错。
  • 是否在conftest.py中定义了同名钩子？conftest.py中的钩子优先级高于外部插件。
• 调试方法：使用pytest --trace-config查看加载的插件和钩子调用者。在钩子函数开头加print语句（或使用日志）是最直接的调试方式。

问题2：pytest_collection_modifyitems中修改items无效

• 原因：可能你的插件加载顺序晚于其他也修改了items的插件。pytest按插件注册顺序调用钩子，后调用的插件会覆盖前者的修改。
• 解决：尝试调整插件加载顺序（比较困难），或者确保你的逻辑能处理已经被其他插件修改过的items列表。使用tryfirst/trylast装饰器可以一定程度上影响钩子执行顺序，但不保证绝对顺序。

问题3：pytest_runtest_setup中抛出的异常导致测试报告混乱

• 现象：测试被标记为ERROR而不是SKIPPED或预期的失败。
• 解决：确保使用正确的pytest异常。想跳过测试，用pytest.skip；想标记测试为失败（在setup阶段），用pytest.fail；对于预期外的异常，应该让其抛出，pytest会将其捕获并记录为ERROR。不要用普通的Exception或AssertionError来代替pytest.skip。

问题4：与pytest-xdist并行执行时行为异常

• 关键：记住，pytest_collection_modifyitems在控制器进程（master）中执行，而pytest_runtest_setup在工作进程（worker）中执行。这意味着：
  • 在pytest_collection_modifyitems中对items的修改，会影响到分发给各个worker的测试集合。
  • 在pytest_runtest_setup中修改全局状态（如环境变量、模块级变量）是无效且危险的，因为每个worker有自己的内存空间。worker间通信需要通过pytest提供的特定机制（如sys.last_value，但很复杂）或外部存储。
  • 如果你的插件逻辑依赖共享状态，必须设计为支持分布式，或者考虑禁用并行。

问题5：钩子函数执行太慢

• 排查：使用pytest --durations=N来查看最慢的测试阶段。如果收集阶段（collection）很慢，问题可能在pytest_collection_modifyitems；如果每个测试的call阶段之前很慢，问题可能在pytest_runtest_setup或相关的autouse fixture。
• 优化：如前所述，缓存、懒加载、减少I/O。对于复杂的pytest_runtest_setup逻辑，考虑是否可以转移到pytest_collection_modifyitems中一次性处理（如果逻辑不依赖于运行时状态）。

6. 综合案例：构建一个智能测试筛选与预热插件

最后，我们综合运用两个钩子，设想一个更复杂的插件pytest-smart-scheduler。它的目标是：

1. 智能筛选：根据代码变更历史（例如通过git diff），只运行受影响的测试。
2. 执行预热：对于需要连接慢速外部资源（如数据库连接池）的测试，在第一个相关测试执行前进行一次性“预热”。

这个案例将展示如何协同使用两个钩子，并引入一些更高级的概念。

# pytest_smart_scheduler.py import pytest import subprocess import sys from pathlib import Path class SmartSchedulerPlugin: """将插件逻辑封装在一个类中，便于管理状态。""" def __init__(self, config): self.config = config self.affected_tests = set() self._warmed_up = False # 资源预热标志 def pytest_addoption(self, parser): group = parser.getgroup("smart-scheduler") group.addoption( "--affected-only", action="store_true", help="Only run tests affected by recent code changes (based on git diff)." ) group.addoption( "--warmup-service", action="store", help="Service endpoint to warm up before the first test that needs it." ) def pytest_configure(self, config): config.pluginmanager.register(SmartSchedulerPlugin(config), "smart-scheduler") if config.getoption("--affected-only"): self._calculate_affected_tests(config) def _calculate_affected_tests(self, config): """通过git diff计算受影响的测试文件/模块。这是一个简化示例。""" try: # 获取最近一次提交的diff result = subprocess.run( ["git", "diff", "--name-only", "HEAD~1"], capture_output=True, text=True, cwd=config.rootdir ) changed_files = result.stdout.strip().split('\n') changed_files = [f for f in changed_files if f] # 移除空行 # 将更改的文件映射到测试文件（简单的启发式规则：同目录下的test_*.py文件） for changed_file in changed_files: changed_path = Path(config.rootdir) / changed_file # 寻找对应的测试文件（实际项目需要更复杂的映射逻辑） test_file = changed_path.parent / f"test_{changed_path.stem}.py" if test_file.exists(): self.affected_tests.add(str(test_file.relative_to(config.rootdir))) except Exception as e: # 如果git命令失败，则回退到运行所有测试 config.issue_config_time_warning( pytest.PytestWarning(f"Failed to calculate affected tests: {e}. Running all tests."), stacklevel=2 ) def pytest_collection_modifyitems(self, config, items): """筛选受影响的测试。""" if not self.affected_tests: return filtered_items = [] for item in items: # 检查测试项所在的文件是否在受影响列表中 item_path = Path(item.location[0]).relative_to(config.rootdir) if str(item_path) in self.affected_tests: filtered_items.append(item) else: # 给未选中的测试添加一个skip标记，并说明原因 item.add_marker(pytest.mark.skip(reason="Test not affected by recent changes")) # 替换原列表 items[:] = filtered_items def pytest_runtest_setup(self, item): """在测试setup阶段进行资源预热。""" warmup_service = self.config.getoption("--warmup-service") if warmup_service and not self._warmed_up: # 检查当前测试是否需要该服务（例如，有特定标记） needs_warmup_marker = item.get_closest_marker("needs_service") if needs_warmup_marker: # 执行预热逻辑（例如，发送一个HTTP HEAD请求） import requests try: print(f"\n[SmartScheduler] Warming up service: {warmup_service}", file=sys.stderr) resp = requests.head(warmup_service, timeout=5) resp.raise_for_status() self._warmed_up = True print(f"[SmartScheduler] Service warm-up successful.\n", file=sys.stderr) except Exception as e: # 预热失败，可以跳过测试或标记为失败 pytest.skip(f"Service warm-up failed: {e}")

这个插件展示了如何将状态保存在插件类实例中，如何在pytest_collection_modifyitems中实现基于复杂条件的过滤，以及如何在pytest_runtest_setup中实现有状态、一次性的前置操作。注意，这个示例中的“受影响测试计算”非常原始，真实项目可能需要集成更复杂的静态分析工具。

开发pytest插件是一个深入理解pytest内部机制的过程。pytest_collection_modifyitems和pytest_runtest_setup这两个钩子为你提供了强大的切入点，让你可以定制化测试流程的骨架和肌肉。记住它们的分工：一个管全局规划，一个管临场准备。用好它们的关键在于清晰界定职责、时刻关注性能、并充分利用pytest已有的fixture机制，避免重新发明轮子。当你遇到一个测试流程上的定制需求时，先问问自己：“这个操作是针对整个测试集的，还是针对单个测试的？需要在执行前就决定好，还是可以等到测试马上开始时再决定？” 答案会清晰地指向该用哪个钩子。