前言

最近在使用pytest做自动化测试，顺便学习pytest的源码，主要想看看框架性的项目是怎么开发的，提高一下姿势水平。Pluggy是pytest里使用的一个插件框架，后来作者也单独拿出来作为一个插件项目，这篇就先来对pluggy进行学习。

主要内容

• pluggy解决了什么问题？
• pluggy是怎么解决这些问题的？

pluggy解决了什么问题

pluggy项目地址 https://github.com/pytest-dev/pluggy
官方文档https://pluggy.readthedocs.io/en/latest/

pluggy是pytest、tox、devpi的核心框架。

它允许用户通过安装“插件”来扩展或修改“主机程序”的行为。插件代码将作为正常程序执行的一部分运行，改变或增强它的某些方面。
本质上，“pluggy”使函数hooking，可以构建“可插拔”系统。

钩子编程（hooking），也称作“挂钩”，是计算机程序设计术语，指通过拦截软件模块间的函数调用、消息传递、事件传递来修改或扩展操作系统、应用程序或其他软件组件的行为的各种技术。处理被拦截的函数调用、事件、消息的代码，被称为钩子（hook）。

相比另外两种改变其他程序或lib行为的方式方法重载和猴子补丁，pluggy这种方式可以解决某部分代码需要被项目里多个地方改变行为的问题。主体host和插件plugin是非常松耦合的关系。

猴子补丁
一种运行时替换功能的方式。
https://www.jianshu.com/p/f1c1eb495f47

总结一下，pluggy要解决的问题就是，提供一种构建插件化代码的方式。这样的代码可以很方便的使用插件进行扩展。

pluggy如何解决这个问题

在pluggy构建的项目中有这样一些角色：

• host ： 以一种类似定义接口方式—hookspecs来定义hook，使用caller在合适的时机调用hook实现,并收集结果
• plugin: 以一种类似接口实现的方式——hookimpl实现hook
• pluggy：负责连接host和plugin
• user: 根据需要安装插件，使用不同功能

pluggy是如何连接host和plugin的呢？

在pluggy中有一个pluginmanager对象，通过这个pm对象来进行hookspecs和plugin的管理。

看个例子：

import pluggy

hookspec = pluggy.HookspecMarker("myproject")
hookimpl = pluggy.HookimplMarker("myproject")


class MySpec:
    """A hook specification namespace."""

    @hookspec
    def myhook(self, arg1, arg2):
        """My special little hook that you can customize."""


class Plugin_1:
    """A hook implementation namespace."""

    @hookimpl
    def myhook(self, arg1, arg2):
        print("inside Plugin_1.myhook()")
        return arg1 + arg2


class Plugin_2:
    """A 2nd hook implementation namespace."""

    @hookimpl
    def myhook(self, arg1, arg2):
        print("inside Plugin_2.myhook()")
        return arg1 - arg2


# create a manager and add the spec
pm = pluggy.PluginManager("myproject")
pm.add_hookspecs(MySpec)
# register plugins
pm.register(Plugin_1())
pm.register(Plugin_2())
# call our `myhook` hook
results = pm.hook.myhook(arg1=1, arg2=2)
print(results)

可以看到，首先指定一个项目名实例化了HookspecMarker、HookimplMarker、HookManager，之后通过Pm的add_hookspecs方法，注册hookspecs，通过register方法注册plugin。这两个方法都是直接以类为参数，同时类中的hook相关方法都加了之前实例化得到的装饰器hookspec或者hookimpl。

所以这一套逻辑的核心就是HookspecMarker、HookimplMarker、add_hookspecs、register这几个类和方法。我们来分别看一下。

HookspecMarker和HookimplMarker

class HookspecMarker(object):
    """ Decorator helper class for marking functions as hook specifications.

    You can instantiate it with a project_name to get a decorator.
    Calling :py:meth:`.PluginManager.add_hookspecs` later will discover all marked functions
    if the :py:class:`.PluginManager` uses the same project_name.
    """

    def __init__(self, project_name):
        self.project_name = project_name

    def __call__(
        self, function=None, firstresult=False, historic=False, warn_on_impl=None
    ):
        """ if passed a function, directly sets attributes on the function
        which will make it discoverable to :py:meth:`.PluginManager.add_hookspecs`.
        If passed no function, returns a decorator which can be applied to a function
        later using the attributes supplied.

        If ``firstresult`` is ``True`` the 1:N hook call (N being the number of registered
        hook implementation functions) will stop at I<=N when the I'th function
        returns a non-``None`` result.

        If ``historic`` is ``True`` calls to a hook will be memorized and replayed
        on later registered plugins.

        """

        def setattr_hookspec_opts(func):
            if historic and firstresult:
                raise ValueError("cannot have a historic firstresult hook")
            setattr(
                func,
                self.project_name + "_spec",
                dict(
                    firstresult=firstresult,
                    historic=historic,
                    warn_on_impl=warn_on_impl,
                ),
            )
            return func

        if function is not None:
            return setattr_hookspec_opts(function)
        else:
            return setattr_hookspec_opts

HookspecMarker类实现了__call__方法，因此实质上是有一个project_name属性的装饰器。

可以看到核心逻辑就是给被装饰的函数加上了project_name+'_spec'属性标记，这样就可以被add_hookspecs方法识别。

类似的hookimplMarke也是给被装饰的函数加上了project_name+'_impl'属性标记,这样就可以被register方法识别。

add_hookspecs和register

    def add_hookspecs(self, module_or_class):
        """ add new hook specifications defined in the given ``module_or_class``.
        Functions are recognized if they have been decorated accordingly. """
        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 = 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(
                "did not find any %r hooks in %r" % (self.project_name, module_or_class)
            )

add_hookspecs首先扫描给过来的module或者类，通过dir(）方法获取类的所有属性。

    def parse_hookspec_opts(self, module_or_class, name):
        method = getattr(module_or_class, name)
        return getattr(method, self.project_name + "_spec", None)

parse_hookspec_opts方法获取方法属性的 project_name+'_spec'属性，如果hook里没有这个hookcaller，则给hook加上以这个hookspec名字为属性名的hookcaller。如果hook里已经有了就更新hookspec。
这里的self.hook是pm实例化时创建的一个属性，是一个私有类_HookRelay，同时也是个空类，他的作用就是存放已经定义的hookcaller。hookcaller是以hookspec内信息构建的对象，作用是将hookspec和hookimpl关联起来，并且定义了hookimpl的实际调用逻辑。

可以看register里的内容：

    def register(self, plugin, name=None):
        """ Register a plugin and return its canonical name or ``None`` if the name
        is blocked from registering.  Raise a :py:class:`ValueError` if the plugin
        is already registered. """
        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)
            )

        # XXX if an error happens we should make sure no state has been
        # changed at point of return
        self._name2plugin[plugin_name] = plugin

        # register matching hook implementations of the plugin
        self._plugin2hookcallers[plugin] = hookcallers = []
        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

这样Hookspec和hookimpl以名字作为纽带，在hookcaller中实现了关联。

通过inspect，pluggy会获取hookspec、hookimpl的参数进行对比，如果不一致则抛出异常。这样也就实现了接口的参数校验。

if hasattr(inspect, "getfullargspec"):

    def _getargspec(func):
        return inspect.getfullargspec(func)


else:

    def _getargspec(func):
        return inspect.getargspec(func)

 def _verify_hook(self, hook, hookimpl):
        if hook.is_historic() and hookimpl.hookwrapper:
            raise PluginValidationError(
                hookimpl.plugin,
                "Plugin %r\nhook %r\nhistoric incompatible to hookwrapper"
                % (hookimpl.plugin_name, hook.name),
            )
        if hook.spec.warn_on_impl:
            _warn_for_function(hook.spec.warn_on_impl, hookimpl.function)
        # positional arg checking
        notinspec = 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,
                ),
            )

好了这一节了解了pluggy的核心逻辑，下一节再来看看pluggy在pytest里的应用以及一些其他功能细节。