wsgi_connection_timeout.py 文件源码

def step(debug):
    output_buffer[:] = []

    server_sock = eventlet.listen(('localhost', 0))
    server_addr = server_sock.getsockname()
    sock_wrap = NaughtySocketAcceptWrap(server_sock)

    eventlet.spawn_n(
        eventlet.wsgi.server,
        debug=debug,
        log=BufferLog,
        max_size=128,
        site=tests.wsgi_test.Site(),
        sock=server_sock,
    )

    try:
        # req #1 - normal
        sock1 = eventlet.connect(server_addr)
        sock1.settimeout(0.1)
        fd1 = sock1.makefile('rwb')
        fd1.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
        fd1.flush()
        tests.wsgi_test.read_http(sock1)

        # let the server socket ops catch up, set bomb
        eventlet.sleep(0)
        output_buffer.append("arming...")
        sock_wrap.arm()

        # req #2 - old conn, post-arm - timeout
        fd1.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
        fd1.flush()
        try:
            tests.wsgi_test.read_http(sock1)
            assert False, 'Expected ConnectionClosed exception'
        except tests.wsgi_test.ConnectionClosed:
            pass

        fd1.close()
        sock1.close()
    finally:
        # reset streams, then output trapped tracebacks
        sock_wrap.unwrap()
    # check output asserts in tests.wsgi_test.TestHttpd
    # test_143_server_connection_timeout_exception

    return output_buffer[:]