[py-dev] Decorators and funcargs in py.test

[holger krekel]

On Wed, Jun 01, 2011 at 17:16 -0400, Vyacheslav Rafalskiy wrote:
> >> #---------->>  in test_1.py
> >> @pytest.mark.timeout(10)
> >> def test_f1():
> >>     # test here
> >>
> >> #---------->>  in conftest.py
> >> def pytest_runtest_call(item):
> >>     if hasattr(item.obj, 'timeout'):
> >>         timeout = item.obj.timeout.args[0]
> >>         item.obj = run_with_timeout(timeout)(item.obj)
> >>
> >> Your comments are welcome.
> >
> > it's basically ok but there are some bits that could
> > be improved.  You are actually implementing the general
> > test item call.  Also I am not sure why you assign
> > "item.obj = ...".
> 
> I replace (or so I think) the original test function by the decorated
> one. It gets called elsewhere.

ah, of course :)

> > I'd probably write something like this:
> >
> >    @pytest.mark.tryfirst
> >    def pytest_pyfunc_call(pyfuncitem, __multicall__):
> >        if 'timeout' in pyfuncitem.keywords:
> >            timeout = pyfuncitem.keywords['timeout'].args[0]
> >            run_with_timeout(timeout)(__multicall__.execute)
> 
> this will take a while to digest

it's actually wrong if run_with_timeout is only decorating
but not running the function.  I think it makes sense to
rather directly call a helper which calls the function 
(note that __multicall__.execute() will execute the remainder
of the hook chain one of which will actually execute the
function).  Such a helper would look like
call_with_timeout(timeout, func) i guess.

best,
holger

> > main differences:
> >
> > * only applies to python test function calls
> > * hook invocation will be "tried first" before other
> >  pytest_pyfunc_call hook impls and it will call those
> >  other hooks through the "__multicall__" bit
> >  which actually represents the ongoing hook call
> > * will call other hook implementations
> >
> > How do you actually implement run_with_timeout, btw?
> 
> def run_with_timeout(timeout=None):
>     def _decorator(f):
>         def _wrapper(*args, **kwargs):
>             def _f(*args, **kwargs):
>                 try:
>                     _result = f(*args, **kwargs)
>                 except Exception as e:
>                     thread._exception = e
>                 else:
>                     thread._result = _result
> 
>             thread = threading.Thread(target=_f, args=args, kwargs=kwargs)
>             thread.daemon = True
>             thread._exception = None
> 
>             thread.start()
>             thread.join(timeout=timeout)
>             if thread.isAlive():
>                 raise RuntimeError('function *%s* exceeded configured
> timeout of %ss' % (f.__name__, timeout))
>             if thread._exception is None:
>                 return thread._result
>             else:
>                 raise thread._exception
>         _wrapper.__name__ = f.__name__
>         return _wrapper
>     return _decorator
>
> 
> 
> >
> > best,
> > holger
> >
> 
> Vyacheslav
>