我们从Python开源项目中，提取了以下39个代码示例，用于说明如何使用tornado.ioloop.IOLoop.initialized()。
def tearDown(self): # Clean up Subprocess, so it can be used again with a new ioloop. Subprocess.uninitialize() self.io_loop.clear_current() if (not IOLoop.initialized() or self.io_loop is not IOLoop.instance()): # Try to clean up any file descriptors left open in the ioloop. # This avoids leaks, especially when tests are run repeatedly # in the same process with autoreload (because curl does not # set FD_CLOEXEC on its file descriptors) self.io_loop.close(all_fds=True) super(AsyncTestCase, self).tearDown() # In case an exception escaped or the StackContext caught an exception # when there wasn't a wait() to re-raise it, do so here. # This is our last chance to raise an exception in a way that the # unittest machinery understands. self.__rethrow()
def run(): app = Application([("/", RootHandler)]) port = random.randrange(options.min_port, options.max_port) app.listen(port, address='127.0.0.1') signal.signal(signal.SIGCHLD, handle_sigchld) args = ["ab"] args.extend(["-n", str(options.n)]) args.extend(["-c", str(options.c)]) if options.keepalive: args.append("-k") if options.quiet: # just stops the progress messages printed to stderr args.append("-q") args.append("http://127.0.0.1:%d/" % port) subprocess.Popen(args) IOLoop.instance().start() IOLoop.instance().close() del IOLoop._instance assert not IOLoop.initialized()
def install_asyncio() -> None: """Install tornado's loop to asyncio.""" try: from tornado.ioloop import IOLoop from tornado.platform.asyncio import AsyncIOMainLoop if not IOLoop.initialized(): AsyncIOMainLoop().install() except ImportError: pass
def tearDown(self): self.http_server.stop() self.io_loop.run_sync(self.http_server.close_all_connections, timeout=get_async_test_timeout()) if (not IOLoop.initialized() or self.http_client.io_loop is not IOLoop.instance()): self.http_client.close() super(AsyncHTTPTestCase, self).tearDown()
def install_asyncio() -> None: """Ensure that asyncio's io-loop is installed to tornado.""" from tornado.ioloop import IOLoop from tornado.platform.asyncio import AsyncIOMainLoop if not IOLoop.initialized(): AsyncIOMainLoop().install()
def tearDown(self): if (not IOLoop.initialized() or self.io_loop is not IOLoop.instance()): # Try to clean up any file descriptors left open in the ioloop. # This avoids leaks, especially when tests are run repeatedly # in the same process with autoreload (because curl does not # set FD_CLOEXEC on its file descriptors) self.io_loop.close(all_fds=True) super(AsyncTestCase, self).tearDown()
def tearDown(self): self.io_loop.clear_current() if (not IOLoop.initialized() or self.io_loop is not IOLoop.instance()): # Try to clean up any file descriptors left open in the ioloop. # This avoids leaks, especially when tests are run repeatedly # in the same process with autoreload (because curl does not # set FD_CLOEXEC on its file descriptors) self.io_loop.close(all_fds=True) super(AsyncTestCase, self).tearDown() # In case an exception escaped or the StackContext caught an exception # when there wasn't a wait() to re-raise it, do so here. # This is our last chance to raise an exception in a way that the # unittest machinery understands. self.__rethrow()
def tearDown(self): self.http_server.stop() if (not IOLoop.initialized() or self.http_client.io_loop is not IOLoop.instance()): self.http_client.close() super(AsyncHTTPTestCase, self).tearDown()
def gen_test(func): """ Helper for running async tests, based on the tornado.testing.gen_test decorator. It wraps the test function with tornado.gen.coroutine and initialises an IOLoop to run the async code using IOLoop.run_sync NOTE: if using this with the mock.patch decorator apply gen_test first, otherwise the patches won't work. ANOTHER NOTE: if you don't yield when calling coroutines in your test you can get false positives, just like any other place where you don't call coroutines correctly. It's always a good idea to see your test fail so you know it's testing something. """ @functools.wraps(func) def wrapper(*args, **kwargs): cofunc = coroutine(func) io_loop = IOLoop.current() try: result = io_loop.run_sync(functools.partial(cofunc, *args, **kwargs)) return result finally: io_loop.clear_current() if not IOLoop.initialized() or io_loop is not IOLoop.instance(): io_loop.close(all_fds=True) return wrapper
