[3.15] gh-155411: Fix test.support.subTests() for asynchronous tests (GH-155412) (GH-155544)

An asynchronous test was wrapped in a synchronous function, which
discarded the coroutine without awaiting it, so the test did not run at
all and was reported as successful.
(cherry picked from commit 198a835)

Co-authored-by: Serhiy Storchaka <storchaka@gmail.com>

Author: miss-islington

Lib/test/support/__init__.py

@@ -1077,16 +1077,29 @@ def subTests(arg_names, arg_values, /, *, _do_cleanups=False):
     def decorator(func):
         if isinstance(func, type):
             raise TypeError('subTests() can only decorate methods, not classes')
-        @functools.wraps(func)
-        def wrapper(self, /, *args, **kwargs):
+
+        def iter_subtest_kwargs():
             for values in arg_values:
-                if single_param:
-                    values = (values,)
-                subtest_kwargs = dict(zip(arg_names, values))
-                with self.subTest(**subtest_kwargs):
-                    func(self, *args, **kwargs, **subtest_kwargs)
-                if _do_cleanups:
-                    self.doCleanups()
+                yield dict(zip(arg_names, (values,) if single_param else values))
+
+        # A synchronous wrapper would discard the coroutine without awaiting
+        # it, so an asynchronous test would not run at all.
+        if inspect.iscoroutinefunction(func):
+            @functools.wraps(func)
+            async def wrapper(self, /, *args, **kwargs):
+                for subtest_kwargs in iter_subtest_kwargs():
+                    with self.subTest(**subtest_kwargs):
+                        await func(self, *args, **kwargs, **subtest_kwargs)
+                    if _do_cleanups:
+                        self.doCleanups()
+        else:
+            @functools.wraps(func)
+            def wrapper(self, /, *args, **kwargs):
+                for subtest_kwargs in iter_subtest_kwargs():
+                    with self.subTest(**subtest_kwargs):
+                        func(self, *args, **kwargs, **subtest_kwargs)
+                    if _do_cleanups:
+                        self.doCleanups()
         return wrapper
     return decorator
 

Lib/test/test_support.py

@@ -1258,5 +1258,71 @@ def test_skipped_without_subprocess_support(self):
         self.assertEqual(calls, [])
 
 
+class TestSubTests(unittest.TestCase):
+
+    def run_test(self, cls):
+        result = unittest.TestResult()
+        cls('test_it').run(result)
+        return result
+
+    def test_sync(self):
+        ran = []
+
+        class Sample(unittest.TestCase):
+            @support.subTests('a', [1, 2, 3])
+            def test_it(self, a):
+                ran.append(a)
+                self.assertNotEqual(a, 2)
+
+        result = self.run_test(Sample)
+        self.assertEqual(ran, [1, 2, 3])
+        self.assertEqual(result.testsRun, 1)
+        self.assertEqual(len(result.failures), 1)
+        self.assertEndsWith(result.failures[0][0].id(), 'test_it (a=2)')
+
+    # Running an asyncio event loop needs a working socket.
+    @support.requires_working_socket()
+    def test_async(self):
+        # Running an event loop sets the event loop policy, which regrtest
+        # reports as a change of the environment.
+        import asyncio.events
+        self.enterContext(
+            support.swap_attr(asyncio.events, '_event_loop_policy', None))
+
+        # An asynchronous test must be awaited: a synchronous wrapper would
+        # make it silently not run at all.
+        ran = []
+
+        class Sample(unittest.IsolatedAsyncioTestCase):
+            @support.subTests('a', [1, 2, 3])
+            async def test_it(self, a):
+                ran.append(a)
+                self.assertNotEqual(a, 2)
+
+        result = self.run_test(Sample)
+        self.assertEqual(ran, [1, 2, 3])
+        self.assertEqual(result.testsRun, 1)
+        self.assertEqual(len(result.failures), 1)
+        self.assertEndsWith(result.failures[0][0].id(), 'test_it (a=2)')
+
+    def test_multiple_parameters(self):
+        ran = []
+
+        class Sample(unittest.TestCase):
+            @support.subTests('a,b', [(1, 'x'), (2, 'y')])
+            def test_it(self, a, b):
+                ran.append((a, b))
+
+        result = self.run_test(Sample)
+        self.assertTrue(result.wasSuccessful(), result.errors)
+        self.assertEqual(ran, [(1, 'x'), (2, 'y')])
+
+    def test_cannot_decorate_class(self):
+        with self.assertRaises(TypeError):
+            @support.subTests('a', [1])
+            class Sample(unittest.TestCase):
+                pass
+
+
 if __name__ == '__main__':
     unittest.main()

Misc/NEWS.d/next/Tests/2026-08-09-14-00-00.gh-issue-155411.Qw8Lm2.rst

@@ -0,0 +1,3 @@
+Fix :func:`!test.support.subTests` for asynchronous test methods.  They were
+wrapped in a synchronous function, which discarded the coroutine without
+awaiting it, so the test silently did not run at all.