Python eventlet 模块，connect() 实例源码

def parent(signal_path, pid):
    eventlet.Timeout(5)
    port = None
    while True:
        try:
            contents = open(signal_path, 'rb').read()
            port = int(contents.strip())
            break
        except Exception:
            eventlet.sleep(0.1)
    eventlet.connect(('127.0.0.1', port))
    while True:
        try:
            contents = open(signal_path, 'rb').read()
            result = contents.split()[1]
            break
        except Exception:
            eventlet.sleep(0.1)
    assert result == b'done', repr(result)
    print('pass')

def test_hub_exceptions(self):
        debug.hub_exceptions(True)
        server = eventlet.listen(('0.0.0.0', 0))
        client = eventlet.connect(('127.0.0.1', server.getsockname()[1]))
        client_2, addr = server.accept()

        def hurl(s):
            s.recv(1)
            {}[1]  # keyerror

        with capture_stderr() as fake:
            gt = eventlet.spawn(hurl, client_2)
            eventlet.sleep(0)
            client.send(b' ')
            eventlet.sleep(0)
            # allow the "hurl" greenlet to trigger the KeyError
            # not sure why the extra context switch is needed
            eventlet.sleep(0)
        self.assertRaises(KeyError, gt.wait)
        debug.hub_exceptions(False)
        # look for the KeyError exception in the traceback
        assert 'KeyError: 1' in fake.getvalue(), "Traceback not in:\n" + fake.getvalue()

def test_raised_multiple_readers(self):
        debug.hub_prevent_multiple_readers(True)

        def handle(sock, addr):
            sock.recv(1)
            sock.sendall(b"a")
            raise eventlet.StopServe()

        listener = eventlet.listen(('127.0.0.1', 0))
        eventlet.spawn(eventlet.serve, listener, handle)

        def reader(s):
            s.recv(1)

        s = eventlet.connect(('127.0.0.1', listener.getsockname()[1]))
        a = eventlet.spawn(reader, s)
        eventlet.sleep(0)
        self.assertRaises(RuntimeError, s.recv, 1)
        s.sendall(b'b')
        a.wait()

def test_zero_timeout_and_back(self):
        listen = eventlet.listen(('', 0))
        # Keep reference to server side of socket
        server = eventlet.spawn(listen.accept)
        client = eventlet.connect(listen.getsockname())

        client.settimeout(0.05)
        # Now must raise socket.timeout
        self.assertRaises(socket.timeout, client.recv, 1)

        client.settimeout(0)
        # Now must raise socket.error with EAGAIN
        try:
            client.recv(1)
            assert False
        except socket.error as e:
            assert get_errno(e) == errno.EAGAIN

        client.settimeout(0.05)
        # Now socket.timeout again
        self.assertRaises(socket.timeout, client.recv, 1)
        server.wait()

def test_socket_file_read_non_int():
    listen_socket = eventlet.listen(('localhost', 0))

    def server():
        conn, _ = listen_socket.accept()
        conn.recv(1)
        conn.sendall(b'response')
        conn.close()

    eventlet.spawn(server)
    sock = eventlet.connect(listen_socket.getsockname())

    fd = sock.makefile('rwb')
    fd.write(b'?')
    fd.flush()
    with eventlet.Timeout(1):
        try:
            fd.read("This shouldn't work")
            assert False
        except TypeError:
            pass

def test_recv_type():
    # https://github.com/eventlet/eventlet/issues/245
    # socket recv returning multiple data types
    # For this test to work, client and server have to be in separate
    # processes or OS threads. Just running two greenthreads gives
    # false test pass.
    threading = eventlet.patcher.original('threading')
    addr = []

    def server():
        sock = eventlet.listen(('127.0.0.1', 0))
        addr[:] = sock.getsockname()
        eventlet.sleep(0.2)

    server_thread = threading.Thread(target=server)
    server_thread.start()
    eventlet.sleep(0.1)
    sock = eventlet.connect(tuple(addr))
    s = sock.recv(1)
    assert isinstance(s, bytes)

def test_correct_upgrade_request_75(self):
        connect = [
            "GET /echo HTTP/1.1",
            "Upgrade: WebSocket",
            "Connection: Upgrade",
            "Host: %s:%s" % self.server_addr,
            "Origin: http://%s:%s" % self.server_addr,
            "WebSocket-Protocol: ws",
        ]
        sock = eventlet.connect(self.server_addr)

        sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n'))
        result = sock.recv(1024)
        # The server responds the correct Websocket handshake
        self.assertEqual(result, six.b('\r\n'.join([
            'HTTP/1.1 101 Web Socket Protocol Handshake',
            'Upgrade: WebSocket',
            'Connection: Upgrade',
            'WebSocket-Origin: http://%s:%s' % self.server_addr,
            'WebSocket-Location: ws://%s:%s/echo\r\n\r\n' % self.server_addr,
        ])))

def test_correct_upgrade_request_76(self):
        connect = [
            "GET /echo HTTP/1.1",
            "Upgrade: WebSocket",
            "Connection: Upgrade",
            "Host: %s:%s" % self.server_addr,
            "Origin: http://%s:%s" % self.server_addr,
            "Sec-WebSocket-Protocol: ws",
            "Sec-WebSocket-Key1: 4 @1  46546xW%0l 1 5",
            "Sec-WebSocket-Key2: 12998 5 Y3 1  .P00",
        ]
        sock = eventlet.connect(self.server_addr)

        sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n^n:ds[4U'))
        result = sock.recv(1024)
        # The server responds the correct Websocket handshake
        self.assertEqual(result, six.b('\r\n'.join([
            'HTTP/1.1 101 WebSocket Protocol Handshake',
            'Upgrade: WebSocket',
            'Connection: Upgrade',
            'Sec-WebSocket-Origin: http://%s:%s' % self.server_addr,
            'Sec-WebSocket-Protocol: ws',
            'Sec-WebSocket-Location: ws://%s:%s/echo\r\n\r\n8jKS\'y:G*Co,Wxa-' % self.server_addr,
        ])))

def test_query_string(self):
        # verify that the query string comes out the other side unscathed
        connect = [
            "GET /echo?query_string HTTP/1.1",
            "Upgrade: WebSocket",
            "Connection: Upgrade",
            "Host: %s:%s" % self.server_addr,
            "Origin: http://%s:%s" % self.server_addr,
            "Sec-WebSocket-Protocol: ws",
            "Sec-WebSocket-Key1: 4 @1  46546xW%0l 1 5",
            "Sec-WebSocket-Key2: 12998 5 Y3 1  .P00",
        ]
        sock = eventlet.connect(self.server_addr)

        sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n^n:ds[4U'))
        result = sock.recv(1024)
        self.assertEqual(result, six.b('\r\n'.join([
            'HTTP/1.1 101 WebSocket Protocol Handshake',
            'Upgrade: WebSocket',
            'Connection: Upgrade',
            'Sec-WebSocket-Origin: http://%s:%s' % self.server_addr,
            'Sec-WebSocket-Protocol: ws',
            'Sec-WebSocket-Location: '
            'ws://%s:%s/echo?query_string\r\n\r\n8jKS\'y:G*Co,Wxa-' % self.server_addr,
        ])))

def test_empty_query_string(self):
        # verify that a single trailing ? doesn't get nuked
        connect = [
            "GET /echo? HTTP/1.1",
            "Upgrade: WebSocket",
            "Connection: Upgrade",
            "Host: %s:%s" % self.server_addr,
            "Origin: http://%s:%s" % self.server_addr,
            "Sec-WebSocket-Protocol: ws",
            "Sec-WebSocket-Key1: 4 @1  46546xW%0l 1 5",
            "Sec-WebSocket-Key2: 12998 5 Y3 1  .P00",
        ]
        sock = eventlet.connect(self.server_addr)

        sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n^n:ds[4U'))
        result = sock.recv(1024)
        self.assertEqual(result, six.b('\r\n'.join([
            'HTTP/1.1 101 WebSocket Protocol Handshake',
            'Upgrade: WebSocket',
            'Connection: Upgrade',
            'Sec-WebSocket-Origin: http://%s:%s' % self.server_addr,
            'Sec-WebSocket-Protocol: ws',
            'Sec-WebSocket-Location: ws://%s:%s/echo?\r\n\r\n8jKS\'y:G*Co,Wxa-' % self.server_addr,
        ])))

def test_sending_messages_to_websocket_75(self):
        connect = [
            "GET /echo HTTP/1.1",
            "Upgrade: WebSocket",
            "Connection: Upgrade",
            "Host: %s:%s" % self.server_addr,
            "Origin: http://%s:%s" % self.server_addr,
            "WebSocket-Protocol: ws",
        ]
        sock = eventlet.connect(self.server_addr)

        sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n'))
        sock.recv(1024)
        sock.sendall(b'\x00hello\xFF')
        result = sock.recv(1024)
        self.assertEqual(result, b'\x00hello\xff')
        sock.sendall(b'\x00start')
        eventlet.sleep(0.001)
        sock.sendall(b' end\xff')
        result = sock.recv(1024)
        self.assertEqual(result, b'\x00start end\xff')
        sock.shutdown(socket.SHUT_RDWR)
        sock.close()
        eventlet.sleep(0.01)

def test_getting_messages_from_websocket_75(self):
        connect = [
            "GET /range HTTP/1.1",
            "Upgrade: WebSocket",
            "Connection: Upgrade",
            "Host: %s:%s" % self.server_addr,
            "Origin: http://%s:%s" % self.server_addr,
            "WebSocket-Protocol: ws",
        ]
        sock = eventlet.connect(self.server_addr)

        sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n'))
        resp = sock.recv(1024)
        headers, result = resp.split(b'\r\n\r\n')
        msgs = [result.strip(b'\x00\xff')]
        cnt = 10
        while cnt:
            msgs.append(sock.recv(20).strip(b'\x00\xff'))
            cnt -= 1
        # Last item in msgs is an empty string
        self.assertEqual(msgs[:-1], [six.b('msg %d' % i) for i in range(10)])

def test_getting_messages_from_websocket_76(self):
        connect = [
            "GET /range HTTP/1.1",
            "Upgrade: WebSocket",
            "Connection: Upgrade",
            "Host: %s:%s" % self.server_addr,
            "Origin: http://%s:%s" % self.server_addr,
            "Sec-WebSocket-Protocol: ws",
            "Sec-WebSocket-Key1: 4 @1  46546xW%0l 1 5",
            "Sec-WebSocket-Key2: 12998 5 Y3 1  .P00",
        ]
        sock = eventlet.connect(self.server_addr)

        sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n^n:ds[4U'))
        resp = sock.recv(1024)
        headers, result = resp.split(b'\r\n\r\n')
        msgs = [result[16:].strip(b'\x00\xff')]
        cnt = 10
        while cnt:
            msgs.append(sock.recv(20).strip(b'\x00\xff'))
            cnt -= 1
        # Last item in msgs is an empty string
        self.assertEqual(msgs[:-1], [six.b('msg %d' % i) for i in range(10)])

def test_server_closing_connect_76(self):
        connect = [
            "GET / HTTP/1.1",
            "Upgrade: WebSocket",
            "Connection: Upgrade",
            "Host: %s:%s" % self.server_addr,
            "Origin: http://%s:%s" % self.server_addr,
            "Sec-WebSocket-Protocol: ws",
            "Sec-WebSocket-Key1: 4 @1  46546xW%0l 1 5",
            "Sec-WebSocket-Key2: 12998 5 Y3 1  .P00",
        ]
        sock = eventlet.connect(self.server_addr)
        sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n^n:ds[4U'))
        resp = sock.recv(1024)
        headers, result = resp.split(b'\r\n\r\n')
        # The remote server should have immediately closed the connection.
        self.assertEqual(result[16:], b'\xff\x00')

def test_close_idle(self):
        pool = eventlet.GreenPool()
        # use log=stderr when test runner can capture it
        self.spawn_server(custom_pool=pool, log=sys.stdout)
        connect = (
            'GET /echo HTTP/1.1',
            'Upgrade: WebSocket',
            'Connection: Upgrade',
            'Host: %s:%s' % self.server_addr,
            'Origin: http://%s:%s' % self.server_addr,
            'Sec-WebSocket-Protocol: ws',
            'Sec-WebSocket-Key1: 4 @1  46546xW%0l 1 5',
            'Sec-WebSocket-Key2: 12998 5 Y3 1  .P00',
        )
        sock = eventlet.connect(self.server_addr)
        sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n^n:ds[4U'))
        sock.recv(1024)
        sock.sendall(b'\x00hello\xff')
        result = sock.recv(1024)
        assert result, b'\x00hello\xff'
        self.killer.kill(KeyboardInterrupt)
        with eventlet.Timeout(1):
            pool.waitall()

def test_ssl_close(self):
        def serve(listener):
            sock, addr = listener.accept()
            sock.recv(8192)
            try:
                self.assertEqual(b'', sock.recv(8192))
            except greenio.SSL.ZeroReturnError:
                pass

        sock = listen_ssl_socket()

        server_coro = eventlet.spawn(serve, sock)

        raw_client = eventlet.connect(sock.getsockname())
        client = ssl.wrap_socket(raw_client)
        client.sendall(b'X')
        greenio.shutdown_safe(client)
        client.close()
        server_coro.wait()

def test_ssl_unwrap(self):
        def serve():
            sock, addr = listener.accept()
            self.assertEqual(sock.recv(6), b'before')
            sock_ssl = ssl.wrap_socket(sock, tests.private_key_file, tests.certificate_file,
                                       server_side=True)
            sock_ssl.do_handshake()
            self.assertEqual(sock_ssl.recv(6), b'during')
            sock2 = sock_ssl.unwrap()
            self.assertEqual(sock2.recv(5), b'after')
            sock2.close()

        listener = eventlet.listen(('127.0.0.1', 0))
        server_coro = eventlet.spawn(serve)
        client = eventlet.connect(listener.getsockname())
        client.sendall(b'before')
        client_ssl = ssl.wrap_socket(client)
        client_ssl.do_handshake()
        client_ssl.sendall(b'during')
        client2 = client_ssl.unwrap()
        client2.sendall(b'after')
        server_coro.wait()

def spawn_server(self, **kwargs):
        """Spawns a new wsgi server with the given arguments using
        :meth:`spawn_thread`.

        Sets `self.server_addr` to (host, port) tuple suitable for `socket.connect`.
        """
        self.logfile = six.StringIO()
        new_kwargs = dict(max_size=128,
                          log=self.logfile,
                          site=self.site)
        new_kwargs.update(kwargs)

        if 'sock' not in new_kwargs:
            new_kwargs['sock'] = eventlet.listen(('localhost', 0))

        self.server_addr = new_kwargs['sock'].getsockname()
        self.spawn_thread(wsgi.server, **new_kwargs)

def test_006_reject_long_urls(self):
        sock = eventlet.connect(self.server_addr)
        path_parts = []
        for ii in range(3000):
            path_parts.append('path')
        path = '/'.join(path_parts)
        request = 'GET /%s HTTP/1.0\r\nHost: localhost\r\n\r\n' % path
        send_expect_close(sock, request.encode())
        fd = sock.makefile('rb')
        result = fd.readline()
        if result:
            # windows closes the socket before the data is flushed,
            # so we never get anything back
            status = result.split(b' ')[1]
            self.assertEqual(status, b'414')
        fd.close()

def test_007_get_arg(self):
        # define a new handler that does a get_arg as well as a read_body
        def new_app(env, start_response):
            body = bytes_to_str(env['wsgi.input'].read())
            a = cgi.parse_qs(body).get('a', [1])[0]
            start_response('200 OK', [('Content-type', 'text/plain')])
            return [six.b('a is %s, body is %s' % (a, body))]

        self.site.application = new_app
        sock = eventlet.connect(self.server_addr)
        request = b'\r\n'.join((
            b'POST / HTTP/1.0',
            b'Host: localhost',
            b'Content-Length: 3',
            b'',
            b'a=a'))
        sock.sendall(request)

        # send some junk after the actual request
        sock.sendall(b'01234567890123456789')
        result = read_http(sock)
        self.assertEqual(result.body, b'a is a, body is a=a')

def test_012_ssl_server(self):
        def wsgi_app(environ, start_response):
            start_response('200 OK', {})
            return [environ['wsgi.input'].read()]

        certificate_file = os.path.join(os.path.dirname(__file__), 'test_server.crt')
        private_key_file = os.path.join(os.path.dirname(__file__), 'test_server.key')

        server_sock = eventlet.wrap_ssl(eventlet.listen(('localhost', 0)),
                                        certfile=certificate_file,
                                        keyfile=private_key_file,
                                        server_side=True)
        self.spawn_server(sock=server_sock, site=wsgi_app)

        sock = eventlet.connect(self.server_addr)
        sock = eventlet.wrap_ssl(sock)
        sock.write(
            b'POST /foo HTTP/1.1\r\nHost: localhost\r\n'
            b'Connection: close\r\nContent-length:3\r\n\r\nabc')
        result = recvall(sock)
        assert result.endswith(b'abc')

def test_013_empty_return(self):
        def wsgi_app(environ, start_response):
            start_response("200 OK", [])
            return [b""]

        certificate_file = os.path.join(os.path.dirname(__file__), 'test_server.crt')
        private_key_file = os.path.join(os.path.dirname(__file__), 'test_server.key')
        server_sock = eventlet.wrap_ssl(eventlet.listen(('localhost', 0)),
                                        certfile=certificate_file,
                                        keyfile=private_key_file,
                                        server_side=True)
        self.spawn_server(sock=server_sock, site=wsgi_app)

        sock = eventlet.connect(('localhost', server_sock.getsockname()[1]))
        sock = eventlet.wrap_ssl(sock)
        sock.write(b'GET /foo HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
        result = recvall(sock)
        assert result[-4:] == b'\r\n\r\n'

def test_016_repeated_content_length(self):
        """content-length header was being doubled up if it was set in
        start_response and could also be inferred from the iterator
        """
        def wsgi_app(environ, start_response):
            start_response('200 OK', [('Content-Length', '7')])
            return [b'testing']
        self.site.application = wsgi_app
        sock = eventlet.connect(self.server_addr)
        fd = sock.makefile('rwb')
        fd.write(b'GET /a HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
        fd.flush()
        header_lines = []
        while True:
            line = fd.readline()
            if line == b'\r\n':
                break
            else:
                header_lines.append(line)
        self.assertEqual(1, len(
            [l for l in header_lines if l.lower().startswith(b'content-length')]))

def test_018_http_10_keepalive(self):
        # verify that if an http/1.0 client sends connection: keep-alive
        # that we don't close the connection
        sock = eventlet.connect(self.server_addr)

        sock.sendall(b'GET / HTTP/1.0\r\nHost: localhost\r\nConnection: keep-alive\r\n\r\n')
        result1 = read_http(sock)
        assert 'connection' in result1.headers_lower
        self.assertEqual('keep-alive', result1.headers_lower['connection'])

        # repeat request to verify connection is actually still open
        sock.sendall(b'GET / HTTP/1.0\r\nHost: localhost\r\nConnection: keep-alive\r\n\r\n')
        result2 = read_http(sock)
        assert 'connection' in result2.headers_lower
        self.assertEqual('keep-alive', result2.headers_lower['connection'])
        sock.close()

def test_019_fieldstorage_compat(self):
        def use_fieldstorage(environ, start_response):
            cgi.FieldStorage(fp=environ['wsgi.input'], environ=environ)
            start_response('200 OK', [('Content-type', 'text/plain')])
            return [b'hello!']

        self.site.application = use_fieldstorage
        sock = eventlet.connect(self.server_addr)

        sock.sendall(b'POST / HTTP/1.1\r\n'
                     b'Host: localhost\r\n'
                     b'Connection: close\r\n'
                     b'Transfer-Encoding: chunked\r\n\r\n'
                     b'2\r\noh\r\n'
                     b'4\r\n hai\r\n0\r\n\r\n')
        assert b'hello!' in recvall(sock)

def test_020_x_forwarded_for(self):
        request_bytes = (
            b'GET / HTTP/1.1\r\nHost: localhost\r\n'
            + b'X-Forwarded-For: 1.2.3.4, 5.6.7.8\r\n\r\n'
        )

        sock = eventlet.connect(self.server_addr)
        sock.sendall(request_bytes)
        sock.recv(1024)
        sock.close()
        assert '1.2.3.4,5.6.7.8,127.0.0.1' in self.logfile.getvalue()

        # turning off the option should work too
        self.logfile = six.StringIO()
        self.spawn_server(log_x_forwarded_for=False)

        sock = eventlet.connect(self.server_addr)
        sock.sendall(request_bytes)
        sock.recv(1024)
        sock.close()
        assert '1.2.3.4' not in self.logfile.getvalue()
        assert '5.6.7.8' not in self.logfile.getvalue()
        assert '127.0.0.1' in self.logfile.getvalue()

def test_021_environ_clobbering(self):
        def clobberin_time(environ, start_response):
            for environ_var in [
                    'wsgi.version', 'wsgi.url_scheme',
                    'wsgi.input', 'wsgi.errors', 'wsgi.multithread',
                    'wsgi.multiprocess', 'wsgi.run_once', 'REQUEST_METHOD',
                    'SCRIPT_NAME', 'RAW_PATH_INFO', 'PATH_INFO', 'QUERY_STRING',
                    'CONTENT_TYPE', 'CONTENT_LENGTH', 'SERVER_NAME', 'SERVER_PORT',
                    'SERVER_PROTOCOL']:
                environ[environ_var] = None
            start_response('200 OK', [('Content-type', 'text/plain')])
            return []
        self.site.application = clobberin_time
        sock = eventlet.connect(self.server_addr)
        sock.sendall(b'GET / HTTP/1.1\r\n'
                     b'Host: localhost\r\n'
                     b'Connection: close\r\n'
                     b'\r\n\r\n')
        assert b'200 OK' in recvall(sock)

def test_023_bad_content_length(self):
        sock = eventlet.connect(self.server_addr)
        sock.sendall(b'GET / HTTP/1.0\r\nHost: localhost\r\nContent-length: argh\r\n\r\n')
        result = recvall(sock)
        assert result.startswith(b'HTTP'), result
        assert b'400 Bad Request' in result, result
        assert b'500' not in result, result

        sock = eventlet.connect(self.server_addr)
        sock.sendall(b'GET / HTTP/1.0\r\nHost: localhost\r\nContent-length:\r\n\r\n')
        result = recvall(sock)
        assert result.startswith(b'HTTP'), result
        assert b'400 Bad Request' in result, result
        assert b'500' not in result, result

        sock = eventlet.connect(self.server_addr)
        sock.sendall(b'GET / HTTP/1.0\r\nHost: localhost\r\nContent-length: \r\n\r\n')
        result = recvall(sock)
        assert result.startswith(b'HTTP'), result
        assert b'400 Bad Request' in result, result
        assert b'500' not in result, result

def test_error_in_chunked_closes_connection(self):
        # From http://rhodesmill.org/brandon/2013/chunked-wsgi/
        self.spawn_server(minimum_chunk_size=1)

        self.site.application = chunked_fail_app
        sock = eventlet.connect(self.server_addr)

        sock.sendall(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')

        result = read_http(sock)
        self.assertEqual(result.status, 'HTTP/1.1 200 OK')
        self.assertEqual(result.headers_lower.get('transfer-encoding'), 'chunked')
        expected_body = (
            b'27\r\nThe dwarves of yore made mighty spells,\r\n'
            b'25\r\nWhile hammers fell like ringing bells\r\n')
        self.assertEqual(result.body, expected_body)

        # verify that socket is closed by server
        self.assertEqual(sock.recv(1), b'')

def test_zero_length_chunked_response(self):
        def zero_chunked_app(env, start_response):
            start_response('200 OK', [('Content-type', 'text/plain')])
            yield b""

        self.site.application = zero_chunked_app
        sock = eventlet.connect(self.server_addr)

        sock.sendall(b'GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
        response = recvall(sock).split(b'\r\n')
        headers = []
        while True:
            h = response.pop(0)
            headers.append(h)
            if h == b'':
                break
        assert b'Transfer-Encoding: chunked' in b''.join(headers), headers
        # should only be one chunk of zero size with two blank lines
        # (one terminates the chunk, one terminates the body)
        self.assertEqual(response, [b'0', b'', b''])

def test_client_disconnect(self):
        """Issue #95 Server must handle disconnect from client in the middle of response
        """
        def long_response(environ, start_response):
            start_response('200 OK', [('Content-Length', '9876')])
            yield b'a' * 9876

        server_sock = eventlet.listen(('localhost', 0))
        self.server_addr = server_sock.getsockname()
        server = wsgi.Server(server_sock, server_sock.getsockname(), long_response,
                             log=self.logfile)

        def make_request():
            sock = eventlet.connect(server_sock.getsockname())
            sock.send(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
            sock.close()

        request_thread = eventlet.spawn(make_request)
        client_sock, addr = server_sock.accept()
        # Next line must not raise IOError -32 Broken pipe
        server.process_request([addr, client_sock, wsgi.STATE_IDLE])
        request_thread.wait()
        server_sock.close()

def test_disable_header_name_capitalization(self):
        # Disable HTTP header name capitalization
        #
        # https://github.com/eventlet/eventlet/issues/80
        random_case_header = ('eTAg', 'TAg-VAluE')

        def wsgi_app(environ, start_response):
            start_response('200 oK', [random_case_header])
            return [b'']

        self.spawn_server(site=wsgi_app, capitalize_response_headers=False)

        sock = eventlet.connect(self.server_addr)
        sock.sendall(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
        result = read_http(sock)
        sock.close()
        self.assertEqual(result.status, 'HTTP/1.1 200 oK')
        self.assertEqual(result.headers_lower[random_case_header[0].lower()], random_case_header[1])
        self.assertEqual(result.headers_original[random_case_header[0]], random_case_header[1])

def test_chunked_readline_wsgi_override_minimum_chunk_size(self):

        fd = self.connect()
        fd.sendall(b"POST /yield_spaces/override_min HTTP/1.1\r\nContent-Length: 0\r\n\r\n")

        resp_so_far = b''
        with eventlet.Timeout(.1):
            while True:
                one_byte = fd.recv(1)
                resp_so_far += one_byte
                if resp_so_far.endswith(b'\r\n\r\n'):
                    break
            self.assertEqual(fd.recv(1), b' ')
        try:
            with eventlet.Timeout(.1):
                fd.recv(1)
        except eventlet.Timeout:
            pass
        else:
            assert False
        self.yield_next_space = True

        with eventlet.Timeout(.1):
            self.assertEqual(fd.recv(1), b' ')

def test_chunked_readline_wsgi_not_override_minimum_chunk_size(self):

        fd = self.connect()
        fd.sendall(b"POST /yield_spaces HTTP/1.1\r\nContent-Length: 0\r\n\r\n")

        resp_so_far = b''
        try:
            with eventlet.Timeout(.1):
                while True:
                    one_byte = fd.recv(1)
                    resp_so_far += one_byte
                    if resp_so_far.endswith(b'\r\n\r\n'):
                        break
                self.assertEqual(fd.recv(1), b' ')
        except eventlet.Timeout:
            pass
        else:
            assert False

def test_connect_tcp(self):
        def accept_once(listenfd):
            try:
                conn, addr = listenfd.accept()
                fd = conn.makefile(mode='wb')
                conn.close()
                fd.write(b'hello\n')
                fd.close()
            finally:
                listenfd.close()

        server = eventlet.listen(('0.0.0.0', 0))
        eventlet.spawn_n(accept_once, server)

        client = eventlet.connect(('127.0.0.1', server.getsockname()[1]))
        fd = client.makefile('rb')
        client.close()
        assert fd.readline() == b'hello\n'
        assert fd.read() == b''
        fd.close()

        check_hub()

def test_001_trampoline_timeout(self):
        server_sock = eventlet.listen(('127.0.0.1', 0))
        bound_port = server_sock.getsockname()[1]

        def server(sock):
            client, addr = sock.accept()
            eventlet.sleep(0.1)
        server_evt = eventlet.spawn(server, server_sock)
        eventlet.sleep(0)
        try:
            desc = eventlet.connect(('127.0.0.1', bound_port))
            hubs.trampoline(desc, read=True, write=False, timeout=0.001)
        except eventlet.Timeout:
            pass  # test passed
        else:
            assert False, "Didn't timeout"

        server_evt.wait()
        check_hub()

def test_concurrency(self):
        evt = eventlet.Event()

        def waiter(sock, addr):
            sock.sendall(b'hi')
            evt.wait()
        l = eventlet.listen(('localhost', 0))
        eventlet.spawn(eventlet.serve, l, waiter, 5)

        def test_client():
            c = eventlet.connect(('localhost', l.getsockname()[1]))
            # verify the client is connected by getting data
            self.assertEqual(b'hi', c.recv(2))
            return c
        [test_client() for i in range(5)]
        # very next client should not get anything
        x = eventlet.with_timeout(
            0.01,
            test_client,
            timeout_value="timed out")
        self.assertEqual(x, "timed out")

def test_correct_upgrade_request_13(self):
        for http_connection in ['Upgrade', 'UpGrAdE', 'keep-alive, Upgrade']:
            connect = [
                "GET /echo HTTP/1.1",
                "Upgrade: websocket",
                "Connection: %s" % http_connection,
                "Host: %s:%s" % self.server_addr,
                "Origin: http://%s:%s" % self.server_addr,
                "Sec-WebSocket-Version: 13",
                "Sec-WebSocket-Key: d9MXuOzlVQ0h+qRllvSCIg==",
            ]
            sock = eventlet.connect(self.server_addr)

            sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n'))
            result = sock.recv(1024)
            # The server responds the correct Websocket handshake
            print('Connection string: %r' % http_connection)
            self.assertEqual(result, six.b('\r\n'.join([
                'HTTP/1.1 101 Switching Protocols',
                'Upgrade: websocket',
                'Connection: Upgrade',
                'Sec-WebSocket-Accept: ywSyWXCPNsDxLrQdQrn5RFNRfBU=\r\n\r\n',
            ])))

def test_send_recv_13(self):
        connect = [
            "GET /echo HTTP/1.1",
            "Upgrade: websocket",
            "Connection: Upgrade",
            "Host: %s:%s" % self.server_addr,
            "Origin: http://%s:%s" % self.server_addr,
            "Sec-WebSocket-Version: 13",
            "Sec-WebSocket-Key: d9MXuOzlVQ0h+qRllvSCIg==",
        ]
        sock = eventlet.connect(self.server_addr)
        sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n'))
        sock.recv(1024)
        ws = websocket.RFC6455WebSocket(sock, {}, client=True)
        ws.send(b'hello')
        assert ws.wait() == b'hello'
        ws.send(b'hello world!\x01')
        ws.send(u'hello world again!')
        assert ws.wait() == b'hello world!\x01'
        assert ws.wait() == u'hello world again!'
        ws.close()
        eventlet.sleep(0.01)

def connect_to(self,endpoint):
       """ Use this to connect to the server if an endpoint is not passed in __init__
       """
       self.tcp_sock            = eventlet.connect(endpoint)
       self.cipher              = crypto.Cipher()
       self.recv_buff           = buffer.Buffer()
       self.compression_enabled = False
       self.protocol_mode       = 0 # did you really think it made sense to set this to anything else you maniac?
       self.send_q              = eventlet.queue.LightQueue(0)
       self.ready               = True
       self.pool.spawn_n(self.read_thread)
       self.pool.spawn_n(self.send_thread)

def test_connect_timeout(self):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(0.1)
        gs = greenio.GreenSocket(s)

        try:
            expect_socket_timeout(gs.connect, ('192.0.2.1', 80))
        except socket.error as e:
            # unreachable is also a valid outcome
            if not get_errno(e) in (errno.EHOSTUNREACH, errno.ENETUNREACH):
                raise

def test_recv_timeout(self):
        listener = greenio.GreenSocket(socket.socket())
        listener.bind(('', 0))
        listener.listen(50)

        evt = event.Event()

        def server():
            # accept the connection in another greenlet
            sock, addr = listener.accept()
            evt.wait()

        gt = eventlet.spawn(server)

        addr = listener.getsockname()

        client = greenio.GreenSocket(socket.socket())
        client.settimeout(0.1)

        client.connect(addr)

        expect_socket_timeout(client.recv, 0)
        expect_socket_timeout(client.recv, 8192)

        evt.send()
        gt.wait()

def test_send_timeout(self):
        self.reset_timeout(2)
        listener = bufsized(eventlet.listen(('', 0)))

        evt = event.Event()

        def server():
            # accept the connection in another greenlet
            sock, addr = listener.accept()
            sock = bufsized(sock)
            evt.wait()

        gt = eventlet.spawn(server)

        addr = listener.getsockname()

        client = bufsized(greenio.GreenSocket(socket.socket()))
        client.connect(addr)

        client.settimeout(0.00001)
        msg = b"A" * 100000  # large enough number to overwhelm most buffers

        # want to exceed the size of the OS buffer so it'll block in a
        # single send
        def send():
            for x in range(10):
                client.send(msg)

        expect_socket_timeout(send)

        evt.send()
        gt.wait()

def test_sendall_timeout(self):
        listener = greenio.GreenSocket(socket.socket())
        listener.bind(('', 0))
        listener.listen(50)

        evt = event.Event()

        def server():
            # accept the connection in another greenlet
            sock, addr = listener.accept()
            evt.wait()

        gt = eventlet.spawn(server)

        addr = listener.getsockname()

        client = greenio.GreenSocket(socket.socket())
        client.settimeout(0.1)
        client.connect(addr)

        # want to exceed the size of the OS buffer so it'll block
        msg = b"A" * (8 << 20)
        expect_socket_timeout(client.sendall, msg)

        evt.send()
        gt.wait()

def test_del_closes_socket(self):
        def accept_once(listener):
            # delete/overwrite the original conn
            # object, only keeping the file object around
            # closing the file object should close everything
            try:
                conn, addr = listener.accept()
                conn = conn.makefile('wb')
                conn.write(b'hello\n')
                conn.close()
                gc.collect()
                self.assertWriteToClosedFileRaises(conn)
            finally:
                listener.close()

        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind(('127.0.0.1', 0))
        server.listen(50)
        killer = eventlet.spawn(accept_once, server)
        client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        client.connect(('127.0.0.1', server.getsockname()[1]))
        fd = client.makefile('rb')
        client.close()
        assert fd.read() == b'hello\n'
        assert fd.read() == b''

        killer.wait()

def test_blocking_accept_mark_as_reopened(self):
        evt_hub = get_hub()
        with mock.patch.object(evt_hub, "mark_as_reopened") as patched_mark_as_reopened:
            def connect_once(listener):
                # delete/overwrite the original conn
                # object, only keeping the file object around
                # closing the file object should close everything
                client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                client.connect(('127.0.0.1', listener.getsockname()[1]))
                client.close()

            server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            server.bind(('127.0.0.1', 0))
            server.listen(50)
            acceptlet = eventlet.spawn(connect_once, server)
            conn, addr = server.accept()
            conn.sendall(b'hello\n')
            connfileno = conn.fileno()
            conn.close()
            assert patched_mark_as_reopened.called
            assert patched_mark_as_reopened.call_count == 3, "3 fds were opened, but the hub was " \
                                                             "only notified {call_count} times" \
                .format(call_count=patched_mark_as_reopened.call_count)
            args, kwargs = patched_mark_as_reopened.call_args
            assert args == (connfileno,), "Expected mark_as_reopened to be called " \
                                          "with {expected_fileno}, but it was called " \
                                          "with {fileno}".format(expected_fileno=connfileno,
                                                                 fileno=args[0])
        server.close()

def test_full_duplex(self):
        large_data = b'*' * 10 * min_buf_size()
        listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        listener.bind(('127.0.0.1', 0))
        listener.listen(50)
        bufsized(listener)

        def send_large(sock):
            sock.sendall(large_data)

        def read_large(sock):
            result = sock.recv(len(large_data))
            while len(result) < len(large_data):
                result += sock.recv(len(large_data))
            self.assertEqual(result, large_data)

        def server():
            (sock, addr) = listener.accept()
            sock = bufsized(sock)
            send_large_coro = eventlet.spawn(send_large, sock)
            eventlet.sleep(0)
            result = sock.recv(10)
            expected = b'hello world'
            while len(result) < len(expected):
                result += sock.recv(10)
            self.assertEqual(result, expected)
            send_large_coro.wait()

        server_evt = eventlet.spawn(server)
        client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        client.connect(('127.0.0.1', listener.getsockname()[1]))
        bufsized(client)
        large_evt = eventlet.spawn(read_large, client)
        eventlet.sleep(0)
        client.sendall(b'hello world')
        server_evt.wait()
        large_evt.wait()
        client.close()

def test_timeout_and_final_write(self):
        # This test verifies that a write on a socket that we've
        # stopped listening for doesn't result in an incorrect switch
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind(('127.0.0.1', 0))
        server.listen(50)
        bound_port = server.getsockname()[1]

        def sender(evt):
            s2, addr = server.accept()
            wrap_wfile = s2.makefile('wb')

            eventlet.sleep(0.02)
            wrap_wfile.write(b'hi')
            s2.close()
            evt.send(b'sent via event')

        evt = event.Event()
        eventlet.spawn(sender, evt)
        # lets the socket enter accept mode, which
        # is necessary for connect to succeed on windows
        eventlet.sleep(0)
        try:
            # try and get some data off of this pipe
            # but bail before any is sent
            eventlet.Timeout(0.01)
            client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            client.connect(('127.0.0.1', bound_port))
            wrap_rfile = client.makefile()
            wrap_rfile.read(1)
            self.fail()
        except eventlet.Timeout:
            pass

        result = evt.wait()
        self.assertEqual(result, b'sent via event')
        server.close()
        client.close()

def test_closure(self):
        def spam_to_me(address):
            sock = eventlet.connect(address)
            while True:
                try:
                    sock.sendall(b'hello world')
                    # Arbitrary delay to not use all available CPU, keeps the test
                    # running quickly and reliably under a second
                    time.sleep(0.001)
                except socket.error as e:
                    if get_errno(e) == errno.EPIPE:
                        return
                    raise

        server = eventlet.listen(('127.0.0.1', 0))
        sender = eventlet.spawn(spam_to_me, server.getsockname())
        client, address = server.accept()
        server.close()

        def reader():
            try:
                while True:
                    data = client.recv(1024)
                    assert data
                    # Arbitrary delay to not use all available CPU, keeps the test
                    # running quickly and reliably under a second
                    time.sleep(0.001)
            except socket.error as e:
                # we get an EBADF because client is closed in the same process
                # (but a different greenthread)
                if get_errno(e) != errno.EBADF:
                    raise

        def closer():
            client.close()

        reader = eventlet.spawn(reader)
        eventlet.spawn_n(closer)
        reader.wait()
        sender.wait()

def test_partial_write_295():
    # https://github.com/eventlet/eventlet/issues/295
    # `socket.makefile('w').writelines()` must send all
    # despite partial writes by underlying socket
    listen_socket = eventlet.listen(('localhost', 0))
    original_accept = listen_socket.accept

    def talk(conn):
        f = conn.makefile('wb')
        line = b'*' * 2140
        f.writelines([line] * 10000)
        conn.close()

    def accept():
        connection, address = original_accept()
        original_send = connection.send

        def slow_send(b, *args):
            b = b[:1031]
            return original_send(b, *args)

        connection.send = slow_send
        eventlet.spawn(talk, connection)
        return connection, address

    listen_socket.accept = accept

    eventlet.spawn(listen_socket.accept)
    sock = eventlet.connect(listen_socket.getsockname())
    with eventlet.Timeout(10):
        bs = sock.makefile('rb').read()
    assert len(bs) == 21400000
    assert bs == (b'*' * 21400000)