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