First step towards deprecating the asyncio policy system.
This deprecates `asyncio.set_event_loop_policy` and will be removed in Python 3.16.
If *policy* is set to ``None``, the default policy is restored.
+ .. deprecated:: next
+ The :func:`set_event_loop_policy` function is deprecated and
+ will be removed in Python 3.16.
+
.. _asyncio-policy-objects:
'AbstractEventLoopPolicy',
'AbstractEventLoop', 'AbstractServer',
'Handle', 'TimerHandle',
- 'get_event_loop_policy', 'set_event_loop_policy',
+ 'get_event_loop_policy',
+ '_set_event_loop_policy',
+ 'set_event_loop_policy',
'get_event_loop', 'set_event_loop', 'new_event_loop',
'_set_running_loop', 'get_running_loop',
'_get_running_loop',
import subprocess
import sys
import threading
+import warnings
from . import format_helpers
return _event_loop_policy
-def set_event_loop_policy(policy):
+def _set_event_loop_policy(policy):
"""Set the current event loop policy.
If policy is None, the default policy is restored."""
raise TypeError(f"policy must be an instance of AbstractEventLoopPolicy or None, not '{type(policy).__name__}'")
_event_loop_policy = policy
+def set_event_loop_policy(policy):
+ warnings._deprecated('set_event_loop_policy', remove=(3,16))
+ _set_event_loop_policy(policy)
def get_event_loop():
"""Return an asyncio event loop.
return support.maybe_get_event_loop_policy()
def restore_asyncio_events__event_loop_policy(self, policy):
asyncio = self.get_module('asyncio')
- asyncio.set_event_loop_policy(policy)
+ asyncio._set_event_loop_policy(policy)
def get_sys_argv(self):
return id(sys.argv), sys.argv, sys.argv[:]
def tearDown(self):
self.loop.close()
self.loop = None
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
def check_async_iterator_anext(self, ait_class):
with self.subTest(anext="pure-Python"):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
def mock_socket_module():
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class ReceiveStuffProto(asyncio.BufferedProtocol):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
@unittest.skipUnless(decimal.HAVE_CONTEXTVAR, "decimal is built with a thread-local context")
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class EagerTaskFactoryLoopTests:
from test.support import socket_helper
from test.support import threading_helper
from test.support import ALWAYS_EQ, LARGEST, SMALLEST
-
+from test.support import warnings_helper
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
def broken_unix_getsockname():
self.assertIs(policy, asyncio.get_event_loop_policy())
def test_set_event_loop_policy(self):
- self.assertRaises(
- TypeError, asyncio.set_event_loop_policy, object())
+ with self.assertWarnsRegex(
+ DeprecationWarning, "'set_event_loop_policy' is deprecated"):
+ self.assertRaises(
+ TypeError, asyncio.set_event_loop_policy, object())
old_policy = asyncio.get_event_loop_policy()
policy = asyncio.DefaultEventLoopPolicy()
- asyncio.set_event_loop_policy(policy)
+ with self.assertWarnsRegex(
+ DeprecationWarning, "'set_event_loop_policy' is deprecated"):
+ asyncio.set_event_loop_policy(policy)
self.assertIs(policy, asyncio.get_event_loop_policy())
self.assertIsNot(policy, old_policy)
old_policy = asyncio.get_event_loop_policy()
try:
- asyncio.set_event_loop_policy(Policy())
+ asyncio._set_event_loop_policy(Policy())
loop = asyncio.new_event_loop()
with self.assertRaises(TestError):
asyncio.get_event_loop()
finally:
- asyncio.set_event_loop_policy(old_policy)
+ asyncio._set_event_loop_policy(old_policy)
if loop is not None:
loop.close()
def test_get_event_loop_returns_running_loop2(self):
old_policy = asyncio.get_event_loop_policy()
try:
- asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
+ asyncio._set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
loop = asyncio.new_event_loop()
self.addCleanup(loop.close)
asyncio.get_event_loop()
finally:
- asyncio.set_event_loop_policy(old_policy)
+ asyncio._set_event_loop_policy(old_policy)
if loop is not None:
loop.close()
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
def _fakefunc(f):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class FutureTests:
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class LockTests(unittest.IsolatedAsyncioTestCase):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
# Test that asyncio.iscoroutine() uses collections.abc.Coroutine
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
def close_transport(transport):
def tearDownModule():
# not needed for the test file but added for uniformness with all other
# asyncio test files for the sake of unified cleanup
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class ProtocolsAbsTests(unittest.TestCase):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class QueueBasicTests(unittest.IsolatedAsyncioTestCase):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
def interrupt_self():
super().setUp()
policy = TestPolicy(self.new_loop)
- asyncio.set_event_loop_policy(policy)
+ asyncio._set_event_loop_policy(policy)
def tearDown(self):
policy = asyncio.get_event_loop_policy()
self.assertTrue(policy.loop.is_closed())
self.assertTrue(policy.loop.shutdown_ag_run)
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
super().tearDown()
loop.set_task_factory(Task)
return loop
- asyncio.set_event_loop_policy(TestPolicy(new_event_loop))
+ asyncio._set_event_loop_policy(TestPolicy(new_event_loop))
with self.assertRaises(asyncio.CancelledError):
asyncio.run(main())
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class TestBaseSelectorEventLoop(BaseSelectorEventLoop):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class MySendfileProto(asyncio.Protocol):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class BaseStartServer(func_tests.FunctionalTestCaseMixin):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class MyProto(asyncio.Protocol):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class MyBaseProto(asyncio.Protocol):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
@unittest.skipIf(ssl is None, 'No ssl module')
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class StaggeredTests(unittest.IsolatedAsyncioTestCase):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class StreamTests(test_utils.TestCase):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class TestSubprocessTransport(base_subprocess.BaseSubprocessTransport):
# To prevent a warning "test altered the execution environment"
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class MyExc(Exception):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
async def coroutine_function():
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class ToThreadTests(unittest.IsolatedAsyncioTestCase):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class TimeoutTests(unittest.IsolatedAsyncioTestCase):
def tearDownModule():
# not needed for the test file but added for uniformness with all other
# asyncio test files for the sake of unified cleanup
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class TransportTests(unittest.TestCase):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
MOCK_ANY = mock.ANY
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
# The following value can be used as a very small timeout:
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class UpperProto(asyncio.Protocol):
old_policy = asyncio.get_event_loop_policy()
try:
- asyncio.set_event_loop_policy(
+ asyncio._set_event_loop_policy(
asyncio.WindowsSelectorEventLoopPolicy())
asyncio.run(main())
finally:
- asyncio.set_event_loop_policy(old_policy)
+ asyncio._set_event_loop_policy(old_policy)
def test_proactor_win_policy(self):
async def main():
old_policy = asyncio.get_event_loop_policy()
try:
- asyncio.set_event_loop_policy(
+ asyncio._set_event_loop_policy(
asyncio.WindowsProactorEventLoopPolicy())
asyncio.run(main())
finally:
- asyncio.set_event_loop_policy(old_policy)
+ asyncio._set_event_loop_policy(old_policy)
if __name__ == '__main__':
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class PipeTests(unittest.TestCase):
asyncio.run(eval(co, globals_))
self.assertEqual(globals_['a'], 1)
finally:
- asyncio.set_event_loop_policy(policy)
+ asyncio._set_event_loop_policy(policy)
def test_compile_top_level_await_invalid_cases(self):
# helper function just to check we can run top=level async-for
mode,
flags=ast.PyCF_ALLOW_TOP_LEVEL_AWAIT)
finally:
- asyncio.set_event_loop_policy(policy)
+ asyncio._set_event_loop_policy(policy)
def test_compile_async_generator(self):
# tests left a policy in place, just in case.
policy = support.maybe_get_event_loop_policy()
assert policy is None, policy
- cls.addClassCleanup(lambda: asyncio.set_event_loop_policy(None))
+ cls.addClassCleanup(lambda: asyncio._set_event_loop_policy(None))
def setUp(self):
super().setUp()
support.requires_working_socket(module=True)
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class TestAbstractAsyncContextManager(unittest.IsolatedAsyncioTestCase):
pass
finally:
loop.close()
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
self.assertEqual(buffer, [1, 2, 'MyException'])
def tearDownModule():
if support.has_socket_support:
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
def signatures_with_lexicographic_keyword_only_parameters():
logging.logAsyncioTasks = False
runner.run(make_record(self.assertIsNone))
finally:
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
@support.requires_working_socket()
def test_taskName_without_asyncio_imported(self):
logging.logAsyncioTasks = False
runner.run(make_record(self.assertIsNone))
finally:
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class BasicConfigTest(unittest.TestCase):
data = f.read().strip()
self.assertRegex(data, r'Task-\d+ - hello world')
finally:
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
if handler:
handler.close()
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class MiscTests(unittest.TestCase):
... loop = asyncio.new_event_loop()
... loop.run_until_complete(test_main())
... loop.close()
- ... asyncio.set_event_loop_policy(None)
+ ... asyncio._set_event_loop_policy(None)
... print("finished")
>>> with PdbTestInput(['step',
... loop = asyncio.new_event_loop()
... loop.run_until_complete(test_main())
... loop.close()
- ... asyncio.set_event_loop_policy(None)
+ ... asyncio._set_event_loop_policy(None)
... print("finished")
>>> with PdbTestInput(['step',
... loop = asyncio.new_event_loop()
... loop.run_until_complete(test_main())
... loop.close()
- ... asyncio.set_event_loop_policy(None)
+ ... asyncio._set_event_loop_policy(None)
... print("finished")
>>> with PdbTestInput(['step',
asyncio.run(func(output))
sys.settrace(None)
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
self.compare_jump_output(expected, output)
def jump_test(jumpFrom, jumpTo, expected, error=None, event='line', warning=None):
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class TestCM:
self.assertTrue(result.wasSuccessful())
def test_loop_factory(self):
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class TestCase1(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.EventLoop
def tearDownModule():
- asyncio.set_event_loop_policy(None)
+ asyncio._set_event_loop_policy(None)
class AsyncClass:
projects / thirdparty / Python / cpython.git / commitdiff
