【pytest框架源码分析一】pluggy源码分析之hook常用方法

简单看一下pytest的源码，其实很多地方是依赖pluggy来实现的。这里我们先看一下pluggy的源码。
pluggy的目录结构如下：

这里主要介绍下_callers.py, _hooks.py, _manager.py，其中_callers.py主要是提供具体调用的功能，_hooks.py提供hook的对象类，_manager.py则是负责控制流程的功能。
这里我们从_hooks.py开始：

首先HookspecOpts，HookimplOpts这两个类是定义了hook的specification和implementation的一些参数。具体参数意义我们后面再看。
接下来是HookspecMarker和HookimplMarker类，看注释可知，这两个类都是装饰器，HookspecMarker具体代码如下：

@final
class HookspecMarker:"""Decorator for marking functions as hook specifications.Instantiate it with a project_name to get a decorator.Calling :meth:`PluginManager.add_hookspecs` later will discover all markedfunctions if the :class:`PluginManager` uses the same project name."""__slots__ = ("project_name",)def __init__(self, project_name: str) -> None:self.project_name: Final = project_name@overloaddef __call__(self,function: _F,firstresult: bool = False,historic: bool = False,warn_on_impl: Warning | None = None,warn_on_impl_args: Mapping[str, Warning] | None = None,) -> _F: ...@overload  # noqa: F811def __call__(  # noqa: F811self,function: None = ...,firstresult: bool = ...,historic: bool = ...,warn_on_impl: Warning | None = ...,warn_on_impl_args: Mapping[str, Warning] | None = ...,) -> Callable[[_F], _F]: ...def __call__(  # noqa: F811self,function: _F | None = None,firstresult: bool = False,historic: bool = False,warn_on_impl: Warning | None = None,warn_on_impl_args: Mapping[str, Warning] | None = None,) -> _F | Callable[[_F], _F]:"""If passed a function, directly sets attributes on the functionwhich will make it discoverable to :meth:`PluginManager.add_hookspecs`.If passed no function, returns a decorator which can be applied to afunction later using the attributes supplied.:param firstresult:If ``True``, the 1:N hook call (N being the number of registeredhook implementation functions) will stop at I<=N when the I'thfunction returns a non-``None`` result. See :ref:`firstresult`.:param historic:If ``True``, every call to the hook will be memorized and replayedon plugins registered after the call was made. See :ref:`historic`.:param warn_on_impl:If given, every implementation of this hook will trigger the givenwarning. See :ref:`warn_on_impl`.:param warn_on_impl_args:If given, every implementation of this hook which requests one ofthe arguments in the dict will trigger the corresponding warning.See :ref:`warn_on_impl`... versionadded:: 1.5"""def setattr_hookspec_opts(func: _F) -> _F:if historic and firstresult:raise ValueError("cannot have a historic firstresult hook")opts: HookspecOpts = {"firstresult": firstresult,"historic": historic,"warn_on_impl": warn_on_impl,"warn_on_impl_args": warn_on_impl_args,}setattr(func, self.project_name + "_spec", opts)return funcif function is not None:return setattr_hookspec_opts(function)else:return setattr_hookspec_opts

这是一个装饰器类，其中主要的实现就是__call__下方的setattr_hookspec_opts方法，除去一个if判断，主要的操作是 setattr(func, self.project_name + “_spec”, opts)，即给我们的方法加一个属性，属性名为self.project_name + “_spec”，value为上面定义的opts，里面更具体的属性则为默认值。这里我们简单调用一个例子，可以看出只要加了装饰器的方法都会给其加上这个属性：

HookimplMarker代码如下：

@final
class HookimplMarker:"""Decorator for marking functions as hook implementations.Instantiate it with a ``project_name`` to get a decorator.Calling :meth:`PluginManager.register` later will discover all markedfunctions if the :class:`PluginManager` uses the same project name."""__slots__ = ("project_name",)def __init__(self, project_name: str) -> None:self.project_name: Final = project_name@overloaddef __call__(self,function: _F,hookwrapper: bool = ...,optionalhook: bool = ...,tryfirst: bool = ...,trylast: bool = ...,specname: str | None = ...,wrapper: bool = ...,) -> _F: ...@overload  # noqa: F811def __call__(  # noqa: F811self,function: None = ...,hookwrapper: bool = ...,optionalhook: bool = ...,tryfirst: bool = ...,trylast: bool = ...,specname: str | None = ...,wrapper: bool = ...,) -> Callable[[_F], _F]: ...def __call__(  # noqa: F811self,function: _F | None = None,hookwrapper: bool = False,optionalhook: bool = False,tryfirst: bool = False,trylast: bool = False,specname: str | None = None,wrapper: bool = False,) -> _F | Callable[[_F], _F]:"""If passed a function, directly sets attributes on the functionwhich will make it discoverable to :meth:`PluginManager.register`.If passed no function, returns a decorator which can be applied to afunction later using the attributes supplied.:param optionalhook:If ``True``, a missing matching hook specification will not resultin an error (by default it is an error if no matching spec isfound). See :ref:`optionalhook`.:param tryfirst:If ``True``, this hook implementation will run as early as possiblein the chain of N hook implementations for a specification. See:ref:`callorder`.:param trylast:If ``True``, this hook implementation will run as late as possiblein the chain of N hook implementations for a specification. See:ref:`callorder`.:param wrapper:If ``True`` ("new-style hook wrapper"), the hook implementationneeds to execute exactly one ``yield``. The code before the``yield`` is run early before any non-hook-wrapper function is run.The code after the ``yield`` is run after all non-hook-wrapperfunctions have run. The ``yield`` receives the result value of theinner calls, or raises the exception of inner calls (includingearlier hook wrapper calls). The return value of the functionbecomes the return value of the hook, and a raised exception becomesthe exception of the hook. See :ref:`hookwrapper`.:param hookwrapper:If ``True`` ("old-style hook wrapper"), the hook implementationneeds to execute exactly one ``yield``. The code before the``yield`` is run early before any non-hook-wrapper function is run.The code after the ``yield`` is run after all non-hook-wrapperfunction have run  The ``yield`` receives a :class:`Result` objectrepresenting the exception or result outcome of the inner calls(including earlier hook wrapper calls). This option is mutuallyexclusive with ``wrapper``. See :ref:`old_style_hookwrapper`.:param specname:If provided, the given name will be used instead of the functionname when matching this hook implementation to a hook specificationduring registration. See :ref:`specname`... versionadded:: 1.2.0The ``wrapper`` parameter."""def setattr_hookimpl_opts(func: _F) -> _F:opts: HookimplOpts = {"wrapper": wrapper,"hookwrapper": hookwrapper,"optionalhook": optionalhook,"tryfirst": tryfirst,"trylast": trylast,"specname": specname,}setattr(func, self.project_name + "_impl", opts)return funcif function is None:return setattr_hookimpl_optselse:return setattr_hookimpl_opts(function)

这个和上面的HookspecMarker的作用类似，是给添加这个装饰器的函数增加几个参数

再下面到normalize_hookimpl_opts方法，这个方法就是把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)

varnames方法如下，这个方法主要就是返回方法的位参和关键字参数：

def varnames(func: object) -> tuple[tuple[str, ...], tuple[str, ...]]:"""Return tuple of positional and keywrord argument names for a function,method, class or callable.In case of a class, its ``__init__`` method is considered.For methods the ``self`` parameter is not included."""if inspect.isclass(func):try:func = func.__init__except AttributeError:return (), ()elif not inspect.isroutine(func):  # callable object?try:func = getattr(func, "__call__", func)except Exception:return (), ()try:# func MUST be a function or method here or we won't parse any args.sig = inspect.signature(func.__func__ if inspect.ismethod(func) else func  # type:ignore[arg-type])except TypeError:return (), ()_valid_param_kinds = (inspect.Parameter.POSITIONAL_ONLY,inspect.Parameter.POSITIONAL_OR_KEYWORD,)_valid_params = {name: paramfor name, param in sig.parameters.items()if param.kind in _valid_param_kinds}args = tuple(_valid_params)defaults = (tuple(param.defaultfor param in _valid_params.values()if param.default is not param.empty)or None)if defaults:index = -len(defaults)args, kwargs = args[:index], tuple(args[index:])else:kwargs = ()# strip any implicit instance arg# pypy3 uses "obj" instead of "self" for default dunder methodsif not _PYPY:implicit_names: tuple[str, ...] = ("self",)else:implicit_names = ("self", "obj")if args:qualname: str = getattr(func, "__qualname__", "")if inspect.ismethod(func) or ("." in qualname and args[0] in implicit_names):args = args[1:]return args, kwargs

HookRelay定义如下，目前是个空类，后面使用的时候会进行处理。

@final
class HookRelay:"""Hook holder object for performing 1:N hook calls where N is the numberof registered plugins."""__slots__ = ("__dict__",)def __init__(self) -> None:""":meta private:"""if TYPE_CHECKING:def __getattr__(self, name: str) -> HookCaller: ...

该类中HookImpl和HookSpec也属于定义类，无实际操作代码

@final
class HookImpl:"""A hook implementation in a :class:`HookCaller`."""__slots__ = ("function","argnames","kwargnames","plugin","opts","plugin_name","wrapper","hookwrapper","optionalhook","tryfirst","trylast",)def __init__(self,plugin: _Plugin,plugin_name: str,function: _HookImplFunction[object],hook_impl_opts: HookimplOpts,) -> None:""":meta private:"""#: The hook implementation function.self.function: Final = functionargnames, kwargnames = varnames(self.function)#: The positional parameter names of ``function```.self.argnames: Final = argnames#: The keyword parameter names of ``function```.self.kwargnames: Final = kwargnames#: The plugin which defined this hook implementation.self.plugin: Final = plugin#: The :class:`HookimplOpts` used to configure this hook implementation.self.opts: Final = hook_impl_opts#: The name of the plugin which defined this hook implementation.self.plugin_name: Final = plugin_name#: Whether the hook implementation is a :ref:`wrapper <hookwrapper>`.self.wrapper: Final = hook_impl_opts["wrapper"]#: Whether the hook implementation is an :ref:`old-style wrapper#: <old_style_hookwrappers>`.self.hookwrapper: Final = hook_impl_opts["hookwrapper"]#: Whether validation against a hook specification is :ref:`optional#: <optionalhook>`.self.optionalhook: Final = hook_impl_opts["optionalhook"]#: Whether to try to order this hook implementation :ref:`first#: <callorder>`.self.tryfirst: Final = hook_impl_opts["tryfirst"]#: Whether to try to order this hook implementation :ref:`last#: <callorder>`.self.trylast: Final = hook_impl_opts["trylast"]def __repr__(self) -> str:return f"<HookImpl plugin_name={self.plugin_name!r}, plugin={self.plugin!r}>"@final
class HookSpec:__slots__ = ("namespace","function","name","argnames","kwargnames","opts","warn_on_impl","warn_on_impl_args",)def __init__(self, namespace: _Namespace, name: str, opts: HookspecOpts) -> None:self.namespace = namespaceself.function: Callable[..., object] = getattr(namespace, name)self.name = nameself.argnames, self.kwargnames = varnames(self.function)self.opts = optsself.warn_on_impl = opts.get("warn_on_impl")self.warn_on_impl_args = opts.get("warn_on_impl_args")

接下来是HookCaller，这个简单过下代码，具体可结合后面的_manager一起看：

class HookCaller:"""A caller of all registered implementations of a hook specification."""__slots__ = ("name","spec","_hookexec","_hookimpls","_call_history",)def __init__(self,name: str,hook_execute: _HookExec,specmodule_or_class: _Namespace | None = None,spec_opts: HookspecOpts | None = None,) -> None:""":meta private:"""#: Name of the hook getting called.self.name: Final = nameself._hookexec: Final = hook_execute# The hookimpls list. The caller iterates it *in reverse*. Format:# 1. trylast nonwrappers# 2. nonwrappers# 3. tryfirst nonwrappers# 4. trylast wrappers# 5. wrappers# 6. tryfirst wrappersself._hookimpls: Final[list[HookImpl]] = []self._call_history: _CallHistory | None = None# TODO: Document, or make private.self.spec: HookSpec | None = Noneif specmodule_or_class is not None:assert spec_opts is not Noneself.set_specification(specmodule_or_class, spec_opts)# TODO: Document, or make private.def has_spec(self) -> bool:return self.spec is not None# TODO: Document, or make private.def set_specification(self,specmodule_or_class: _Namespace,spec_opts: HookspecOpts,) -> None:  # 设置specif 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 = []def is_historic(self) -> bool:"""Whether this caller is :ref:`historic <historic>`."""return self._call_history is not Nonedef _remove_plugin(self, plugin: _Plugin) -> None:for i, method in enumerate(self._hookimpls):if method.plugin == plugin:del self._hookimpls[i]returnraise ValueError(f"plugin {plugin!r} not found")def get_hookimpls(self) -> list[HookImpl]:"""Get all registered hook implementations for this hook."""return self._hookimpls.copy()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)def __repr__(self) -> str:return f"<HookCaller {self.name!r}>"def _verify_all_args_are_provided(self, kwargs: Mapping[str, object]) -> None:# This is written to avoid expensive operations when not needed.if self.spec:for argname in self.spec.argnames:if argname not in kwargs:notincall = ", ".join(repr(argname)for argname in self.spec.argnames# Avoid self.spec.argnames - kwargs.keys() - doesn't preserve order.if argname not in kwargs.keys())warnings.warn("Argument(s) {} which are declared in the hookspec ""cannot be found in this hook call".format(notincall),stacklevel=2,)breakdef __call__(self, **kwargs: object) -> Any:"""Call the hook.Only accepts keyword arguments, which should match the hookspecification.Returns the result(s) of calling all registered plugins, see:ref:`calling`."""assert (not self.is_historic()), "Cannot directly call a historic hook - use call_historic instead."self._verify_all_args_are_provided(kwargs)firstresult = self.spec.opts.get("firstresult", False) if self.spec else False# Copy because plugins may register other plugins during iteration (#438).return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)def call_historic(self,result_callback: Callable[[Any], None] | None = None,kwargs: Mapping[str, object] | None = None,) -> None:"""Call the hook with given ``kwargs`` for all registered plugins andfor all plugins which will be registered afterwards, see:ref:`historic`.:param result_callback:If provided, will be called for each non-``None`` result obtainedfrom a hook implementation."""assert self._call_history is not Nonekwargs = kwargs or {}self._verify_all_args_are_provided(kwargs)self._call_history.append((kwargs, result_callback))# Historizing hooks don't return results.# Remember firstresult isn't compatible with historic.# Copy because plugins may register other plugins during iteration (#438).res = self._hookexec(self.name, self._hookimpls.copy(), kwargs, False)if result_callback is None:returnif isinstance(res, list):for x in res:result_callback(x)def call_extra(self, methods: Sequence[Callable[..., object]], kwargs: Mapping[str, object]) -> Any:"""Call the hook with some additional temporarily participatingmethods using the specified ``kwargs`` as call parameters, see:ref:`call_extra`."""assert (not self.is_historic()), "Cannot directly call a historic hook - use call_historic instead."self._verify_all_args_are_provided(kwargs)opts: HookimplOpts = {"wrapper": False,"hookwrapper": False,"optionalhook": False,"trylast": False,"tryfirst": False,"specname": None,}hookimpls = self._hookimpls.copy()for method in methods:hookimpl = HookImpl(None, "<temp>", method, opts)# Find last non-tryfirst nonwrapper method.i = len(hookimpls) - 1while i >= 0 and (# Skip wrappers.(hookimpls[i].hookwrapper or hookimpls[i].wrapper)# Skip tryfirst nonwrappers.or hookimpls[i].tryfirst):i -= 1hookimpls.insert(i + 1, hookimpl)firstresult = self.spec.opts.get("firstresult", False) if self.spec else Falsereturn self._hookexec(self.name, hookimpls, kwargs, firstresult)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])

其中_add_hookimpl方法是在添加hook的实现方法。这里添加时先判断了下hookwrapper，hookwrapper为True的放在后面，否则放在前面；tryfirst的放在后面，trylast的放在前面，否则放在最前面的tryfirst的前一个。注意这里是放在后面的先执行，放在前面的后执行，先进后出。所以这边tryfirst和trylast只是try第一个或最后一个执行，并不保证最先或最后执行。
_verify_all_args_are_provided方法判断了下在self.spec.argnames中但是不在入参kwargs中的参数，warn了下，未做其它操作。
__call__方法中校验了下是不是historic_hook，然后获取了下firstresult，最后调用了_hookexec方法。
这个类中还有不少参数没有指明是什么意思，后面我们结合manager一起看下。