Pytest 源码解读 [3] - [pluggy] plugin 注册逻辑分析

pytest 源码解读

今天我们再仔细分析一下 plugin 的注册逻辑，这里面包含了pluggy 框架中的一些核心设计元素。代码在 manager.py中的PluginManager的register方法

首先是先判断一下这个plugin是否已经注册过了，

1
2
3
4
5
6
7
8
9
plugin_name = name or self.get_canonical_name(plugin)

        if plugin_name in self._name2plugin or plugin in self._plugin2hookcallers:
            if self._name2plugin.get(plugin_name, -1) is None:
                return  # blocked plugin, return None to indicate no registration
            raise ValueError(
                "Plugin already registered: %s=%s\n%s"
                % (plugin_name, plugin, self._name2plugin)
            )

用于做 contains 判断的主要是 PluginManager对象中的两个dict， self._name2plugin和self._plugin2hookcaller，前者是用 plugin_name 做key，后者是用 plugin object 做key，不过都可以用来判断是否已经注册过重复的plugin

判定完毕后，确认是一个新的plugin了，那就开始进入添加新的plugin逻辑, 先看一下self._plugin2hookcallers　的数据结构, 一个plugin 对象对应多个 _HookCaller对象

self._plugin2hookcallers[plhaougin] = hookcallers = []

再看下pluggy是如何分析 plugin 对象，并完成 _HookCaller的映射的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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)
return plugin_name

类似add_spechooks的逻辑，先通过dir函数遍历 plugin 的所有元素，主要是 method，通过self.parse_hookimpl_opts方法来找到被特定装饰器修饰过的method，找到以后更新一下hookimpl 的 options，然后通过getattr 获取这个函数对象，然后用HookImpl进行封装，我们看一下HookImpl的实现

1
2
3
4
5
6
7
8
9
10
11
class HookImpl(object):
    def __init__(self, plugin, plugin_name, function, hook_impl_opts):
        self.function = function
        self.argnames, self.kwargnames = varnames(self.function)
        self.plugin = plugin
        self.opts = hook_impl_opts
        self.plugin_name = plugin_name
        self.__dict__.update(hook_impl_opts)

    def __repr__(self):
        return "<HookImpl plugin_name=%r, plugin=%r>" % (self.plugin_name, self.plugin)

没有任何逻辑，纯粹就是一个对象封装，那我们继续分析，后面就是检查一下 pm.hook.xxxx 的这个对象有没有绑定，如果没有，那么就重新绑定一个_HookCaller对象

1
2
3
if hook is None:
                    hook = _HookCaller(name, self._hookexec)
                    setattr(self.hook, name, hook)

_HookCaller是整个pluggy的核心封装，我们看一下它的构造函数

1
2
3
4
5
6
7
8
9
10
11
12
13
class _HookCaller(object):
    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)

_HookCaller 这里的 self._hookexec 表示是一个函数对象，负责实际的plugin 调用，这个下次会分析，主要看到它这里把 plugin 分成了两类, wrappers 和 nonwrappers , 这个区分是依据HookimplMarker的那个hookwrapper属性

刚才说了也存在 hook 没有的情况，如果有的话，那么就再检查一下 plugin 和 hook 是否 valid match，并且看是否要出发 historical 的调用，基于 HookspeckMarker中定义的historical属性

1
2
3
elif hook.has_spec():
                   self._verify_hook(hook, hookimpl)
                   hook._maybe_apply_history(hookimpl)

最后我们把这个 刚才封装好的HookImpl对象添加到 _HookCaller中去，我们看一下 HookCaller的_add_hookimpl的实现

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
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)
    else:
        # find last non-tryfirst method
        i = len(methods) - 1
        while i >= 0 and methods[i].tryfirst:
            i -= 1
        methods.insert(i + 1, hookimpl)

    if "__multicall__" in hookimpl.argnames:
        warnings.warn(
            "Support for __multicall__ is now deprecated and will be"
            "removed in an upcoming release.",
            DeprecationWarning,
        )
        self.multicall = _legacymultical

主要是根据HookimplMarker 做一下顺序的调整，这里就不再叙述了