Merge branch 'master' into fix/accept-pathlike-subprocess-args

Author: achimnol

MANIFEST.in

@@ -6,4 +6,4 @@ recursive-include vendor/libuv *
 recursive-exclude vendor/libuv/.git *
 recursive-exclude vendor/libuv/docs *
 recursive-exclude vendor/libuv/img *
-include LICENSE-MIT LICENSE-APACHE README.rst Makefile performance.png
+include LICENSE-MIT LICENSE-APACHE README.rst Makefile performance.png .flake8

tests/test_base.py

@@ -541,6 +541,46 @@ async def coro():
         self.assertFalse(isinstance(task, MyTask))
         self.loop.run_until_complete(task)
 
+    def test_set_task_name(self):
+        if self.implementation == 'asyncio' and sys.version_info < (3, 8, 0):
+            raise unittest.SkipTest('unsupported task name')
+
+        self.loop._process_events = mock.Mock()
+
+        result = None
+
+        class MyTask(asyncio.Task):
+            def set_name(self, name):
+                nonlocal result
+                result = name + "!"
+
+            def get_name(self):
+                return result
+
+        async def coro():
+            pass
+
+        factory = lambda loop, coro: MyTask(coro, loop=loop)
+
+        self.assertIsNone(self.loop.get_task_factory())
+        task = self.loop.create_task(coro(), name="mytask")
+        self.assertFalse(isinstance(task, MyTask))
+        if sys.version_info >= (3, 8, 0):
+            self.assertEqual(task.get_name(), "mytask")
+        self.loop.run_until_complete(task)
+
+        self.loop.set_task_factory(factory)
+        self.assertIs(self.loop.get_task_factory(), factory)
+
+        task = self.loop.create_task(coro(), name="mytask")
+        self.assertTrue(isinstance(task, MyTask))
+        self.assertEqual(result, "mytask!")
+        self.assertEqual(task.get_name(), "mytask!")
+        self.loop.run_until_complete(task)
+
+        self.loop.set_task_factory(None)
+        self.assertIsNone(self.loop.get_task_factory())
+
     def _compile_agen(self, src):
         try:
             g = {}

tests/test_udp.py

@@ -56,7 +56,8 @@ def datagram_received(self, data, addr):
 
         s_transport, server = self.loop.run_until_complete(coro)
 
-        host, port, *_ = s_transport.get_extra_info('sockname')
+        remote_addr = s_transport.get_extra_info('sockname')
+        host, port, *_ = remote_addr
 
         self.assertIsInstance(server, TestMyDatagramProto)
         self.assertEqual('INITIALIZED', server.state)
@@ -86,6 +87,36 @@ def datagram_received(self, data, addr):
         # received
         self.assertEqual(8, client.nbytes)
 
+        # https://github.com/MagicStack/uvloop/issues/319
+        # uvloop should behave the same as asyncio when given remote_addr
+        transport.sendto(b'xxx', remote_addr)
+        tb.run_until(
+            self.loop, lambda: server.nbytes > 3 or client.done.done())
+        self.assertEqual(6, server.nbytes)
+        tb.run_until(self.loop, lambda: client.nbytes > 8)
+
+        # received
+        self.assertEqual(16, client.nbytes)
+
+        # reject sendto with a different port
+        with self.assertRaisesRegex(
+            ValueError, "Invalid address.*" + repr(remote_addr)
+        ):
+            bad_addr = list(remote_addr)
+            bad_addr[1] += 1
+            bad_addr = tuple(bad_addr)
+            transport.sendto(b"xxx", bad_addr)
+
+        # reject sento with unresolved hostname
+        if remote_addr[0] != lc_addr[0]:
+            with self.assertRaisesRegex(
+                ValueError, "Invalid address.*" + repr(remote_addr)
+            ):
+                bad_addr = list(remote_addr)
+                bad_addr[0] = lc_addr[0]
+                bad_addr = tuple(bad_addr)
+                transport.sendto(b"xxx", bad_addr)
+
         # extra info is available
         self.assertIsNotNone(transport.get_extra_info('sockname'))
 
@@ -100,8 +131,11 @@ def test_create_datagram_endpoint_addrs_ipv4(self):
         self._test_create_datagram_endpoint_addrs(
             socket.AF_INET, ('127.0.0.1', 0))
 
-    @unittest.skipUnless(tb.has_IPv6, 'no IPv6')
-    def test_create_datagram_endpoint_addrs_ipv6(self):
+    def test_create_datagram_endpoint_addrs_ipv4_nameaddr(self):
+        self._test_create_datagram_endpoint_addrs(
+            socket.AF_INET, ('localhost', 0))
+
+    def _test_create_datagram_endpoint_addrs_ipv6(self):
         self._test_create_datagram_endpoint_addrs(
             socket.AF_INET6, ('::1', 0))
 
@@ -250,6 +284,50 @@ async def run():
 
         self.loop.run_until_complete(run())
 
+    def test_socketpair(self):
+        peername = asyncio.Future(loop=self.loop)
+
+        class Proto(MyDatagramProto):
+            def datagram_received(self, data, addr):
+                super().datagram_received(data, addr)
+                peername.set_result(addr)
+
+        s1, s2 = socket.socketpair(socket.AF_UNIX, socket.SOCK_DGRAM, 0)
+
+        with s1, s2:
+            try:
+                f = self.loop.create_datagram_endpoint(
+                    lambda: Proto(loop=self.loop), sock=s1)
+            except TypeError as ex:
+                # asyncio in 3.5.0 doesn't have the 'sock' argument
+                if 'got an unexpected keyword argument' not in ex.args[0]:
+                    raise
+            else:
+                tr, pr = self.loop.run_until_complete(f)
+                self.assertIsInstance(pr, Proto)
+
+                s2.send(b'hello, socketpair')
+                addr = self.loop.run_until_complete(
+                    asyncio.wait_for(peername, 1, loop=self.loop))
+                if sys.platform.startswith('linux'):
+                    self.assertEqual(addr, None)
+                else:
+                    self.assertEqual(addr, '')
+                self.assertEqual(pr.nbytes, 17)
+
+                if not self.is_asyncio_loop():
+                    # asyncio doesn't support sendto(xx) on UDP sockets
+                    # https://git.io/Jfqbw
+                    data = b'from uvloop'
+                    tr.sendto(data)
+                    result = self.loop.run_until_complete(asyncio.wait_for(
+                        self.loop.run_in_executor(None, s2.recv, 1024),
+                        1, loop=self.loop))
+                    self.assertEqual(data, result)
+
+                tr.close()
+                self.loop.run_until_complete(pr.done)
+
 
 class Test_UV_UDP(_TestUDP, tb.UVTestCase):
 
@@ -291,6 +369,16 @@ def test_send_after_close(self):
         self.loop.run_until_complete(asyncio.sleep(0.01))
         s_transport.sendto(b'aaaa', ('127.0.0.1', 80))
 
+    @unittest.skipUnless(tb.has_IPv6, 'no IPv6')
+    def test_create_datagram_endpoint_addrs_ipv6(self):
+        self._test_create_datagram_endpoint_addrs_ipv6()
+
 
 class Test_AIO_UDP(_TestUDP, tb.AIOTestCase):
-    pass
+    @unittest.skipUnless(tb.has_IPv6, 'no IPv6')
+    @unittest.skipIf(
+        sys.version_info[:3] < (3, 6, 7) or sys.version_info[:3] == (3, 7, 0),
+        'bpo-27500: bug fixed in Python 3.6.7, 3.7.1 and above.',
+    )
+    def test_create_datagram_endpoint_addrs_ipv6(self):
+        self._test_create_datagram_endpoint_addrs_ipv6()

uvloop/handles/udp.pxd

@@ -2,8 +2,10 @@ cdef class UDPTransport(UVBaseTransport):
     cdef:
         bint __receiving
         int _family
+        object _address
 
     cdef _init(self, Loop loop, unsigned int family)
+    cdef _set_address(self, system.addrinfo *addr)
 
     cdef _connect(self, system.sockaddr* addr, size_t addr_len)
 

uvloop/handles/udp.pyx

@@ -55,6 +55,7 @@ cdef class UDPTransport(UVBaseTransport):
     def __cinit__(self):
         self._family = uv.AF_UNSPEC
         self.__receiving = 0
+        self._address = None
 
     cdef _init(self, Loop loop, unsigned int family):
         cdef int err
@@ -78,6 +79,9 @@ cdef class UDPTransport(UVBaseTransport):
 
         self._finish_init()
 
+    cdef _set_address(self, system.addrinfo *addr):
+        self._address = __convert_sockaddr_to_pyaddr(addr.ai_addr)
+
     cdef _connect(self, system.sockaddr* addr, size_t addr_len):
         cdef int err
         err = uv.uv_udp_connect(<uv.uv_udp_t*>self._handle, addr)
@@ -279,6 +283,16 @@ cdef class UDPTransport(UVBaseTransport):
             # Replicating asyncio logic here.
             return
 
+        if self._address:
+            if addr not in (None, self._address):
+                # Replicating asyncio logic here.
+                raise ValueError(
+                    'Invalid address: must be None or %s' % (self._address,))
+
+            # Instead of setting addr to self._address below like what asyncio
+            # does, we depend on previous uv_udp_connect() to set the address
+            addr = None
+
         if self._conn_lost:
             # Replicating asyncio logic here.
             if self._conn_lost >= LOG_THRESHOLD_FOR_CONNLOST_WRITES:
@@ -330,6 +344,12 @@ cdef void __uv_udp_on_receive(uv.uv_udp_t* handle,
 
     if addr is NULL:
         pyaddr = None
+    elif addr.sa_family == uv.AF_UNSPEC:
+        # https://github.com/MagicStack/uvloop/issues/304
+        IF UNAME_SYSNAME == "Linux":
+            pyaddr = None
+        ELSE:
+            pyaddr = ''
     else:
         try:
             pyaddr = __convert_sockaddr_to_pyaddr(addr)
@@ -342,13 +362,6 @@ cdef void __uv_udp_on_receive(uv.uv_udp_t* handle,
         udp._on_receive(None, exc, pyaddr)
         return
 
-    if pyaddr is None:
-        udp._fatal_error(
-            RuntimeError(
-                'uv_udp.receive callback: addr is NULL and nread >= 0'),
-            False)
-        return
-
     if nread == 0:
         data = b''
     else:

uvloop/loop.pyx

@@ -1386,16 +1386,29 @@ cdef class Loop:
         """Create a Future object attached to the loop."""
         return self._new_future()
 
-    def create_task(self, coro):
+    def create_task(self, coro, *, name=None):
         """Schedule a coroutine object.
 
         Return a task object.
+
+        If name is not None, task.set_name(name) will be called if the task
+        object has the set_name attribute, true for default Task in Python 3.8.
         """
         self._check_closed()
         if self._task_factory is None:
             task = aio_Task(coro, loop=self)
         else:
             task = self._task_factory(self, coro)
+
+        # copied from asyncio.tasks._set_task_name (bpo-34270)
+        if name is not None:
+            try:
+                set_name = task.set_name
+            except AttributeError:
+                pass
+            else:
+                set_name(name)
+
         return task
 
     def set_task_factory(self, factory):
@@ -3004,6 +3017,7 @@ cdef class Loop:
                     rai = (<AddrInfo>rads).data
                     udp._init(self, rai.ai_family)
                     udp._connect(rai.ai_addr, rai.ai_addrlen)
+                    udp._set_address(rai)
                 else:
                     if family not in (uv.AF_INET, uv.AF_INET6):
                         raise ValueError('unexpected address family')
@@ -3047,6 +3061,7 @@ cdef class Loop:
                             rai = rai.ai_next
                             continue
                         udp._connect(rai.ai_addr, rai.ai_addrlen)
+                        udp._set_address(rai)
                         break
                     else:
                         raise OSError(