Fix test

Author: CoolCat467

src/trio/__init__.py

@@ -64,6 +64,7 @@
 )
 from ._highlevel_open_tcp_stream import open_tcp_stream as open_tcp_stream
 from ._highlevel_open_unix_listeners import (
+    UnixSocketListener as UnixSocketListener,
     open_unix_listeners as open_unix_listeners,
     serve_unix as serve_unix,
 )
@@ -72,7 +73,6 @@
 from ._highlevel_socket import (
     SocketListener as SocketListener,
     SocketStream as SocketStream,
-    UnixSocketListener as UnixSocketListener,
 )
 from ._highlevel_ssl_helpers import (
     open_ssl_over_tcp_listeners as open_ssl_over_tcp_listeners,

src/trio/_tests/test_highlevel_open_unix_listeners.py

@@ -3,32 +3,66 @@
 import sys
 from typing import TYPE_CHECKING
 
+import pytest
+
 import trio.socket as tsocket
-from trio import SocketListener
 from trio._highlevel_open_unix_listeners import (
+    UnixSocketListener,
     open_unix_listeners,
 )
-from trio.testing import open_stream_to_socket_listener
+from trio._highlevel_socket import SocketStream
+
+assert not TYPE_CHECKING or sys.platform != "win32"
 
-assert (  # Skip type checking when on Windows
-    sys.platform != "win32" or not TYPE_CHECKING
+skip_if_not_unix = pytest.mark.skipif(
+    not hasattr(tsocket, "AF_UNIX"),
+    reason="Needs unix socket support",
 )
 
 
+async def open_stream_to_unix_socket_listener(
+    socket_listener: UnixSocketListener,
+    sockaddr: str,
+) -> SocketStream:
+    """Connect to the given :class:`~trio.UnixSocketListener`.
+
+    This is particularly useful in tests when you want to let a server pick
+    its own port, and then connect to it::
+
+        listeners = await trio.open_tcp_listeners(0)
+        client = await trio.testing.open_stream_to_socket_listener(listeners[0])
+
+    Args:
+      socket_listener (~trio.UnixSocketListener): The
+          :class:`~trio.UnixSocketListener` to connect to.
+
+    Returns:
+      SocketStream: a stream connected to the given listener.
+
+    """
+    family = socket_listener.socket.family
+    assert family == tsocket.AF_UNIX
+
+    sock = tsocket.socket(family=family)
+    await sock.connect(sockaddr)
+    return SocketStream(sock)
+
+
+@skip_if_not_unix
 async def test_open_unix_listeners_basic() -> None:
     # Since we are on unix, we can use fun things like /tmp
-    listeners = await open_unix_listeners("/tmp/test_socket.sock", backlog=0)
+    path = "/tmp/test_socket.sock"
+    listeners = await open_unix_listeners(path, backlog=0)
     assert isinstance(listeners, list)
     for obj in listeners:
-        assert isinstance(obj, SocketListener)
-        # Binds to wildcard address by default
-        assert obj.socket.family in [tsocket.AF_INET, tsocket.AF_INET6]
-        assert obj.socket.getsockname()[0] in ["0.0.0.0", "::"]
+        assert obj.socket.family == tsocket.AF_UNIX
+        # Does not work because of atomic overwrite
+        # assert obj.socket.getsockname() == path
 
     listener = listeners[0]
     # Make sure the backlog is at least 2
-    c1 = await open_stream_to_socket_listener(listener)
-    c2 = await open_stream_to_socket_listener(listener)
+    c1 = await open_stream_to_unix_socket_listener(listener, path)
+    c2 = await open_stream_to_unix_socket_listener(listener, path)
 
     s1 = await listener.accept()
     s2 = await listener.accept()