我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用tornado.testing.bind_unused_port()。
def test_handler_callback_file_object(self): """The handler callback receives the same fd object it passed in.""" server_sock, port = bind_unused_port() fds = [] def handle_connection(fd, events): fds.append(fd) conn, addr = server_sock.accept() conn.close() self.stop() self.io_loop.add_handler(server_sock, handle_connection, IOLoop.READ) with contextlib.closing(socket.socket()) as client_sock: client_sock.connect(('127.0.0.1', port)) self.wait() self.io_loop.remove_handler(server_sock) self.io_loop.add_handler(server_sock.fileno(), handle_connection, IOLoop.READ) with contextlib.closing(socket.socket()) as client_sock: client_sock.connect(('127.0.0.1', port)) self.wait() self.assertIs(fds[0], server_sock) self.assertEqual(fds[1], server_sock.fileno()) self.io_loop.remove_handler(server_sock.fileno()) server_sock.close()
def test_multi_line_headers(self): # Multi-line http headers are rare but rfc-allowed # http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.2 sock, port = bind_unused_port() with closing(sock): def write_response(stream, request_data): if b"HTTP/1." not in request_data: self.skipTest("requires HTTP/1.x") stream.write(b"""\ HTTP/1.1 200 OK X-XSS-Protection: 1; \tmode=block """.replace(b"\n", b"\r\n"), callback=stream.close) def accept_callback(conn, address): stream = IOStream(conn, io_loop=self.io_loop) stream.read_until(b"\r\n\r\n", functools.partial(write_response, stream)) netutil.add_accept_handler(sock, accept_callback, self.io_loop) self.http_client.fetch("http://127.0.0.1:%d/" % port, self.stop) resp = self.wait() resp.rethrow() self.assertEqual(resp.headers['X-XSS-Protection'], "1; mode=block") self.io_loop.remove_handler(sock.fileno())
def setUp(self): if IOLoop.configured_class().__name__ in ('TwistedIOLoop', 'AsyncIOMainLoop'): # TwistedIOLoop only supports the global reactor, so we can't have # separate IOLoops for client and server threads. # AsyncIOMainLoop doesn't work with the default policy # (although it could with some tweaks to this test and a # policy that created loops for non-main threads). raise unittest.SkipTest( 'Sync HTTPClient not compatible with TwistedIOLoop or ' 'AsyncIOMainLoop') self.server_ioloop = IOLoop() sock, self.port = bind_unused_port() app = Application([('/', HelloWorldHandler)]) self.server = HTTPServer(app, io_loop=self.server_ioloop) self.server.add_socket(sock) self.server_thread = threading.Thread(target=self.server_ioloop.start) self.server_thread.start() self.http_client = HTTPClient()
def refusing_port(): """Returns a local port number that will refuse all connections. Return value is (cleanup_func, port); the cleanup function must be called to free the port to be reused. """ # On travis-ci, port numbers are reassigned frequently. To avoid # collisions with other tests, we use an open client-side socket's # ephemeral port number to ensure that nothing can listen on that # port. server_socket, port = bind_unused_port() server_socket.setblocking(1) client_socket = socket.socket() client_socket.connect(("127.0.0.1", port)) conn, client_addr = server_socket.accept() conn.close() server_socket.close() return (client_socket.close, client_addr[1])
def make_iostream_pair(self, **kwargs): listener, port = bind_unused_port() streams = [None, None] def accept_callback(connection, address): streams[0] = self._make_server_iostream(connection, **kwargs) self.stop() def connect_callback(): streams[1] = client_stream self.stop() netutil.add_accept_handler(listener, accept_callback, io_loop=self.io_loop) client_stream = self._make_client_iostream(socket.socket(), **kwargs) client_stream.connect(('127.0.0.1', port), callback=connect_callback) self.wait(condition=lambda: all(streams)) self.io_loop.remove_handler(listener.fileno()) listener.close() return streams
def setUp(self): try: super(TestIOStreamStartTLS, self).setUp() self.listener, self.port = bind_unused_port() self.server_stream = None self.server_accepted = Future() netutil.add_accept_handler(self.listener, self.accept) self.client_stream = IOStream(socket.socket()) self.io_loop.add_future(self.client_stream.connect( ('127.0.0.1', self.port)), self.stop) self.wait() self.io_loop.add_future(self.server_accepted, self.stop) self.wait() except Exception as e: print(e) raise
def connect_to_server(self, server_cls): server = client = None try: sock, port = bind_unused_port() server = server_cls(ssl_options=_server_ssl_options()) server.add_socket(sock) client = SSLIOStream(socket.socket(), ssl_options=dict(cert_reqs=ssl.CERT_NONE)) yield client.connect(('127.0.0.1', port)) self.assertIsNotNone(client.socket.cipher()) finally: if server is not None: server.stop() if client is not None: client.close()
def setUp(self): try: super(TestIOStreamStartTLS, self).setUp() self.listener, self.port = bind_unused_port() self.server_stream = None self.server_accepted = Future() netutil.add_accept_handler(self.listener, self.accept) self.client_stream = MicroProxyIOStream(socket.socket()) self.io_loop.add_future(self.client_stream.connect( ('127.0.0.1', self.port)), self.stop) self.wait() self.io_loop.add_future(self.server_accepted, self.stop) self.wait() except Exception as e: print(e) raise
def create_iostream_pair(self): _lock = Event() server_streams = [] def accept_callback(conn, addr): server_stream = MicroProxyIOStream(conn) server_streams.append(server_stream) # self.addCleanup(server_stream.close) _lock.set() listener, port = bind_unused_port() add_accept_handler(listener, accept_callback) client_stream = MicroProxyIOStream(socket.socket()) yield [client_stream.connect(('127.0.0.1', port)), _lock.wait()] self.io_loop.remove_handler(listener) listener.close() raise Return((client_stream, server_streams[0]))
def test_connection_refused(self): server_socket, port = bind_unused_port() server_socket.close() with ExpectLog(gen_log, ".*", required=False): self.http_client.fetch("http://localhost:%d/" % port, self.stop) response = self.wait() self.assertEqual(599, response.code) if sys.platform != 'cygwin': # cygwin returns EPERM instead of ECONNREFUSED here self.assertTrue(str(errno.ECONNREFUSED) in str(response.error), response.error) # This is usually "Connection refused". # On windows, strerror is broken and returns "Unknown error". expected_message = os.strerror(errno.ECONNREFUSED) self.assertTrue(expected_message in str(response.error), response.error)
def test_connection_refused(self): # When a connection is refused, the connect callback should not # be run. (The kqueue IOLoop used to behave differently from the # epoll IOLoop in this respect) server_socket, port = bind_unused_port() server_socket.close() stream = IOStream(socket.socket(), self.io_loop) self.connect_called = False def connect_callback(): self.connect_called = True stream.set_close_callback(self.stop) # log messages vary by platform and ioloop implementation with ExpectLog(gen_log, ".*", required=False): stream.connect(("localhost", port), connect_callback) self.wait() self.assertFalse(self.connect_called) self.assertTrue(isinstance(stream.error, socket.error), stream.error) if sys.platform != 'cygwin': # cygwin's errnos don't match those used on native windows python self.assertEqual(stream.error.args[0], errno.ECONNREFUSED)
def setUp(self): if IOLoop.configured_class().__name__ == 'TwistedIOLoop': # TwistedIOLoop only supports the global reactor, so we can't have # separate IOLoops for client and server threads. raise unittest.SkipTest( 'Sync HTTPClient not compatible with TwistedIOLoop') self.server_ioloop = IOLoop() sock, self.port = bind_unused_port() app = Application([('/', HelloWorldHandler)]) server = HTTPServer(app, io_loop=self.server_ioloop) server.add_socket(sock) self.server_thread = threading.Thread(target=self.server_ioloop.start) self.server_thread.start() self.http_client = HTTPClient()
def test_connection_refused(self): server_socket, port = bind_unused_port() server_socket.close() with ExpectLog(gen_log, ".*"): self.http_client.fetch("http://localhost:%d/" % port, self.stop) response = self.wait() self.assertEqual(599, response.code) if sys.platform != 'cygwin': # cygwin returns EPERM instead of ECONNREFUSED here self.assertTrue(str(errno.ECONNREFUSED) in str(response.error), response.error) # This is usually "Connection refused". # On windows, strerror is broken and returns "Unknown error". expected_message = os.strerror(errno.ECONNREFUSED) self.assertTrue(expected_message in str(response.error), response.error)