asyncio: honor pending-task, unawaited, and get_referrers (#8689)

* asyncio: honor pending-task, unawaited, and get_referrers

Report Task destructor as 'Task was destroyed but it is pending!'.
Warn when a never-started coroutine is collected. Preserve signal()
errno. Attribute generator-owned frame locals to the owner in
gc.get_referrers. Remove the expectedFailure markers that now pass.

Assisted-by: Grok:grok-4.6

* tests: drop expectedFailure after unawaited warning

The never-started coroutine RuntimeWarning unblocks
test_bpo_45813_1 and the AsyncMock assertNeverAwaited cases.

Assisted-by: Grok:grok-4.6

Lib/test/test_asyncio/test_events.py

@@ -273,7 +273,6 @@ class EventLoopTestsMixin:
         support.gc_collect()
         super().tearDown()
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; - RuntimeWarning for unawaited coroutine not triggered
     def test_run_until_complete_nesting(self):
         async def coro1():
             await asyncio.sleep(0)
@@ -560,7 +559,6 @@ class EventLoopTestsMixin:
         r.close()
         self.assertEqual(read, data)
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; - signal handler implementation differs
     @unittest.skipUnless(hasattr(signal, 'SIGKILL'), 'No SIGKILL')
     def test_add_signal_handler(self):
         caught = 0

Lib/test/test_asyncio/test_futures.py

@@ -734,10 +734,6 @@ class CFutureTests(BaseFutureTests, test_utils.TestCase):
         fut.remove_done_callback(f2)
         self.assertIsNone(fut._callbacks)
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; - gc.get_referents not implemented
-    def test_future_iter_get_referents_segfault(self):
-        return super().test_future_iter_get_referents_segfault()
-
 
 @unittest.skipUnless(hasattr(futures, '_CFuture'),
                      'requires the C _asyncio module')

Lib/test/test_asyncio/test_runners.py

@@ -402,7 +402,6 @@ class RunnerTests(BaseTest):
 
             self.assertEqual(2, runner.run(get_context()).get(cvar))
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; - RuntimeWarning for unawaited coroutine not triggered
     def test_recursive_run(self):
         async def g():
             pass

Lib/test/test_asyncio/test_streams.py

@@ -1239,7 +1239,6 @@ class StreamTests(test_utils.TestCase):
         messages = self._basetest_unhandled_exceptions(handle_echo)
         self.assertEqual(messages, [])
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; NotImplementedError
     def test_open_connection_happy_eyeball_refcycles(self):
         port = socket_helper.find_unused_port()
         async def main():

Lib/test/test_asyncio/test_taskgroups.py

@@ -1106,29 +1106,6 @@ class BaseTestTaskGroup:
 class TestTaskGroup(BaseTestTaskGroup, unittest.IsolatedAsyncioTestCase):
     loop_factory = asyncio.EventLoop
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Test that TaskGroup deletes propagate_cancellation_error
-    async def test_exception_refcycles_propagate_cancellation_error(self):
-        return await super().test_exception_refcycles_propagate_cancellation_error()
-
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Test that TaskGroup deletes self._base_error
-    async def test_exception_refcycles_base_error(self):
-        return await super().test_exception_refcycles_base_error()
-
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Test that TaskGroup deletes self._errors, and __aexit__ args
-    async def test_exception_refcycles_errors(self):
-        return await super().test_exception_refcycles_errors()
-
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Test that TaskGroup deletes self._parent_task
-    async def test_exception_refcycles_parent_task(self):
-        return await super().test_exception_refcycles_parent_task()
-
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Test that TaskGroup deletes self._parent_task and create_task() deletes task
-    async def test_exception_refcycles_parent_task_wr(self):
-        return await super().test_exception_refcycles_parent_task_wr()
-
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Test that TaskGroup doesn't keep a reference to the raised ExceptionGroup
-    async def test_exception_refcycles_direct(self):
-        return await super().test_exception_refcycles_direct()
 
 class TestEagerTaskTaskGroup(BaseTestTaskGroup, unittest.IsolatedAsyncioTestCase):
     @staticmethod
@@ -1137,30 +1114,6 @@ class TestEagerTaskTaskGroup(BaseTestTaskGroup, unittest.IsolatedAsyncioTestCase
         loop.set_task_factory(asyncio.eager_task_factory)
         return loop
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Test that TaskGroup deletes propagate_cancellation_error
-    async def test_exception_refcycles_propagate_cancellation_error(self):
-        return await super().test_exception_refcycles_propagate_cancellation_error()
-
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Test that TaskGroup deletes self._base_error
-    async def test_exception_refcycles_base_error(self):
-        return await super().test_exception_refcycles_base_error()
-
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Test that TaskGroup deletes self._errors, and __aexit__ args
-    async def test_exception_refcycles_errors(self):
-        return await super().test_exception_refcycles_errors()
-
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Test that TaskGroup deletes self._parent_task
-    async def test_exception_refcycles_parent_task(self):
-        return await super().test_exception_refcycles_parent_task()
-
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Test that TaskGroup deletes self._parent_task and create_task() deletes task
-    async def test_exception_refcycles_parent_task_wr(self):
-        return await super().test_exception_refcycles_parent_task_wr()
-
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Test that TaskGroup doesn't keep a reference to the raised ExceptionGroup
-    async def test_exception_refcycles_direct(self):
-        return await super().test_exception_refcycles_direct()
-
 
 if __name__ == "__main__":
     unittest.main()

Lib/test/test_asyncio/test_tasks.py

@@ -2942,10 +2942,6 @@ class CTask_CFuture_Tests(BaseTaskTests, SetMethodsTest,
         with self.assertRaises(AttributeError):
             del task._log_destroy_pending
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Actual: not called.
-    def test_log_destroyed_pending_task(self):
-        return super().test_log_destroyed_pending_task()
-
 
 @unittest.skipUnless(hasattr(futures, '_CFuture') and
                      hasattr(tasks, '_CTask'),
@@ -2958,10 +2954,6 @@ class CTask_CFuture_SubclassTests(BaseTaskTests, test_utils.TestCase):
     all_tasks = getattr(tasks, '_c_all_tasks', None)
     current_task = staticmethod(getattr(tasks, '_c_current_task', None))
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Actual: not called.
-    def test_log_destroyed_pending_task(self):
-        return super().test_log_destroyed_pending_task()
-
 
 @unittest.skipUnless(hasattr(tasks, '_CTask'),
                      'requires the C _asyncio module')
@@ -2973,10 +2965,6 @@ class CTaskSubclass_PyFuture_Tests(BaseTaskTests, test_utils.TestCase):
     all_tasks = getattr(tasks, '_c_all_tasks', None)
     current_task = staticmethod(getattr(tasks, '_c_current_task', None))
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Actual: not called.
-    def test_log_destroyed_pending_task(self):
-        return super().test_log_destroyed_pending_task()
-
 
 @unittest.skipUnless(hasattr(futures, '_CFuture'),
                      'requires the C _asyncio module')
@@ -2998,10 +2986,6 @@ class CTask_PyFuture_Tests(BaseTaskTests, test_utils.TestCase):
     all_tasks = getattr(tasks, '_c_all_tasks', None)
     current_task = staticmethod(getattr(tasks, '_c_current_task', None))
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; Actual: not called.
-    def test_log_destroyed_pending_task(self):
-        return super().test_log_destroyed_pending_task()
-
 
 @unittest.skipUnless(hasattr(futures, '_CFuture'),
                      'requires the C _asyncio module')

Lib/test/test_coroutines.py

@@ -2230,7 +2230,6 @@ class CoroutineTest(unittest.TestCase):
             return 'end'
         self.assertEqual(run_async(run_gen()), ([], 'end'))
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; This would crash the interpreter in 3.11a2
     def test_bpo_45813_1(self):
         'This would crash the interpreter in 3.11a2'
         async def f():

Lib/test/test_unittest/testmock/testasync.py

@@ -336,7 +336,6 @@ class AsyncSpecTest(unittest.TestCase):
         # only the shape of the spec at the time of mock construction matters
         self.assertNotIsInstance(mock_async_instance.later_async_func_attr, AsyncMock)
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_spec_mock_type_kw(self):
         def inner_test(mock_type):
             async_mock = mock_type(spec=async_func)
@@ -351,7 +350,6 @@ class AsyncSpecTest(unittest.TestCase):
             with self.subTest(f"test spec kwarg with {mock_type}"):
                 inner_test(mock_type)
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_spec_mock_type_positional(self):
         def inner_test(mock_type):
             async_mock = mock_type(async_func)
@@ -800,7 +798,6 @@ class AsyncMockAssert(unittest.TestCase):
     async def _await_coroutine(self, coroutine):
         return await coroutine
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_assert_called_but_not_awaited(self):
         mock = AsyncMock(AsyncClass)
         with assertNeverAwaited(self):
@@ -841,7 +838,6 @@ class AsyncMockAssert(unittest.TestCase):
         self.mock.assert_called_once()
         self.mock.assert_awaited_once()
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_assert_called_twice_and_awaited_once(self):
         mock = AsyncMock(AsyncClass)
         coroutine = mock.async_method()
@@ -881,7 +877,6 @@ class AsyncMockAssert(unittest.TestCase):
         with self.assertRaises(AssertionError):
             self.mock.assert_called()
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_assert_has_calls_not_awaits(self):
         kalls = [call('foo')]
         with assertNeverAwaited(self):
@@ -890,7 +885,6 @@ class AsyncMockAssert(unittest.TestCase):
         with self.assertRaises(AssertionError):
             self.mock.assert_has_awaits(kalls)
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_assert_has_mock_calls_on_async_mock_no_spec(self):
         with assertNeverAwaited(self):
             self.mock()
@@ -904,7 +898,6 @@ class AsyncMockAssert(unittest.TestCase):
         mock_kalls = ([call(), call('foo'), call('baz')])
         self.assertEqual(self.mock.mock_calls, mock_kalls)
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_assert_has_mock_calls_on_async_mock_with_spec(self):
         a_class_mock = AsyncMock(AsyncClass)
         with assertNeverAwaited(self):
@@ -920,7 +913,6 @@ class AsyncMockAssert(unittest.TestCase):
         self.assertEqual(a_class_mock.async_method.mock_calls, method_kalls)
         self.assertEqual(a_class_mock.mock_calls, mock_kalls)
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_async_method_calls_recorded(self):
         with assertNeverAwaited(self):
             self.mock.something(3, fish=None)
@@ -936,7 +928,6 @@ class AsyncMockAssert(unittest.TestCase):
                          [("something", (6,), {'cake': sentinel.Cake})],
                          "method calls not recorded correctly")
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_async_arg_lists(self):
         def assert_attrs(mock):
             names = ('call_args_list', 'method_calls', 'mock_calls')

crates/stdlib/src/_asyncio.rs

@@ -1876,15 +1876,11 @@ pub(crate) mod _asyncio {
             {
                 if let Some(loop_obj) = loop_obj.clone() {
                     let context = PyDict::default().into_ref(&vm.ctx);
-                    let task_repr = zelf
-                        .as_object()
-                        .repr(vm)
-                        .unwrap_or_else(|_| vm.ctx.new_str("<Task>"));
-                    let message =
-                        format!("Task was destroyed but it is pending!\ntask: {task_repr}");
                     context.set_item(
                         vm.ctx.intern_str("message"),
-                        vm.ctx.new_str(message).into(),
+                        vm.ctx
+                            .new_str("Task was destroyed but it is pending!")
+                            .into(),
                         vm,
                     )?;
                     context.set_item(vm.ctx.intern_str("task"), zelf.to_owned().into(), vm)?;

crates/vm/src/builtins/coroutine.rs

@@ -163,6 +163,13 @@ impl Destructor for PyCoroutine {
             return Ok(());
         }
         if zelf.inner.frame().lasti() == 0 {
+            let name = zelf.inner.qualname();
+            let msg = format!("coroutine '{name}' was never awaited");
+            if let Err(e) =
+                crate::stdlib::_warnings::warn(vm.ctx.exceptions.runtime_warning, msg, 1, vm)
+            {
+                vm.run_unraisable(e, None, zelf.as_object().to_owned());
+            }
             zelf.inner.closed.store(true);
             return Ok(());
         }

crates/vm/src/stdlib/_signal.rs

@@ -249,7 +249,7 @@ pub(crate) mod _signal {
         signal::check_signals(vm)?;
 
         unsafe { host_signal::install_handler(signalnum.into(), sig_handler) }
-            .map_err(|_| vm.new_os_error("Failed to set signal"))?;
+            .map_err(|err| err.into_pyexception(vm))?;
 
         let signal_handlers = vm.signal_handlers.get_or_init(SignalHandlers::default);
         let old_handler = signal_handlers.borrow_mut()[signalnum].replace(handler);

crates/vm/src/stdlib/gc.rs

@@ -211,21 +211,34 @@ mod gc {
             stack_frames.insert(obj as *const crate::PyObject as usize);
         });
 
+        let refers_to_target = |obj: &crate::PyObject| -> bool {
+            let referent_ptrs = unsafe { obj.gc_get_referent_ptrs() };
+            referent_ptrs
+                .iter()
+                .any(|child_ptr| targets.contains(&(child_ptr.as_ptr() as usize)))
+        };
+
         let mut result = Vec::new();
 
         // Scan all tracked objects across all generations
         let all_objects = vm.state.gc.get_objects(None);
         for obj in all_objects {
             let obj_ptr = obj.as_ref() as *const crate::PyObject as usize;
+            // Generator/coroutine frames are embedded in their owner, so
+            // locals show up as referents of the owner rather than the frame.
+            if let Some(frame) = obj.downcast_ref::<crate::frame::FrameObject>()
+                && let Some(owner) = frame.iframe().generator.to_owned()
+            {
+                if refers_to_target(obj.as_ref()) {
+                    result.push(owner);
+                }
+                continue;
+            }
             if stack_frames.contains(&obj_ptr) {
                 continue;
             }
-            let referent_ptrs = unsafe { obj.gc_get_referent_ptrs() };
-            for child_ptr in referent_ptrs {
-                if targets.contains(&(child_ptr.as_ptr() as usize)) {
-                    result.push(obj.clone());
-                    break;
-                }
+            if refers_to_target(obj.as_ref()) {
+                result.push(obj.clone());
             }
         }