I want to be able to create utility subclasses of IsolatedAsyncioTestCase that use either uvloop or a specific event loop using the asyncio.Runner(..., loop_factory=...) kwarg, Ideally it would look something like:
Another advantage of setting loop_factory would allow use of IsolatedAsyncioTestCase without using def tearDownModule(): asyncio.set_event_loop_policy(None) because asyncio.Runner does not call asyncio.get_event_loop_policy() when called with a loop_factory
class DefaultIsolatedAsyncioTestCase:
"""
Use the most efficient event loop regardless of what has been configured with asyncio.set_event_loop_policy
does not require `def tearDownModule(): asyncio.set_event_loop_policy(None)`
"""
loop_factory = (
asyncio.ProactorEventLoop
if sys.platform == "win32"
else asyncio.SelectorEventLoop
)
or
class UvloopIsolatedAsyncioTestCase:
"""
runs tests on uvloop
"""
loop_factory = uvloop.new_event_loop
the patch to CPython would look like this:
diff --git a/Lib/unittest/async_case.py b/Lib/unittest/async_case.py
index bd2a471156..beeb7037d9 100644
--- a/Lib/unittest/async_case.py
+++ b/Lib/unittest/async_case.py
@@ -32,6 +32,8 @@ class IsolatedAsyncioTestCase(TestCase):
# should reset a policy in every test module
# by calling asyncio.set_event_loop_policy(None) in tearDownModule()
+ loop_factory = None
+
def __init__(self, methodName='runTest'):
super().__init__(methodName)
self._asyncioRunner = None
@@ -118,7 +120,7 @@ def _callMaybeAsync(self, func, /, *args, **kwargs):
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):