mirror of
https://github.com/python/cpython.git
synced 2025-07-21 10:15:46 +00:00
4dd3e3f9bb
Add explicit `asyncSetUp` and `asyncTearDown` methods. The rest is the same as for #13228 `AsyncTestCase` create a loop instance for every test for the sake of test isolation. Sometimes a loop shared between all tests can speed up tests execution time a lot but it requires control of closed resources after every test finish. Basically, it requires nested supervisors support that was discussed with @1st1 many times. Sorry, asyncio supervisors have no chance to land on Python 3.8. The PR intentionally does not provide API for changing the used event loop or getting the test loop: use `asyncio.set_event_loop_policy()` and `asyncio.get_event_loop()` instead. The PR adds four overridable methods to base `unittest.TestCase` class: ``` def _callSetUp(self): self.setUp() def _callTestMethod(self, method): method() def _callTearDown(self): self.tearDown() def _callCleanup(self, function, /, *args, **kwargs): function(*args, **kwargs) ``` It allows using asyncio facilities with minimal influence on the unittest code. The last but not least: the PR respects contextvars. The context variable installed by `asyncSetUp` is available on test, `tearDown` and a coroutine scheduled by `addCleanup`. https://bugs.python.org/issue32972
195 lines
6.1 KiB
Python
195 lines
6.1 KiB
Python
import asyncio
|
|
import unittest
|
|
|
|
|
|
def tearDownModule():
|
|
asyncio.set_event_loop_policy(None)
|
|
|
|
|
|
class TestAsyncCase(unittest.TestCase):
|
|
def test_full_cycle(self):
|
|
events = []
|
|
|
|
class Test(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
self.assertEqual(events, [])
|
|
events.append('setUp')
|
|
|
|
async def asyncSetUp(self):
|
|
self.assertEqual(events, ['setUp'])
|
|
events.append('asyncSetUp')
|
|
|
|
async def test_func(self):
|
|
self.assertEqual(events, ['setUp',
|
|
'asyncSetUp'])
|
|
events.append('test')
|
|
self.addAsyncCleanup(self.on_cleanup)
|
|
|
|
async def asyncTearDown(self):
|
|
self.assertEqual(events, ['setUp',
|
|
'asyncSetUp',
|
|
'test'])
|
|
events.append('asyncTearDown')
|
|
|
|
def tearDown(self):
|
|
self.assertEqual(events, ['setUp',
|
|
'asyncSetUp',
|
|
'test',
|
|
'asyncTearDown'])
|
|
events.append('tearDown')
|
|
|
|
async def on_cleanup(self):
|
|
self.assertEqual(events, ['setUp',
|
|
'asyncSetUp',
|
|
'test',
|
|
'asyncTearDown',
|
|
'tearDown'])
|
|
events.append('cleanup')
|
|
|
|
test = Test("test_func")
|
|
test.run()
|
|
self.assertEqual(events, ['setUp',
|
|
'asyncSetUp',
|
|
'test',
|
|
'asyncTearDown',
|
|
'tearDown',
|
|
'cleanup'])
|
|
|
|
def test_exception_in_setup(self):
|
|
events = []
|
|
|
|
class Test(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
events.append('asyncSetUp')
|
|
raise Exception()
|
|
|
|
async def test_func(self):
|
|
events.append('test')
|
|
self.addAsyncCleanup(self.on_cleanup)
|
|
|
|
async def asyncTearDown(self):
|
|
events.append('asyncTearDown')
|
|
|
|
async def on_cleanup(self):
|
|
events.append('cleanup')
|
|
|
|
|
|
test = Test("test_func")
|
|
test.run()
|
|
self.assertEqual(events, ['asyncSetUp'])
|
|
|
|
def test_exception_in_test(self):
|
|
events = []
|
|
|
|
class Test(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
events.append('asyncSetUp')
|
|
|
|
async def test_func(self):
|
|
events.append('test')
|
|
raise Exception()
|
|
self.addAsyncCleanup(self.on_cleanup)
|
|
|
|
async def asyncTearDown(self):
|
|
events.append('asyncTearDown')
|
|
|
|
async def on_cleanup(self):
|
|
events.append('cleanup')
|
|
|
|
test = Test("test_func")
|
|
test.run()
|
|
self.assertEqual(events, ['asyncSetUp', 'test', 'asyncTearDown'])
|
|
|
|
def test_exception_in_test_after_adding_cleanup(self):
|
|
events = []
|
|
|
|
class Test(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
events.append('asyncSetUp')
|
|
|
|
async def test_func(self):
|
|
events.append('test')
|
|
self.addAsyncCleanup(self.on_cleanup)
|
|
raise Exception()
|
|
|
|
async def asyncTearDown(self):
|
|
events.append('asyncTearDown')
|
|
|
|
async def on_cleanup(self):
|
|
events.append('cleanup')
|
|
|
|
test = Test("test_func")
|
|
test.run()
|
|
self.assertEqual(events, ['asyncSetUp', 'test', 'asyncTearDown', 'cleanup'])
|
|
|
|
def test_exception_in_tear_down(self):
|
|
events = []
|
|
|
|
class Test(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
events.append('asyncSetUp')
|
|
|
|
async def test_func(self):
|
|
events.append('test')
|
|
self.addAsyncCleanup(self.on_cleanup)
|
|
|
|
async def asyncTearDown(self):
|
|
events.append('asyncTearDown')
|
|
raise Exception()
|
|
|
|
async def on_cleanup(self):
|
|
events.append('cleanup')
|
|
|
|
test = Test("test_func")
|
|
test.run()
|
|
self.assertEqual(events, ['asyncSetUp', 'test', 'asyncTearDown', 'cleanup'])
|
|
|
|
|
|
def test_exception_in_tear_clean_up(self):
|
|
events = []
|
|
|
|
class Test(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
events.append('asyncSetUp')
|
|
|
|
async def test_func(self):
|
|
events.append('test')
|
|
self.addAsyncCleanup(self.on_cleanup)
|
|
|
|
async def asyncTearDown(self):
|
|
events.append('asyncTearDown')
|
|
|
|
async def on_cleanup(self):
|
|
events.append('cleanup')
|
|
raise Exception()
|
|
|
|
test = Test("test_func")
|
|
test.run()
|
|
self.assertEqual(events, ['asyncSetUp', 'test', 'asyncTearDown', 'cleanup'])
|
|
|
|
def test_cleanups_interleave_order(self):
|
|
events = []
|
|
|
|
class Test(unittest.IsolatedAsyncioTestCase):
|
|
async def test_func(self):
|
|
self.addAsyncCleanup(self.on_sync_cleanup, 1)
|
|
self.addAsyncCleanup(self.on_async_cleanup, 2)
|
|
self.addAsyncCleanup(self.on_sync_cleanup, 3)
|
|
self.addAsyncCleanup(self.on_async_cleanup, 4)
|
|
|
|
async def on_sync_cleanup(self, val):
|
|
events.append(f'sync_cleanup {val}')
|
|
|
|
async def on_async_cleanup(self, val):
|
|
events.append(f'async_cleanup {val}')
|
|
|
|
test = Test("test_func")
|
|
test.run()
|
|
self.assertEqual(events, ['async_cleanup 4',
|
|
'sync_cleanup 3',
|
|
'async_cleanup 2',
|
|
'sync_cleanup 1'])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|