python类engine()的实例源码

def test_yield_in_with(self):
        @gen.engine
        def f():
            self.callback = yield gen.Callback('a')
            with StackContext(functools.partial(self.context, 'c1')):
                # This yield is a problem: the generator will be suspended
                # and the StackContext's __exit__ is not called yet, so
                # the context will be left on _state.contexts for anything
                # that runs before the yield resolves.
                yield gen.Wait('a')

        with self.assertRaises(StackContextInconsistentError):
            f()
            self.wait()
        # Cleanup: to avoid GC warnings (which for some reason only seem
        # to show up on py33-asyncio), invoke the callback (which will do
        # nothing since the gen.Runner is already finished) and delete it.
        self.callback()
        del self.callback

def test_stack_context_leak_exception(self):
        # same as previous, but with a function that exits with an exception
        @gen.engine
        def inner(callback):
            yield gen.Task(self.io_loop.add_callback)
            1 / 0

        @gen.engine
        def outer():
            for i in range(10):
                try:
                    yield gen.Task(inner)
                except ZeroDivisionError:
                    pass
            stack_increase = len(stack_context._state.contexts) - initial_stack_depth
            self.assertTrue(stack_increase <= 2)
            self.stop()
        initial_stack_depth = len(stack_context._state.contexts)
        self.run_gen(outer)

def test_task_refcounting(self):
        # On CPython, tasks and their arguments should be released immediately
        # without waiting for garbage collection.
        @gen.engine
        def f():
            class Foo(object):
                pass
            arg = Foo()
            self.arg_ref = weakref.ref(arg)
            task = gen.Task(self.io_loop.add_callback, arg=arg)
            self.task_ref = weakref.ref(task)
            yield task
            self.stop()

        self.run_gen(f)
        self.assertIs(self.arg_ref(), None)
        self.assertIs(self.task_ref(), None)

def test_future_traceback(self):
        @return_future
        @gen.engine
        def f(callback):
            yield gen.Task(self.io_loop.add_callback)
            try:
                1 / 0
            except ZeroDivisionError:
                self.expected_frame = traceback.extract_tb(
                    sys.exc_info()[2], limit=1)[0]
                raise
        try:
            yield f()
            self.fail("didn't get expected exception")
        except ZeroDivisionError:
            tb = traceback.extract_tb(sys.exc_info()[2])
            self.assertIn(self.expected_frame, tb)

# The following series of classes demonstrate and test various styles
# of use, with and without generators and futures.

def test_yield_in_with(self):
        @gen.engine
        def f():
            self.callback = yield gen.Callback('a')
            with StackContext(functools.partial(self.context, 'c1')):
                # This yield is a problem: the generator will be suspended
                # and the StackContext's __exit__ is not called yet, so
                # the context will be left on _state.contexts for anything
                # that runs before the yield resolves.
                yield gen.Wait('a')

        with self.assertRaises(StackContextInconsistentError):
            f()
            self.wait()
        # Cleanup: to avoid GC warnings (which for some reason only seem
        # to show up on py33-asyncio), invoke the callback (which will do
        # nothing since the gen.Runner is already finished) and delete it.
        self.callback()
        del self.callback

def test_task_refcounting(self):
        # On CPython, tasks and their arguments should be released immediately
        # without waiting for garbage collection.
        @gen.engine
        def f():
            class Foo(object):
                pass
            arg = Foo()
            self.arg_ref = weakref.ref(arg)
            task = gen.Task(self.io_loop.add_callback, arg=arg)
            self.task_ref = weakref.ref(task)
            yield task
            self.stop()

        self.run_gen(f)
        self.assertIs(self.arg_ref(), None)
        self.assertIs(self.task_ref(), None)

def test_yield_in_with(self):
        @gen.engine
        def f():
            self.callback = yield gen.Callback('a')
            with StackContext(functools.partial(self.context, 'c1')):
                # This yield is a problem: the generator will be suspended
                # and the StackContext's __exit__ is not called yet, so
                # the context will be left on _state.contexts for anything
                # that runs before the yield resolves.
                yield gen.Wait('a')

        with self.assertRaises(StackContextInconsistentError):
            f()
            self.wait()
        # Cleanup: to avoid GC warnings (which for some reason only seem
        # to show up on py33-asyncio), invoke the callback (which will do
        # nothing since the gen.Runner is already finished) and delete it.
        self.callback()
        del self.callback

def test_stack_context_leak_exception(self):
        # same as previous, but with a function that exits with an exception
        @gen.engine
        def inner(callback):
            yield gen.Task(self.io_loop.add_callback)
            1 / 0

        @gen.engine
        def outer():
            for i in range(10):
                try:
                    yield gen.Task(inner)
                except ZeroDivisionError:
                    pass
            stack_increase = len(stack_context._state.contexts) - initial_stack_depth
            self.assertTrue(stack_increase <= 2)
            self.stop()
        initial_stack_depth = len(stack_context._state.contexts)
        self.run_gen(outer)

def test_task_refcounting(self):
        # On CPython, tasks and their arguments should be released immediately
        # without waiting for garbage collection.
        @gen.engine
        def f():
            class Foo(object):
                pass
            arg = Foo()
            self.arg_ref = weakref.ref(arg)
            task = gen.Task(self.io_loop.add_callback, arg=arg)
            self.task_ref = weakref.ref(task)
            yield task
            self.stop()

        self.run_gen(f)
        self.assertIs(self.arg_ref(), None)
        self.assertIs(self.task_ref(), None)

def test_future_traceback(self):
        @return_future
        @gen.engine
        def f(callback):
            yield gen.Task(self.io_loop.add_callback)
            try:
                1 / 0
            except ZeroDivisionError:
                self.expected_frame = traceback.extract_tb(
                    sys.exc_info()[2], limit=1)[0]
                raise
        try:
            yield f()
            self.fail("didn't get expected exception")
        except ZeroDivisionError:
            tb = traceback.extract_tb(sys.exc_info()[2])
            self.assertIn(self.expected_frame, tb)

# The following series of classes demonstrate and test various styles
# of use, with and without generators and futures.

def test_yield_in_with(self):
        @gen.engine
        def f():
            self.callback = yield gen.Callback('a')
            with StackContext(functools.partial(self.context, 'c1')):
                # This yield is a problem: the generator will be suspended
                # and the StackContext's __exit__ is not called yet, so
                # the context will be left on _state.contexts for anything
                # that runs before the yield resolves.
                yield gen.Wait('a')

        with self.assertRaises(StackContextInconsistentError):
            f()
            self.wait()
        # Cleanup: to avoid GC warnings (which for some reason only seem
        # to show up on py33-asyncio), invoke the callback (which will do
        # nothing since the gen.Runner is already finished) and delete it.
        self.callback()
        del self.callback

def test_stack_context_leak_exception(self):
        # same as previous, but with a function that exits with an exception
        @gen.engine
        def inner(callback):
            yield gen.Task(self.io_loop.add_callback)
            1 / 0

        @gen.engine
        def outer():
            for i in range(10):
                try:
                    yield gen.Task(inner)
                except ZeroDivisionError:
                    pass
            stack_increase = len(stack_context._state.contexts) - initial_stack_depth
            self.assertTrue(stack_increase <= 2)
            self.stop()
        initial_stack_depth = len(stack_context._state.contexts)
        self.run_gen(outer)

def test_task_refcounting(self):
        # On CPython, tasks and their arguments should be released immediately
        # without waiting for garbage collection.
        @gen.engine
        def f():
            class Foo(object):
                pass
            arg = Foo()
            self.arg_ref = weakref.ref(arg)
            task = gen.Task(self.io_loop.add_callback, arg=arg)
            self.task_ref = weakref.ref(task)
            yield task
            self.stop()

        self.run_gen(f)
        self.assertIs(self.arg_ref(), None)
        self.assertIs(self.task_ref(), None)

def test_future_traceback(self):
        @return_future
        @gen.engine
        def f(callback):
            yield gen.Task(self.io_loop.add_callback)
            try:
                1 / 0
            except ZeroDivisionError:
                self.expected_frame = traceback.extract_tb(
                    sys.exc_info()[2], limit=1)[0]
                raise
        try:
            yield f()
            self.fail("didn't get expected exception")
        except ZeroDivisionError:
            tb = traceback.extract_tb(sys.exc_info()[2])
            self.assertIn(self.expected_frame, tb)

# The following series of classes demonstrate and test various styles
# of use, with and without generators and futures.

def test_yield_in_with_exception_stack_context(self):
        # As above, but with ExceptionStackContext instead of StackContext.
        @gen.engine
        def f():
            with ExceptionStackContext(lambda t, v, tb: False):
                yield gen.Task(self.io_loop.add_callback)

        with self.assertRaises(StackContextInconsistentError):
            f()
            self.wait()

def test_no_yield(self):
        @gen.engine
        def f():
            self.stop()
        self.run_gen(f)

def test_inline_cb(self):
        @gen.engine
        def f():
            (yield gen.Callback("k1"))()
            res = yield gen.Wait("k1")
            self.assertTrue(res is None)
            self.stop()
        self.run_gen(f)

def test_ioloop_cb(self):
        @gen.engine
        def f():
            self.io_loop.add_callback((yield gen.Callback("k1")))
            yield gen.Wait("k1")
            self.stop()
        self.run_gen(f)

def test_exception_phase2(self):
        @gen.engine
        def f():
            self.io_loop.add_callback((yield gen.Callback("k1")))
            yield gen.Wait("k1")
            1 / 0
        self.assertRaises(ZeroDivisionError, self.run_gen, f)

def test_exception_in_task_phase1(self):
        def fail_task(callback):
            1 / 0

        @gen.engine
        def f():
            try:
                yield gen.Task(fail_task)
                raise Exception("did not get expected exception")
            except ZeroDivisionError:
                self.stop()
        self.run_gen(f)

def test_exception_in_task_phase2(self):
        # This is the case that requires the use of stack_context in gen.engine
        def fail_task(callback):
            self.io_loop.add_callback(lambda: 1 / 0)

        @gen.engine
        def f():
            try:
                yield gen.Task(fail_task)
                raise Exception("did not get expected exception")
            except ZeroDivisionError:
                self.stop()
        self.run_gen(f)

def test_with_arg(self):
        @gen.engine
        def f():
            (yield gen.Callback("k1"))(42)
            res = yield gen.Wait("k1")
            self.assertEqual(42, res)
            self.stop()
        self.run_gen(f)

def test_with_arg_tuple(self):
        @gen.engine
        def f():
            (yield gen.Callback((1, 2)))((3, 4))
            res = yield gen.Wait((1, 2))
            self.assertEqual((3, 4), res)
            self.stop()
        self.run_gen(f)

def test_key_reuse_tuple(self):
        @gen.engine
        def f():
            yield gen.Callback((1, 2))
            yield gen.Callback((1, 2))
            self.stop()
        self.assertRaises(gen.KeyReuseError, self.run_gen, f)

def test_key_mismatch(self):
        @gen.engine
        def f():
            yield gen.Callback("k1")
            yield gen.Wait("k2")
            self.stop()
        self.assertRaises(gen.UnknownKeyError, self.run_gen, f)

def test_key_mismatch_tuple(self):
        @gen.engine
        def f():
            yield gen.Callback((1, 2))
            yield gen.Wait((2, 3))
            self.stop()
        self.assertRaises(gen.UnknownKeyError, self.run_gen, f)

def test_leaked_callback(self):
        @gen.engine
        def f():
            yield gen.Callback("k1")
            self.stop()
        self.assertRaises(gen.LeakedCallbackError, self.run_gen, f)

def test_leaked_callback_tuple(self):
        @gen.engine
        def f():
            yield gen.Callback((1, 2))
            self.stop()
        self.assertRaises(gen.LeakedCallbackError, self.run_gen, f)

def test_bogus_yield(self):
        @gen.engine
        def f():
            yield 42
        self.assertRaises(gen.BadYieldError, self.run_gen, f)

def test_bogus_yield_tuple(self):
        @gen.engine
        def f():
            yield (1, 2)
        self.assertRaises(gen.BadYieldError, self.run_gen, f)