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

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

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

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._name2pluginself._plugin2hookcaller,前者是用 plugin_name 做key,后者是用 plugin object 做key,不过都可以用来判断是否已经注册过重复的plugin

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

self._plugin2hookcallers[plhaougin] = hookcallers = []

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

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

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

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

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

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 分成了两类, wrappersnonwrappers , 这个区分是依据HookimplMarker的那个hookwrapper属性

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

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

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

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 做一下顺序的调整,这里就不再叙述了

Author: 妞爸爸
Link: http://markshao.github.io/2019/10/03/register-pluggy/
Copyright Notice: All articles in this blog are licensed under CC BY-NC-SA 4.0 unless stating additionally.