【pytest框架源码分析二】pluggy源码分析之add_hookspecs和register

这里我们看一下_manager.py里的类和方法，最主要的是PluginManager类，类的初始化函数如下：

class PluginManager:"""Core class which manages registration of plugin objects and 1:N hookcalling.You can register new hooks by calling :meth:`add_hookspecs(module_or_class)<PluginManager.add_hookspecs>`.You can register plugin objects (which contain hook implementations) bycalling :meth:`register(plugin) <PluginManager.register>`.For debugging purposes you can call :meth:`PluginManager.enable_tracing`which will subsequently send debug information to the trace helper.:param project_name:The short project name. Prefer snake case. Make sure it's unique!"""def __init__(self, project_name: str) -> None:#: The project name.self.project_name: Final = project_nameself._name2plugin: Final[dict[str, _Plugin]] = {}self._plugin_distinfo: Final[list[tuple[_Plugin, DistFacade]]] = []#: The "hook relay", used to call a hook on all registered plugins.#: See :ref:`calling`.self.hook: Final = HookRelay()#: The tracing entry point. See :ref:`tracing`.self.trace: Final[_tracing.TagTracerSub] = _tracing.TagTracer().get("pluginmanage")self._inner_hookexec = _multicall

简单介绍下几个参数：
project_name： 项目名，同一个项目使用一个project name即可
_name2plugin：字典类型，用于存放插件名和插件的对应关系
_plugin_distinfo：存放了plugin及其distributions
hook：hook_relay实例，用于调用plugin
trace：与主流程无关，暂时用不上
_inner_hookexec ：调用的_multicall函数，其在_call.py里，被注册的插件执行及结果返回逻辑都在这里面，我们后面具体看下。
接下来我们看下add_hookspecs和register方法，这两个是添加plugin的主要方法。前面我们介绍了hook中的HookspecMarker和HookimplMarker这两个装饰器，其实对hook来说，spec可以看作接口，impl是接口的实现，对应到add_hookspecs是添加hook的接口，register是添加实现的方法。这里我们简单举个例子：

import pluggyspec_test = pluggy.HookspecMarker('test11')
impl_test = pluggy.HookimplMarker('test11')# 定义spec类，相当于接口类，无具体内容
class Spec:@spec_testdef pluggy_test(self, arg1):print(f'this is spec test and arg is {arg1}')pass# 定义impl类，相当于实现类，具体内容在这里定义
class Impl1:@impl_testdef pluggy_test(self, arg1):print(f'this is test1 and arg is {arg1}')return arg1class Impl2:@impl_testdef pluggy_test(self, arg1):print(f'this is test2 and arg is {arg1}')return -1 * arg1pm = pluggy.PluginManager('test11')
pm.add_hookspecs(Spec)
pm.register(Impl1())
pm.register(Impl2())
res = pm.hook.pluggy_test(arg1=1)
print(res)

返回为

这里可以看出初始化pm后，先add_hookspecs，再register具体的实现，调用时，未打印sepc类中的内容，只打印了impl类中的内容，并且后注册的先执行。这就是pluggin的注册执行的大概流程，接下来看下具体的代码。

    def add_hookspecs(self, module_or_class: _Namespace) -> None:"""Add new hook specifications defined in the given ``module_or_class``.Functions are recognized as hook specifications if they have beendecorated with a matching :class:`HookspecMarker`."""names = []for name in dir(module_or_class):spec_opts = self.parse_hookspec_opts(module_or_class, name)if spec_opts is not None:hc: HookCaller | None = getattr(self.hook, name, None)if hc is None:hc = HookCaller(name, self._hookexec, module_or_class, spec_opts)setattr(self.hook, name, hc)else:# Plugins registered this hook without knowing the spec.hc.set_specification(module_or_class, spec_opts)for hookfunction in hc.get_hookimpls():self._verify_hook(hc, hookfunction)names.append(name)if not names:raise ValueError(f"did not find any {self.project_name!r} hooks in {module_or_class!r}")

add_hookspecs方法输入参数为module_or_class，即为module或者class，上面我举的例子是一个class，也可以是一个模块，如pytest就是入参了一个_pytest.hookspec，hookspec为pytest的一个py文件，用于添加spec。

接下来是具体的方法，定义了一个names的list，然后dir(module_or_class) 查看我们入参模块或者类的所有属性和方法（dir()内置函数，可查看所有属性和方法，注意这里是属性名和方法名，是str类型的，不是直接的方法），然后进入parse_hookspec_opts(module_or_class, name)方法

    def parse_hookspec_opts(self, module_or_class: _Namespace, name: str) -> HookspecOpts | None:"""Try to obtain a hook specification from an item with the given namein the given module or class which is being searched for hook specs.:returns:The parsed hookspec options for defining a hook, or None to skip thegiven item.This method can be overridden by ``PluginManager`` subclasses tocustomize how hook specifications are picked up. By default, returns theoptions for items decorated with :class:`HookspecMarker`."""method = getattr(module_or_class, name)opts: HookspecOpts | None = getattr(method, self.project_name + "_spec", None)return opts

这个方法首先根据name名获取了对应的方法，然后根据获取了方法的project_name + "_spec"的属性，HookspecOpts在_hook中有定义，这里我们前面讲过在HookspecMarker的__call__方法中有设置这个属性，所有加了HookspecMarker注解的方法都自动加了这个属性。

有这个属性的方法会返回对应的HookspecOpts，没的则返回None。add_hookspecs方法接着判断spec_opts如果不是None，去hook中根据名称获取对应的HookCaller，如果没有就返回None，一般我们添加时都是没有的，会返回None，这是再判断hc也是None，我们就给hook里添加一个。

hc = HookCaller(name, self._hookexec, module_or_class, spec_opts)
setattr(self.hook, name, hc)

如果hc不是None，这里其实就是plugin已经注册了，但是spec不是我们当前添加的这个spec。则进一步处理，进入set_specification(module_or_class, spec_opts)方法。

    def set_specification(self,specmodule_or_class: _Namespace,spec_opts: HookspecOpts,) -> None:if self.spec is not None:raise ValueError(f"Hook {self.spec.name!r} is already registered "f"within namespace {self.spec.namespace}")self.spec = HookSpec(specmodule_or_class, self.name, spec_opts)if spec_opts.get("historic"):self._call_history = []

如果spec不是None，则抛错（刚初始化的HookCaller其spec都是None，一般设置过才不是None），如果是None，则设置对应的spec。这边还涉及到historic参数，我们后面再捋捋。回到add_hookspecs方法，下一步就是校验_verify_hook，这个方法register时再一起看。最后把所有符合要求的name放到names里（即加了spec装饰器的方法），最后判断下names是不是为空，为空则抛错。
接下来看下register方法

    def register(self, plugin: _Plugin, name: str | None = None) -> str | None:"""Register a plugin and return its name.:param name:The name under which to register the plugin. If not specified, aname is generated using :func:`get_canonical_name`.:returns:The plugin name. If the name is blocked from registering, returns``None``.If the plugin is already registered, raises a :exc:`ValueError`."""plugin_name = name or self.get_canonical_name(plugin)if plugin_name in self._name2plugin:if self._name2plugin.get(plugin_name, -1) is None:return None  # blocked plugin, return None to indicate no registrationraise ValueError("Plugin name already registered: %s=%s\n%s"% (plugin_name, plugin, self._name2plugin))if plugin in self._name2plugin.values():raise ValueError("Plugin already registered under a different name: %s=%s\n%s"% (plugin_name, plugin, self._name2plugin))# XXX if an error happens we should make sure no state has been# changed at point of returnself._name2plugin[plugin_name] = plugin# register matching hook implementations of the pluginfor name in dir(plugin):hookimpl_opts = self.parse_hookimpl_opts(plugin, name)if hookimpl_opts is not None:normalize_hookimpl_opts(hookimpl_opts)method: _HookImplFunction[object] = getattr(plugin, name)hookimpl = HookImpl(plugin, plugin_name, method, hookimpl_opts)name = hookimpl_opts.get("specname") or namehook: HookCaller | None = 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)return plugin_name

方法入参为一个plugin对象和name，如果有name传入，则plugin_name取name，否则根据plugin取name

    def get_canonical_name(self, plugin: _Plugin) -> str:"""Return a canonical name for a plugin object.Note that a plugin may be registered under a different namespecified by the caller of :meth:`register(plugin, name) <register>`.To obtain the name of a registered plugin use :meth:`get_name(plugin)<get_name>` instead."""name: str | None = getattr(plugin, "__name__", None)return name or str(id(plugin))

get_canonical_name中直接取了plugin的__name__,如果这个__name__不存在则取其对象的id（id是python中判断身份的唯一标识，任何对象都会有自己的id，判断两个对象是否为同一个就是通过id这个内置函数判断的)。接下来判断plugin_name是否在_name2plugin中，如果在且其值为None时，说明被block了，直接返回，不为None时则抛错：插件已注册。
第二个判断plugin是否在_name2plugin中，由于上面已经判断过其name是否在_name2plugin这个dict中，走到这里说明其name这个key不在，如果plugin这个value在，说明plugin使用其他name注册了，抛出报错。
接下来self._name2plugin[plugin_name] = plugin往_name2plugin存ket-value。
最后遍历plugin中的所有方法，根据方法名在 self.parse_hookimpl_opts(plugin, name)获取对应的opts参数，

    def parse_hookimpl_opts(self, plugin: _Plugin, name: str) -> HookimplOpts | None:"""Try to obtain a hook implementation from an item with the given namein the given plugin which is being searched for hook impls.:returns:The parsed hookimpl options, or None to skip the given item.This method can be overridden by ``PluginManager`` subclasses tocustomize how hook implementation are picked up. By default, returns theoptions for items decorated with :class:`HookimplMarker`."""method: object = getattr(plugin, name)if not inspect.isroutine(method):return Nonetry:res: HookimplOpts | None = getattr(method, self.project_name + "_impl", None)except Exception:res = {}  # type: ignore[assignment]if res is not None and not isinstance(res, dict):# false positiveres = None  # type:ignore[unreachable]return res

这个方法和上面parse_hookspec_opts方法类似，也是先获取方法，然后获取方法中的self.project_name + "_impl"属性，如果没有则返回None，加了HookimplMarker装饰器的都会有这个属性。所以hookspec_opts不是None时，进入下一步，normalize_hookimpl_opts设置hookspec_opts的默认值。

def normalize_hookimpl_opts(opts: HookimplOpts) -> None:opts.setdefault("tryfirst", False)opts.setdefault("trylast", False)opts.setdefault("wrapper", False)opts.setdefault("hookwrapper", False)opts.setdefault("optionalhook", False)opts.setdefault("specname", None)

接下来去plugin根据name获取对应的方法，找到方法后，实例化HookImpl对象，name取hookimpl_opts.get(“specname”) or name，使用装饰器时，我们可以添加这个specname参数（添加则取它，不添则不取），HookCaller则根据name到self.hook中取。
如果hook为空（即未根据name获取到HookCaller），则重新实例化一个HookCaller对象，并且添加到self.hook中去，如果hook存在，则进入_verify_hook方法

    def _verify_hook(self, hook: HookCaller, hookimpl: HookImpl) -> None:if hook.is_historic() and (hookimpl.hookwrapper or hookimpl.wrapper):raise PluginValidationError(hookimpl.plugin,"Plugin %r\nhook %r\nhistoric incompatible with yield/wrapper/hookwrapper"% (hookimpl.plugin_name, hook.name),)assert hook.spec is not Noneif hook.spec.warn_on_impl:_warn_for_function(hook.spec.warn_on_impl, hookimpl.function)# positional arg checkingnotinspec = set(hookimpl.argnames) - set(hook.spec.argnames)if notinspec:raise PluginValidationError(hookimpl.plugin,"Plugin %r for hook %r\nhookimpl definition: %s\n""Argument(s) %s are declared in the hookimpl but ""can not be found in the hookspec"% (hookimpl.plugin_name,hook.name,_formatdef(hookimpl.function),notinspec,),)if hook.spec.warn_on_impl_args:for hookimpl_argname in hookimpl.argnames:argname_warning = hook.spec.warn_on_impl_args.get(hookimpl_argname)if argname_warning is not None:_warn_for_function(argname_warning, hookimpl.function)if (hookimpl.wrapper or hookimpl.hookwrapper) and not inspect.isgeneratorfunction(hookimpl.function):raise PluginValidationError(hookimpl.plugin,"Plugin %r for hook %r\nhookimpl definition: %s\n""Declared as wrapper=True or hookwrapper=True ""but function is not a generator function"% (hookimpl.plugin_name, hook.name, _formatdef(hookimpl.function)),)if hookimpl.wrapper and hookimpl.hookwrapper:raise PluginValidationError(hookimpl.plugin,"Plugin %r for hook %r\nhookimpl definition: %s\n""The wrapper=True and hookwrapper=True options are mutually exclusive"% (hookimpl.plugin_name, hook.name, _formatdef(hookimpl.function)),)

这个方法有如下几个判断：
1.首先判断了hook的is_historic和wrapper参数，这些参数不能同时为true。
2.然后判断了下hook.spec不能为None，这个在register方法中已经判断过，可能是担心其他地方又使用到，重新判断了下。
3.warn_on_impl如果为true，则warn一下。
4.接下来判断了下hookimpl中的参数是否比spec中的多，如果多会报错。
5.再下面还是关于warn的，这个和上面那个warn类似，只是这个是针对特定参数的。
6.判断hookwrapper参数为true的时候，其实现的方法是不是生成器方法，即方法中是不是yield返回的
7.最后判断了下hookwrapper和wrapper是不是同时为true，如果是，则报错。hookwrapper和wrapper是新旧版本的不同名称，只要用一个参数即可
然后进入_maybe_apply_history方法

    def _maybe_apply_history(self, method: HookImpl) -> None:"""Apply call history to a new hookimpl if it is marked as historic."""if self.is_historic():assert self._call_history is not Nonefor kwargs, result_callback in self._call_history:res = self._hookexec(self.name, [method], kwargs, False)if res and result_callback is not None:# XXX: remember firstresult isn't compat with historicassert isinstance(res, list)result_callback(res[0])

开始判断了下is_historic是否为true，是的话确认下_call_history 是否为None（一般is_historic为true时，_call_history 会被初始为[]），然后遍历了下_call_history，在res and result_callback都不是None的情况下，result_callback(res[0])。historic这部分后面一起串起来看。
这时主方法到了_add_hookimpl这里。

    def _add_hookimpl(self, hookimpl: HookImpl) -> None:"""Add an implementation to the callback chain."""for i, method in enumerate(self._hookimpls):if method.hookwrapper or method.wrapper:splitpoint = ibreakelse:splitpoint = len(self._hookimpls)if hookimpl.hookwrapper or hookimpl.wrapper:start, end = splitpoint, len(self._hookimpls)else:start, end = 0, splitpointif hookimpl.trylast:self._hookimpls.insert(start, hookimpl)elif hookimpl.tryfirst:self._hookimpls.insert(end, hookimpl)else:# find last non-tryfirst methodi = end - 1while i >= start and self._hookimpls[i].tryfirst:i -= 1self._hookimpls.insert(i + 1, hookimpl)

1.这个方法开始先是查询有多少wrapper为true的impl方法，在wrapper和非wrapper方法数组中找到分界点。这个数组中wrapper和非wrapper方法是分在前后两部分的，互不交叉。
2.然后判断新增的impl的wrapper是否为true，如果是true，则插入到分界点到数组末尾这段，如果不是true，则插到0到分界点这个位置。
3.在判断有无设置trylast或者tryfirst，如果有trylast则放到数组开头，如果是tryfirst，则放到数组末尾；如果都没有，则放到当前段tryfirst那一段的前一个。（越靠前的后执行，越靠后的先执行）
到这里就register完成了。