Allow create_connection & create_server accept AF_UNIX

Internally, they will delegate to create_unix_connection and
create_unix_server.

Author: 1st1

tests/test_tcp.py

@@ -808,20 +808,20 @@ def client():
 
     @unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'no Unix sockets')
     def test_create_connection_wrong_sock(self):
-        sock = socket.socket(socket.AF_UNIX)
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         with sock:
             coro = self.loop.create_connection(MyBaseProto, sock=sock)
             with self.assertRaisesRegex(ValueError,
-                                        'A TCP Stream Socket was expected'):
+                                        'A Stream Socket was expected'):
                 self.loop.run_until_complete(coro)
 
     @unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'no Unix sockets')
     def test_create_server_wrong_sock(self):
-        sock = socket.socket(socket.AF_UNIX)
+        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         with sock:
             coro = self.loop.create_server(MyBaseProto, sock=sock)
             with self.assertRaisesRegex(ValueError,
-                                        'A TCP Stream Socket was expected'):
+                                        'A Stream Socket was expected'):
                 self.loop.run_until_complete(coro)
 
     @unittest.skipUnless(hasattr(socket, 'SOCK_NONBLOCK'),

tests/test_unix.py

@@ -83,7 +83,7 @@ async def start_server():
                 # asyncio doesn't cleanup the sock file
                 self.assertTrue(os.path.exists(sock_name))
 
-        async def start_server_sock():
+        async def start_server_sock(start_server):
             nonlocal CNT
             CNT = 0
 
@@ -92,11 +92,7 @@ async def start_server_sock():
                 sock = socket.socket(socket.AF_UNIX)
                 sock.bind(sock_name)
 
-                srv = await asyncio.start_unix_server(
-                    handle_client,
-                    None,
-                    loop=self.loop,
-                    sock=sock)
+                srv = await start_server(sock)
 
                 try:
                     srv_socks = srv.sockets
@@ -121,11 +117,27 @@ async def start_server_sock():
                 # asyncio doesn't cleanup the sock file
                 self.assertTrue(os.path.exists(sock_name))
 
-        self.loop.run_until_complete(start_server())
-        self.assertEqual(CNT, TOTAL_CNT)
+        with self.subTest(func='start_unix_server(host, port)'):
+            self.loop.run_until_complete(start_server())
+            self.assertEqual(CNT, TOTAL_CNT)
 
-        self.loop.run_until_complete(start_server_sock())
-        self.assertEqual(CNT, TOTAL_CNT)
+        with self.subTest(func='start_unix_server(sock)'):
+            self.loop.run_until_complete(start_server_sock(
+                 lambda sock: asyncio.start_unix_server(
+                    handle_client,
+                    None,
+                    loop=self.loop,
+                    sock=sock)))
+            self.assertEqual(CNT, TOTAL_CNT)
+
+        with self.subTest(func='start_server(sock)'):
+            self.loop.run_until_complete(start_server_sock(
+                 lambda sock: asyncio.start_server(
+                    handle_client,
+                    None, None,
+                    loop=self.loop,
+                    sock=sock)))
+            self.assertEqual(CNT, TOTAL_CNT)
 
     def test_create_unix_server_2(self):
         with tempfile.TemporaryDirectory() as td:
@@ -187,6 +199,24 @@ async def client_2(addr):
 
             writer.close()
 
+        async def client_3(addr):
+            sock = socket.socket(socket.AF_UNIX)
+            sock.connect(addr)
+            reader, writer = await asyncio.open_connection(
+                sock=sock,
+                loop=self.loop)
+
+            writer.write(b'AAAA')
+            self.assertEqual(await reader.readexactly(2), b'OK')
+
+            writer.write(b'BBBB')
+            self.assertEqual(await reader.readexactly(4), b'SPAM')
+
+            nonlocal CNT
+            CNT += 1
+
+            writer.close()
+
         def run(coro):
             nonlocal CNT
             CNT = 0
@@ -208,6 +238,7 @@ def run(coro):
 
         run(client)
         run(client_2)
+        run(client_3)
 
     def test_create_unix_connection_2(self):
         with tempfile.NamedTemporaryFile() as tmp:

uvloop/loop.pyx

@@ -1,6 +1,6 @@
 # cython: language_level=3, embedsignature=True
 
-
+import asyncio
 cimport cython
 
 from .includes.debug cimport UVLOOP_DEBUG
@@ -38,10 +38,6 @@ include "includes/stdlib.pxi"
 include "errors.pyx"
 
 
-cdef _is_sock_ip(sock_family):
-    return sock_family == uv.AF_INET or sock_family == uv.AF_INET6
-
-
 cdef _is_sock_stream(sock_type):
     # Linux's socket.type is a bitmask that can include extra info
     # about socket, therefore we can't do simple
@@ -1298,7 +1294,16 @@ cdef class Loop:
         cdef:
             TCPServer tcp
             system.addrinfo *addrinfo
-            Server server = Server(self)
+            Server server
+
+        if sock is not None and sock.family == uv.AF_UNIX:
+            if host is not None or port is not None:
+                raise ValueError(
+                    'host/port and sock can not be specified at the same time')
+            return await self.create_unix_server(
+                protocol_factory, sock=sock, ssl=ssl)
+
+        server = Server(self)
 
         if ssl is not None and not isinstance(ssl, ssl_SSLContext):
             raise TypeError('ssl argument must be an SSLContext or None')
@@ -1349,10 +1354,10 @@ cdef class Loop:
         else:
             if sock is None:
                 raise ValueError('Neither host/port nor sock were specified')
-            if (not _is_sock_stream(sock.type) or
-                    not _is_sock_ip(sock.family)):
+            if not _is_sock_stream(sock.type):
                 raise ValueError(
-                    'A TCP Stream Socket was expected, got {!r}'.format(sock))
+                    'A Stream Socket was expected, got {!r}'.format(sock))
+
             tcp = TCPServer.new(self, protocol_factory, server, ssl,
                                 uv.AF_UNSPEC)
 
@@ -1404,6 +1409,14 @@ cdef class Loop:
             object protocol
             object ssl_waiter
 
+        if sock is not None and sock.family == uv.AF_UNIX:
+            if host is not None or port is not None:
+                raise ValueError(
+                    'host/port and sock can not be specified at the same time')
+            return await self.create_unix_connection(
+                protocol_factory, None,
+                sock=sock, ssl=ssl, server_hostname=server_hostname)
+
         app_protocol = protocol = protocol_factory()
         ssl_waiter = None
         if ssl:
@@ -1423,6 +1436,10 @@ cdef class Loop:
                 raise ValueError('server_hostname is only meaningful with ssl')
 
         if host is not None or port is not None:
+            if sock is not None:
+                raise ValueError(
+                    'host/port and sock can not be specified at the same time')
+
             fs = []
             f1 = f2 = None
 
@@ -1536,10 +1553,9 @@ cdef class Loop:
             if sock is None:
                 raise ValueError(
                     'host and port was not specified and no sock specified')
-            if (not _is_sock_stream(sock.type) or
-                    not _is_sock_ip(sock.family)):
+            if not _is_sock_stream(sock.type):
                 raise ValueError(
-                    'A TCP Stream Socket was expected, got {!r}'.format(sock))
+                    'A Stream Socket was expected, got {!r}'.format(sock))
 
             waiter = self._new_future()
             tr = TCPTransport.new(self, protocol, None, waiter)