test_ssl.py 文件源码

def ssl_echo_serve_sync(sock, *, expect_fail=False):
    try:
        wrapped = SERVER_CTX.wrap_socket(
            sock, server_side=True, suppress_ragged_eofs=False
        )
        wrapped.do_handshake()
        while True:
            data = wrapped.recv(4096)
            if not data:
                # other side has initiated a graceful shutdown; we try to
                # respond in kind but it's legal for them to have already gone
                # away.
                exceptions = (BrokenPipeError,)
                # Under unclear conditions, CPython sometimes raises
                # SSLWantWriteError here. This is a bug (bpo-32219), but it's
                # not our bug, so ignore it.
                exceptions += (stdlib_ssl.SSLWantWriteError,)
                if WORKAROUND_PYPY_BUG:
                    exceptions += (stdlib_ssl.SSLEOFError,)
                try:
                    wrapped.unwrap()
                except exceptions:
                    pass
                return
            wrapped.sendall(data)
    except Exception as exc:
        if expect_fail:
            print("ssl_echo_serve_sync got error as expected:", exc)
        else:  # pragma: no cover
            raise
    else:
        if expect_fail:  # pragma: no cover
            raise RuntimeError("failed to fail?")


# Fixture that gives a raw socket connected to a trio-test-1 echo server
# (running in a thread). Useful for testing making connections with different
# SSLContexts.
#
# This way of writing it is pretty janky, with the nursery hidden inside the
# fixture and no proper parental supervision. Don't copy this code; it was
# written this way before we knew better.