Function _run_test_server

Lib/test/test_asyncio/utils.py:165–197

(*, address, use_ssl=False, server_cls, server_ssl_cls)

def _run_test_server(*, address, use_ssl=False, server_cls, server_ssl_cls):

    def loop(environ):
        size = int(environ['CONTENT_LENGTH'])
        while size:
            data = environ['wsgi.input'].read(min(size, 0x10000))
            yield data
            size -= len(data)

    def app(environ, start_response):
        status = '200 OK'
        headers = [('Content-type', 'text/plain')]
        start_response(status, headers)
        if environ['PATH_INFO'] == '/loop':
            return loop(environ)
        else:
            return [b'Test message']

    # Run the test WSGI server in a separate thread in order not to
    # interfere with event handling in the main thread
    server_class = server_ssl_cls if use_ssl else server_cls
    httpd = server_class(address, SilentWSGIRequestHandler)
    httpd.set_app(app)
    httpd.address = httpd.server_address
    server_thread = threading.Thread(
        target=lambda: httpd.serve_forever(poll_interval=0.05))
    server_thread.start()
    try:
        yield httpd
    finally:
        httpd.shutdown()
        httpd.server_close()
        server_thread.join()


if hasattr(socket, 'AF_UNIX'):