pytest源码走读-钩子函数
pytest源码走读-钩子函数
钩子函数的实现
研究对象：pytest的_com.py
细化目标：Muticall、Registry、Hook，是一个递进关系
学习方式：实现一遍Hook
Muticall
将多个方法对象（python 万物即对象）存放在list中，在调用这些方法时，执行execute方法，实现多方法的一次性调用输出，输出结果以list结构输出
以下是Muticall的实现，
#coding=utf-8
#使用list实现muticall
class Muticall: def __init__(self,methods,*args,**kwargs):
self.methods = methods[:] #以list的数据格式入参
self.args = args
self.kwargs = kwargs
self.result = [] def execute(self): while self.methods:
currentmethod = self.methods.pop()
res = self.execute_method(currentmethod)
self.result.append(res) if self.result: return self.result def execute_method(self,currentmethod):
self.currentmethod = currentmethod
varnames = currentmethod.func_code.co_varnames #varnames0 = currentmethod.__code__.co_varnames
#调用标识位
needscall = varnames[:1] == ('__call__',) #print needscall
#print "func_code属性:"
#print varnames
#print "__code__属性:"
#print varnames0
return currentmethod(*self.args, **self.kwargs)调用过程，
（1）首先初始化Muticall对象，将需要调用的方法，以list的格式赋值给mc的属性self.methods
（2）调用execute()方法，由该方法循环遍历对象，实现挨个执行调用方法的能力，
（3）将返回结果append到list中返回
调用方式（目的是熟悉，固定入参，*args 的tuple 可变长度入参，kwargs 可变长度的key-value 入参该如何入参，muticall 初始化时，如何打包固定参数和tuple入参，具体到调用函数时，又如何解包）
from muticall import Muticall class Test: def mytest(self): return "111" def mytest1(self): return "222" class Test1: def mytest(self,x,*args): print args return x+1 def mytest1(self,x,*args): print args return x+2 class Test2: def mytest(self,**kwargs): return kwargs class Test3: def mytest(self,x,*args,**kwargs): print args,kwargs return x+1 def mytest1(self,x,*args,**kwargs): print args,kwargs return x+2 class Tool: @staticmethod def call_execute(methods,*args,**kwargs): pass if __name__=='__main__': switch = sys.argv[1] if switch == 'None': test = Test() methods = [test.mytest,test.mytest1] mc = Muticall(methods) print mc.execute() elif switch == 'args': test = Test1() args = 1 methods = [test.mytest,test.mytest1] mc = Muticall(methods,3,args) print mc.execute() elif switch == 'kwargs': test = Test2() methods = [test.mytest] mc = Muticall(methods,kw='a') print mc.execute() elif switch == 'args_kwargs': test = Test3() args = 1 methods = [test.mytest,test.mytest1] mc = Muticall(methods,3,args,kw1=1,kw2=2) print mc.execute() else: pass
1、switch == 'None'，以空入参方法，调用，返回 ['222', '111']
2、switch == 'args'，有固定入参&有可选入参，调用，返回 [5, 4]
3、switch == 'kwargs'，有可选key-value 的dict入参，调用,返回 [{'kw': 'a'}]
4、switch == 'args_kwargs'，有可选key-value 的dict入参，调用,返回 [5, 4]
Registry
1、将类作为插件，存放在list中，通过操作list的append、remove、 in ，实现插件的注册、注销、判断插件是否注册；以及通过listattr获取指定的属性或者每个插件的指定属性。
2、支持批量注册
#coding=utf-8 #对插件做管理，包括注册、移除插件，遍历插件，检查插件是否注册 #注册对象：类(、方法、实例、属性) #数据结构：list import muticall as Muticall class Registry: Muticall = Muticall #无特殊意义和用途，只是用来做sys.argv[1]=='None'的测试之用 def __init__(self,plugins=None): if plugins is None: plugins = [] self.plugins = plugins def register(self,plugin): self.plugins.append(plugin) def unregister(self,plugin): self.plugins.remove(plugin) def isregister(self,plugin): return plugin in self.plugins def __iter__(self): return iter(self.plugins) def listattr(self,attrname,plugins=None,extra=(),isreverse=False): l = [] if plugins is None: plugins = self.plugins for plugin in list(plugins)+list(extra): try: l.append(getattr(plugin,attrname)) except AttributeError: continue if isreverse: l.reverse() return l
调用过程，
初始化的时候，获得一个注册机对象，包含注册插件的list容器plugins。
可通过register方法，一个一个的注册插件，通过unregister实现注销，isregister判断是否已经注册，listattr可以指定插件，获取插件属性，不指定的情况下，默认输出所有插件的该属性，最终以list格式返回；
可通过初始化注册机对象，实现批量注册
调用方式，
1、实例化一个空注册机，没有任何注册类
2、通过注册函数，注册单个插件
3、注销
4、判断是否注册
5、获取注册插件的指定属性
6、通过初始化，批量注册插件
#coding=utf-8
import sys from register import Registry class test_Registry: pass
class Test: pass
class Api1:
x = 1 class Api2:
x = 2 class Api3:
x = 3 if __name__ == '__main__':
test = test_Registry()
test1 = Test()
api1 = Api1()
api2 = Api2()
api3 = Api3() if sys.argv[1] == 'None':
registry = Registry() if hasattr(registry,'Muticall'): print 'Muticall'
if sys.argv[1] == 'register':
registry = Registry()
registry.register(test) print "注册test"
flag = (list(registry) == [test]) print flag print list(registry) print [registry] if sys.argv[1] == 'unregister':
registry = Registry()
registry.register(test)
registry.register(test1) print "注册[test,test1]"
flag = (list(registry)==[test,test1]) print flag print list(registry) print [test,test1] print "注销test1"
registry.unregister(test1)
flag = (list(registry)==[test]) print flag print list(registry) print [test] if sys.argv[1] == 'isregister':
registry = Registry()
registry.register(test) print "注册test"
flag = (list(registry)==[test]) print flag print list(registry) print [test] print "test已注册:",
flag = registry.isregister(test) print flag print "test1已注册:",
flag1 = registry.isregister(test1) print flag1 if sys.argv[1] == 'listattr':
registry = Registry()
registry.register(api1)
registry.register(api2)
registry.register(api3) print '指定属性&所有插件'
print registry.listattr('x') print '指定属性&指定插件：'
print registry.listattr('x',[api1]) if sys.argv[1] == 'mutiregister':
registry = Registry([api1,api2,api3]) print "初始化时注册api1,api2,api3:"
flag = (list(registry)==[api1,api2,api3]) print flag print list(registry) print [api1,api2,api3]Hook
第一步需要实现注册，利用注册机对待钩的函数进行登记注册；
第二步需要实现调用，利用Muticall 对注册的钩子函数进行批量执行；
以下是对1.0.06b的钩子实现的解读：
class Hooks: def __init__(self, hookspecs, registry=None):
self._hookspecs = hookspecs if registry is None:
registry = py._com.comregistry
self.registry = registry for name, method in vars(hookspecs).items():#是class Object的dict的key和value(key- name，value-method)
#只注册方法
if name[:1] != "_":#找函数名（所以强烈建议甚至禁止，函数名开头使用下划线）
#print method.__dict__ #空方法对象的是个空dict
firstresult = getattr(method, 'firstresult', False) #print "注册一个方法需要的信息，注册插件，注册函数名，注册属性是否存在"
#print registry.__dict__
#print name
#print firstresult
mm = HookCall(registry, name, firstresult=firstresult) #打包成一个钩子对象
print '\033[1;31m 打包的钩子对象:\033[0m'
print mm.__dict__
setattr(self, name, mm) print('\033[0;31m hookspecs='+ str(self._hookspecs) + '\033[0m') print self.__dict__
#for key in self:
#print(key+":"+self[key])
def __repr__(self): return "<Hooks %r %r>" %(self._hookspecs, self.registry) class HookCall: def __init__(self, registry, name, firstresult, extralookup=None):
self.registry = registry
self.name = name
self.firstresult = firstresult
self.extralookup = extralookup and [extralookup] or () def clone(self, extralookup): return HookCall(self.registry, self.name, self.firstresult, extralookup) def __repr__(self):
mode = self.firstresult and "firstresult" or "each"
return "<HookCall %r mode=%s %s>" %(self.name, mode, self.registry) def __call__(self, *args, **kwargs): if args: raise TypeError("only keyword arguments allowed "
"for api call to %r" % self.name)
attr = self.registry.listattr(self.name, extra=self.extralookup)
mc = MultiCall(attr, **kwargs) # XXX this should be doable from a hook impl:
if self.registry.logfile:
self.registry.logfile.write("%s(**%s) # firstresult=%s\n" %
(self.name, kwargs, self.firstresult))
self.registry.logfile.flush() return mc.execute(firstresult=self.firstresult)
comregistry = Registry()在使用的时候，除了第一个类需要初始化一次钩子对象，其他的具有相同方法的类，直接通过注册，即可实现钩子函数的身份登记
然后，直接使用实例化钩子对象，调用暴露出来的钩子函数，即可完成调用。调用过程使用了listattr()方法遍历该类该方法先的属性（参数值），还原该方法的入参格式，供Muticall 调用
class TestHooks: def test_happypath(self):
registry = Registry() class Api: def hello(self, arg): return arg
mcm = Hooks(hookspecs=Api, registry=registry) #registry.register(Api())
#r = mcm.hello(arg="222")
#print "调用钩子函数:"
#print r
assert hasattr(mcm, 'hello') assert repr(mcm.hello).find("hello") != -1 class Plugin: def hello(self, arg): return arg+"1"
registry.register(Plugin())
l = mcm.hello(arg="3") print l #assert l == [4]
assert not hasattr(mcm, 'world') print mcm.__dict__
class Test: def hello(self,arg): return arg+"5"
registry.register(Test()) #t = mcm.hello(arg='abc')
t = mcm.hello(arg='a') print "调用钩子函数:"
print t #assert t == "abc"
__EOF__
作 者：XiaoLee-C
出 处：https://www.cnblogs.com/ansonwan/p/14637692.html
关于博主：编程路上的小学生，热爱技术，喜欢专研。评论和私信会在第一时间回复。或者直接私信我。
版权声明：署名 - 非商业性使用 - 禁止演绎，协议普通文本 | 协议法律文本。
声援博主：如果您觉得文章对您有帮助，可以点击文章右下角【推荐】一下。您的鼓励是博主的最大动力！