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

[Vyacheslav Rafalskiy]

>> #---------->>  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.

>
> 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

> 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