summaryrefslogtreecommitdiffstats
path: root/Lib
diff options
context:
space:
mode:
Diffstat (limited to 'Lib')
-rw-r--r--Lib/test/test_unittest/test_async_case.py14
-rw-r--r--Lib/unittest/async_case.py7
2 files changed, 19 insertions, 2 deletions
diff --git a/Lib/test/test_unittest/test_async_case.py b/Lib/test/test_unittest/test_async_case.py
index a465103..ba1ab83 100644
--- a/Lib/test/test_unittest/test_async_case.py
+++ b/Lib/test/test_unittest/test_async_case.py
@@ -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()
diff --git a/Lib/unittest/async_case.py b/Lib/unittest/async_case.py
index bd2a471..63ff6a5 100644
--- a/Lib/unittest/async_case.py
+++ b/Lib/unittest/async_case.py
@@ -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):