我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用tornado.ioloop.TimeoutError()。
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
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
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
def test_timeout(self): @gen.coroutine def f(): yield gen.Task(self.io_loop.add_timeout, self.io_loop.time() + 1) self.assertRaises(TimeoutError, self.io_loop.run_sync, f, timeout=0.01)
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
def run_sync(self, function: typing.Callable, timeout: int=5): """ Execute the async function and return its result. This method takes care of starting and stopping the ioloop. The main use for this function is to use the inmanta internal rpc to communicate with the server. :param function: The async function to execute. This function should return a yieldable object. :param timeout: A timeout for the async function. :return: The result of the async call. :raises ConnectionRefusedError: When the function timeouts this exception is raised. """ from tornado.ioloop import IOLoop, TimeoutError try: return IOLoop.current().run_sync(function, timeout) except TimeoutError: raise ConnectionRefusedError()
