test_ok1/py/test/funcargs.py

import py

def getfuncargnames(function):
    argnames = py.std.inspect.getargs(function.func_code)[0]
    startindex = hasattr(function, 'im_self') and 1 or 0 
    numdefaults = len(function.func_defaults or ()) 
    if numdefaults:
        return argnames[startindex:-numdefaults]
    return argnames[startindex:]
    
def fillfuncargs(function):
    """ fill missing funcargs. """ 
    argnames = getfuncargnames(function.obj)
    if argnames:
        assert not function._args, "yielded functions cannot have funcargs" 
        for argname in argnames:
            if argname not in function.funcargs:
                request = FuncargRequest(pyfuncitem=function, argname=argname) 
                try:
                    function.funcargs[argname] = request.call_next_provider()
                except request.Error:
                    request._raiselookupfailed()


_notexists = object()
class CallSpec:
    def __init__(self, id, param):
        self.id = id
        if param is not _notexists:
            self.param = param 

class Metafunc:
    def __init__(self, function, config=None, cls=None, module=None):
        self.config = config
        self.module = module 
        self.function = function
        self.funcargnames = getfuncargnames(function)
        self.cls = cls
        self.module = module
        self._calls = []
        self._ids = py.builtin.set()

    def addcall(self, id=None, param=_notexists):
        if id is None:
            id = len(self._calls)
        id = str(id)
        if id in self._ids:
            raise ValueError("duplicate id %r" % id)
        self._ids.add(id)
        self._calls.append(CallSpec(id, param))

class FunctionCollector(py.test.collect.Collector):
    def __init__(self, name, parent, calls):
        super(FunctionCollector, self).__init__(name, parent)
        self.calls = calls 
        self.obj = getattr(self.parent.obj, name) 
       
    def collect(self):
        l = []
        for call in self.calls:
            function = self.parent.Function(name="%s[%s]" %(self.name, call.id),
                parent=self, requestparam=call.param, callobj=self.obj)
            l.append(function)
        return l 

class FuncargRequest:
    _argprefix = "pytest_funcarg__"

    class Error(LookupError):
        """ error on performing funcarg request. """ 

    def __init__(self, pyfuncitem, argname):
        self._pyfuncitem = pyfuncitem
        self.argname = argname 
        self.function = pyfuncitem.obj
        self.module = pyfuncitem._getparent(py.test.collect.Module).obj
        self.cls = getattr(self.function, 'im_class', None)
        self.config = pyfuncitem.config
        self.fspath = pyfuncitem.fspath
        if hasattr(pyfuncitem, '_requestparam'):
            self.param = pyfuncitem._requestparam 
        self._plugins = self.config.pluginmanager.getplugins()
        self._plugins.append(self.module)
        self._provider = self.config.pluginmanager.listattr(
            plugins=self._plugins, 
            attrname=self._argprefix + str(argname)
        )

    def call_next_provider(self):
        if not self._provider:
            raise self.Error("no provider methods left")
        next_provider = self._provider.pop()
        return next_provider(request=self)

    def addfinalizer(self, finalizer):
        self._pyfuncitem.addfinalizer(finalizer)

    def __repr__(self):
        return "<FuncargRequest %r for %r>" %(self.argname, self._pyfuncitem)

    def _raiselookupfailed(self):
        available = []
        for plugin in self._plugins:
            for name in vars(plugin.__class__):
                if name.startswith(self._argprefix):
                    name = name[len(self._argprefix):]
                    if name not in available:
                        available.append(name) 
        fspath, lineno, msg = self._pyfuncitem.reportinfo()
        line = "%s:%s" %(fspath, lineno)
        msg = "funcargument %r not found for: %s" %(self.argname, line)
        msg += "\n available funcargs: %s" %(", ".join(available),)
        raise LookupError(msg)