gh-110774: allow setting the Runner(loop_factory=...) from IsolatedAsyncioTestCase (#110776)

Co-authored-by: Łukasz Langa <lukasz@langa.pl>
Co-authored-by: Carol Willing <carolcode@willingconsulting.com>
This commit is contained in:
Thomas Grainger 2023-10-31 15:03:54 -07:00 committed by GitHub
parent f6a02327b5
commit 770530679e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 28 additions and 2 deletions

View File

@ -1571,6 +1571,14 @@ Test cases
.. versionadded:: 3.8
.. attribute:: loop_factory
The *loop_factory* passed to :class:`asyncio.Runner`. Override
in subclasses with :class:`asyncio.EventLoop` to avoid using the
asyncio policy system.
.. versionadded:: 3.13
.. coroutinemethod:: asyncSetUp()
Method called to prepare the test fixture. This is called after :meth:`setUp`.

View File

@ -484,5 +484,19 @@ class TestAsyncCase(unittest.TestCase):
result = test.run()
self.assertTrue(result.wasSuccessful())
def test_loop_factory(self):
asyncio.set_event_loop_policy(None)
class TestCase1(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.EventLoop
async def test_demo1(self):
pass
test = TestCase1('test_demo1')
result = test.run()
self.assertTrue(result.wasSuccessful())
self.assertIsNone(support.maybe_get_event_loop_policy())
if __name__ == "__main__":
unittest.main()

View File

@ -25,12 +25,15 @@ class IsolatedAsyncioTestCase(TestCase):
# them inside the same task.
# Note: the test case modifies event loop policy if the policy was not instantiated
# yet.
# yet, unless loop_factory=asyncio.EventLoop is set.
# asyncio.get_event_loop_policy() creates a default policy on demand but never
# returns None
# I believe this is not an issue in user level tests but python itself for testing
# should reset a policy in every test module
# by calling asyncio.set_event_loop_policy(None) in tearDownModule()
# or set loop_factory=asyncio.EventLoop
loop_factory = None
def __init__(self, methodName='runTest'):
super().__init__(methodName)
@ -118,7 +121,7 @@ class IsolatedAsyncioTestCase(TestCase):
def _setupAsyncioRunner(self):
assert self._asyncioRunner is None, 'asyncio runner is already initialized'
runner = asyncio.Runner(debug=True)
runner = asyncio.Runner(debug=True, loop_factory=self.loop_factory)
self._asyncioRunner = runner
def _tearDownAsyncioRunner(self):

View File

@ -0,0 +1 @@
Support setting the :class:`asyncio.Runner` loop_factory kwarg in :class:`unittest.IsolatedAsyncioTestCase`