Function run_udp_echo_server

Lib/test/test_asyncio/utils.py:282–298

(*, host='127.0.0.1', port=0)

@contextlib.contextmanager
def run_udp_echo_server(*, host='127.0.0.1', port=0):
    addr_info = socket.getaddrinfo(host, port, type=socket.SOCK_DGRAM)
    family, type, proto, _, sockaddr = addr_info[0]
    sock = socket.socket(family, type, proto)
    sock.bind((host, port))
    sockname = sock.getsockname()
    thread = threading.Thread(target=lambda: echo_datagrams(sock))
    thread.start()
    try:
        yield sockname
    finally:
        # gh-122187: use a separate socket to send the stop message to avoid
        # TSan reported race on the same socket.
        sock2 = socket.socket(family, type, proto)
        sock2.sendto(b'STOP', sockname)
        sock2.close()
        thread.join()


def make_test_protocol(base):