pytest源碼走讀-pluggy的前身之_com.py

背景

pluggy僅有幾千行代碼，但是是實現pytest框架的一個核心組成。

1.0.06b版本的pytest中，pluggy框架還只是作為一個.py模塊出現，沒有被獨立打包成一個插件，而這個模塊就是_com.py。接下來主要讀一讀_com.py這段源碼

鈎子函數的實現

研究對象：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"

 參考資料：http://markshao.github.io/categories/pytest-%E6%BA%90%E7%A0%81%E8%A7%A3%E8%AF%BB/（將pytest中hook和pluggy的關系，pytest中hook的作用指明）