1

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

 2 years ago
source link: http://markshao.github.io/2019/10/05/hook-caller-logic/
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.

pluggy "_HookCall" 调用链分析

Toggle site
Catalog
You've read100%
Pytest 源码解读 [4] - pluggy "_HookCall" 调用链分析

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

python
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的实现就知道了

python
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函数的封装，

python
self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
           methods,
           kwargs,
           firstresult=hook.spec.opts.get("firstresult") if hook.spec else False,
       )

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

python
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的函数挺长的，我们分别看一下
python
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，就直接返回

我们看看代码的实现

python
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那一段逻辑

python
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 返回给到外部调用方

report

Recommend

About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK