Method test_set_default_executor

tornado/test/ioloop_test.py:554–572

(self)

    @gen_test
    def test_set_default_executor(self):
        count = [0]

        class MyExecutor(futures.ThreadPoolExecutor):
            def submit(self, func, *args):
                count[0] += 1
                return super().submit(func, *args)

        event = threading.Event()

        def sync_func():
            event.set()

        executor = MyExecutor(1)
        loop = IOLoop.current()
        loop.set_default_executor(executor)
        yield loop.run_in_executor(None, sync_func)
        self.assertEqual(1, count[0])
        self.assertTrue(event.is_set())


class TestIOLoopRunSync(unittest.TestCase):