|
|
@@ -1,12 +1,18 @@ |
|
|
import contextvars |
|
|
import enum |
|
|
import functools |
|
|
import os |
|
|
import signal |
|
|
import sys |
|
|
import threading |
|
|
import unittest |
|
|
import warnings |
|
|
from unittest import mock |
|
|
|
|
|
import asyncio |
|
|
import asyncio.runners |
|
|
from asyncio import base_subprocess |
|
|
from asyncio import coroutines |
|
|
from asyncio import events |
|
|
from asyncio import subprocess |
|
|
from test.test_asyncio import utils as test_utils |
|
|
from test import support |
|
Expand Down
Expand Up
|
@@ -636,6 +642,147 @@ async def execute(): |
|
|
self.assertIsNone(self.loop.run_until_complete(execute())) |
|
|
|
|
|
|
|
|
class _State(enum.Enum): |
|
|
CREATED = "created" |
|
|
INITIALIZED = "initialized" |
|
|
CLOSED = "closed" |
|
|
|
|
|
class Runner: |
|
|
"""A context manager that controls event loop life cycle. |
|
|
|
|
|
The context manager always creates a new event loop, |
|
|
allows to run async functions inside it, |
|
|
and properly finalizes the loop at the context manager exit. |
|
|
|
|
|
If debug is True, the event loop will be run in debug mode. |
|
|
If loop_factory is passed, it is used for new event loop creation. |
|
|
|
|
|
asyncio.run(main(), debug=True) |
|
|
|
|
|
is a shortcut for |
|
|
|
|
|
with asyncio.Runner(debug=True) as runner: |
|
|
runner.run(main()) |
|
|
|
|
|
The run() method can be called multiple times within the runner's context. |
|
|
|
|
|
This can be useful for interactive console (e.g. IPython), |
|
|
unittest runners, console tools, -- everywhere when async code |
|
|
is called from existing sync framework and where the preferred single |
|
|
asyncio.run() call doesn't work. |
|
|
|
|
|
""" |
|
|
|
|
|
# Note: the class is final, it is not intended for inheritance. |
|
|
|
|
|
def __init__(self, *, debug=None, loop_factory=None): |
|
|
self._state = _State.CREATED |
|
|
self._debug = debug |
|
|
self._loop_factory = loop_factory |
|
|
self._loop = None |
|
|
self._context = None |
|
|
self._interrupt_count = 0 |
|
|
self._set_event_loop = False |
|
|
|
|
|
def __enter__(self): |
|
|
self._lazy_init() |
|
|
return self |
|
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb): |
|
|
self.close() |
|
|
|
|
|
def close(self): |
|
|
"""Shutdown and close event loop.""" |
|
|
if self._state is not _State.INITIALIZED: |
|
|
return |
|
|
try: |
|
|
loop = self._loop |
|
|
asyncio.runners._cancel_all_tasks(loop) |
|
|
loop.run_until_complete(loop.shutdown_asyncgens()) |
|
|
loop.run_until_complete(loop.shutdown_default_executor()) |
|
|
finally: |
|
|
if self._set_event_loop: |
|
|
events.set_event_loop(None) |
|
|
loop.close() |
|
|
self._loop = None |
|
|
self._state = _State.CLOSED |
|
|
|
|
|
def get_loop(self): |
|
|
"""Return embedded event loop.""" |
|
|
self._lazy_init() |
|
|
return self._loop |
|
|
|
|
|
def run(self, coro, *, context=None): |
|
|
"""Run a coroutine inside the embedded event loop.""" |
|
|
if not coroutines.iscoroutine(coro): |
|
|
raise ValueError("a coroutine was expected, got {!r}".format(coro)) |
|
|
|
|
|
if events._get_running_loop() is not None: |
|
|
# fail fast with short traceback |
|
|
raise RuntimeError( |
|
|
"Runner.run() cannot be called from a running event loop") |
|
|
|
|
|
self._lazy_init() |
|
|
|
|
|
if context is None: |
|
|
context = self._context |
|
|
task = self._loop.create_task(coro) |
|
|
|
|
|
if (threading.current_thread() is threading.main_thread() |
|
|
and signal.getsignal(signal.SIGINT) is signal.default_int_handler |
|
|
): |
|
|
sigint_handler = functools.partial(self._on_sigint, main_task=task) |
|
|
try: |
|
|
signal.signal(signal.SIGINT, sigint_handler) |
|
|
except ValueError: |
|
|
# `signal.signal` may throw if `threading.main_thread` does |
|
|
# not support signals (e.g. embedded interpreter with signals |
|
|
# not registered - see gh-91880) |
|
|
sigint_handler = None |
|
|
else: |
|
|
sigint_handler = None |
|
|
|
|
|
self._interrupt_count = 0 |
|
|
try: |
|
|
if self._set_event_loop: |
|
|
events.set_event_loop(self._loop) |
|
|
return self._loop.run_until_complete(task) |
|
|
except exceptions.CancelledError: |
|
|
if self._interrupt_count > 0 and task.uncancel() == 0: |
|
|
raise KeyboardInterrupt() |
|
|
else: |
|
|
raise # CancelledError |
|
|
finally: |
|
|
if (sigint_handler is not None |
|
|
and signal.getsignal(signal.SIGINT) is sigint_handler |
|
|
): |
|
|
signal.signal(signal.SIGINT, signal.default_int_handler) |
|
|
|
|
|
def _lazy_init(self): |
|
|
if self._state is _State.CLOSED: |
|
|
raise RuntimeError("Runner is closed") |
|
|
if self._state is _State.INITIALIZED: |
|
|
return |
|
|
if self._loop_factory is None: |
|
|
self._loop = events.new_event_loop() |
|
|
self._set_event_loop = True |
|
|
else: |
|
|
self._loop = self._loop_factory() |
|
|
if self._debug is not None: |
|
|
self._loop.set_debug(self._debug) |
|
|
self._context = contextvars.copy_context() |
|
|
self._state = _State.INITIALIZED |
|
|
|
|
|
def _on_sigint(self, signum, frame, main_task): |
|
|
self._interrupt_count += 1 |
|
|
if self._interrupt_count == 1 and not main_task.done(): |
|
|
main_task.cancel() |
|
|
# wakeup loop if it is blocked by select() with long timeout |
|
|
self._loop.call_soon_threadsafe(lambda: None) |
|
|
return |
|
|
raise KeyboardInterrupt() |
|
|
|
|
|
|
|
|
if sys.platform != 'win32': |
|
|
# Unix |
|
|
class SubprocessWatcherMixin(SubprocessMixin): |
|
Expand Down
Expand Up
|
@@ -699,34 +846,40 @@ class SubprocessPidfdWatcherTests(SubprocessWatcherMixin, |
|
|
test_utils.TestCase): |
|
|
Watcher = unix_events.PidfdChildWatcher |
|
|
|
|
|
else: |
|
|
# Windows |
|
|
class SubprocessProactorTests(SubprocessMixin, test_utils.TestCase): |
|
|
|
|
|
def setUp(self): |
|
|
super().setUp() |
|
|
self.loop = asyncio.ProactorEventLoop() |
|
|
self.set_event_loop(self.loop) |
|
|
|
|
|
|
|
|
class GenericWatcherTests: |
|
|
class GenericWatcherTests(test_utils.TestCase): |
|
|
|
|
|
def test_create_subprocess_fails_with_inactive_watcher(self): |
|
|
def test_create_subprocess_fails_with_inactive_watcher(self): |
|
|
watcher = mock.create_autospec( |
|
|
asyncio.AbstractChildWatcher, |
|
|
**{"__enter__.return_value.is_active.return_value": False} |
|
|
) |
|
|
|
|
|
async def execute(): |
|
|
watcher = mock.create_authspec(asyncio.AbstractChildWatcher) |
|
|
watcher.is_active.return_value = False |
|
|
asyncio.set_child_watcher(watcher) |
|
|
async def execute(): |
|
|
asyncio.set_child_watcher(watcher) |
|
|
|
|
|
with self.assertRaises(RuntimeError): |
|
|
await subprocess.create_subprocess_exec( |
|
|
os_helper.FakePath(sys.executable), '-c', 'pass') |
|
|
with self.assertRaises(RuntimeError): |
|
|
await subprocess.create_subprocess_exec( |
|
|
os_helper.FakePath(sys.executable), '-c', 'pass') |
|
|
|
|
|
watcher.add_child_handler.assert_not_called() |
|
|
watcher.add_child_handler.assert_not_called() |
|
|
|
|
|
self.assertIsNone(self.loop.run_until_complete(execute())) |
|
|
with Runner(loop_factory=asyncio.new_event_loop) as runner: |
|
|
self.assertIsNone(runner.run(execute())) |
Comment thread
Outdated
Comment on lines
867
to
868
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why not just use self.loop.run_until_complete() as in other tests?
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
the GenericWatcherTests don't have a self.loop, and SubprocessWatcherMixin (which sets self.loop) calls get_event_loop_policy().attach_loop and this test case is testing what happens when attach_loop isn't called
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Many classes set self.loop. For this test you can inherit from test_utils.TestCase and add
def setUp(self):
super().setUp()
self.loop = asyncio.new_event_loop()
self.set_event_loop(self.loop)
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
|
|
|
self.assertListEqual(watcher.mock_calls, [ |
|
|
mock.call.__enter__(), |
|
|
mock.call.__enter__().is_active(), |
|
|
mock.call.__exit__(RuntimeError, mock.ANY, mock.ANY), |
|
|
]) |
|
|
|
|
|
else: |
|
|
# Windows |
|
|
class SubprocessProactorTests(SubprocessMixin, test_utils.TestCase): |
|
|
|
|
|
def setUp(self): |
|
|
super().setUp() |
|
|
self.loop = asyncio.ProactorEventLoop() |
|
|
self.set_event_loop(self.loop) |
|
|
|
|
|
|
|
|
if __name__ == '__main__': |
|
Expand Down
|
|