test_helpers.py 文件源码

def gen_test(func):
    """
    Helper for running async tests, based on the tornado.testing.gen_test
    decorator.
    It wraps the test function with tornado.gen.coroutine and initialises an
    IOLoop to run the async code using IOLoop.run_sync
    NOTE: if using this with the mock.patch decorator apply gen_test first,
    otherwise the patches won't work.
    ANOTHER NOTE: if you don't yield when calling coroutines in your test you
    can get false positives, just like any other place where you don't call
    coroutines correctly. It's always a good idea to see your test fail so
    you know it's testing something.
    """
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        cofunc = coroutine(func)
        io_loop = IOLoop.current()

        try:
            result = io_loop.run_sync(functools.partial(cofunc, *args, **kwargs))
            return result

        finally:
            io_loop.clear_current()
            if not IOLoop.initialized() or io_loop is not IOLoop.instance():
                io_loop.close(all_fds=True)


    return wrapper