前言

最近在使用pytest做自動化測試，順便學習pytest的源碼，主要想看看框架性的項目是怎么開發(fā)的，提高一下姿勢水平。Pluggy是pytest里使用的一個插件框架，后來作者也單獨拿出來作為一個插件項目，這篇就先來對pluggy進行學習。

主要內(nèi)容

• pluggy解決了什么問題？
• pluggy是怎么解決這些問題的？

pluggy解決了什么問題

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

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

它允許用戶通過安裝“插件”來擴展或修改“主機程序”的行為。插件代碼將作為正常程序執(zhí)行的一部分運行，改變或增強它的某些方面。
本質(zhì)上，“pluggy”使函數(shù)hooking，可以構(gòu)建“可插拔”系統(tǒng)。

鉤子編程（hooking），也稱作“掛鉤”，是計算機程序設計術(shù)語，指通過攔截軟件模塊間的函數(shù)調(diào)用、消息傳遞、事件傳遞來修改或擴展操作系統(tǒng)、應用程序或其他軟件組件的行為的各種技術(shù)。處理被攔截的函數(shù)調(diào)用、事件、消息的代碼，被稱為鉤子（hook）。

相比另外兩種改變其他程序或lib行為的方式方法重載和猴子補丁，pluggy這種方式可以解決某部分代碼需要被項目里多個地方改變行為的問題。主體host和插件plugin是非常松耦合的關系。

猴子補丁
一種運行時替換功能的方式。
http://www.itdecent.cn/p/f1c1eb495f47

總結(jié)一下，pluggy要解決的問題就是，提供一種構(gòu)建插件化代碼的方式。這樣的代碼可以很方便的使用插件進行擴展。

pluggy如何解決這個問題

在pluggy構(gòu)建的項目中有這樣一些角色：

• host ： 以一種類似定義接口方式—hookspecs來定義hook，使用caller在合適的時機調(diào)用hook實現(xiàn),并收集結(jié)果
• plugin: 以一種類似接口實現(xiàn)的方式——hookimpl實現(xiàn)hook
• pluggy：負責連接host和plugin
• user: 根據(jù)需要安裝插件，使用不同功能

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。這兩個方法都是直接以類為參數(shù)，同時類中的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類實現(xiàn)了__call__方法，因此實質(zhì)上是有一個project_name屬性的裝飾器。

可以看到核心邏輯就是給被裝飾的函數(shù)加上了project_name+'_spec'屬性標記，這樣就可以被add_hookspecs方法識別。

類似的hookimplMarke也是給被裝飾的函數(shù)加上了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里已經(jīng)有了就更新hookspec。
這里的self.hook是pm實例化時創(chuàng)建的一個屬性，是一個私有類_HookRelay，同時也是個空類，他的作用就是存放已經(jīng)定義的hookcaller。hookcaller是以hookspec內(nèi)信息構(gòu)建的對象，作用是將hookspec和hookimpl關聯(lián)起來，并且定義了hookimpl的實際調(diào)用邏輯。

可以看register里的內(nèi)容：

    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中實現(xiàn)了關聯(lián)。

通過inspect，pluggy會獲取hookspec、hookimpl的參數(shù)進行對比，如果不一致則拋出異常。這樣也就實現(xiàn)了接口的參數(shù)校驗。

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,
                ),
            )

好了這一節(jié)了解了pluggy的核心邏輯，下一節(jié)再來看看pluggy在pytest里的應用以及一些其他功能細節(jié)。