3

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

 2 years ago
source link: http://markshao.github.io/2019/10/03/register-pluggy/
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] plugin 注册逻辑分析

Toggle site
Catalog
You've read100%
Pytest 源码解读 [3] - [pluggy] plugin 注册逻辑分析

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

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

python
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的映射的

python
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的实现

python
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对象

python
if hook is None:
                    hook = _HookCaller(name, self._hookexec)
                    setattr(self.hook, name, hook)

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

python
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属性

python
elif hook.has_spec():
                   self._verify_hook(hook, hookimpl)
                   hook._maybe_apply_history(hookimpl)

最后我们把这个 刚才封装好的HookImpl对象添加到 _HookCaller中去，我们看一下 HookCaller的_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)
    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 做一下顺序的调整，这里就不再叙述了


report

Recommend

About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK