def __getattr__(name: str):
import warnings
- deprecated = {
- "AbstractEventLoopPolicy",
- "DefaultEventLoopPolicy",
- "WindowsSelectorEventLoopPolicy",
- "WindowsProactorEventLoopPolicy",
- }
- if name in deprecated:
- warnings._deprecated(f"asyncio.{name}", remove=(3, 16))
- # deprecated things have underscores in front of them
- return globals()["_" + name]
+ match name:
+ case "AbstractEventLoopPolicy":
+ warnings._deprecated(f"asyncio.{name}", remove=(3, 16))
+ return events._AbstractEventLoopPolicy
+ case "DefaultEventLoopPolicy":
+ warnings._deprecated(f"asyncio.{name}", remove=(3, 16))
+ if sys.platform == 'win32':
+ return windows_events._DefaultEventLoopPolicy
+ return unix_events._DefaultEventLoopPolicy
+ case "WindowsSelectorEventLoopPolicy":
+ if sys.platform == 'win32':
+ warnings._deprecated(f"asyncio.{name}", remove=(3, 16))
+ return windows_events._WindowsSelectorEventLoopPolicy
+ # Else fall through to the AttributeError below.
+ case "WindowsProactorEventLoopPolicy":
+ if sys.platform == 'win32':
+ warnings._deprecated(f"asyncio.{name}", remove=(3, 16))
+ return windows_events._WindowsProactorEventLoopPolicy
+ # Else fall through to the AttributeError below.
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
# SPDX-FileCopyrightText: Copyright (c) 2015-2021 MagicStack Inc. http://magic.io
__all__ = (
- "_AbstractEventLoopPolicy",
"AbstractEventLoop",
"AbstractServer",
"Handle",
"TimerHandle",
- "_get_event_loop_policy",
"get_event_loop_policy",
- "_set_event_loop_policy",
"set_event_loop_policy",
"get_event_loop",
"set_event_loop",
global _event_loop_policy
with _lock:
if _event_loop_policy is None: # pragma: no branch
- from . import _DefaultEventLoopPolicy
+ if sys.platform == 'win32':
+ from .windows_events import _DefaultEventLoopPolicy
+ else:
+ from .unix_events import _DefaultEventLoopPolicy
_event_loop_policy = _DefaultEventLoopPolicy()
__all__ = (
'SelectorEventLoop',
- '_DefaultEventLoopPolicy',
'EventLoop',
)
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.events._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.events._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.events._set_event_loop_policy(None)
def mock_socket_module():
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class ReceiveStuffProto(asyncio.BufferedProtocol):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._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.events._set_event_loop_policy(None)
class EagerTaskFactoryLoopTests:
from test.support import ALWAYS_EQ, LARGEST, SMALLEST
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
def broken_unix_getsockname():
self.assertIsInstance(policy, asyncio.DefaultEventLoopPolicy)
def test_event_loop_policy(self):
- policy = asyncio._AbstractEventLoopPolicy()
+ policy = asyncio.events._AbstractEventLoopPolicy()
self.assertRaises(NotImplementedError, policy.get_event_loop)
self.assertRaises(NotImplementedError, policy.set_event_loop, object())
self.assertRaises(NotImplementedError, policy.new_event_loop)
def test_get_event_loop(self):
- policy = asyncio._DefaultEventLoopPolicy()
+ policy = test_utils.DefaultEventLoopPolicy()
self.assertIsNone(policy._local._loop)
with self.assertRaises(RuntimeError):
self.assertIsNone(policy._local._loop)
def test_get_event_loop_does_not_call_set_event_loop(self):
- policy = asyncio._DefaultEventLoopPolicy()
+ policy = test_utils.DefaultEventLoopPolicy()
with mock.patch.object(
policy, "set_event_loop",
m_set_event_loop.assert_not_called()
def test_get_event_loop_after_set_none(self):
- policy = asyncio._DefaultEventLoopPolicy()
+ policy = test_utils.DefaultEventLoopPolicy()
policy.set_event_loop(None)
self.assertRaises(RuntimeError, policy.get_event_loop)
def test_get_event_loop_thread(self, m_current_thread):
def f():
- policy = asyncio._DefaultEventLoopPolicy()
+ policy = test_utils.DefaultEventLoopPolicy()
self.assertRaises(RuntimeError, policy.get_event_loop)
th = threading.Thread(target=f)
th.join()
def test_new_event_loop(self):
- policy = asyncio._DefaultEventLoopPolicy()
+ policy = test_utils.DefaultEventLoopPolicy()
loop = policy.new_event_loop()
self.assertIsInstance(loop, asyncio.AbstractEventLoop)
loop.close()
def test_set_event_loop(self):
- policy = asyncio._DefaultEventLoopPolicy()
+ policy = test_utils.DefaultEventLoopPolicy()
old_loop = policy.new_event_loop()
policy.set_event_loop(old_loop)
with self.assertWarnsRegex(
DeprecationWarning, "'asyncio.get_event_loop_policy' is deprecated"):
policy = asyncio.get_event_loop_policy()
- self.assertIsInstance(policy, asyncio._AbstractEventLoopPolicy)
+ self.assertIsInstance(policy, asyncio.events._AbstractEventLoopPolicy)
self.assertIs(policy, asyncio.get_event_loop_policy())
def test_set_event_loop_policy(self):
DeprecationWarning, "'asyncio.get_event_loop_policy' is deprecated"):
old_policy = asyncio.get_event_loop_policy()
- policy = asyncio._DefaultEventLoopPolicy()
+ policy = test_utils.DefaultEventLoopPolicy()
with self.assertWarnsRegex(
DeprecationWarning, "'asyncio.set_event_loop_policy' is deprecated"):
asyncio.set_event_loop_policy(policy)
class TestError(Exception):
pass
- class Policy(asyncio._DefaultEventLoopPolicy):
+ class Policy(test_utils.DefaultEventLoopPolicy):
def get_event_loop(self):
raise TestError
- old_policy = asyncio._get_event_loop_policy()
+ old_policy = asyncio.events._get_event_loop_policy()
try:
- asyncio._set_event_loop_policy(Policy())
+ asyncio.events._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.events._set_event_loop_policy(old_policy)
if loop is not None:
loop.close()
self.assertIs(asyncio._get_running_loop(), None)
def test_get_event_loop_returns_running_loop2(self):
- old_policy = asyncio._get_event_loop_policy()
+ old_policy = asyncio.events._get_event_loop_policy()
try:
- asyncio._set_event_loop_policy(asyncio._DefaultEventLoopPolicy())
+ asyncio.events._set_event_loop_policy(test_utils.DefaultEventLoopPolicy())
loop = asyncio.new_event_loop()
self.addCleanup(loop.close)
asyncio.get_event_loop()
finally:
- asyncio._set_event_loop_policy(old_policy)
+ asyncio.events._set_event_loop_policy(old_policy)
if loop is not None:
loop.close()
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class TestFreeThreading:
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
def _fakefunc(f):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class FutureTests:
# To prevent a warning "test altered the execution environment"
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
def capture_test_stack(*, fut=None, depth=1):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class LockTests(unittest.IsolatedAsyncioTestCase):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
# Test that asyncio.iscoroutine() uses collections.abc.Coroutine
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._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.events._set_event_loop_policy(None)
class ProtocolsAbsTests(unittest.TestCase):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class QueueBasicTests(unittest.IsolatedAsyncioTestCase):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
def interrupt_self():
_thread.interrupt_main()
-class TestPolicy(asyncio._AbstractEventLoopPolicy):
+class TestPolicy(asyncio.events._AbstractEventLoopPolicy):
def __init__(self, loop_factory):
self.loop_factory = loop_factory
super().setUp()
policy = TestPolicy(self.new_loop)
- asyncio._set_event_loop_policy(policy)
+ asyncio.events._set_event_loop_policy(policy)
def tearDown(self):
- policy = asyncio._get_event_loop_policy()
+ policy = asyncio.events._get_event_loop_policy()
if policy.loop is not None:
self.assertTrue(policy.loop.is_closed())
self.assertTrue(policy.loop.shutdown_ag_run)
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
super().tearDown()
await asyncio.sleep(0)
return 42
- policy = asyncio._get_event_loop_policy()
+ policy = asyncio.events._get_event_loop_policy()
policy.set_event_loop = mock.Mock()
asyncio.run(main())
self.assertTrue(policy.set_event_loop.called)
loop.set_task_factory(Task)
return loop
- asyncio._set_event_loop_policy(TestPolicy(new_event_loop))
+ asyncio.events._set_event_loop_policy(TestPolicy(new_event_loop))
with self.assertRaises(asyncio.CancelledError):
asyncio.run(main())
async def coro():
pass
- policy = asyncio._get_event_loop_policy()
+ policy = asyncio.events._get_event_loop_policy()
policy.set_event_loop = mock.Mock()
runner = asyncio.Runner()
runner.run(coro())
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class TestBaseSelectorEventLoop(BaseSelectorEventLoop):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class MySendfileProto(asyncio.Protocol):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class BaseStartServer(func_tests.FunctionalTestCaseMixin):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class MyProto(asyncio.Protocol):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class MyBaseProto(asyncio.Protocol):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
@unittest.skipIf(ssl is None, 'No ssl module')
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class StaggeredTests(unittest.IsolatedAsyncioTestCase):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class StreamTests(test_utils.TestCase):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._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.events._set_event_loop_policy(None)
class MyExc(Exception):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
async def coroutine_function():
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class ToThreadTests(unittest.IsolatedAsyncioTestCase):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._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.events._set_event_loop_policy(None)
class TransportTests(unittest.TestCase):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
MOCK_ANY = mock.ANY
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._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.events._set_event_loop_policy(None)
class UpperProto(asyncio.Protocol):
async def main():
self.assertIsInstance(asyncio.get_running_loop(), asyncio.SelectorEventLoop)
- old_policy = asyncio._get_event_loop_policy()
+ old_policy = asyncio.events._get_event_loop_policy()
try:
with self.assertWarnsRegex(
DeprecationWarning,
"'asyncio.WindowsSelectorEventLoopPolicy' is deprecated",
):
- asyncio._set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+ asyncio.events._set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
asyncio.run(main())
finally:
- asyncio._set_event_loop_policy(old_policy)
+ asyncio.events._set_event_loop_policy(old_policy)
def test_proactor_win_policy(self):
async def main():
asyncio.get_running_loop(),
asyncio.ProactorEventLoop)
- old_policy = asyncio._get_event_loop_policy()
+ old_policy = asyncio.events._get_event_loop_policy()
try:
with self.assertWarnsRegex(
DeprecationWarning,
"'asyncio.WindowsProactorEventLoopPolicy' is deprecated",
):
- asyncio._set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
+ asyncio.events._set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
asyncio.run(main())
finally:
- asyncio._set_event_loop_policy(old_policy)
+ asyncio.events._set_event_loop_policy(old_policy)
if __name__ == '__main__':
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class PipeTests(unittest.TestCase):
await asyncio.sleep(0)
if exc is not None:
raise exc
+
+
+if sys.platform == 'win32':
+ DefaultEventLoopPolicy = asyncio.windows_events._DefaultEventLoopPolicy
+else:
+ DefaultEventLoopPolicy = asyncio.unix_events._DefaultEventLoopPolicy
# 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.events._set_event_loop_policy(None))
def setUp(self):
super().setUp()
pass
finally:
loop.close()
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
self.assertEqual(buffer, [1, 2, 'MyException'])
@classmethod
def tearDownClass(cls):
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
def _asyncgenstate(self):
return inspect.getasyncgenstate(self.asyncgen)
logging.logAsyncioTasks = False
runner.run(make_record(self.assertIsNone))
finally:
- asyncio._set_event_loop_policy(None)
+ asyncio.events._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.events._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.events._set_event_loop_policy(None)
if handler:
handler.close()
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class MiscTests(unittest.TestCase):
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class TestCM:
class TestCase1(unittest.IsolatedAsyncioTestCase):
def setUp(self):
- asyncio._get_event_loop_policy().get_event_loop()
+ asyncio.events._get_event_loop_policy().get_event_loop()
async def test_demo1(self):
pass
self.assertTrue(result.wasSuccessful())
def test_loop_factory(self):
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class TestCase1(unittest.IsolatedAsyncioTestCase):
loop_factory = asyncio.EventLoop
def tearDownModule():
- asyncio._set_event_loop_policy(None)
+ asyncio.events._set_event_loop_policy(None)
class AsyncClass:
--- /dev/null
+:mod:`asyncio`: Remove some private names from ``asyncio.__all__``.
projects / thirdparty / Python / cpython.git / commitdiff
