Merge branch 'master' into iscoroutinefunction-deprecation

Author: fantix

tests/test_dns.py

@@ -217,6 +217,10 @@ def test_getaddrinfo_22(self):
         self._test_getaddrinfo(payload, 80)
         self._test_getaddrinfo(payload, 80, type=socket.SOCK_STREAM)
 
+    def test_getaddrinfo_broadcast(self):
+        self._test_getaddrinfo('<broadcast>', 80)
+        self._test_getaddrinfo('<broadcast>', 80, type=socket.SOCK_STREAM)
+
     ######
 
     def test_getnameinfo_1(self):

tests/test_fs_event.py

@@ -1,100 +1,87 @@
 import asyncio
+import contextlib
 import os.path
 import tempfile
 
 from uvloop import _testbase as tb
 from uvloop.loop import FileSystemEvent
 
 
-class Test_UV_FS_EVENT_CHANGE(tb.UVTestCase):
-    async def _file_writer(self):
-        f = await self.q.get()
-        while True:
-            f.write('hello uvloop\n')
-            f.flush()
-            x = await self.q.get()
-            if x is None:
-                return
-
-    def fs_event_setup(self):
-        self.change_event_count = 0
-        self.fname = ''
-        self.q = asyncio.Queue()
-
-    def event_cb(self, ev_fname: bytes, evt: FileSystemEvent):
-        _d, fn = os.path.split(self.fname)
-        self.assertEqual(ev_fname, fn)
-        self.assertEqual(evt, FileSystemEvent.CHANGE)
-        self.change_event_count += 1
-        if self.change_event_count < 4:
-            self.q.put_nowait(0)
-        else:
-            self.q.put_nowait(None)
+class Test_UV_FS_Event(tb.UVTestCase):
+    def setUp(self):
+        super().setUp()
+        self.exit_stack = contextlib.ExitStack()
+        self.tmp_dir = self.exit_stack.enter_context(
+            tempfile.TemporaryDirectory()
+        )
+
+    def tearDown(self):
+        self.exit_stack.close()
+        super().tearDown()
 
     def test_fs_event_change(self):
-        self.fs_event_setup()
-
-        async def run(write_task):
-            self.q.put_nowait(tf)
-            try:
-                await asyncio.wait_for(write_task, 4)
-            except asyncio.TimeoutError:
-                write_task.cancel()
-
-        with tempfile.NamedTemporaryFile('wt') as tf:
-            self.fname = tf.name.encode()
-            h = self.loop._monitor_fs(tf.name, self.event_cb)
+        change_event_count = 0
+        filename = "fs_event_change.txt"
+        path = os.path.join(self.tmp_dir, filename)
+        q = asyncio.Queue()
+
+        with open(path, 'wt') as f:
+            async def file_writer():
+                while True:
+                    f.write('hello uvloop\n')
+                    f.flush()
+                    x = await q.get()
+                    if x is None:
+                        return
+
+            def event_cb(ev_fname: bytes, evt: FileSystemEvent):
+                nonlocal change_event_count
+                self.assertEqual(ev_fname, filename.encode())
+                self.assertEqual(evt, FileSystemEvent.CHANGE)
+                change_event_count += 1
+                if change_event_count < 4:
+                    q.put_nowait(0)
+                else:
+                    q.put_nowait(None)
+
+            h = self.loop._monitor_fs(path, event_cb)
+            self.loop.run_until_complete(
+                asyncio.sleep(0.1)  # let monitor start
+            )
             self.assertFalse(h.cancelled())
 
-            self.loop.run_until_complete(run(
-                self.loop.create_task(self._file_writer())))
+            self.loop.run_until_complete(asyncio.wait_for(file_writer(), 4))
             h.cancel()
             self.assertTrue(h.cancelled())
 
-        self.assertEqual(self.change_event_count, 4)
-
-
-class Test_UV_FS_EVENT_RENAME(tb.UVTestCase):
-    async def _file_renamer(self):
-        await self.q.get()
-        os.rename(os.path.join(self.dname, self.changed_name),
-                  os.path.join(self.dname, self.changed_name + "-new"))
-        await self.q.get()
-
-    def fs_event_setup(self):
-        self.dname = ''
-        self.changed_name = "hello_fs_event.txt"
-        self.changed_set = {self.changed_name, self.changed_name + '-new'}
-        self.q = asyncio.Queue()
-
-    def event_cb(self, ev_fname: bytes, evt: FileSystemEvent):
-        ev_fname = ev_fname.decode()
-        self.assertEqual(evt, FileSystemEvent.RENAME)
-        self.changed_set.remove(ev_fname)
-        if len(self.changed_set) == 0:
-            self.q.put_nowait(None)
+        self.assertEqual(change_event_count, 4)
 
     def test_fs_event_rename(self):
-        self.fs_event_setup()
-
-        async def run(write_task):
-            self.q.put_nowait(0)
-            try:
-                await asyncio.wait_for(write_task, 4)
-            except asyncio.TimeoutError:
-                write_task.cancel()
-
-        with tempfile.TemporaryDirectory() as td_name:
-            self.dname = td_name
-            f = open(os.path.join(td_name, self.changed_name), 'wt')
+        orig_name = "hello_fs_event.txt"
+        new_name = "hello_fs_event_rename.txt"
+        changed_set = {orig_name, new_name}
+        event = asyncio.Event()
+
+        async def file_renamer():
+            os.rename(os.path.join(self.tmp_dir, orig_name),
+                      os.path.join(self.tmp_dir, new_name))
+            await event.wait()
+
+        def event_cb(ev_fname: bytes, evt: FileSystemEvent):
+            ev_fname = ev_fname.decode()
+            self.assertEqual(evt, FileSystemEvent.RENAME)
+            changed_set.discard(ev_fname)
+            if len(changed_set) == 0:
+                event.set()
+
+        with open(os.path.join(self.tmp_dir, orig_name), 'wt') as f:
             f.write('hello!')
-            f.close()
-            h = self.loop._monitor_fs(td_name, self.event_cb)
-            self.assertFalse(h.cancelled())
+        h = self.loop._monitor_fs(self.tmp_dir, event_cb)
+        self.loop.run_until_complete(asyncio.sleep(0.5))  # let monitor start
+        self.assertFalse(h.cancelled())
 
-            self.loop.run_until_complete(run(
-                self.loop.create_task(self._file_renamer())))
-            h.cancel()
-            self.assertTrue(h.cancelled())
+        self.loop.run_until_complete(asyncio.wait_for(file_renamer(), 4))
+        h.cancel()
+        self.assertTrue(h.cancelled())
 
-        self.assertEqual(len(self.changed_set), 0)
+        self.assertEqual(len(changed_set), 0)

tests/test_process.py

@@ -685,6 +685,12 @@ async def cancel_make_transport():
             self.loop.run_until_complete(cancel_make_transport())
 
     def test_cancel_post_init(self):
+        if sys.version_info >= (3, 13) and self.implementation == 'asyncio':
+            # https://github.com/python/cpython/issues/103847#issuecomment-3736561321
+            # This test started to flake on CPython 3.13 and later,
+            # so we skip it for asyncio tests until the issue is resolved.
+            self.skipTest('flaky test on CPython 3.13+')
+
         async def cancel_make_transport():
             coro = self.loop.subprocess_exec(asyncio.SubprocessProtocol,
                                              *self.PROGRAM_BLOCKED)

tests/test_tcp.py

@@ -405,7 +405,7 @@ async def client(addr):
             self.assertEqual(await reader.readexactly(2), b'OK')
 
             re = r'(a bytes-like object)|(must be byte-ish)'
-            if sys.version_info >= (3, 14):
+            if sys.version_info >= (3, 13, 9):
                 re += r'|(must be a bytes, bytearray, or memoryview object)'
             with self.assertRaisesRegex(TypeError, re):
                 writer.write('AAAA')

tests/test_udp.py

@@ -378,6 +378,22 @@ def test_udp_sendto_dns(self):
         s_transport.close()
         self.loop.run_until_complete(asyncio.sleep(0.01))
 
+    def test_udp_sendto_broadcast(self):
+        coro = self.loop.create_datagram_endpoint(
+            asyncio.DatagramProtocol,
+            local_addr=('127.0.0.1', 0),
+            family=socket.AF_INET)
+
+        s_transport, server = self.loop.run_until_complete(coro)
+
+        try:
+            s_transport.sendto(b'aaaa', ('<broadcast>', 80))
+        except ValueError as exc:
+            raise AssertionError('sendto raises {}.'.format(exc))
+
+        s_transport.close()
+        self.loop.run_until_complete(asyncio.sleep(0.01))
+
     def test_send_after_close(self):
         coro = self.loop.create_datagram_endpoint(
             asyncio.DatagramProtocol,

uvloop/handles/udp.pyx

@@ -208,6 +208,10 @@ cdef class UDPTransport(UVBaseTransport):
         if addr is None:
             saddr = NULL
         else:
+            # resolve special hostname <broadcast> to the broadcast address before use
+            if self._family == uv.AF_INET and addr[0] == '<broadcast>':
+                addr = (b'255.255.255.255', addr[1])
+
             try:
                 __convert_pyaddr_to_sockaddr(self._family, addr,
                                              <system.sockaddr*>&saddr_st)