Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 22 additions & 9 deletions Lib/test/support/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -1098,16 +1098,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

Expand Down
60 changes: 60 additions & 0 deletions Lib/test/test_support.py
Original file line number Diff line number Diff line change
Expand Up @@ -1247,5 +1247,65 @@ 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):
# 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()
Original file line number Diff line number Diff line change
@@ -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.
Loading