1from .. import socket as tsocket
2from .._highlevel_socket import SocketStream
3
4
5async def open_stream_to_socket_listener(socket_listener):
6    """Connect to the given :class:`~trio.SocketListener`.
7
8    This is particularly useful in tests when you want to let a server pick
9    its own port, and then connect to it::
10
11        listeners = await trio.open_tcp_listeners(0)
12        client = await trio.testing.open_stream_to_socket_listener(listeners[0])
13
14    Args:
15      socket_listener (~trio.SocketListener): The
16          :class:`~trio.SocketListener` to connect to.
17
18    Returns:
19      SocketStream: a stream connected to the given listener.
20
21    """
22    family = socket_listener.socket.family
23    sockaddr = socket_listener.socket.getsockname()
24    if family in (tsocket.AF_INET, tsocket.AF_INET6):
25        sockaddr = list(sockaddr)
26        if sockaddr[0] == "0.0.0.0":
27            sockaddr[0] = "127.0.0.1"
28        if sockaddr[0] == "::":
29            sockaddr[0] = "::1"
30        sockaddr = tuple(sockaddr)
31
32    sock = tsocket.socket(family=family)
33    await sock.connect(sockaddr)
34    return SocketStream(sock)
35