Method do_test_connect

tornado/test/tcpclient_test.py:88–102

(self, family, host, source_ip=None, source_port=None)

    @gen_test
    def do_test_connect(self, family, host, source_ip=None, source_port=None):
        port = self.start_server(family)
        stream = yield self.client.connect(
            host,
            port,
            source_ip=source_ip,
            source_port=source_port,
            af=family,
        )
        assert self.server is not None
        server_stream = yield self.server.queue.get()
        with closing(stream):
            stream.write(b"hello")
            data = yield server_stream.read_bytes(5)
            self.assertEqual(data, b"hello")

    def test_connect_ipv4_ipv4(self):
        self.do_test_connect(socket.AF_INET, "127.0.0.1")