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
>
_______________________________________________
py-dev mailing list
py-dev@codespeak.net
http://codespeak.net/mailman/listinfo/py-dev
