mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-03 16:39:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

gh-100160: Restore and deprecate implicit creation of an event loop (GH-100410)

Browse source 
Partially revert changes made in GH-93453.

asyncio.DefaultEventLoopPolicy.get_event_loop() now emits a
DeprecationWarning and creates and sets a new event loop instead of
raising a RuntimeError if there is no current event loop set.

Co-authored-by: Guido van Rossum <gvanrossum@gmail.com>
This commit is contained in:
Serhiy Storchaka 2023-01-13 14:40:29 +02:00 • committed by GitHub
parent 468c3bf798
commit e5bd5ad70d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 77 additions and 41 deletions
Download patch file Download diff file Expand all files Collapse all files

35
Lib/test/test_asyncio/test_events.py
View file

@ -2614,8 +2614,33 @@ class PolicyTests(unittest.TestCase):
def test_get_event_loop(self):
policy = asyncio.DefaultEventLoopPolicy()
self.assertIsNone(policy._local._loop)
with self.assertRaisesRegex(RuntimeError, 'no current event loop'):
policy.get_event_loop()
with self.assertWarns(DeprecationWarning) as cm:
loop = policy.get_event_loop()
self.assertEqual(cm.filename, __file__)
self.assertIsInstance(loop, asyncio.AbstractEventLoop)
self.assertIs(policy._local._loop, loop)
self.assertIs(loop, policy.get_event_loop())
loop.close()
def test_get_event_loop_calls_set_event_loop(self):
policy = asyncio.DefaultEventLoopPolicy()
with mock.patch.object(
policy, "set_event_loop",
wraps=policy.set_event_loop) as m_set_event_loop:
with self.assertWarns(DeprecationWarning) as cm:
loop = policy.get_event_loop()
self.addCleanup(loop.close)
self.assertEqual(cm.filename, __file__)
# policy._local._loop must be set through .set_event_loop()
# (the unix DefaultEventLoopPolicy needs this call to attach
# the child watcher correctly)
m_set_event_loop.assert_called_with(loop)
loop.close()
def test_get_event_loop_after_set_none(self):
policy = asyncio.DefaultEventLoopPolicy()
@ -2801,8 +2826,10 @@ class GetEventLoopTestsMixin:
loop = asyncio.new_event_loop()
self.addCleanup(loop.close)
with self.assertRaisesRegex(RuntimeError, 'no current'):
asyncio.get_event_loop()
with self.assertWarns(DeprecationWarning) as cm:
loop2 = asyncio.get_event_loop()
self.addCleanup(loop2.close)
self.assertEqual(cm.filename, __file__)
asyncio.set_event_loop(None)
with self.assertRaisesRegex(RuntimeError, 'no current'):
asyncio.get_event_loop()