Python tornado.testing 模块，gen_test() 实例源码

我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用tornado.testing.gen_test()。

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_timeout(self):
        # Set a short timeout and exceed it.
        @gen_test(timeout=0.1)
        def test(self):
            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)

        # This can't use assertRaises because we need to inspect the
        # exc_info triple (and not just the exception object)
        try:
            test(self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            # The stack trace should blame the add_timeout line, not just
            # unrelated IOLoop/testing internals.
            self.assertIn(
                "gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
                traceback.format_exc())

        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_future_delayed_close_callback(self):
        # Same as test_delayed_close_callback, but with the future interface.
        server, client = self.make_iostream_pair()

        # We can't call make_iostream_pair inside a gen_test function
        # because the ioloop is not reentrant.
        @gen_test
        def f(self):
            server.write(b"12")
            chunks = []
            chunks.append((yield client.read_bytes(1)))
            server.close()
            chunks.append((yield client.read_bytes(1)))
            self.assertEqual(chunks, [b"1", b"2"])
        try:
            f(self)
        finally:
            server.close()
            client.close()

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_timeout(self):
        # Set a short timeout and exceed it.
        @gen_test(timeout=0.1)
        def test(self):
            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)

        # This can't use assertRaises because we need to inspect the
        # exc_info triple (and not just the exception object)
        try:
            test(self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            # The stack trace should blame the add_timeout line, not just
            # unrelated IOLoop/testing internals.
            self.assertIn(
                "gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
                traceback.format_exc())

        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_timeout(self):
        # Set a short timeout and exceed it.
        @gen_test(timeout=0.1)
        def test(self):
            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)

        # This can't use assertRaises because we need to inspect the
        # exc_info triple (and not just the exception object)
        try:
            test(self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            # The stack trace should blame the add_timeout line, not just
            # unrelated IOLoop/testing internals.
            self.assertIn(
                "gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
                traceback.format_exc())

        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_future_delayed_close_callback(self):
        # Same as test_delayed_close_callback, but with the future interface.
        server, client = self.make_iostream_pair()

        # We can't call make_iostream_pair inside a gen_test function
        # because the ioloop is not reentrant.
        @gen_test
        def f(self):
            server.write(b"12")
            chunks = []
            chunks.append((yield client.read_bytes(1)))
            server.close()
            chunks.append((yield client.read_bytes(1)))
            self.assertEqual(chunks, [b"1", b"2"])
        try:
            f(self)
        finally:
            server.close()
            client.close()

项目：microProxy 作者：mike820324 | 项目源码 | 文件源码

def test_future_delayed_close_callback(self):
        # Same as test_delayed_close_callback, but with the future interface.
        server, client = self.make_iostream_pair()

        # We can't call make_iostream_pair inside a gen_test function
        # because the ioloop is not reentrant.
        @gen_test
        def f(self):
            server.write(b"12")
            chunks = []
            chunks.append((yield client.read_bytes(1)))
            server.close()
            chunks.append((yield client.read_bytes(1)))
            self.assertEqual(chunks, [b"1", b"2"])
        try:
            f(self)
        finally:
            server.close()
            client.close()

项目：My-Web-Server-Framework-With-Python2.7 作者：syjsu | 项目源码 | 文件源码

def test_timeout(self):
        # Set a short timeout and exceed it.
        @gen_test(timeout=0.1)
        def test(self):
            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)

        # This can't use assertRaises because we need to inspect the
        # exc_info triple (and not just the exception object)
        try:
            test(self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            # The stack trace should blame the add_timeout line, not just
            # unrelated IOLoop/testing internals.
            self.assertIn(
                "gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
                traceback.format_exc())

        self.finished = True

项目：My-Web-Server-Framework-With-Python2.7 作者：syjsu | 项目源码 | 文件源码

def test_future_delayed_close_callback(self):
        # Same as test_delayed_close_callback, but with the future interface.
        server, client = self.make_iostream_pair()

        # We can't call make_iostream_pair inside a gen_test function
        # because the ioloop is not reentrant.
        @gen_test
        def f(self):
            server.write(b"12")
            chunks = []
            chunks.append((yield client.read_bytes(1)))
            server.close()
            chunks.append((yield client.read_bytes(1)))
            self.assertEqual(chunks, [b"1", b"2"])
        try:
            f(self)
        finally:
            server.close()
            client.close()

项目：annotated-py-tornado 作者：hhstore | 项目源码 | 文件源码

def test_timeout(self):
        # Set a short timeout and exceed it.
        @gen_test(timeout=0.1)
        def test(self):
            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)

        # This can't use assertRaises because we need to inspect the
        # exc_info triple (and not just the exception object)
        try:
            test(self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            # The stack trace should blame the add_timeout line, not just
            # unrelated IOLoop/testing internals.
            self.assertIn(
                "gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
                traceback.format_exc())

        self.finished = True

项目：annotated-py-tornado 作者：hhstore | 项目源码 | 文件源码

def test_future_delayed_close_callback(self):
        # Same as test_delayed_close_callback, but with the future interface.
        server, client = self.make_iostream_pair()

        # We can't call make_iostream_pair inside a gen_test function
        # because the ioloop is not reentrant.
        @gen_test
        def f(self):
            server.write(b"12")
            chunks = []
            chunks.append((yield client.read_bytes(1)))
            server.close()
            chunks.append((yield client.read_bytes(1)))
            self.assertEqual(chunks, [b"1", b"2"])
        try:
            f(self)
        finally:
            server.close()
            client.close()

项目：projects-2017-2 作者：ncss | 项目源码 | 文件源码

def test_timeout(self):
        # Set a short timeout and exceed it.
        @gen_test(timeout=0.1)
        def test(self):
            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)

        # This can't use assertRaises because we need to inspect the
        # exc_info triple (and not just the exception object)
        try:
            test(self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            # The stack trace should blame the add_timeout line, not just
            # unrelated IOLoop/testing internals.
            self.assertIn(
                "gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
                traceback.format_exc())

        self.finished = True

项目：projects-2017-2 作者：ncss | 项目源码 | 文件源码

def test_future_delayed_close_callback(self):
        # Same as test_delayed_close_callback, but with the future interface.
        server, client = self.make_iostream_pair()

        # We can't call make_iostream_pair inside a gen_test function
        # because the ioloop is not reentrant.
        @gen_test
        def f(self):
            server.write(b"12")
            chunks = []
            chunks.append((yield client.read_bytes(1)))
            server.close()
            chunks.append((yield client.read_bytes(1)))
            self.assertEqual(chunks, [b"1", b"2"])
        try:
            f(self)
        finally:
            server.close()
            client.close()

项目：aweasome_learning 作者：Knight-ZXW | 项目源码 | 文件源码

def test_timeout(self):
        # Set a short timeout and exceed it.
        @gen_test(timeout=0.1)
        def test(self):
            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)

        # This can't use assertRaises because we need to inspect the
        # exc_info triple (and not just the exception object)
        try:
            test(self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            # The stack trace should blame the add_timeout line, not just
            # unrelated IOLoop/testing internals.
            self.assertIn(
                "gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
                traceback.format_exc())

        self.finished = True

项目：aweasome_learning 作者：Knight-ZXW | 项目源码 | 文件源码

def test_future_delayed_close_callback(self):
        # Same as test_delayed_close_callback, but with the future interface.
        server, client = self.make_iostream_pair()

        # We can't call make_iostream_pair inside a gen_test function
        # because the ioloop is not reentrant.
        @gen_test
        def f(self):
            server.write(b"12")
            chunks = []
            chunks.append((yield client.read_bytes(1)))
            server.close()
            chunks.append((yield client.read_bytes(1)))
            self.assertEqual(chunks, [b"1", b"2"])
        try:
            f(self)
        finally:
            server.close()
            client.close()

项目：zenchmarks 作者：squeaky-pl | 项目源码 | 文件源码

def test_timeout(self):
        # Set a short timeout and exceed it.
        @gen_test(timeout=0.1)
        def test(self):
            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)

        # This can't use assertRaises because we need to inspect the
        # exc_info triple (and not just the exception object)
        try:
            test(self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            # The stack trace should blame the add_timeout line, not just
            # unrelated IOLoop/testing internals.
            self.assertIn(
                "gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
                traceback.format_exc())

        self.finished = True

项目：zenchmarks 作者：squeaky-pl | 项目源码 | 文件源码

def test_future_delayed_close_callback(self):
        # Same as test_delayed_close_callback, but with the future interface.
        server, client = self.make_iostream_pair()

        # We can't call make_iostream_pair inside a gen_test function
        # because the ioloop is not reentrant.
        @gen_test
        def f(self):
            server.write(b"12")
            chunks = []
            chunks.append((yield client.read_bytes(1)))
            server.close()
            chunks.append((yield client.read_bytes(1)))
            self.assertEqual(chunks, [b"1", b"2"])
        try:
            f(self)
        finally:
            server.close()
            client.close()

项目：browser_vuln_check 作者：lcatro | 项目源码 | 文件源码

def test_timeout(self):
        # Set a short timeout and exceed it.
        @gen_test(timeout=0.1)
        def test(self):
            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)

        # This can't use assertRaises because we need to inspect the
        # exc_info triple (and not just the exception object)
        try:
            test(self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            # The stack trace should blame the add_timeout line, not just
            # unrelated IOLoop/testing internals.
            self.assertIn(
                "gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
                traceback.format_exc())

        self.finished = True

项目：browser_vuln_check 作者：lcatro | 项目源码 | 文件源码

def test_future_delayed_close_callback(self):
        # Same as test_delayed_close_callback, but with the future interface.
        server, client = self.make_iostream_pair()

        # We can't call make_iostream_pair inside a gen_test function
        # because the ioloop is not reentrant.
        @gen_test
        def f(self):
            server.write(b"12")
            chunks = []
            chunks.append((yield client.read_bytes(1)))
            server.close()
            chunks.append((yield client.read_bytes(1)))
            self.assertEqual(chunks, [b"1", b"2"])
        try:
            f(self)
        finally:
            server.close()
            client.close()

项目：PyQYT 作者：collinsctk | 项目源码 | 文件源码

def test_timeout(self):
        # Set a short timeout and exceed it.
        @gen_test(timeout=0.1)
        def test(self):
            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)

        # This can't use assertRaises because we need to inspect the
        # exc_info triple (and not just the exception object)
        try:
            test(self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            # The stack trace should blame the add_timeout line, not just
            # unrelated IOLoop/testing internals.
            self.assertIn(
                "gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
                traceback.format_exc())

        self.finished = True

项目：PyQYT 作者：collinsctk | 项目源码 | 文件源码

def test_future_delayed_close_callback(self):
        # Same as test_delayed_close_callback, but with the future interface.
        server, client = self.make_iostream_pair()

        # We can't call make_iostream_pair inside a gen_test function
        # because the ioloop is not reentrant.
        @gen_test
        def f(self):
            server.write(b"12")
            chunks = []
            chunks.append((yield client.read_bytes(1)))
            server.close()
            chunks.append((yield client.read_bytes(1)))
            self.assertEqual(chunks, [b"1", b"2"])
        try:
            f(self)
        finally:
            server.close()
            client.close()

项目：ProgrameFacil 作者：Gpzim98 | 项目源码 | 文件源码

def test_timeout(self):
        # Set a short timeout and exceed it.
        @gen_test(timeout=0.1)
        def test(self):
            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)

        # This can't use assertRaises because we need to inspect the
        # exc_info triple (and not just the exception object)
        try:
            test(self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            # The stack trace should blame the add_timeout line, not just
            # unrelated IOLoop/testing internals.
            self.assertIn(
                "gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
                traceback.format_exc())

        self.finished = True

项目：ProgrameFacil 作者：Gpzim98 | 项目源码 | 文件源码

def test_future_delayed_close_callback(self):
        # Same as test_delayed_close_callback, but with the future interface.
        server, client = self.make_iostream_pair()

        # We can't call make_iostream_pair inside a gen_test function
        # because the ioloop is not reentrant.
        @gen_test
        def f(self):
            server.write(b"12")
            chunks = []
            chunks.append((yield client.read_bytes(1)))
            server.close()
            chunks.append((yield client.read_bytes(1)))
            self.assertEqual(chunks, [b"1", b"2"])
        try:
            f(self)
        finally:
            server.close()
            client.close()

项目：ProgrameFacil 作者：Gpzim98 | 项目源码 | 文件源码

def test_timeout(self):
        # Set a short timeout and exceed it.
        @gen_test(timeout=0.1)
        def test(self):
            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)

        # This can't use assertRaises because we need to inspect the
        # exc_info triple (and not just the exception object)
        try:
            test(self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            # The stack trace should blame the add_timeout line, not just
            # unrelated IOLoop/testing internals.
            self.assertIn(
                "gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)",
                traceback.format_exc())

        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_no_timeout(self):
        # A test that does not exceed its timeout should succeed.
        @gen_test(timeout=1)
        def test(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 0.1)

        test(self)
        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_timeout_environment_variable(self):
        @gen_test(timeout=0.5)
        def test_long_timeout(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 0.25)

        # Uses provided timeout of 0.5 seconds, doesn't time out.
        with set_environ('ASYNC_TEST_TIMEOUT', '0.1'):
            test_long_timeout(self)

        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_no_timeout_environment_variable(self):
        @gen_test(timeout=0.01)
        def test_short_timeout(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 1)

        # Uses environment-variable timeout of 0.1, times out.
        with set_environ('ASYNC_TEST_TIMEOUT', '0.1'):
            with self.assertRaises(ioloop.TimeoutError):
                test_short_timeout(self)

        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_with_method_args(self):
        @gen_test
        def test_with_args(self, *args):
            self.assertEqual(args, ('test',))
            yield gen.Task(self.io_loop.add_callback)

        test_with_args(self, 'test')
        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_native_coroutine(self):
        namespace = exec_test(globals(), locals(), """
        @gen_test
        async def test(self):
            self.finished = True
        """)

        namespace['test'](self)

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_native_coroutine_timeout(self):
        # Set a short timeout and exceed it.
        namespace = exec_test(globals(), locals(), """
        @gen_test(timeout=0.1)
        async def test(self):
            await gen.sleep(1)
        """)

        try:
            namespace['test'](self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_no_timeout(self):
        # A test that does not exceed its timeout should succeed.
        @gen_test(timeout=1)
        def test(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 0.1)

        test(self)
        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_no_timeout_environment_variable(self):
        @gen_test(timeout=0.01)
        def test_short_timeout(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 1)

        # Uses environment-variable timeout of 0.1, times out.
        with set_environ('ASYNC_TEST_TIMEOUT', '0.1'):
            with self.assertRaises(ioloop.TimeoutError):
                test_short_timeout(self)

        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_with_method_args(self):
        @gen_test
        def test_with_args(self, *args):
            self.assertEqual(args, ('test',))
            yield gen.Task(self.io_loop.add_callback)

        test_with_args(self, 'test')
        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_with_method_kwargs(self):
        @gen_test
        def test_with_kwargs(self, **kwargs):
            self.assertDictEqual(kwargs, {'test': 'test'})
            yield gen.Task(self.io_loop.add_callback)

        test_with_kwargs(self, test='test')
        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_native_coroutine(self):
        namespace = exec_test(globals(), locals(), """
        @gen_test
        async def test(self):
            self.finished = True
        """)

        namespace['test'](self)

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_native_coroutine_timeout(self):
        # Set a short timeout and exceed it.
        namespace = exec_test(globals(), locals(), """
        @gen_test(timeout=0.1)
        async def test(self):
            await gen.sleep(1)
        """)

        try:
            namespace['test'](self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_no_timeout(self):
        # A test that does not exceed its timeout should succeed.
        @gen_test(timeout=1)
        def test(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 0.1)

        test(self)
        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_timeout_environment_variable(self):
        @gen_test(timeout=0.5)
        def test_long_timeout(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 0.25)

        # Uses provided timeout of 0.5 seconds, doesn't time out.
        with set_environ('ASYNC_TEST_TIMEOUT', '0.1'):
            test_long_timeout(self)

        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_no_timeout_environment_variable(self):
        @gen_test(timeout=0.01)
        def test_short_timeout(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 1)

        # Uses environment-variable timeout of 0.1, times out.
        with set_environ('ASYNC_TEST_TIMEOUT', '0.1'):
            with self.assertRaises(ioloop.TimeoutError):
                test_short_timeout(self)

        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_with_method_args(self):
        @gen_test
        def test_with_args(self, *args):
            self.assertEqual(args, ('test',))
            yield gen.Task(self.io_loop.add_callback)

        test_with_args(self, 'test')
        self.finished = True

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_native_coroutine(self):
        namespace = exec_test(globals(), locals(), """
        @gen_test
        async def test(self):
            self.finished = True
        """)

        namespace['test'](self)

项目：noc-orchestrator 作者：DirceuSilvaLabs | 项目源码 | 文件源码

def test_native_coroutine_timeout(self):
        # Set a short timeout and exceed it.
        namespace = exec_test(globals(), locals(), """
        @gen_test(timeout=0.1)
        async def test(self):
            await gen.sleep(1)
        """)

        try:
            namespace['test'](self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            self.finished = True

项目：auth-srv 作者：openpermissions | 项目源码 | 文件源码

def test_invalid_action(self):
        # parametrize doesn't work with gen_test?
        scopes = [
            'invalid',
            READ + ' invalid',
            'invalid[read]'
        ]

        for scope in scopes:
            with pytest.raises(oauth2.InvalidScope):
                yield Scope(scope).validate(self.client)

项目：My-Web-Server-Framework-With-Python2.7 作者：syjsu | 项目源码 | 文件源码

def test_no_timeout(self):
        # A test that does not exceed its timeout should succeed.
        @gen_test(timeout=1)
        def test(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 0.1)

        test(self)
        self.finished = True

项目：My-Web-Server-Framework-With-Python2.7 作者：syjsu | 项目源码 | 文件源码

def test_timeout_environment_variable(self):
        @gen_test(timeout=0.5)
        def test_long_timeout(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 0.25)

        # Uses provided timeout of 0.5 seconds, doesn't time out.
        with set_environ('ASYNC_TEST_TIMEOUT', '0.1'):
            test_long_timeout(self)

        self.finished = True

项目：My-Web-Server-Framework-With-Python2.7 作者：syjsu | 项目源码 | 文件源码

def test_no_timeout_environment_variable(self):
        @gen_test(timeout=0.01)
        def test_short_timeout(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 1)

        # Uses environment-variable timeout of 0.1, times out.
        with set_environ('ASYNC_TEST_TIMEOUT', '0.1'):
            with self.assertRaises(ioloop.TimeoutError):
                test_short_timeout(self)

        self.finished = True

项目：My-Web-Server-Framework-With-Python2.7 作者：syjsu | 项目源码 | 文件源码

def test_with_method_args(self):
        @gen_test
        def test_with_args(self, *args):
            self.assertEqual(args, ('test',))
            yield gen.Task(self.io_loop.add_callback)

        test_with_args(self, 'test')
        self.finished = True

项目：My-Web-Server-Framework-With-Python2.7 作者：syjsu | 项目源码 | 文件源码

def test_native_coroutine(self):
        namespace = exec_test(globals(), locals(), """
        @gen_test
        async def test(self):
            self.finished = True
        """)

        namespace['test'](self)

项目：My-Web-Server-Framework-With-Python2.7 作者：syjsu | 项目源码 | 文件源码

def test_native_coroutine_timeout(self):
        # Set a short timeout and exceed it.
        namespace = exec_test(globals(), locals(), """
        @gen_test(timeout=0.1)
        async def test(self):
            await gen.sleep(1)
        """)

        try:
            namespace['test'](self)
            self.fail("did not get expected exception")
        except ioloop.TimeoutError:
            self.finished = True

项目：annotated-py-tornado 作者：hhstore | 项目源码 | 文件源码

def test_timeout(self):
        # Set a short timeout and exceed it.
        @gen_test(timeout=0.1)
        def test(self):
            yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1)

        with self.assertRaises(ioloop.TimeoutError):
            test(self)

        self.finished = True

项目：annotated-py-tornado 作者：hhstore | 项目源码 | 文件源码

def test_no_timeout(self):
        # A test that does not exceed its timeout should succeed.
        @gen_test(timeout=1)
        def test(self):
            time = self.io_loop.time
            yield gen.Task(self.io_loop.add_timeout, time() + 0.1)

        test(self)
        self.finished = True