Pytest 源码解读 [2] - [pluggy] 核心设计理念和代码结构

pytest 源码解读

代码结构

pluggy 的核心代码非常简介，把 repo 克隆到本地目录后，它的核心代码就是在 src目录下的4个文件

  • _tracing.py - 调试作用，把 hookspec 和 plugin 调用链路的分析打印出来，一般不开启
  • callers.py - 主要是 _multicall 这个方法，用来实现对于 plugin 的调用逻辑
  • hooks.py - 核心是 HookspeckMarker 和 HookimplMarker, 用来装饰 hook 和 plugin 的方法
  • manager.py - PluginManager 是整个 pluggy 的核心类，负责 hook 和 plugin 的管理和核心调用逻辑

核心设计理念

整个 pluggy 的核心逻辑就是这么几行代码

1
2
3
4
5
pm = PluginManager("pluggy_demo_1")
pm.add_hookspecs(HookSpec)
pm.register(HookImpl1())
pm.register(HookImpl2())
print(pm.hook.calculate(a=1, b=2))
  • 创建一个 PluginManager 对象
  • 调用 add_hookspecs，注册一个新的 hook object
  • 调用 register 注册一个新的 plugin object
  • 通过 pm.hook.calculate 来实现对 plugin 的调用

这里就有几个核心设计的问题，通过解答这些问题可以帮助我们对于整个 pluggy的核心设计原理有进一步的了解

  1. pluggy 是怎么知道 Spec 里面哪些方法是定义的 hook ？

  2. pluggy 是怎么把 Plugin 的具体实现和 hook 做关联的

  3. 实际调用的时候，pm.hook 是个什么, 如何实现最终的 plugin 调用的

pluggy 是怎么知道 Spec 里面哪些方法是定义的 hook ？

这个就要全靠 HookspecMarker 和 HookimplMarker 这两个装饰器了，因为两者的实现逻辑类似，这里就只解释一下HookspecMarker, 我们先看上篇博文的demo中，我们一开始就定义了2个装饰器，并用他们来注释了 hook的方法

1
2
3
4
5
6
7
8
spec = HookspecMarker("pluggy_demo_1")
impl = HookimplMarker("pluggy_demo_1")


class HookSpec:
    @spec(firstresult=True)
    def calculate(self, a, b):
        pass

这里HookspecMarker是一个基于class实现的装饰器，我们看一下代码里它的具体实现逻辑 , __call__方法，

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def __call__(
    self, function=None, firstresult=False, historic=False, warn_on_impl=None
):
    """ if passed a function, directly sets attributes on the function
    which will make it discoverable to add_hookspecs().  If passed no
    function, returns a decorator which can be applied to a function
    later using the attributes supplied.

    If firstresult is True the 1:N hook call (N being the number of registered
    hook implementation functions) will stop at I<=N when the I'th function
    returns a non-None result.

    If historic is True calls to a hook will be memorized and replayed
    on later registered plugins.

    """

    def setattr_hookspec_opts(func):
        if historic and firstresult:
            raise ValueError("cannot have a historic firstresult hook")
        setattr(
            func,
            self.project_name + "_spec",
            dict(
                firstresult=firstresult,
                historic=historic,
                warn_on_impl=warn_on_impl,
            ),
        )
        return func

    if function is not None:
        return setattr_hookspec_opts(function)
    else:
        return setattr_hookspec_opts

原来主要的功能就是给被装饰的函数增加一个特别的属性, 属性的名字是 Project_name + _spec, 属性的value 就是装饰器的参数取值，

这样在PluginManger的 add_hookspecs函数的逻辑中，我们就可以看到通过遍历对象的所以属性，找到含有这这个特殊的attribute 的方法就好了

1
2
3
for name in dir(module_or_class):
    spec_opts = self.parse_hookspec_opts(module_or_class, name)
    if spec_opts is not None:

当然这里就有一个前提，全局的project_name 必须是一致的且唯一，因为这个key是拼出来的，如果不一致就找不到 对应的 hook 和 impl 定义了

pluggy 是怎么把 Plugin 的具体实现和 hook 做关联的 ?

首先我们先看一下 pm.hook, 这个 hook 是个什么东西，我们在 PluginManager 的构造函数看到了hook的定义

1
self.hook = _HookRelay()

看了一下_HookReplay 的代码，只有一个类的定义，看起来就是一个桩 ，这里可以猜测一下，当我们真实调用pm.hook.calculate(a=1,b=2) 的时候，实际的调用可能是这样的

1
getattr(pm.hook, "calculate")(a=1, b=2)

为了印证我的猜测，我在 PluginManager 的 register方法这找到了这段逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
for name in dir(plugin):
    hookimpl_opts = self.parse_hookimpl_opts(plugin, name)
    if hookimpl_opts is not None:
        normalize_hookimpl_opts(hookimpl_opts)
        method = getattr(plugin, name)
        hookimpl = HookImpl(plugin, plugin_name, method, hookimpl_opts)
        hook = getattr(self.hook, name, None)
        if hook is None:
            hook = _HookCaller(name, self._hookexec)
            setattr(self.hook, name, hook)
        elif hook.has_spec():
            self._verify_hook(hook, hookimpl)
            hook._maybe_apply_history(hookimpl)
        hook._add_hookimpl(hookimpl)
        hookcallers.append(hook)
  • 遍历 plugin 对象的所有属性（method）, 通过 self.parse_hookimpl_opts这个方法找到含有特殊 attribute 的方法，就是被 impl　装饰器装饰过的方法
  • 然后再把一个_HookCaller的对象添加到hook对象中, setattr(self.hook, name, hook), 所以看起来真实的调用是 _HookCaller(a=1, b=2),　具体我们下次再分析
  • 如何实现1:N的调用关系呢，实际上一个hook 可以添加多个 hookimpl, 看这行代码就清楚了，它是在找到了 hook 绑定后调用的 hook._add_hookimpl(hookimpl)

实际调用的时候，pm.hook 是个什么, 如何实现最终的 plugin 调用的

前面说了实际上 pm.hook.calculate获取的是一个 _HookCaller对象，所以真实的是调用了它的__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)

核心代码就是最后一行, self._hookexec ，把具体的kwargs传入，来调用实际的 plugin 实现，具体的这个 self._hookexec方法，是在构造_HookCaller的时候作为一个参数传入的，它的定义看起来也只是一个封装

1
2
3
4
def _hookexec(self, hook, methods, kwargs):
    # called from all hookcaller instances.
    # enable_tracing will set its own wrapping function at self._inner_hookexec
    return self._inner_hookexec(hook, methods, kwargs)

真正的逻辑是 PluginManger构造时候封装的 _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,
        )

总结

到这里我们把pluggy主要的设计思路都介绍完了，后面会对于几个核心类的实现做进一步的分析，欢迎有兴趣的同学和我留言