test_ssl.py 文件源码

def ssl_echo_server_raw(**kwargs):
    a, b = stdlib_socket.socketpair()
    async with trio.open_nursery() as nursery:
        # Exiting the 'with a, b' context manager closes the sockets, which
        # causes the thread to exit (possibly with an error), which allows the
        # nursery context manager to exit too.
        with a, b:
            nursery.start_soon(
                trio.run_sync_in_worker_thread,
                partial(ssl_echo_serve_sync, b, **kwargs)
            )

            await yield_(SocketStream(tsocket.from_stdlib_socket(a)))


# Fixture that gives a properly set up SSLStream connected to a trio-test-1
# echo server (running in a thread)