https://github.com/python/cpython/commit/f6dc38b51d2e10292157f81c0b766180da80b51a commit: f6dc38b51d2e10292157f81c0b766180da80b51a branch: 3.13 author: Miss Islington (bot) <[email protected]> committer: serhiy-storchaka <[email protected]> date: 2026-08-12T14:27:41Z summary:
[3.13] gh-155411: Fix test.support.subTests() for asynchronous tests (GH-155412) (GH-155546) 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 198a83581cfdcf9f6c4050e3d1567ab49af754eb) Co-authored-by: Serhiy Storchaka <[email protected]> files: A Misc/NEWS.d/next/Tests/2026-08-09-14-00-00.gh-issue-155411.Qw8Lm2.rst M Lib/test/support/__init__.py M Lib/test/test_support.py diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py index 91bc7bfd226fee9..51c7d2e86927206 100644 --- a/Lib/test/support/__init__.py +++ b/Lib/test/support/__init__.py @@ -6,6 +6,7 @@ import contextlib import dataclasses import functools +import inspect import logging import _opcode import os @@ -976,16 +977,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 diff --git a/Lib/test/test_support.py b/Lib/test/test_support.py index 49788ef7b8a1b83..eeafa41bd33cfe8 100644 --- a/Lib/test/test_support.py +++ b/Lib/test/test_support.py @@ -968,5 +968,71 @@ def test_skipped_without_subprocess_support(self): self.assertEqual(calls, []) +class TestSubTests(ExtraAssertions, 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() diff --git a/Misc/NEWS.d/next/Tests/2026-08-09-14-00-00.gh-issue-155411.Qw8Lm2.rst b/Misc/NEWS.d/next/Tests/2026-08-09-14-00-00.gh-issue-155411.Qw8Lm2.rst new file mode 100644 index 000000000000000..50f684052526051 --- /dev/null +++ b/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. _______________________________________________ Python-checkins mailing list -- [email protected] To unsubscribe send an email to [email protected] https://mail.python.org/mailman3//lists/python-checkins.python.org Member address: [email protected]
