concurrent_test.py 文件源码

def test_no_calling(self):
        class Object(object):
            def __init__(self, io_loop):
                self.io_loop = io_loop
                self.executor = futures.thread.ThreadPoolExecutor(1)

            @run_on_executor
            def f(self):
                return 42

        o = Object(io_loop=self.io_loop)
        answer = yield o.f()
        self.assertEqual(answer, 42)