Pytest 源码解读 [4] - pluggy "_HookCall" 调用链分析

pytest 源码解读

我们知道pm.hook.xxx(**kwargs)　，最后实际是调用了绑定的 _HookCaller对象的__call__方法，那么今天我们来看一下这个方法的逻辑是什么样的，废话少说，先上代码, 我们先看下_HookCaller的构造函数

1
2
3
4
5
6
7
8
9
10
11
12
def __init__(self, name, hook_execute, specmodule_or_class=None, spec_opts=None):
    self.name = name
    self._wrappers = []
    self._nonwrappers = []
    self._hookexec = hook_execute
    self.argnames = None
    self.kwargnames = None
    self.multicall = _multicall
    self.spec = None
    if specmodule_or_class is not None:
        assert spec_opts is not None
        self.set_specification(specmodule_or_class, spec_opts)

这里有两个List是用来存放 hook 对应的 plugin 的，分为 wrapper 和 nonwrapper 两种类型，当我们通过调用_HookCaller的方法_add_hookimpl来注册plugin的时候，会根据HookimplMarker装饰器的hookwrapper属性来区分不同的 plugin list , 简单看几行_add_hookimpl的实现就知道了

1
2
3
4
5
6
7
8
9
10
11
12
def _add_hookimpl(self, hookimpl):
    """Add an implementation to the callback chain.
    """
    if hookimpl.hookwrapper:
        methods = self._wrappers
    else:
        methods = self._nonwrappers

    if hookimpl.trylast:
        methods.insert(0, hookimpl)
    elif hookimpl.tryfirst:
        methods.append(hookimpl)

，我们再继续回来看_HookCaller的构造函数，有两个属性看起来很类似，一个是self._hookexec，一个是self.multicall, _multicall是_HookCaller自己的 method , _hookexec 是通过构造函数传入的，我们看下PluginManager在绑定_HookCaller对象的时候的代码,hook = _HookCaller(name, self._hookexec),传入的是一个_hookexec, 再继续网上爬代码，发现这个所谓的 self._hookexec ，实际就是multicall函数的封装，

1
2
3
4
5
self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
           methods,
           kwargs,
           firstresult=hook.spec.opts.get("firstresult") if hook.spec else False,
       )

看到这里我们知道了，原来这两个属性最后底层都是同一个_multicall函数对象 。 前面铺垫了那么多，那我们来看一下核心调用函数__call__到底再干嘛

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def __call__(self, *args, **kwargs):
    if args:
        raise TypeError("hook calling supports only keyword arguments")
    assert not self.is_historic()
    if self.spec and self.spec.argnames:
        notincall = (
            set(self.spec.argnames) - set(["__multicall__"]) - set(kwargs.keys())
        )
        if notincall:
            warnings.warn(
                "Argument(s) {} which are declared in the hookspec "
                "can not be found in this hook call".format(tuple(notincall)),
                stacklevel=2,
            )
    return self._hookexec(self, self.get_hookimpls(), kwargs)
  • 先判断下调用参数的类型，不支持非kwargs类型的参数
  • 判断一下是否是 historical call 的，这里回头要研究一下为啥不允许historical call
  • 判断下调用的参数类型是否符合 spec 定义的要求，如果不符合会给 warning ，这里有点困惑, 既然不符合spec 规定，为什么不报错停止，只是给一个 warning ，有了解的同学欢迎给我留言
  • 最后是调用self._hookexec，把所有的注册的plugin 作为参数传入，根据上面的介绍，就是调用了_multicall的函数，_multicall的函数挺长的，我们分别看一下
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def _multicall(hook_impls, caller_kwargs, firstresult=False):
    """Execute a call into multiple python functions/methods and return the
    result(s).

    ``caller_kwargs`` comes from _HookCaller.__call__().
    """
    __tracebackhide__ = True
    results = []
    excinfo = None
    try:  # run impl and wrapper setup functions in a loop
        teardowns = []
        try:
            for hook_impl in reversed(hook_impls):
                try:
                    args = [caller_kwargs[argname] for argname in hook_impl.argnames]
                except KeyError:
                    for argname in hook_impl.argnames:
                        if argname not in caller_kwargs:
                            raise HookCallError(
                                "hook call must provide argument %r" % (argname,)
                            )
  • 原来这里还是会再检查调用的参数，如果spec定义的参数没有在kwargs中，这次是报错了

然后就是根据是否是hookwrapper来区分调用逻辑，我们先看下也应该正确的调用逻辑

  • hookwrapper
    • 先调用 hookwrapper的plugin,先执行yield之前的代码
    • 然后执行其他 plugin
    • 最后把其他的plugin的执行结果作为参数传回到hookwrapper的plugin的 yield point ,继续执行
  • nonwrapper
    • 直接调用 plugin ，并把结果返回到 result list中
    • 如果有 firstresult，就直接返回

我们看看代码的实现

1
2
3
4
5
6
7
8
9
10
11
12
13
if hook_impl.hookwrapper:
    try:
        gen = hook_impl.function(*args)
        next(gen)  # first yield
        teardowns.append(gen)
    except StopIteration:
        _raise_wrapfail(gen, "did not yield")
else:
    res = hook_impl.function(*args)
    if res is not None:
        results.append(res)
        if firstresult:  # halt further impl calls
            break
  • 同上面的逻辑，next(gen), 先执行yield前的逻辑，把gen存放在teardown 里，用作后续的callback
  • nonwrapper,　就直接调用，把结果存在result里

最后finllay那一段逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
finally:
    if firstresult:  # first result hooks return a single value
        outcome = _Result(results[0] if results else None, excinfo)
    else:
        outcome = _Result(results, excinfo)

    # 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()

  • 所有的结果都要用 _Result来封装返回
  • 如果是 hookwrapper的，还要遍历之前存的generator, 把nonwrapper的结果回调回去 gen.send(output),为了避免实现错误，例如在plugin里写了两个yield, 直接在外部调用逻辑里抛出异常终止逻辑_raise_wrapfail(gen, "has second yield")
  • 最后是把outcome 返回给到外部调用方