preface
This article is not suitable for mobile reading, please quit
The company architecture group developed a set of testing framework SSTEST based on PyTest. The purpose is to let the business group (that is, the group I work in) write unit tests better, so as to improve the code quality. The purpose of unit tests is to regression verification, so as to avoid the new submitted code affecting the old functions in the project.
I was the first student in my group to access SSTEST. Therefore, I became interested in the source code of PyTest. In the process of reading the source code of PyTest, I found that pluggy plug-in system is actually the core of PyTest. It can be said that PyTest is just a project built by multiple plug-ins using Pluggy, so I analyzed Pluggy first.
As always, here are a few things you want to know before you start your analysis:
- 1. How to use pluggy?
- 2. How can plug-in code be flexible and pluggable?
- 3. How does an external system invoke plug-in logic?
As the analysis progresses, new questions are thrown up to help clarify our purpose and avoid getting lost in the source code.
The whole point
The pluggy system is different from the Python system I studied earlier. Pluggy can’t be plugged in dynamically, meaning you can’t add new functionality while the program is running.
Pluggy has three main concepts:
- PluginManager: Used to manage the plug-in specification and the plug-in itself
- 2.HookspecMarker: Define the plug-in call specification, each specification can correspond to 1~N plug-ins, each plug-in meets the specification, otherwise it cannot be successfully called externally
- 3.HookimplMarker: Define the plug-in. The specific implementation of the plug-in logic is in the method of this type of decoration
To use it briefly, the code is as follows.
import pluggy
Create a plug-in specification class decorator
hookspac = pluggy.HookspecMarker('example')
Create a plug-in class decorator
hookimpl = pluggy.HookimplMarker('example')
class MySpec(object):
Create a plug-in specification
@hookspac
def myhook(self, a, b):
pass
class Plugin_1(object):
# define plug-in
@hookimpl
def myhook(self, a, b):
return a + b
class Plugin_2(object):
@hookimpl
def myhook(self, a, b):
return a - b
# create manger and add hook specification
pm = pluggy.PluginManager('example')
pm.add_hookspecs(MySpec)
# register plugin
pm.register(Plugin_1())
pm.register(Plugin_2())
Call the myhook method in the plugin
results = pm.hook.myhook(a=10, b=20)
print(results)
Copy the codeThe entire code simply creates the corresponding class decorators to decorate the methods in the class, from which the plug-in specification and the plug-in itself are built.
First, instantiate the PluginManager class, passing in a globally unique project name, and use the same project name for instantiation of the HookspecMarker and HookimplMarker classes.
Once the plug-in manager is created, you can add the plug-in specification through the add_hookspecs method and the plug-in itself through the register method.
Once you have added the plug-in invocation specification and the plug-in itself, you can call the plug-in directly through the hook attribute of the plug-in manager.
Here is the answer to the question “1,2,3”.
The process used by Pluggy can be broken down into four steps:
- 1. Define the plug-in invocation specification through the HookspecMarker class decorator
- 2. Define the plug-in logic through the HookimplMarker class decorator
- 3. Create the PluginManager and bind the plug-in invocation specification to the plug-in itself
- 4. Invoke the plug-in
Pluginmanager. add_hookspecs and pluginManager. register can be combined with class decorators to implement pluginmanager. pluginManager. register. The corresponding add_hookspecs and register methods use this attribute information to determine whether it is a plug-in specification or the plug-in itself.
To use a plug-in in an external system, simply call the pm.hook. Any_hook_function method. Any registered plug-in can be easily called.
But this raises a new question:
- 4. How does a class decorator set a method in a class to be a plug-in?
- 5. How does Pluggy relate the plug-in specification to the plug-in itself?
- 6. How exactly is the logic in the plug-in invoked?
These three questions focus on implementation details, which are further examined in the following steps.
Hookspac versus Hookimpl decorator
In the code, the Hookspac class decorator is used to define the plug-in invocation specification, and the Hookimpl class decorator is used to define the plug-in itself, which is essentially “adding new attributes to the decorated method.” Because the logic is similar, we will only analyze the hookspac class decorator code, which looks like this:
class HookspecMarker(object):
def __init__(self, project_name):
self.project_name = project_name
def __call__( self, function=None, firstresult=False, historic=False, warn_on_impl=None ):
def setattr_hookspec_opts(func):
if historic and firstresult:
raise ValueError("cannot have a historic firstresult hook")
Add new attributes to the decorated method
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
Copy the codeThe class decorator overrides the class’s __call__ method. The core logic of the __call__ method above is to use the setattr method to add a new property to the decorated func method. The property name is the current project name suffixed with _spec, and the value of the property is a dictionary object.
The information added by the Hookspac class decorator is used when the pluginManager.add_hookspecs method is called
The HookimplMarker class is similar, except that the attributes are added differently. The core code is as follows.
setattr(
func,
self.project_name + "_impl",
dict(
hookwrapper=hookwrapper,
optionalhook=optionalhook,
tryfirst=tryfirst,
trylast=trylast,
specname=specname,
),
)
Copy the codeSo how does a class decorator set a method in a class to be a plug-in? The setattr method is used to set new properties for the current method. These properties provide information that PluginManager uses to determine whether the method is a plug-in, essentially the same as the following example.
In [1] :def fuc1(a):. : print('hh')
...:
In [2]: setattr(fuc1, 'fuc1' + '_impl', dict(a=1, b=2))
In [3]: fuc1.fuc1_impl
Out[3] : {'a': 1.'b': 2}
Copy the codeAdd plug-in specification with registered plug-in behind
Once you have instantiated the plugy.pluginManager class, you can add the plug-in specification through the add_hookSpecs method and register the plug-in through the register method.
To figure out “How does Pluggy relate the plug-in specification to the plug-in itself?” , you need to go deep into their source code.
Instantiate the PluginManager class by calling its __init__ method.
def __init__(self, project_name, implprefix=None):
"""If ``implprefix`` is given implementation functions will be recognized if their name matches the ``implprefix``. """
self.project_name = project_name
#... Ellipsis...
# key
self.hook = _HookRelay()
self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
methods,
kwargs,
firstresult=hook.spec.opts.get("firstresult") if hook.spec else False.)Copy the codeThe key is to define the self.hook and self._inner_hookexec attributes. It is an anonymous method that takes hook, methods, and kwargs parameters and passes them to the hook.multicall method.
The add_hookspecs method is then called to add the plug-in specification as follows.
class PluginManager(object):
Get information about the corresponding attribute in the decorated method (HookspecMarker)
def parse_hookspec_opts(self, module_or_class, name):
method = getattr(module_or_class, name)
return getattr(method, self.project_name + "_spec".None)
def add_hookspecs(self, module_or_class):
names = []
for name in dir(module_or_class):
Get plugin specification information
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)
#... Omit some code...
Copy the codeIn the above code, the parse_hookspec_opts method is used to obtain the parameters of the corresponding property in the method. If the parameter is not None, the information about the decorated method in the _HookRelay class is obtained (this method is myhook of MySpec). From the source code can be found that the _HookRelay class is actually empty, its purpose is to receive new attributes, analysis to the later you will find that the _HookRelay class is actually used to connect the plug-in specification with the plug-in itself.
If there is no myhook attribute information in the _HookRelay class, instantiate the _HookCaller class as a self.hook attribute. Specifically, use an instance of the _HookCaller class as the value of the myhook attribute in the _HookRelay class.
The _HookCaller class is important, and its code is as follows.
class _HookCaller(object):
def __init__(self, name, hook_execute, specmodule_or_class=None, spec_opts=None):
self.name = name
#... Omit...
self._hookexec = hook_execute
self.spec = None
if specmodule_or_class is not None:
assert spec_opts is not None
self.set_specification(specmodule_or_class, spec_opts)
def has_spec(self):
return self.spec is not None
def set_specification(self, specmodule_or_class, spec_opts):
assert not self.has_spec()
self.spec = HookSpec(specmodule_or_class, self.name, spec_opts)
if spec_opts.get("historic"):
self._call_history = []
Copy the codeThe key is the set_specification method, which instantiates the HookSpec class and copies it to self.spec.
At this point, the plug-in specification is added and the plug-in itself is registered with the register method, with the following core code.
def register(self, plugin, name=None):
# omit some code
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)
# instantiate the plug-in
hookimpl = HookImpl(plugin, plugin_name, method, hookimpl_opts)
name = hookimpl_opts.get("specname") or name
hook = getattr(self.hook, name, None) Get the hookSpec plugin specification
if hook is None:
hook = _HookCaller(name, self._hookexec)
setattr(self.hook, name, hook)
elif hook.has_spec():
Check that the name and parameters of the plug-in method are the same as the plug-in specification
self._verify_hook(hook, hookimpl)
hook._maybe_apply_history(hookimpl)
Add to the plug-in specification to complete the binding of the plug-in to the plug-in specification
hook._add_hookimpl(hookimpl)
hookcallers.append(hook)
Copy the codeFirst get the information added by the Hookimpl decorator through the self.parse_hookimpl_opts method, then get the method name (myhook) through the getattr(plugin, name) method, and finally initialize the Hookimpl class, which is the plug-in itself. Bind it to the corresponding plug-in specification through the _add_hookimpl method.
The _add_hookimpl method determines the insertion position based on the attributes in the hookimpl instance. The call order varies depending on the position, as shown in the following code.
def _add_hookimpl(self, hookimpl):
"""Add an implementation to the callback chain. """
# Whether there is a wrapper (that is, the yield keyword is used in the plug-in logic)
if hookimpl.hookwrapper:
methods = self._wrappers
else:
methods = self._nonwrappers
Call first or code later
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)
Copy the codeSo far “5. How does Pluggy relate the plug-in specification to the plug-in itself?” In short, both the plug-in specification and the plug-in itself have special information added by the decorator. By finding and distributing these special information, the _HookCaller class (the plug-in specification) and the HookImpl class (the plug-in itself) are initialized using the values of these attributes. The binding is finally done with the _add_hookimpl method.
How exactly is the logic in the plug-in invoked?
Myhook (a=10, b=20); pm.hook (a=10, b=20);
What’s behind it?
Pluginmanager. hook is the _HookRelay class, and the _HookRelay class schema is an empty class. By using the add_hookspecs and register methods, the _HookRelay class has a property named myhook. This property corresponds to the _HookCaller class instance.
Pm.hook. Myhook (a=10, b=20) calls _hookcaller.__call__.
def __call__(self, *args, **kwargs):
# omit some code
if self.spec and self.spec.argnames:
Calculate whether the parameters accepted in the plug-in specification are the same as those accepted by the plug-in itself
notincall = (
set(self.spec.argnames) - set(["__multicall__"]) - set(kwargs.keys())
)
if notincall:
# omit code
# call method
return self._hookexec(self, self.get_hookimpls(), kwargs)
Copy the codeThe main purpose of the __call__ method is to determine whether the plug-in specification matches the plug-in itself and then execute it through the self._hookexec method.
Through analysis, the complete call chain is: _hookexec -> pluginManager._inner_hookexec -> _hookcaller. multicall -> the _multicall method in the callers file
The most critical code snippet in the _multicall method is as follows.
def _multicall(hook_impls, caller_kwargs, firstresult=False):
for hook_impl in reversed(hook_impls):
try:
Call myhook
args = [caller_kwargs[argname] for argname in hook_impl.argnames]
# omit code
If yeild is used in the plug-in, it is called this way
if hook_impl.hookwrapper:
try:
gen = hook_impl.function(*args)
next(gen) # first yield
teardowns.append(gen)
except StopIteration:
_raise_wrapfail(gen, "did not yield")
Copy the codeThat’s the end of pluggy’s core logic.
The tail
If you are finished, congratulations, but this is only the simplest model of Pluggy. There are some more important methods that I have not posted for space reasons. If you are interested, you can study them by yourself or discuss them with me.
After taking time out to talk about some of the source code.