Pytest 源码解读 [6] - pluggy 的插件执行顺序

其实本来想介绍下 tryfirst , trylast 这几个特性和实现的，不过仔细看了下实现，发现 pytest 在插件的执行顺序上还有挺多巧妙的设计可以讲的。为了更好理解插件存储的顺序，我主要想介绍下下面几个部分

• 什么是 wrapper 类型的 hook
• 什么是 historical 类型的hook
• 最后来看下 hook 执行顺序

什么是 wrapper 类型的 hook

hookwrapper=True 的时候，就是申明了一个 wrapper 类型的 hook ，看这个例子就明白了

hookimpl = HookImplMarker("pluggy")

hookimpl(hookwrapper=True)
def hookfunc(**kwargs):
	print("before hook execution")
	result = yield # get results from other normal hooks 
	print(result.get_result())

一个 wrapper 类型的 hook，可以看作是其他类型的 hook的装饰器，它会先执行一段代码，然后通过 yield 切换协程。当其他 hook 都执行完毕后后，重新唤起当前协程，并拿到执行结果再完成剩下的指令。

它的实现其实也很简单直白，我们看下_caller.py 中的实现

# part1
if hook_impl.hookwrapper:
     try:
          # If this cast is not valid, a type error is raised below,
          # which is the desired response.
            res = hook_impl.function(*args)
            gen = cast(Generator[None, _Result[object], None], res)
            next(gen)  # first yield
            teardowns.append(gen)
      except StopIteration:
            _raise_wrapfail(gen, "did not yield")

...
# part2
# run all wrapper post-yield blocks
        for gen in reversed(teardowns):
            try:
                gen.send(outcome)
                _raise_wrapfail(gen, "has second yield")
            except StopIteration:
                pass

return outcome.get_result()

先看 part1，本质就是先判断下这个 hook 是不是 wrapper 类型。如果是话就先通过 .function 创建一个协程，这个时候返回值 res 实际是一个协程，第二行的 cast 只是一个 typing 的语法，把这个对象专程一个协程类型的对象 gen 。所有只有在执行 next(gen) 的时候，我们才执行了 wrapper hook 中 yield 语句之前的那部分代码。再把这个协程放到一个 tearndown 的 list 里面去

再看 part2 ，当我们执行完了所有的其他普通 hook 后，我们再通过遍历这个 teardown ，通过 send 语句，唤醒之前的协程，并把结果传递过去。我们看后面 _raise_wrapfail , 其实就是规定这个协程唤醒后就应该结束了 raise StopIteration ，所以如果执行到了这行代码，就会 raise expcetion

什么是 historical 类型的hook

historical=True 是 HookspecMarker 上的一个属性，所以它其实会影响到和这个 hook 相关的所有的 HookImpl 上。当我们通过特定的 call_historic 来进行 hook 调用的时候，它带记忆能力，会记录下这次执行的历史。当下次有新的 HookImpl 注册的时候，它会自动回放过去的调用记录。

它的实现也是比较简单的，我们先看下 _hooks.py 的 call_historic 实现

self._call_history.append((kwargs, result_callback))

它其实是记录了当时调用 hook 的参数，这样就可以用于后续的回放了。我们再来看 _manager.py 当我们注册新的 HookImpl 的时候会尝试去回放过去的call_historic 记录，为什么要用 maybe 呢，因为我也不确定之前有没有记录呀0

hook._maybe_apply_history(hookimpl)

再看具体的实现, 就是把历史记录取出来，一一执行

def _maybe_apply_history(self, method: "HookImpl") -> None:
    """Apply call history to a new hookimpl if it is marked as historic."""
    if self.is_historic():
        assert self._call_history is not None
        for kwargs, result_callback in self._call_history:
            res = self._hookexec(self.name, [method], kwargs, False)
            if res and result_callback is not None:
                # XXX: remember firstresult isn't compat with historic
                assert isinstance(res, list)
                result_callback(res[0])

最后来看下 hook 执行顺序

前面介绍了2个不同的 hook 类型，一个是 historical ，另外一个是 wrapper，其实 historical 会特殊些，它需要通过特定的函数来触发。其实在 HookImplMarker 的实现中还有2个 option ，分别是 tryfirst 和 trylast ，分别用来控制执行的顺序在前还是在后 。 我们知道 pluggy 的 hook 实现本质是在内部实现了一个 1:N 的关系，但因为这些特殊的 hook 属性，实际上在我们通过 register 插入 hook 的时候，pluggy 会帮我们维持一个特定的顺序。我们先看下 _hooks.py 中 _add_hookimpl 的实现

def _add_hookimpl(self, hookimpl: "HookImpl") -> None:
    """Add an implementation to the callback chain."""
    for i, method in enumerate(self._hookimpls):
        if method.hookwrapper:
            splitpoint = i
            break
    else:
        splitpoint = len(self._hookimpls)
    if hookimpl.hookwrapper:
        start, end = splitpoint, len(self._hookimpls)
    else:
        start, end = 0, splitpoint

if hookimpl.trylast:
        self._hookimpls.insert(start, hookimpl)
    elif hookimpl.tryfirst:
        self._hookimpls.insert(end, hookimpl)
    else:
        # find last non-tryfirst method
        i = end - 1
        while i >= start and self._hookimpls[i].tryfirst:
            i -= 1
        self._hookimpls.insert(i + 1, hookimpl)

它的最终效果就是如图，无论你用什么样的顺序插入，最后都会变成类似这样

前面说过，hookwrapper 会先于其他的 hook 先执行，所以放在整个 list 一边是很合理的，但为啥感觉顺序有点怪怪呢？你看到 _callers.py 的这里就豁然开朗了, 哈哈，人家直接用了一个 reversed

def _multicall(
    hook_name: str,
    hook_impls: Sequence["HookImpl"],
    caller_kwargs: Mapping[str, object],
    firstresult: bool,
) -> Union[object, List[object]]:
    """Execute a call into multiple python functions/methods and return the
    result(s).

``caller_kwargs`` comes from _HookCaller.__call__().
    """
    __tracebackhide__ = True
    results: List[object] = []
    excinfo = None
    try:  # run impl and wrapper setup functions in a loop
        teardowns = []
        try:
            for hook_impl in reversed(hook_impls):

好了，关于 pluggy 整体的介绍就到这里了，本来还想介绍下 tracing 但看了下意义不大，后面就重点开始 pytest 实现的分析了