Merge branch 'main' into multi_inputs

Author: s-ball

Doc/library/asyncio-graph.rst

@@ -59,13 +59,13 @@ and debuggers.
 
       async def main():
           async with asyncio.TaskGroup() as g:
-              g.create_task(test())
+              g.create_task(test(), name='test')
 
       asyncio.run(main())
 
    will print::
 
-      * Task(name='Task-2', id=0x1039f0fe0)
+      * Task(name='test', id=0x1039f0fe0)
       + Call stack:
       |   File 't2.py', line 4, in async test()
       + Awaited by:

Lib/_pyrepl/fancy_termios.py

@@ -40,7 +40,9 @@ def as_list(self):
             self.lflag,
             self.ispeed,
             self.ospeed,
-            self.cc,
+            # Always return a copy of the control characters list to ensure
+            # there are not any additional references to self.cc
+            self.cc[:],
         ]
 
     def copy(self):

Lib/asyncio/staggered.py

@@ -66,8 +66,27 @@ async def staggered_race(coro_fns, delay, *, loop=None):
     enum_coro_fns = enumerate(coro_fns)
     winner_result = None
     winner_index = None
+    unhandled_exceptions = []
     exceptions = []
-    running_tasks = []
+    running_tasks = set()
+    on_completed_fut = None
+
+    def task_done(task):
+        running_tasks.discard(task)
+        if (
+            on_completed_fut is not None
+            and not on_completed_fut.done()
+            and not running_tasks
+        ):
+            on_completed_fut.set_result(None)
+
+        if task.cancelled():
+            return
+
+        exc = task.exception()
+        if exc is None:
+            return
+        unhandled_exceptions.append(exc)
 
     async def run_one_coro(ok_to_start, previous_failed) -> None:
         # in eager tasks this waits for the calling task to append this task
@@ -91,11 +110,11 @@ async def run_one_coro(ok_to_start, previous_failed) -> None:
         this_failed = locks.Event()
         next_ok_to_start = locks.Event()
         next_task = loop.create_task(run_one_coro(next_ok_to_start, this_failed))
-        running_tasks.append(next_task)
+        running_tasks.add(next_task)
+        next_task.add_done_callback(task_done)
         # next_task has been appended to running_tasks so next_task is ok to
         # start.
         next_ok_to_start.set()
-        assert len(running_tasks) == this_index + 2
         # Prepare place to put this coroutine's exceptions if not won
         exceptions.append(None)
         assert len(exceptions) == this_index + 1
@@ -120,31 +139,36 @@ async def run_one_coro(ok_to_start, previous_failed) -> None:
             # up as done() == True, cancelled() == False, exception() ==
             # asyncio.CancelledError. This behavior is specified in
             # https://bugs.python.org/issue30048
-            for i, t in enumerate(running_tasks):
-                if i != this_index:
+            current_task = tasks.current_task(loop)
+            for t in running_tasks:
+                if t is not current_task:
                     t.cancel()
 
-    ok_to_start = locks.Event()
-    first_task = loop.create_task(run_one_coro(ok_to_start, None))
-    running_tasks.append(first_task)
-    # first_task has been appended to running_tasks so first_task is ok to start.
-    ok_to_start.set()
+    propagate_cancellation_error = None
     try:
-        # Wait for a growing list of tasks to all finish: poor man's version of
-        # curio's TaskGroup or trio's nursery
-        done_count = 0
-        while done_count != len(running_tasks):
-            done, _ = await tasks.wait(running_tasks)
-            done_count = len(done)
+        ok_to_start = locks.Event()
+        first_task = loop.create_task(run_one_coro(ok_to_start, None))
+        running_tasks.add(first_task)
+        first_task.add_done_callback(task_done)
+        # first_task has been appended to running_tasks so first_task is ok to start.
+        ok_to_start.set()
+        propagate_cancellation_error = None
+        # Make sure no tasks are left running if we leave this function
+        while running_tasks:
+            on_completed_fut = loop.create_future()
+            try:
+                await on_completed_fut
+            except exceptions_mod.CancelledError as ex:
+                propagate_cancellation_error = ex
+                for task in running_tasks:
+                    task.cancel(*ex.args)
+            on_completed_fut = None
+        if __debug__ and unhandled_exceptions:
             # If run_one_coro raises an unhandled exception, it's probably a
             # programming error, and I want to see it.
-            if __debug__:
-                for d in done:
-                    if d.done() and not d.cancelled() and d.exception():
-                        raise d.exception()
+            raise ExceptionGroup("staggered race failed", unhandled_exceptions)
+        if propagate_cancellation_error is not None:
+            raise propagate_cancellation_error
         return winner_result, winner_index, exceptions
     finally:
-        del exceptions
-        # Make sure no tasks are left running if we leave this function
-        for t in running_tasks:
-            t.cancel()
+        del exceptions, propagate_cancellation_error, unhandled_exceptions

Lib/idlelib/idle_test/test_configdialog.py

@@ -98,8 +98,8 @@ def test_click_help(self):
             dialog.buttons['Help'].invoke()
             title, contents = view.kwds['title'], view.kwds['contents']
         self.assertEqual(title, 'Help for IDLE preferences')
-        self.assertTrue(contents.startswith('When you click') and
-                        contents.endswith('a different name.\n'))
+        self.assertStartsWith(contents, 'When you click')
+        self.assertEndsWith(contents,'a different name.\n')
 
 
 class FontPageTest(unittest.TestCase):

Lib/idlelib/idle_test/test_debugger.py

@@ -256,7 +256,7 @@ def test_init(self):
         flist = None
         master_window = self.root
         sv = debugger.StackViewer(master_window, flist, gui)
-        self.assertTrue(hasattr(sv, 'stack'))
+        self.assertHasAttr(sv, 'stack')
 
     def test_load_stack(self):
         # Test the .load_stack() method against a fixed test stack.

Lib/idlelib/idle_test/test_grep.py

@@ -143,7 +143,7 @@ def test_found(self):
         self.assertIn(pat, lines[0])
         self.assertIn('py: 1:', lines[1])  # line number 1
         self.assertIn('2', lines[3])  # hits found 2
-        self.assertTrue(lines[4].startswith('(Hint:'))
+        self.assertStartsWith(lines[4], '(Hint:')
 
 
 class Default_commandTest(unittest.TestCase):

Lib/idlelib/idle_test/test_multicall.py

@@ -27,7 +27,7 @@ def tearDownClass(cls):
     def test_creator(self):
         mc = self.mc
         self.assertIs(multicall._multicall_dict[Text], mc)
-        self.assertTrue(issubclass(mc, Text))
+        self.assertIsSubclass(mc, Text)
         mc2 = multicall.MultiCallCreator(Text)
         self.assertIs(mc, mc2)
 

Lib/idlelib/idle_test/test_query.py

@@ -134,10 +134,10 @@ def test_c_source_name(self):
 
     def test_good_module_name(self):
         dialog = self.Dummy_ModuleName('idlelib')
-        self.assertTrue(dialog.entry_ok().endswith('__init__.py'))
+        self.assertEndsWith(dialog.entry_ok(), '__init__.py')
         self.assertEqual(dialog.entry_error['text'], '')
         dialog = self.Dummy_ModuleName('idlelib.idle')
-        self.assertTrue(dialog.entry_ok().endswith('idle.py'))
+        self.assertEndsWith(dialog.entry_ok(), 'idle.py')
         self.assertEqual(dialog.entry_error['text'], '')
 
 
@@ -389,7 +389,7 @@ def test_click_module_name(self):
         self.assertEqual(dialog.text0, 'idlelib')
         self.assertEqual(dialog.entry.get(), 'idlelib')
         dialog.button_ok.invoke()
-        self.assertTrue(dialog.result.endswith('__init__.py'))
+        self.assertEndsWith(dialog.result, '__init__.py')
         root.destroy()
 
 

Lib/idlelib/idle_test/test_redirector.py

@@ -34,7 +34,7 @@ def test_close(self):
         redir.register('insert', Func)
         redir.close()
         self.assertEqual(redir._operations, {})
-        self.assertFalse(hasattr(self.text, 'widget'))
+        self.assertNotHasAttr(self.text, 'widget')
 
 
 class WidgetRedirectorTest(unittest.TestCase):

Lib/idlelib/idle_test/test_sidebar.py

@@ -725,7 +725,7 @@ def test_copy(self):
 
         text.tag_add('sel', f'{first_line}.0', 'end-1c')
         selected_text = text.get('sel.first', 'sel.last')
-        self.assertTrue(selected_text.startswith('if True:\n'))
+        self.assertStartsWith(selected_text, 'if True:\n')
         self.assertIn('\n1\n', selected_text)
 
         text.event_generate('<<copy>>')
@@ -749,7 +749,7 @@ def test_copy_with_prompts(self):
 
         text.tag_add('sel', f'{first_line}.3', 'end-1c')
         selected_text = text.get('sel.first', 'sel.last')
-        self.assertTrue(selected_text.startswith('True:\n'))
+        self.assertStartsWith(selected_text, 'True:\n')
 
         selected_lines_text = text.get('sel.first linestart', 'sel.last')
         selected_lines = selected_lines_text.split('\n')