test_wsgi.py 文件源码

def _setup_wsgi_server():
    stop_event = multiprocessing.Event()
    process = multiprocessing.Process(
        target=_run_server,
        args=(stop_event,)
    )

    process.start()

    # NOTE(kgriffs): Let the server start up
    time.sleep(0.2)

    yield

    stop_event.set()

    # NOTE(kgriffs): Pump the request handler loop in case execution
    # made it to the next server.handle_request() before we sent the
    # event.
    try:
        requests.get(_SERVER_BASE_URL)
    except Exception:
        pass  # Thread already exited

    process.join()